diff --git a/tui_gateway/hosted_room_driver.py b/tui_gateway/hosted_room_driver.py index 3e89c3d879..fb40791b14 100644 --- a/tui_gateway/hosted_room_driver.py +++ b/tui_gateway/hosted_room_driver.py @@ -1,12 +1,10 @@ """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 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. +The durable state machine lives in :mod:`gateway.hosted_room_driver`; this module owns the +process-local worker and an injected session adapter, never the gateway server or agents. +One bounded supervisor schedules independent room workers: profile turn locks serialize Bots +sharing a profile while a room waiting for approval cannot stall unrelated rooms. Member +sessions reuse ``Group: `` so a local-to-hosted migration keeps one transcript. """ from __future__ import annotations @@ -33,33 +31,26 @@ _STOP_PENDING = "stop retry remains pending: {exc}" class InternalSessionRPC(Protocol): - """Normalized in-process session operations required by the room driver.""" + """Normalized in-process session operations required by the room driver. - def resolve_exact(self, *, profile: str, title: str, source: str) -> Mapping[str, Any] | None: - """Return the exact titled session under ``profile``, if it exists.""" - - def create(self, *, profile: str, title: str, source: str) -> Mapping[str, Any]: - """Create a session without model or provider overrides.""" - - def resume(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]: - """Resume the canonical room session.""" + ``submit`` durably reports one fenced turn's terminal result via ``on_terminal``; + ``interrupt`` acts only while the current turn still matches ``expected_task_id``. + """ + def resolve_exact( + self, *, profile: str, title: str, source: str) -> Mapping[str, Any] | None: ... + def create(self, *, profile: str, title: str, source: str) -> Mapping[str, Any]: ... + def resume(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]: ... 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], - ) -> Mapping[str, Any]: - """Submit one fenced room turn and durably report its terminal result.""" - - def history(self, *, profile: str, session_id: str, source: str) -> Sequence[Mapping[str, Any]]: - """Return normalized session messages in durable order.""" - - def info(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]: - """Return normalized live status for a session.""" - + ) -> Mapping[str, Any]: ... + def history( + self, *, profile: str, session_id: str, source: str) -> Sequence[Mapping[str, Any]]: ... + def info(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]: ... def interrupt( - self, *, profile: str, session_id: str, source: str, expected_task_id: str - ) -> Mapping[str, Any] | None: - """Interrupt only when the current turn still matches the expected task.""" + self, *, profile: str, session_id: str, source: str, expected_task_id: str, + ) -> Mapping[str, Any] | None: ... # Resolve the local or peer session transport for one durable room task. @@ -97,9 +88,8 @@ def _session_kw(profile: str, session_id: str) -> dict[str, str]: def _fences(task: Mapping[str, Any]) -> dict[str, int]: - return { - "expected_execution_generation": int(task["execution_generation"]), - "expected_cancel_generation": int(task["cancel_generation"])} + return {"expected_execution_generation": int(task["execution_generation"]), + "expected_cancel_generation": int(task["cancel_generation"])} class HostedRoomRuntime: @@ -130,8 +120,7 @@ class HostedRoomRuntime: if ( not isinstance(max_concurrent_rooms, int) or isinstance(max_concurrent_rooms, bool) - or max_concurrent_rooms < 1 - ): + 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") @@ -141,20 +130,18 @@ 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 = float(lease_ttl_seconds) - self.poll_interval_seconds = float(poll_interval_seconds) + 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) self.max_concurrent_rooms = max_concurrent_rooms - self.unavailable_retry_min_seconds = float(unavailable_retry_min_seconds) - self.unavailable_retry_max_seconds = float(unavailable_retry_max_seconds) + self.unavailable_retry_min_seconds, self.unavailable_retry_max_seconds = ( + float(unavailable_retry_min_seconds), float(unavailable_retry_max_seconds)) self.process_generation = process_generation or uuid.uuid4().hex self._rooms_provider: Callable[[], Iterable[HostedRoomBinding]] = ( - cast(Callable[[], Iterable[HostedRoomBinding]], rooms) - if callable(rooms) + 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._room_threads: dict[str, threading.Thread] = {} @@ -165,8 +152,7 @@ class HostedRoomRuntime: self._ambiguous_rooms: 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] = {} + self._status_lock, self._current_tasks = threading.Lock(), {} self._room_schedule_cursor, self._cycles = 0, 0 self._last_error: str | None = None @@ -201,9 +187,8 @@ class HostedRoomRuntime: def wakeup(self) -> None: """Wake the worker after task admission or a room-state change.""" with self._status_lock: - # Rooms that still own a worker slot are revisited once that thread - # exits, closing the race between terminal publication/route repair - # and the longer idle fallback without busy-looping idle rooms. + # Rooms still owning a worker slot are revisited once that thread exits, closing + # the terminal-publication/route-repair race without busy-looping idle rooms. self._rooms_needing_reschedule.update(self._room_threads) self._wake.set() @@ -213,24 +198,19 @@ class HostedRoomRuntime: thread = self._thread current_tasks = tuple(self._current_tasks.values()) return { - "running": bool(thread and thread.is_alive()), - "stopping": self._stop.is_set(), + "running": bool(thread and thread.is_alive()), "stopping": self._stop.is_set(), "process_generation": self.process_generation, "current_task": current_tasks[0] if current_tasks else None, - "current_tasks": current_tasks, - "leased_rooms": tuple(sorted(self._leases)), + "current_tasks": current_tasks, "leased_rooms": tuple(sorted(self._leases)), "blocked_rooms": tuple(sorted(self._blocked_rooms)), - "last_error": self._last_error, - "cycles": self._cycles} + "last_error": self._last_error, "cycles": self._cycles} # ------------------------------------------------------------------ public ops 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 -> - terminal), so the status read is only a routing hint: every fast-path - failure caused by a concurrent transition re-reads and re-routes instead - of surfacing a transient `InvalidTaskTransitionError`/`StaleTaskError`. + The worker transitions tasks concurrently, so the status read is only a routing + hint: a fast-path fence failure re-reads and re-routes instead of surfacing it. """ for _ in range(_CANCEL_ROUTE_RETRIES): before = state.get_task(self.db_path, identity) @@ -265,8 +245,7 @@ class HostedRoomRuntime: self._record_error(f"stop remains pending: {exc}") self.wakeup() return state.get_task(self.db_path, identity) - # Exhausted routing retries under sustained contention: surface the - # live status honestly rather than a transient transition error. + # Routing retries exhausted under contention: surface the live status honestly. final = state.get_task(self.db_path, identity) if final["status"] == "cancelled": return final @@ -346,12 +325,6 @@ class HostedRoomRuntime: 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, - **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: @@ -366,10 +339,9 @@ class HostedRoomRuntime: # ------------------------------------------------------------------ session probes def _resume_exact( self, transport: InternalSessionRPC, room_id: str, profile: str) -> str | None: - """Resolve the canonical room session and return its resumed runtime id. + """Resume the canonical room session and return its runtime id (None when absent). - Callers must use the returned id (not the stored one) for every - subsequent history/info probe; resume may hand back a different id. + Probes must use the returned id, not the stored one: resume may hand back another. """ session = self._resolve_or_create(transport, profile, room_id, create=False) return None if session is None else _session_id(session) @@ -377,10 +349,8 @@ class HostedRoomRuntime: 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. - """ + """Return ``(transport, profile, resumed session id)``; no id when the transport is + missing (or local under ``peer_only``) or the session is absent.""" transport = self._transport_for(binding, task) if transport is None or (peer_only and transport is self.rpc): return transport, None, None @@ -395,23 +365,18 @@ class HostedRoomRuntime: 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.""" + def _peer_stop_acknowledged(self, binding: HostedRoomBinding, task: Mapping[str, Any]) -> bool: + """Probe a peer's exact durable terminal Stop receipt before reading history.""" + transport, profile, session_id = self._open_session(binding, task, peer_only=True) + if session_id is None: + return False + info = transport.info(**_session_kw(profile, session_id)) 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"])) - def _peer_stop_acknowledged(self, binding: HostedRoomBinding, task: Mapping[str, Any]) -> bool: - """Probe a peer's exact durable terminal status before reading history.""" - transport, profile, session_id = self._open_session(binding, task, peer_only=True) - if session_id is None: - return False - 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, profile, session_id = self._open_session(binding, task) if session_id is None: @@ -429,9 +394,7 @@ class HostedRoomRuntime: return False result = transport.interrupt( **_session_kw(profile, session_id), expected_task_id=task["identity"].task_id) - if result is None: - return False - return ( + return result is not None and ( result.get("interrupted") is True or str(result.get("status") or "") in _STOP_ACK_STATUSES) @@ -445,7 +408,7 @@ class HostedRoomRuntime: receipt = self._terminal_from_history(transport, profile, session_id, task) if receipt is None: return False - self._settle_stopping(binding, task, lease, **asdict(receipt)) + self._fenced(state.settle_stopping_task, binding, task, lease, **asdict(receipt)) return True def _report_pending_action( @@ -458,17 +421,14 @@ class HostedRoomRuntime: 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, + "kind": "approval", "task_id": task["identity"].task_id, "execution_generation": int(task["execution_generation"]), - "run_id": info.get("run_id"), - "session_id": session_id, - "request_id": safe_approval.get("request_id"), - "approval": safe_approval} + "run_id": info.get("run_id"), "session_id": session_id, + "request_id": safe_approval.get("request_id"), "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: - for task in state.list_tasks(self.db_path, room_id=binding.room_id, status="stopping"): + for task in self._tasks(binding, "stopping"): try: lease = self._renew_lease_if_needed(binding, lease) if self._peer_stop_acknowledged(binding, task): @@ -516,9 +476,7 @@ class HostedRoomRuntime: return with self._status_lock: self._room_threads = { - room_id: thread - for room_id, thread in self._room_threads.items() - if thread.is_alive()} + room_id: t for room_id, t in self._room_threads.items() if t.is_alive()} available = self.max_concurrent_rooms - len(self._room_threads) active_rooms = set(self._room_threads) if available <= 0: @@ -546,7 +504,7 @@ class HostedRoomRuntime: try: self._process_room(binding) except state.LeaseHeldError: - return + pass except Exception as exc: if isinstance(exc, (state.RoomUnavailableError, state.StaleLeaseError)): self._drop_lease(binding.room_id) @@ -568,8 +526,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") - if running and self.clock() < deferred_until: + if self._tasks(binding, "running") and self.clock() < deferred_until: return self._ambiguous_rooms.pop(binding.room_id, None) lease = self._ensure_lease(binding) @@ -582,7 +539,7 @@ 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"): + for task in self._tasks(binding, "queued"): if self._stop.is_set() or self._route_retry_is_deferred(task): return lease = self._renew_lease_if_needed(binding, lease) @@ -612,9 +569,6 @@ class HostedRoomRuntime: "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: @@ -670,9 +624,8 @@ class HostedRoomRuntime: try: with self.turn_lock(profile): session = self._resolve_or_create(transport, profile, binding.room_id) - # An in-process submit should fail before admission or return - # after it, but an unexpected exception at that boundary is - # still ambiguous. Never terminalize it as a proven failure. + # 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) deadline_monotonic = time.monotonic() + self.turn_timeout_seconds @@ -680,7 +633,7 @@ class HostedRoomRuntime: **_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) + self._unavailable_route_retries.pop(self._route_retry_key(task), None) receipt = self._wait_for_terminal( binding, profile=profile, session_id=session_id, attempt=attempt, transport=transport, deadline_monotonic=deadline_monotonic) @@ -689,34 +642,30 @@ class HostedRoomRuntime: 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}") + self._record_task_error(attempt, f"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) 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}") + self._record_task_error( + attempt, f"not-admitted proof lost 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") + self._record_task_error( + attempt, f"was not admitted; 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}") + self._record_task_error(attempt, f"observation failed after submit: {exc}") else: self._settle_failure_if_current(attempt, exc) finally: with self._status_lock: self._current_tasks.pop(binding.room_id, None) - # The task may have published a reply, deferred a member, or - # exposed the next turn while this room thread still occupied - # its slot. Schedule exactly one immediate follow-up after the - # thread leaves; idle room scans never set this marker. + # The task may have published a reply or exposed the next turn while this + # thread held its slot: schedule exactly one follow-up after it 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: @@ -743,17 +692,16 @@ class HostedRoomRuntime: 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, **asdict(terminal), + self._fenced( + state.settle_stopping_task, 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 - # correct fence into a worker thread exception. + # Cancellation, disband, or authority transfer won the durable race: the model + # result is discarded rather than turning a correct fence into a thread exception. pass except state.DriverStateError as exc: - # A malformed terminal receipt must not escape the callback and - # hold the profile lock until the deadline. + # A malformed receipt must not escape the callback and hold the profile lock. self._settle_failure_if_current( attempt, RuntimeError(f"terminal result could not be committed: {exc}")) self.wakeup() @@ -802,8 +750,8 @@ class HostedRoomRuntime: """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, + return self._fenced( + state.settle_stopping_task, 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.", @@ -825,14 +773,13 @@ class HostedRoomRuntime: if not self._is_deadline_stop(task): return lease = self._renew_lease_if_needed(binding, lease, force=True) - if self._finish_stop(binding, task, lease): - return - self._record_error( - f"task {task['identity'].task_id} exceeded its deadline; stop remains pending") + if not self._finish_stop(binding, task, lease): + self._record_error( + f"task {task['identity'].task_id} exceeded its deadline; stop remains pending") # ------------------------------------------------------------------ recovery def _inspect_abandoned_attempts(self, binding: HostedRoomBinding) -> None: - for task in state.list_tasks(self.db_path, room_id=binding.room_id, status="running"): + for task in self._tasks(binding, "running"): if task["run_process_generation"] == self.process_generation: continue inspection = ( @@ -871,13 +818,11 @@ class HostedRoomRuntime: return self._inspect_session(transport, task, session_id, read_history=True) def _inspect_local_recovery_session(self, task: Mapping[str, Any]) -> _RecoveryInspection: - """Check only live process state before explicit local recovery. + """Check only live process state (no resume, no history) before explicit local recovery. - A restart loses the in-process terminal callback identity. Session - history is a display projection that cannot prove which durable task - attempt authored a row, so never hydrate or infer completion from it. - An inactive abandoned attempt remains indeterminate until the user - explicitly retries it under a new fenced generation. + A restart loses the in-process terminal callback identity and history cannot prove + which attempt authored a row, so an inactive abandoned attempt stays indeterminate + until the user retries it under a new fenced generation. """ profile = task["payload"]["target_profile"] with self.turn_lock(profile): @@ -890,7 +835,7 @@ class HostedRoomRuntime: def _reconcile_indeterminate( self, binding: HostedRoomBinding, lease: state.DriverLease) -> bool: - unresolved = state.list_tasks(self.db_path, room_id=binding.room_id, status="indeterminate") + unresolved = self._tasks(binding, "indeterminate") if not unresolved: self._set_blocked(binding.room_id, False) return False @@ -909,12 +854,9 @@ class HostedRoomRuntime: if inspection.active: self._set_blocked(binding.room_id, True) return True - deferred_at = float( - task.get("indeterminate_at") - or task.get("updated_at") - or task.get("created_at") + deadline = self.indeterminate_defer_seconds + float( + task.get("indeterminate_at") or task.get("updated_at") or task.get("created_at") or self.clock()) - deadline = deferred_at + self.indeterminate_defer_seconds inspection = _NO_INSPECTION if attempt_key not in inspected or self.clock() >= deadline: try: @@ -948,21 +890,22 @@ class HostedRoomRuntime: self, binding: HostedRoomBinding, task: Mapping[str, Any], receipt: _TerminalReceipt ) -> None: previous_attempt = state.TaskAttempt( - identity=task["identity"], + identity=task["identity"], execution_generation=task["execution_generation"], + cancel_generation=task["cancel_generation"], lease=state.DriverLease( 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), - execution_generation=task["execution_generation"], - 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. + lease_generation=task["run_lease_generation"], expires_at=0.0)) + # Once the previous proof has expired there is deliberately no "trust this historical + # output" escape hatch; fenced recovery leaves the task indeterminate for the user. 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) + 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) @@ -991,7 +934,10 @@ class HostedRoomRuntime: self.db_path, attempt, settlement_id=f"failure:{attempt.identity.task_id}:{attempt.execution_generation}", status="failed", result={"error": str(exc)}, clock=self.clock) - self._record_error(f"task {attempt.identity.task_id} failed: {exc}") + self._record_task_error(attempt, f"failed: {exc}") + + def _record_task_error(self, attempt: state.TaskAttempt, message: str) -> None: + self._record_error(f"task {attempt.identity.task_id} {message}") def _record_error(self, message: str) -> None: with self._status_lock: @@ -1004,8 +950,8 @@ def room_session_title(room_id: str) -> str: def _member_id(task: Mapping[str, Any]) -> str: - payload = task.get("payload") or {} - return str(payload.get("target_member_id") or payload.get("target_profile") or "") + p = task.get("payload") or {} + return str(p.get("target_member_id") or p.get("target_profile") or "") def _session_id(session: Mapping[str, Any]) -> str: @@ -1016,8 +962,7 @@ def _session_id(session: Mapping[str, Any]) -> str: def _truncate_utf8(value: Any, *, max_bytes: int) -> tuple[str, bool]: - text = str(value or "") - encoded = text.encode("utf-8") + 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") @@ -1055,8 +1000,7 @@ def _find_terminal_receipt( if not isinstance(receipt_id, str) or not receipt_id: receipt_id = f"reply:{identity.task_id}:{execution_generation}" return _TerminalReceipt( - status=cast(state.TerminalStatus, status), - settlement_id=receipt_id, + status=cast(state.TerminalStatus, status), settlement_id=receipt_id, result=_bounded_terminal_result( {"message_id": receipt_id, "text": message.get("content", "")})) return None @@ -1071,6 +1015,5 @@ def _info_is_active_for( if not _info_active(info): return False active_task_id = info.get("task_id") - if require_exact: - return active_task_id == identity.task_id - return active_task_id in {None, identity.task_id} + return active_task_id == identity.task_id if require_exact else ( + active_task_id in {None, identity.task_id}) diff --git a/tui_gateway/hosted_room_peer_http.py b/tui_gateway/hosted_room_peer_http.py index e3af7e3377..308352b64e 100644 --- a/tui_gateway/hosted_room_peer_http.py +++ b/tui_gateway/hosted_room_peer_http.py @@ -25,13 +25,11 @@ logger = logging.getLogger(__name__) _NOT_ADMITTED_ERRNOS = frozenset( - value - for name in ("ECONNREFUSED", "ENETDOWN", "ENETUNREACH", "EHOSTDOWN", "EHOSTUNREACH") + value for name in ("ECONNREFUSED", "ENETDOWN", "ENETUNREACH", "EHOSTDOWN", "EHOSTUNREACH") 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 -# from scaling memory use without limit. +# A replay page may legitimately hold many bounded 64 KiB room events; cap it so a peer-sized +# response cannot scale memory use without limit. MAX_PEER_RESPONSE_BYTES = 16 * 1024 * 1024 MAX_PEER_ERROR_RESPONSE_BYTES = 16 * 1024 _PEER_RESPONSE_CHUNK_BYTES = 64 * 1024 @@ -43,8 +41,8 @@ _RECEIPT_SCOPE_FIELDS = ( _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 -# Older target gateways wrap these conditions inside the generic dispatch -# error; normalize locally until their wire code becomes specific. +_RUN_STATUS_KEYS = ("run_id", "status", "output", "error", "approval", "last_event") +# Older target gateways wrap these inside the generic dispatch error; normalize locally. _LEGACY_DISPATCH_MESSAGE_CODES = ( ("room grant", "invalid_room_grant"), ("capability catalog changed", "room_capability_catalog_changed"), @@ -131,6 +129,18 @@ def _read_bounded_response(response: Any, *, max_bytes: int, deadline: float) -> raise _PeerResponseTooLarge +def _read_body(response: Any, *, max_bytes: int, deadline: float, kind: str, **flags: Any) -> str: + """Read a bounded body as text; budget overruns become classified ``PeerRunsHTTPError``.""" + try: + return _read_bounded_response( + response, max_bytes=max_bytes, deadline=deadline).decode("utf-8", "replace") + except _PeerResponseTooLarge as exc: + raise PeerRunsHTTPError(_BUDGET_MESSAGES["size"].format(kind=kind), **flags) from exc + except _PeerResponseDeadlineExceeded as exc: + raise PeerRunsHTTPError( + _BUDGET_MESSAGES["time"].format(kind=kind), retryable=True, **flags) from exc + + def _is_proven_pre_admission_failure(exc: BaseException) -> bool: """Return whether no HTTP connection could have carried the request.""" reason: Any = exc @@ -181,11 +191,8 @@ class PeerRunsHTTPError(RuntimeError): 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 - self.not_admitted = not_admitted - self.status_code = status_code - self.error_code = error_code + 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) @@ -225,15 +232,13 @@ class PeerRunsHTTPClient: 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") - self.base_url = base_url - self.api_key = api_key + self.base_url, self.api_key, self.clock = base_url, api_key, clock self.timeout_seconds = float(timeout_seconds) self.receipt_db_path = Path(receipt_db_path) if receipt_db_path else None if poll_min_seconds <= 0 or poll_max_seconds < poll_min_seconds: raise ValueError("peer polling bounds are invalid") self.poll_min_seconds = float(poll_min_seconds) self.poll_max_seconds = float(poll_max_seconds) - self.clock = clock self._runs: dict[tuple[str, int], dict[str, Any]] = {} self._observation_key: tuple[str, int] | None = None self._status_cache: dict[str, dict[str, Any]] = {} @@ -254,8 +259,7 @@ class PeerRunsHTTPClient: """Fence every in-memory and durable receipt to one room authority.""" epoch = int(authority_epoch or 0) names = [ - str(value or "") - for value in ( + 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): @@ -307,31 +311,18 @@ class PeerRunsHTTPClient: deadline = time.monotonic() + self.timeout_seconds ambiguous = method == "POST" request = urllib.request.Request( - f"{self.base_url}{path}", - data=( - json.dumps(body, separators=(",", ":")).encode("utf-8") - if body is not None - else None), - method=method, + f"{self.base_url}{path}", method=method, + data=None if body is None else json.dumps(body, separators=(",", ":")).encode("utf-8"), 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", + "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( - response, max_bytes=MAX_PEER_RESPONSE_BYTES, deadline=deadline - ).decode("utf-8", "replace") - except _PeerResponseTooLarge as exc: - raise PeerRunsHTTPError( - _BUDGET_MESSAGES["size"].format(kind=""), ambiguous=ambiguous, - ) from exc - except _PeerResponseDeadlineExceeded as exc: - raise PeerRunsHTTPError( - _BUDGET_MESSAGES["time"].format(kind=""), retryable=True, ambiguous=ambiguous - ) from exc + raw = _read_body( + response, max_bytes=MAX_PEER_RESPONSE_BYTES, deadline=deadline, kind="", + ambiguous=ambiguous) 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: @@ -358,17 +349,11 @@ class PeerRunsHTTPClient: "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( - _BUDGET_MESSAGES["size"].format(kind=" error"), **flags, - ) from body_exc - except _PeerResponseDeadlineExceeded as body_exc: - raise PeerRunsHTTPError( - _BUDGET_MESSAGES["time"].format(kind=" error"), retryable=True, **flags - ) from body_exc + detail = _read_body( + exc, max_bytes=MAX_PEER_ERROR_RESPONSE_BYTES, deadline=deadline, kind=" error", + **flags)[:500] + except PeerRunsHTTPError: + raise except Exception: detail = "" error_code = _response_error_code(detail) @@ -434,12 +419,9 @@ class PeerRunsHTTPClient: checked: HostedMemberDispatch, *, run_id: str, session_id: str, replayed: bool, ) -> dict[str, Any]: return { - "status": "accepted", - "task_id": checked.task_id, - "execution_generation": checked.execution_generation, - "run_id": run_id, - "session_id": session_id, - "replayed": replayed} + "status": "accepted", "task_id": checked.task_id, + "execution_generation": checked.execution_generation, "run_id": run_id, + "session_id": session_id, "replayed": replayed} def _admit_dispatch(self, checked: HostedMemberDispatch, *, grant: str) -> Mapping[str, Any]: session_id = self._session_id(checked, grant=grant) @@ -462,11 +444,9 @@ class PeerRunsHTTPClient: if not run_id: raise PeerRunsHTTPError("peer did not return a run id") receipt = { - "run_id": run_id, - "session_id": session_id, + "run_id": run_id, "session_id": session_id, **{field: getattr(checked, field) for field in _RECEIPT_SCOPE_FIELDS}, - "task_id": checked.task_id, - "execution_generation": checked.execution_generation} + "task_id": checked.task_id, "execution_generation": checked.execution_generation} if self.receipt_db_path is not None: from gateway import hosted_rooms @@ -490,9 +470,7 @@ class PeerRunsHTTPClient: def _observation_receipt( 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) + record = None if self._observation_key is None else self._receipt(*self._observation_key) if record is None: return None scope = (record["room_id"], record["target_profile"], record["session_id"]) @@ -500,27 +478,16 @@ class PeerRunsHTTPClient: raise PeerRunsHTTPError("peer observation receipt changed scope") return record - @staticmethod - def _compact_run_status(status: Mapping[str, Any]) -> dict[str, Any]: - return { - key: status[key] - for key in ("run_id", "status", "output", "error", "approval", "last_event") - 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 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]: run_id, now = str(record["run_id"]), self.clock() cached = self._status_cache.get(run_id) if cached is not None: status = cached["status"] - if self._run_is_terminal(status): + if status.get("status") in _TERMINAL_RUN_STATES: return status if now < float(cached["next_poll_at"]): error = cached.get("error") @@ -530,8 +497,8 @@ class PeerRunsHTTPClient: delay = self._next_poll_delay(cached) entry = {"delay": delay, "next_poll_at": now + delay} try: - status = self._compact_run_status( - self._request(_run_path(record), room_grant=self._require_room_grant(grant))) + full = self._request(_run_path(record), room_grant=self._require_room_grant(grant)) + status = {key: full[key] for key in _RUN_STATUS_KEYS if key in full} if ( str(status.get("run_id") or "") != run_id or status.get("status") not in _KNOWN_RUN_STATES): @@ -541,7 +508,7 @@ class PeerRunsHTTPClient: self._status_cache = {run_id: {"status": previous, "error": exc, **entry}} raise self._status_cache = {run_id: {"status": status, **entry}} - if self._run_is_terminal(status): + if status.get("status") in _TERMINAL_RUN_STATES: self._terminal_receipts.add( (str(record["task_id"]), int(record["execution_generation"]))) return status @@ -557,8 +524,7 @@ class PeerRunsHTTPClient: if state not in {"completed", "failed", "interrupted"}: return [] return [{ - "role": "assistant", - "task_id": receipt["task_id"], + "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')}", @@ -571,11 +537,9 @@ class PeerRunsHTTPClient: return {"active": False, "task_id": None} status = self._poll_receipt(receipt, grant=grant) return { - "active": status.get("status") in _ACTIVE_RUN_STATES, - "task_id": receipt["task_id"], + "active": status.get("status") in _ACTIVE_RUN_STATES, "task_id": receipt["task_id"], "execution_generation": receipt["execution_generation"], - "status": status.get("status"), - "run_id": status.get("run_id"), + "status": status.get("status"), "run_id": status.get("run_id"), "approval": status.get("approval")} def approve_receipt( @@ -614,7 +578,7 @@ class PeerRunsHTTPClient: return None result = self._post_run_action( record, "stop", body={}, grant=self._require_room_grant(grant)) - if self._run_is_terminal(result): + if result.get("status") in _TERMINAL_RUN_STATES: self._terminal_receipts.add((str(task_id), int(execution_generation))) return result @@ -628,17 +592,11 @@ class PeerRunsHTTPClient: return self._request( "/v1/room-members/invitations", method="POST", body={ - "room_id": room_id, - "home_install_id": home_install_id, - "authority_gateway_id": authority_gateway_id, - "authority_epoch": authority_epoch, - "member_id": member_id, - "grant_id": grant_id, - "ttl_seconds": ttl_seconds, - **( - {"status_ttl_seconds": status_ttl_seconds} - if status_ttl_seconds is not None - else {})}) + "room_id": room_id, "home_install_id": home_install_id, + "authority_gateway_id": authority_gateway_id, "authority_epoch": authority_epoch, + "member_id": member_id, "grant_id": grant_id, "ttl_seconds": ttl_seconds, + **({} if status_ttl_seconds is None else { + "status_ttl_seconds": status_ttl_seconds})}) def refresh_grant( self, *, grant: str, ttl_seconds: float = 24 * 60 * 60, @@ -650,8 +608,7 @@ class PeerRunsHTTPClient: replacement = str(refreshed.get("grant") or "") if not replacement: raise PeerRunsHTTPError("peer returned no refreshed room grant") - # Persist only after the target proves the replacement can authorize - # the same scoped capability endpoint. + # Persist only after the target proves the replacement authorizes the scoped endpoint. probe = self.probe(grant=replacement) error = digest_reauthorization_error( GatewayRoomCatalog.from_mapping(probe.get("catalog")), diff --git a/tui_gateway/hosted_room_service.py b/tui_gateway/hosted_room_service.py index 8923dde6bd..98d6a1977d 100644 --- a/tui_gateway/hosted_room_service.py +++ b/tui_gateway/hosted_room_service.py @@ -71,8 +71,7 @@ class HostedRoomService: self, server: ModuleType, *, db_path: Path | str | None = None, peer_routes: Mapping[tuple[str, str], PeerMemberRoute] | None = None, peer_clients: Mapping[Any, HostedRoomPeerClient] | None = None) -> None: - self.server = server - self.db_path = Path(db_path or hosted_rooms.default_db_path()) + self.server, self.db_path = server, Path(db_path or hosted_rooms.default_db_path()) hosted_rooms.prune_disbanded_rooms(self.db_path) self._policy_lock = threading.RLock() self._pending_actions: dict[tuple[str, str], dict[str, Any]] = {} @@ -89,9 +88,7 @@ class HostedRoomService: supplied_clients = dict(peer_clients or {}) for key, route in dict(peer_routes or {}).items(): self.peer_routes[key] = route - client = supplied_clients.get(key) - if client is None: - client = supplied_clients.get(route.target_install_id) + client = supplied_clients.get(key, supplied_clients.get(route.target_install_id)) if client is not None: self.peer_clients[key] = client self.runtime = HostedRoomRuntime( @@ -131,8 +128,7 @@ class HostedRoomService: return self.db_path.parent def local_profiles(self) -> tuple[str, ...]: - profiles = {"default"} - profiles_dir = self.root / "profiles" + profiles, profiles_dir = {"default"}, self.root / "profiles" if profiles_dir.is_dir(): profiles.update(path.name for path in profiles_dir.iterdir() if path.is_dir()) return tuple(sorted(profiles)) @@ -213,9 +209,8 @@ class HostedRoomService: 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. Remote revocation is the - boundary: an unreachable target leaves the room intact for retry rather than - reporting a false disband with a live grant.""" + """Revoke and forget every scoped peer route for one room; an unreachable target + leaves the room intact for retry rather than a false disband with a live grant.""" with self._policy_lock: routes = [(key, route) for key, route in self.peer_routes.items() if key[0] == room_id] for key, route in routes: @@ -237,31 +232,31 @@ class HostedRoomService: def _resolve_member_transport(self, binding: HostedRoomBinding, task: Mapping[str, Any]): payload = task.get("payload", {}) member_id = str(payload.get("target_member_id") or payload.get("target_profile") or "") - route = self.peer_routes.get((binding.room_id, member_id)) + key = (binding.room_id, member_id) + route = self.peer_routes.get(key) if route is None: if self._member_is_peer(binding.room_id, member_id): raise RuntimeError("peer room route is unavailable") return self.rpc - client = self.peer_clients.get((binding.room_id, member_id)) + client = self.peer_clients.get(key) if client is None: raise RuntimeError("peer room client is unavailable") identity = task.get("identity") execution_generation = int(task.get("execution_generation") or 0) bind_observation = _hook(client, "bind_observation") if ( - bind_observation is not None - and isinstance(identity, driver.TaskIdentity) + bind_observation is not None and isinstance(identity, driver.TaskIdentity) and execution_generation > 0): bind_observation(task_id=identity.task_id, execution_generation=execution_generation) def set_status(status: str): - return lambda: self._set_route_status(binding.room_id, member_id, status) + return lambda: self._set_route_status(*key, status) tracked_client = _RouteStatusPeerClient( client, on_ready=set_status("ready"), on_reauthorization=set_status("needs_reauthorization"), on_unavailable=set_status("unavailable"), on_refreshed=lambda grant, catalog=None: self._rotate_route_grant( - binding.room_id, member_id, grant, catalog)) + *key, grant, catalog)) self._recover_peer_admission(binding, task, route, tracked_client) return PeerHostedRoomTransport( binding=binding, route=route, client=tracked_client, @@ -273,14 +268,11 @@ class HostedRoomService: client: Any) -> None: """Rediscover an admitted peer run without advancing its generation.""" recover = _hook(client, "recover_dispatch") - identity = task.get("identity") - payload = task.get("payload") + identity, payload = task.get("identity"), task.get("payload") execution_generation = int(task.get("execution_generation") or 0) if ( - recover is None - or not isinstance(identity, driver.TaskIdentity) - or not isinstance(payload, Mapping) - or execution_generation < 1 + recover is None or not isinstance(identity, driver.TaskIdentity) + or not isinstance(payload, Mapping) or execution_generation < 1 or task.get("status") not in {"running", "indeterminate", "stopping"}): return prompt = payload.get("prompt") @@ -294,11 +286,10 @@ class HostedRoomService: recover(dispatch=dispatch.as_mapping(), grant=route.grant) def _member_is_peer(self, room_id: str, member_id: str) -> bool: - for member in self._room(room_id).get("members") or []: - if ( - isinstance(member, Mapping) - and str(member.get("member_id") or member.get("profile") or "") == member_id): - target = member.get("target") + for m in self._room(room_id).get("members") or []: + if isinstance(m, Mapping) and str( + m.get("member_id") or m.get("profile") or "") == member_id: + target = m.get("target") return isinstance(target, Mapping) and target.get("kind") == "peer" return False @@ -313,12 +304,11 @@ class HostedRoomService: def _set_pending_action( self, room_id: str, member_id: str, action: Mapping[str, Any] | None) -> None: - key = (room_id, member_id) with self._policy_lock: if action is None: - self._pending_actions.pop(key, None) + self._pending_actions.pop((room_id, member_id), None) else: - self._pending_actions[key] = {**action, "member_id": member_id} + self._pending_actions[(room_id, member_id)] = {**action, "member_id": member_id} def _rotate_route_grant( self, room_id: str, member_id: str, grant: str, catalog: GatewayRoomCatalog | None = None @@ -338,8 +328,7 @@ class HostedRoomService: catalog.installation_id != route.target_install_id or catalog.execution_policy.target_profile != route.target_profile or PROTOCOL_VERSION not in catalog.protocol_versions - or "direct" not in catalog.link_modes - or not catalog.text + or "direct" not in catalog.link_modes or not catalog.text or catalog.execution_policy.policy_digest != route.execution_policy_digest): self._set_route_status(room_id, member_id, "needs_reauthorization") raise RuntimeError( @@ -347,12 +336,11 @@ class HostedRoomService: digests = { "capability_digest": catalog.catalog_digest, "execution_policy_digest": catalog.execution_policy.policy_digest} - rotated_route = replace(route, grant=grant, **digests) self._save_link( room_id=room_id, member_id=member_id, target_url=stored.target_url, target_profile=stored.target_profile, grant=grant, catalog=catalog or stored.catalog, cancellation_scope_id=stored.cancellation_scope_id, trace_id=stored.trace_id) - self._publish_route(key, rotated_route) + self._publish_route(key, replace(route, grant=grant, **digests)) def _route_statuses(self, room_id: str | None = None) -> list[dict[str, str]]: with self._policy_lock: @@ -382,9 +370,7 @@ class HostedRoomService: room_id=str(room["room_id"]), latest_seq=int(room["latest_seq"])) def _publish_terminal_tasks(self, room: Mapping[str, Any]) -> bool: - changed = False - room_id = str(room["room_id"]) - local_profiles = self.local_profiles() + 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"]) @@ -439,9 +425,8 @@ class HostedRoomService: driver.admit_task( self.db_path, decision.task.identity, payload=decision.task.payload, clock=time.time) - # A stop can race the policy read from another process. Re-read after - # admission and cancel before the runtime can execute a task whose - # source event is now behind the room stop fence. + # A stop can race the policy read from another process: re-read after admission + # and cancel a task whose source event is now behind the room stop fence. fence = self._policy_snapshot(self._room(binding.room_id)).stopped_through_seq if decision.source_event_seq is not None and decision.source_event_seq < fence: self.runtime.cancel(decision.task.identity, cancel_id=f"stop-fence:{fence}") @@ -455,9 +440,7 @@ class HostedRoomService: def create_room(self, *, room_id: str, name: str, members: Any) -> dict[str, Any]: normalized = discussion.validate_roster(members, local_profiles=self.local_profiles()) room = hosted_rooms.create_room( - self.db_path, - room_id=room_id, - name=name, + self.db_path, room_id=room_id, name=name, members=[ { "member_id": member.member_id, "profile": member.profile, @@ -494,17 +477,15 @@ class HostedRoomService: (task["identity"].room_id, task["identity"].task_id): task for task in self._list_tasks(room_id, _STOPPABLE_STATUSES)} for task in tasks.values(): - task_cancel_id = ( - str(task.get("cancel_id") or "") if task.get("status") == "stopping" else "") - result = self.runtime.cancel( - task["identity"], cancel_id=task_cancel_id or cancel_id) + own_cancel_id = ( + task.get("status") == "stopping" and str(task.get("cancel_id") or "")) + result = self.runtime.cancel(task["identity"], cancel_id=own_cancel_id or cancel_id) if result["status"] == "stopping": pending += 1 - cancelled = len(tasks) if require_acknowledged and pending: raise RuntimeError("room work is still stopping; retry deletion after Stop completes") self.runtime.wakeup() - return cancelled + return len(tasks) def retry_room_task(self, room_id: str, *, task_id: str) -> dict[str, Any]: """Retry one uncertain or deferred task only after explicit user action.""" @@ -519,8 +500,7 @@ class HostedRoomService: choice: str, request_id: str | None = None) -> Mapping[str, Any]: """Resolve one exact local or peer approval and wake room observation.""" key = (room_id, member_id) - route = self.peer_routes.get(key) - client = self.peer_clients.get(key) + route, client = self.peer_routes.get(key), self.peer_clients.get(key) with self._policy_lock: action = self._pending_actions.get(key) requested_approval_id = str(request_id or "") @@ -572,12 +552,10 @@ class HostedRoomService: 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), + "running": runtime["running"], "working": any(counts.get(s) for s in _LIVE_STATUSES), "blocked": room_id in runtime["blocked_rooms"] or bool(counts.get("indeterminate") or counts.get("stopping")), - "counts": dict(counts), - "pending_actions": pending_actions, + "counts": dict(counts), "pending_actions": pending_actions, "peer_routes": self._route_statuses(room_id)} @@ -586,11 +564,8 @@ class _RouteStatusPeerClient: def __init__( self, client, *, on_ready, on_reauthorization, on_unavailable, on_refreshed) -> None: - self._client = client - self._on_ready = on_ready - self._on_reauthorization = on_reauthorization - self._on_unavailable = on_unavailable - self._on_refreshed = on_refreshed + self._client, self._on_ready, self._on_refreshed = client, on_ready, on_refreshed + self._on_reauthorization, self._on_unavailable = on_reauthorization, on_unavailable def _refresh_grant(self, kwargs: dict) -> dict: """Rotate an expiring grant before dispatch; return the kwargs to send. Refresh @@ -620,11 +595,11 @@ class _RouteStatusPeerClient: refreshed_catalog = None if refreshed.get("catalog") is not None: refreshed_catalog = GatewayRoomCatalog.from_mapping(refreshed.get("catalog")) - drift = None - if refreshed_catalog.execution_policy.policy_digest != checked.execution_policy_digest: - drift = _POLICY_DRIFT - elif refreshed_catalog.catalog_digest != checked.capability_digest: - drift = _CATALOG_DRIFT + 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) if drift is not None: self._on_reauthorization() raise PeerRunsHTTPError(