ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(tui_gateway): hosted-room driver/peer-http/service — fold wrappers, unify bounded-body reads, pack payloads

Browse Source
This commit is contained in:
Teknium
2026-09-03 03:39:11 -07:00
parent 7041d8ff9d
commit 89fe549af4
3 changed files with 197 additions and 322 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tui_gateway/hosted_room_driver.py
+105 -162
View File
@@ -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: <room_id>`` 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: <room_id>`` 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})
tui_gateway/hosted_room_peer_http.py
+52 -95
View File
@@ -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")),
tui_gateway/hosted_room_service.py
+40 -65
View File
@@ -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(