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

Merge branch 'simp/r3-36-D' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-03 00:10:30 -07:00
parent 3b46faca73 9ea83e7f96
commit c32eb2c458
4 changed files with 495 additions and 1231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tui_gateway/hosted_room_driver.py
+278 -692
View File
File diff suppressed because it is too large Load Diff
tui_gateway/hosted_room_peer_http.py
+167 -408
View File
@@ -13,14 +13,12 @@ import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Callable, Mapping, Sequence
from contextlib import suppress
from pathlib import Path
from typing import Any, NoReturn
from gateway.hosted_room_peer import (
GatewayRoomCatalog,
HostedMemberDispatch,
validate_room_link_url,
)
GatewayRoomCatalog, HostedMemberDispatch, validate_room_link_url)
logger = logging.getLogger(__name__)
@@ -29,8 +27,7 @@ logger = logging.getLogger(__name__)
_NOT_ADMITTED_ERRNOS = frozenset(
value
for name in ("ECONNREFUSED", "ENETDOWN", "ENETUNREACH", "EHOSTDOWN", "EHOSTUNREACH")
if (value := getattr(errno, name, None)) is not None
)
if (value := getattr(errno, name, None)) is not None)
_ERROR_CODE_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
# A replay page may legitimately contain many bounded 64 KiB room events. Keep
# enough room for the largest normal page while preventing peer-sized responses
@@ -41,14 +38,8 @@ _PEER_RESPONSE_CHUNK_BYTES = 64 * 1024
# Receipt fields that must match between a stored receipt and a dispatch.
_RECEIPT_SCOPE_FIELDS = (
"room_id",
"home_install_id",
"authority_gateway_id",
"authority_epoch",
"member_id",
"target_install_id",
"target_profile",
)
"room_id", "home_install_id", "authority_gateway_id", "authority_epoch",
"member_id", "target_install_id", "target_profile")
_TERMINAL_RUN_STATES = frozenset({"completed", "failed", "interrupted", "cancelled"})
_ACTIVE_RUN_STATES = frozenset({"queued", "running", "waiting_for_approval", "stopping"})
_KNOWN_RUN_STATES = _TERMINAL_RUN_STATES | _ACTIVE_RUN_STATES
@@ -57,23 +48,23 @@ _KNOWN_RUN_STATES = _TERMINAL_RUN_STATES | _ACTIVE_RUN_STATES
_LEGACY_DISPATCH_MESSAGE_CODES = (
("room grant", "invalid_room_grant"),
("capability catalog changed", "room_capability_catalog_changed"),
("execution policy changed", "room_execution_policy_changed"),
)
_REAUTHORIZATION_CODES = frozenset({
"invalid_room_grant",
"room_capability_catalog_changed",
"room_execution_policy_changed",
"room_reauthorization_required",
})
("execution policy changed", "room_execution_policy_changed"))
_GRANT_RENEWAL_CODES = frozenset({"invalid_room_grant", "room_reauthorization_required"})
# (error_code, human message) for the two digest checks after a grant refresh.
_EXECUTION_POLICY_CHANGED = (
"room_execution_policy_changed",
"peer room execution policy needs reauthorization",
)
"room_execution_policy_changed", "peer room execution policy needs reauthorization")
_CAPABILITY_CHANGED = (
"room_capability_catalog_changed",
"peer room capabilities need reauthorization",
)
"room_capability_catalog_changed", "peer room capabilities need reauthorization")
_REAUTHORIZATION_CODES = _GRANT_RENEWAL_CODES | {
_EXECUTION_POLICY_CHANGED[0], _CAPABILITY_CHANGED[0]}
# Human messages for 401/403 reauthorization codes (any other status keeps the generic text).
_REAUTHORIZATION_MESSAGES = {
**dict.fromkeys(_GRANT_RENEWAL_CODES, "peer room authorization needs renewal"),
_EXECUTION_POLICY_CHANGED[0]: _EXECUTION_POLICY_CHANGED[1],
_CAPABILITY_CHANGED[0]: _CAPABILITY_CHANGED[1]}
_BUDGET_MESSAGES = {
"size": "peer{kind} response exceeded the RoomLink size limit",
"time": "peer{kind} response exceeded the RoomLink time budget"}
class _PeerResponseTooLarge(ValueError):
@@ -85,19 +76,15 @@ class _PeerResponseDeadlineExceeded(TimeoutError):
def _content_length(response: Any) -> int | None:
headers = getattr(response, "headers", None)
if headers is None or not hasattr(headers, "get"):
return None
try:
value = int(headers.get("Content-Length"))
except (TypeError, ValueError):
value = int(response.headers.get("Content-Length"))
except (AttributeError, TypeError, ValueError):
return None
return value if value >= 0 else None
def _set_response_socket_timeout(response: Any, remaining: float) -> None:
"""Best-effort urllib socket timeout tightened to the remaining budget."""
frontier = [response]
seen: set[int] = set()
for _depth in range(5):
@@ -108,14 +95,10 @@ def _set_response_socket_timeout(response: Any, remaining: float) -> None:
seen.add(id(value))
setter = getattr(value, "settimeout", None)
if callable(setter):
try:
with suppress(OSError, ValueError):
setter(max(0.001, remaining))
except (OSError, ValueError):
pass
return
next_frontier.extend(
getattr(value, field, None) for field in ("fp", "raw", "_sock")
)
next_frontier.extend(getattr(value, field, None) for field in ("fp", "raw", "_sock"))
frontier = next_frontier
@@ -123,7 +106,6 @@ def _read_bounded_response(response: Any, *, max_bytes: int, deadline: float) ->
declared = _content_length(response)
if declared is not None and declared > max_bytes:
raise _PeerResponseTooLarge
reader = getattr(response, "read1", None)
if not callable(reader):
reader = response.read
@@ -184,15 +166,10 @@ def _response_error_code(detail: str) -> str | None:
def _http_error_message(method: str, path: str, status: int, error_code: str | None) -> str:
if status in {401, 403} and error_code in {
"invalid_room_grant",
"room_reauthorization_required",
}:
return "peer room authorization needs renewal"
if status == 403 and error_code == _EXECUTION_POLICY_CHANGED[0]:
return _EXECUTION_POLICY_CHANGED[1]
if status == 403 and error_code == _CAPABILITY_CHANGED[0]:
return _CAPABILITY_CHANGED[1]
renewal = status in {401, 403} and error_code in _GRANT_RENEWAL_CODES
drift = status == 403 and error_code in {_EXECUTION_POLICY_CHANGED[0], _CAPABILITY_CHANGED[0]}
if renewal or drift:
return _REAUTHORIZATION_MESSAGES[error_code]
return f"peer rejected {method} {path} with HTTP {status}"
@@ -200,16 +177,9 @@ class PeerRunsHTTPError(RuntimeError):
"""Controlled peer HTTP failure."""
def __init__(
self,
message: str,
*,
retryable: bool = False,
ambiguous: bool = False,
not_admitted: bool = False,
status_code: int | None = None,
error_code: str | None = None,
error_message: str | None = None,
) -> None:
self, message: str, *, retryable: bool = False, ambiguous: bool = False,
not_admitted: bool = False, status_code: int | None = None, error_code: str | None = None,
error_message: str | None = None) -> None:
super().__init__(message)
self.retryable = retryable
self.ambiguous = ambiguous
@@ -217,56 +187,41 @@ class PeerRunsHTTPError(RuntimeError):
self.status_code = status_code
self.error_code = error_code
self.error_message = error_message
self.needs_reauthorization = bool(
status_code in {401, 403} and error_code in _REAUTHORIZATION_CODES
)
self.needs_capability_refresh = bool(
status_code == 403 and error_code == _CAPABILITY_CHANGED[0]
)
self.needs_execution_policy_refresh = bool(
status_code == 403 and error_code == _EXECUTION_POLICY_CHANGED[0]
)
self.needs_reauthorization = (
status_code in {401, 403} and error_code in _REAUTHORIZATION_CODES)
self.needs_capability_refresh = status_code == 403 and error_code == _CAPABILITY_CHANGED[0]
self.needs_execution_policy_refresh = (
status_code == 403 and error_code == _EXECUTION_POLICY_CHANGED[0])
def digest_reauthorization_error(
catalog: GatewayRoomCatalog,
*,
capability_digest: str | None,
execution_policy_digest: str | None,
) -> PeerRunsHTTPError | None:
catalog: GatewayRoomCatalog, *, capability_digest: str | None,
execution_policy_digest: str | None) -> PeerRunsHTTPError | None:
"""Return the fail-closed error when a refreshed catalog drifted from the frozen digests.
Execution policy is checked before capabilities; a ``None`` digest skips its check.
"""
for expected, actual, (code, message) in (
(
execution_policy_digest,
catalog.execution_policy.policy_digest,
_EXECUTION_POLICY_CHANGED,
),
(capability_digest, catalog.catalog_digest, _CAPABILITY_CHANGED),
):
execution_policy_digest, catalog.execution_policy.policy_digest,
_EXECUTION_POLICY_CHANGED),
(capability_digest, catalog.catalog_digest, _CAPABILITY_CHANGED)):
if expected is not None and actual != expected:
return PeerRunsHTTPError(
message, status_code=403, error_code=code, not_admitted=True
)
return PeerRunsHTTPError(message, status_code=403, error_code=code, not_admitted=True)
return None
def _run_path(record: Mapping[str, Any], *suffix: str) -> str:
return "/".join(("/v1/runs", urllib.parse.quote(str(record["run_id"]), safe=""), *suffix))
class PeerRunsHTTPClient:
"""Drive a peer's dedicated group session via scoped async Runs APIs."""
def __init__(
self,
*,
base_url: str,
api_key: str,
timeout_seconds: float = 30,
receipt_db_path: Path | str | None = None,
poll_min_seconds: float = 0.1,
poll_max_seconds: float = 2.0,
clock: Callable[[], float] = time.monotonic,
) -> None:
self, *, base_url: str, api_key: str, timeout_seconds: float = 30,
receipt_db_path: Path | str | None = None, poll_min_seconds: float = 0.1,
poll_max_seconds: float = 2.0, clock: Callable[[], float] = time.monotonic) -> None:
base_url, self.transport_security = validate_room_link_url(base_url)
if api_key and len(api_key) < 16:
raise ValueError("peer API key is missing or too short")
@@ -294,31 +249,20 @@ class PeerRunsHTTPClient:
self.receipt_db_path = path
def bind_room_scope(
self,
*,
room_id: str,
home_install_id: str,
authority_gateway_id: str,
authority_epoch: int,
member_id: str,
target_install_id: str,
target_profile: str,
) -> None:
self, *, room_id: str, home_install_id: str, authority_gateway_id: str,
authority_epoch: int, member_id: str, target_install_id: str, target_profile: str) -> None:
"""Fence every in-memory and durable receipt to one room authority."""
scope = {
"room_id": str(room_id or ""),
"home_install_id": str(home_install_id or ""),
"authority_gateway_id": str(authority_gateway_id or ""),
"authority_epoch": int(authority_epoch or 0),
"member_id": str(member_id or ""),
"target_install_id": str(target_install_id or ""),
"target_profile": str(target_profile or ""),
}
if not all(value for key, value in scope.items() if key != "authority_epoch"):
epoch = int(authority_epoch or 0)
names = [
str(value or "")
for value in (
room_id, home_install_id, authority_gateway_id, member_id, target_install_id,
target_profile)]
if not all(names):
raise PeerRunsHTTPError("peer room receipt scope is incomplete")
if scope["authority_epoch"] < 1:
if epoch < 1:
raise PeerRunsHTTPError("peer room receipt authority epoch is invalid")
scope = dict(zip(_RECEIPT_SCOPE_FIELDS, names[:3] + [epoch] + names[3:]))
if self._room_scope == scope:
return
self._room_scope = scope
@@ -329,9 +273,7 @@ class PeerRunsHTTPClient:
self._terminal_receipts.clear()
def _bind_dispatch_scope(self, dispatch: HostedMemberDispatch) -> None:
self.bind_room_scope(
**{field: getattr(dispatch, field) for field in _RECEIPT_SCOPE_FIELDS}
)
self.bind_room_scope(**{field: getattr(dispatch, field) for field in _RECEIPT_SCOPE_FIELDS})
def _receipt(self, task_id: str, execution_generation: int) -> dict[str, Any] | None:
"""Return the in-memory receipt, falling back to the durable store."""
@@ -340,14 +282,9 @@ class PeerRunsHTTPClient:
return record
from gateway import hosted_rooms
identity = {"task_id": task_id, "execution_generation": execution_generation}
return hosted_rooms.remote_run_receipt(
self.receipt_db_path,
record={
**self._room_scope,
"task_id": task_id,
"execution_generation": execution_generation,
},
)
self.receipt_db_path, record={**self._room_scope, **identity})
def bind_observation(self, *, task_id: str, execution_generation: int) -> None:
"""Pin history/status reads to one exact logical task attempt."""
@@ -363,36 +300,25 @@ class PeerRunsHTTPClient:
self._recovery_backoff.clear()
def _request(
self,
path: str,
*,
method: str = "GET",
body: Mapping[str, Any] | None = None,
headers: Mapping[str, str] | None = None,
room_grant: str | None = None,
) -> dict[str, Any]:
self, path: str, *, method: str = "GET", body: Mapping[str, Any] | None = None,
headers: Mapping[str, str] | None = None, room_grant: str | None = None) -> dict[str, Any]:
from hermes_cli.urllib_security import open_credentialed_url
deadline = time.monotonic() + self.timeout_seconds
ambiguous = method == "POST"
request_headers = {
"Authorization": (
f"HermesRoom {room_grant}" if room_grant else f"Bearer {self.api_key}"
),
"Content-Type": "application/json",
"User-Agent": "Hermes-RoomLink/1.0",
**(headers or {}),
}
request = urllib.request.Request(
f"{self.base_url}{path}",
data=(
json.dumps(body, separators=(",", ":")).encode("utf-8")
if body is not None
else None
),
else None),
method=method,
headers=request_headers,
)
headers={
"Authorization": (
f"HermesRoom {room_grant}" if room_grant else f"Bearer {self.api_key}"),
"Content-Type": "application/json",
"User-Agent": "Hermes-RoomLink/1.0",
**(headers or {})})
try:
with open_credentialed_url(request, timeout=self.timeout_seconds) as response:
raw = _read_bounded_response(
@@ -400,23 +326,19 @@ class PeerRunsHTTPClient:
).decode("utf-8", "replace")
except _PeerResponseTooLarge as exc:
raise PeerRunsHTTPError(
"peer response exceeded the RoomLink size limit", ambiguous=ambiguous
_BUDGET_MESSAGES["size"].format(kind=""), ambiguous=ambiguous,
) from exc
except _PeerResponseDeadlineExceeded as exc:
raise PeerRunsHTTPError(
"peer response exceeded the RoomLink time budget",
retryable=True,
ambiguous=ambiguous,
_BUDGET_MESSAGES["time"].format(kind=""), retryable=True, ambiguous=ambiguous
) from exc
except urllib.error.HTTPError as exc:
self._raise_http_error(exc, method=method, path=path, deadline=deadline)
except (urllib.error.URLError, TimeoutError, OSError) as exc:
not_admitted = ambiguous and _is_proven_pre_admission_failure(exc)
raise PeerRunsHTTPError(
"peer RoomLink endpoint is unreachable",
retryable=True,
ambiguous=ambiguous and not not_admitted,
not_admitted=not_admitted,
"peer RoomLink endpoint is unreachable", retryable=True,
ambiguous=ambiguous and not not_admitted, not_admitted=not_admitted,
) from exc
try:
payload = json.loads(raw)
@@ -428,134 +350,88 @@ class PeerRunsHTTPClient:
@staticmethod
def _raise_http_error(
exc: urllib.error.HTTPError, *, method: str, path: str, deadline: float
) -> NoReturn:
exc: urllib.error.HTTPError, *, method: str, path: str, deadline: float) -> NoReturn:
"""Raise the classified PeerRunsHTTPError for an HTTP error response."""
ambiguous = method == "POST" and exc.code >= 500
# A 4xx on admission proves the peer never admitted the run.
pre_admission = method == "POST" and path == "/v1/runs" and 400 <= exc.code < 500
flags = {"ambiguous": ambiguous, "not_admitted": pre_admission, "status_code": exc.code}
flags = {
"ambiguous": method == "POST" and exc.code >= 500,
"not_admitted": method == "POST" and path == "/v1/runs" and 400 <= exc.code < 500,
"status_code": exc.code}
try:
detail = _read_bounded_response(
exc, max_bytes=MAX_PEER_ERROR_RESPONSE_BYTES, deadline=deadline
).decode("utf-8", "replace")[:500]
except _PeerResponseTooLarge as body_exc:
raise PeerRunsHTTPError(
"peer error response exceeded the RoomLink size limit", **flags
_BUDGET_MESSAGES["size"].format(kind=" error"), **flags,
) from body_exc
except _PeerResponseDeadlineExceeded as body_exc:
raise PeerRunsHTTPError(
"peer error response exceeded the RoomLink time budget",
retryable=True,
**flags,
_BUDGET_MESSAGES["time"].format(kind=" error"), retryable=True, **flags
) from body_exc
except Exception:
detail = ""
error_code = _response_error_code(detail)
logger.debug(
"Peer RoomLink request returned HTTP %s (%s)", exc.code, error_code or "no-code"
)
"Peer RoomLink request returned HTTP %s (%s)", exc.code, error_code or "no-code")
raise PeerRunsHTTPError(
_http_error_message(method, path, exc.code, error_code),
retryable=exc.code in {408, 425, 429} or exc.code >= 500,
error_code=error_code,
**flags,
error_code=error_code, **flags,
) from exc
def prepare(
self,
*,
room_id: str,
profile: str,
source: str,
grant: str,
create: bool,
expected_session_id: str | None = None,
) -> Mapping[str, Any] | None:
self, *, room_id: str, profile: str, source: str, grant: str, create: bool,
expected_session_id: str | None = None) -> Mapping[str, Any] | None:
if source != "bot_room":
raise PeerRunsHTTPError("peer room source must be bot_room")
self._require_room_grant(grant)
logical_session = (
"roomlink_"
+ hashlib.sha256(f"{room_id}\0{profile}".encode("utf-8")).hexdigest()[:32]
)
"roomlink_" + hashlib.sha256(f"{room_id}\0{profile}".encode("utf-8")).hexdigest()[:32])
if expected_session_id and expected_session_id != logical_session:
raise PeerRunsHTTPError("peer room session identity changed")
return {
"session_id": logical_session,
"title": f"Group: {room_id}",
"source": source,
}
return {"session_id": logical_session, "title": f"Group: {room_id}", "source": source}
def _checked_dispatch(
self, dispatch: Mapping[str, Any], grant: str
) -> HostedMemberDispatch:
def _checked_dispatch(self, dispatch: Mapping[str, Any], grant: str) -> HostedMemberDispatch:
"""Validate a dispatch, its grant, and pin scope + observation to it."""
checked = HostedMemberDispatch.from_mapping(dispatch)
self._require_room_grant(grant)
self._bind_dispatch_scope(checked)
self.bind_observation(
task_id=checked.task_id,
execution_generation=checked.execution_generation,
)
task_id=checked.task_id, execution_generation=checked.execution_generation)
return checked
def dispatch(
self,
*,
dispatch: Mapping[str, Any],
grant: str,
) -> Mapping[str, Any]:
checked = self._checked_dispatch(dispatch, grant)
return self._admit_dispatch(checked, grant=grant)
def dispatch(self, *, dispatch: Mapping[str, Any], grant: str) -> Mapping[str, Any]:
return self._admit_dispatch(self._checked_dispatch(dispatch, grant), grant=grant)
def recover_dispatch(
self,
*,
dispatch: Mapping[str, Any],
grant: str,
) -> Mapping[str, Any]:
def recover_dispatch(self, *, dispatch: Mapping[str, Any], grant: str) -> Mapping[str, Any]:
"""Recover one exact admission by receipt or idempotent POST replay."""
checked = self._checked_dispatch(dispatch, grant)
existing = self._receipt_for_dispatch(checked)
existing = self._receipt(checked.task_id, checked.execution_generation)
if existing is not None:
if any(
existing[field] != getattr(checked, field)
for field in _RECEIPT_SCOPE_FIELDS
):
raise PeerRunsHTTPError(
"peer run receipt conflicts with the recovered dispatch"
)
if any(existing[field] != getattr(checked, field) for field in _RECEIPT_SCOPE_FIELDS):
raise PeerRunsHTTPError("peer run receipt conflicts with the recovered dispatch")
return self._accepted(
checked,
run_id=str(existing["run_id"]),
session_id=str(existing["session_id"]),
replayed=True,
)
key = (checked.task_id, checked.execution_generation)
now = self.clock()
checked, run_id=str(existing["run_id"]),
session_id=str(existing["session_id"]), replayed=True)
key, now = (checked.task_id, checked.execution_generation), self.clock()
backoff = self._recovery_backoff.get(key)
if backoff is not None and now < float(backoff["next_attempt_at"]):
raise PeerRunsHTTPError(
"peer admission recovery is backing off",
retryable=True,
ambiguous=True,
)
"peer admission recovery is backing off", retryable=True, ambiguous=True)
try:
recovered = self._admit_dispatch(checked, grant=grant)
except PeerRunsHTTPError as exc:
if exc.retryable or exc.ambiguous:
delay = self._next_poll_delay(backoff)
self._recovery_backoff = {
key: {"delay": delay, "next_attempt_at": now + delay}
}
self._recovery_backoff = {key: {"delay": delay, "next_attempt_at": now + delay}}
raise
self._recovery_backoff.pop(key, None)
return recovered
@staticmethod
def _accepted(
checked: HostedMemberDispatch, *, run_id: str, session_id: str, replayed: bool
checked: HostedMemberDispatch, *, run_id: str, session_id: str, replayed: bool,
) -> dict[str, Any]:
return {
"status": "accepted",
@@ -563,37 +439,25 @@ class PeerRunsHTTPClient:
"execution_generation": checked.execution_generation,
"run_id": run_id,
"session_id": session_id,
"replayed": replayed,
}
"replayed": replayed}
def _admit_dispatch(
self,
checked: HostedMemberDispatch,
*,
grant: str,
) -> Mapping[str, Any]:
def _admit_dispatch(self, checked: HostedMemberDispatch, *, grant: str) -> Mapping[str, Any]:
session_id = self._session_id(checked, grant=grant)
idempotency_key = f"room:{checked.task_id}:{checked.execution_generation}"
def admit(dispatch: HostedMemberDispatch) -> dict[str, Any]:
def admit() -> dict[str, Any]:
return self._request(
"/v1/runs",
method="POST",
body={
"input": dispatch.prompt,
"hosted_room_dispatch": dispatch.as_mapping(),
},
headers={"Idempotency-Key": idempotency_key},
room_grant=grant,
)
"/v1/runs", method="POST",
body={"input": checked.prompt, "hosted_room_dispatch": checked.as_mapping()},
headers={
"Idempotency-Key": f"room:{checked.task_id}:{checked.execution_generation}"},
room_grant=grant)
try:
result = admit(checked)
result = admit()
except PeerRunsHTTPError as exc:
if exc.ambiguous:
result = admit(checked)
else:
if not exc.ambiguous:
raise
result = admit()
run_id = str(result.get("run_id") or "")
if not run_id:
raise PeerRunsHTTPError("peer did not return a run id")
@@ -602,8 +466,7 @@ class PeerRunsHTTPClient:
"session_id": session_id,
**{field: getattr(checked, field) for field in _RECEIPT_SCOPE_FIELDS},
"task_id": checked.task_id,
"execution_generation": checked.execution_generation,
}
"execution_generation": checked.execution_generation}
if self.receipt_db_path is not None:
from gateway import hosted_rooms
@@ -611,40 +474,29 @@ class PeerRunsHTTPClient:
self._runs[(checked.task_id, checked.execution_generation)] = receipt
self._status_cache.pop(run_id, None)
return self._accepted(
checked,
run_id=run_id,
session_id=session_id,
replayed=bool(result.get("replayed", False)),
)
checked, run_id=run_id, session_id=session_id,
replayed=bool(result.get("replayed", False)))
def _session_id(self, dispatch: HostedMemberDispatch, *, grant: str) -> str:
existing = self._receipt_for_dispatch(dispatch)
existing = self._receipt(dispatch.task_id, dispatch.execution_generation)
if existing:
return str(existing["session_id"])
prepared = self.prepare(
room_id=dispatch.room_id,
profile=dispatch.target_profile,
source="bot_room",
grant=grant,
create=True,
)
room_id=dispatch.room_id, profile=dispatch.target_profile, source="bot_room",
grant=grant, create=True)
if prepared is None:
raise PeerRunsHTTPError("peer room session is unavailable")
return str(prepared.get("session_id") or prepared.get("id") or "")
def _observation_receipt(
self, *, room_id: str, profile: str, session_id: str
) -> dict[str, Any] | None:
self, *, room_id: str, profile: str, session_id: str) -> dict[str, Any] | None:
if self._observation_key is None:
return None
record = self._receipt(*self._observation_key)
if record is None:
return None
if (
record["room_id"] != room_id
or record["target_profile"] != profile
or record["session_id"] != session_id
):
scope = (record["room_id"], record["target_profile"], record["session_id"])
if scope != (room_id, profile, session_id):
raise PeerRunsHTTPError("peer observation receipt changed scope")
return record
@@ -653,27 +505,18 @@ class PeerRunsHTTPClient:
return {
key: status[key]
for key in ("run_id", "status", "output", "error", "approval", "last_event")
if key in status
}
if key in status}
def _next_poll_delay(self, cached: Mapping[str, Any] | None) -> float:
previous = (
float(cached["delay"]) if cached is not None else self.poll_min_seconds / 2
)
previous = float(cached["delay"]) if cached is not None else self.poll_min_seconds / 2
return min(self.poll_max_seconds, max(self.poll_min_seconds, previous * 2))
@staticmethod
def _run_is_terminal(status: Mapping[str, Any]) -> bool:
return status.get("status") in _TERMINAL_RUN_STATES
def _poll_receipt(
self,
record: Mapping[str, Any],
*,
grant: str,
) -> dict[str, Any]:
run_id = str(record["run_id"])
now = self.clock()
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"]
@@ -685,81 +528,45 @@ class PeerRunsHTTPClient:
raise error
return status
delay = self._next_poll_delay(cached)
entry = {"delay": delay, "next_poll_at": now + delay}
try:
status = self._compact_run_status(
self._request(
f"/v1/runs/{urllib.parse.quote(run_id, safe='')}",
room_grant=self._require_room_grant(grant),
)
)
self._request(_run_path(record), room_grant=self._require_room_grant(grant)))
if (
str(status.get("run_id") or "") != run_id
or status.get("status") not in _KNOWN_RUN_STATES
):
or status.get("status") not in _KNOWN_RUN_STATES):
raise PeerRunsHTTPError("peer returned a mismatched run status")
except PeerRunsHTTPError as exc:
self._status_cache = {
run_id: {
"status": cached["status"] if cached is not None else {},
"error": exc,
"delay": delay,
"next_poll_at": now + delay,
}
}
previous = cached["status"] if cached is not None else {}
self._status_cache = {run_id: {"status": previous, "error": exc, **entry}}
raise
self._status_cache = {
run_id: {"status": status, "delay": delay, "next_poll_at": now + delay}
}
self._status_cache = {run_id: {"status": status, **entry}}
if self._run_is_terminal(status):
self._terminal_receipts.add(
(str(record["task_id"]), int(record["execution_generation"]))
)
(str(record["task_id"]), int(record["execution_generation"])))
return status
def _receipt_for_dispatch(
self, dispatch: HostedMemberDispatch
) -> dict[str, Any] | None:
return self._receipt(dispatch.task_id, dispatch.execution_generation)
def history(
self,
*,
room_id: str,
profile: str,
session_id: str,
grant: str,
self, *, room_id: str, profile: str, session_id: str, grant: str,
) -> Sequence[Mapping[str, Any]]:
receipt = self._observation_receipt(
room_id=room_id, profile=profile, session_id=session_id
)
receipt = self._observation_receipt(room_id=room_id, profile=profile, session_id=session_id)
if receipt is None:
return []
status = self._poll_receipt(receipt, grant=grant)
state = str(status.get("status") or "")
if state not in {"completed", "failed", "interrupted"}:
return []
return [
{
"role": "assistant",
"task_id": receipt["task_id"],
"execution_generation": receipt["execution_generation"],
"status": "settled" if state == "completed" else "failed",
"message_id": f"peer-run:{status.get('run_id')}",
"content": status.get("output") or status.get("error") or "",
}
]
return [{
"role": "assistant",
"task_id": receipt["task_id"],
"execution_generation": receipt["execution_generation"],
"status": "settled" if state == "completed" else "failed",
"message_id": f"peer-run:{status.get('run_id')}",
"content": status.get("output") or status.get("error") or ""}]
def status(
self,
*,
room_id: str,
profile: str,
session_id: str,
grant: str,
) -> Mapping[str, Any]:
receipt = self._observation_receipt(
room_id=room_id, profile=profile, session_id=session_id
)
self, *, room_id: str, profile: str, session_id: str, grant: str) -> Mapping[str, Any]:
receipt = self._observation_receipt(room_id=room_id, profile=profile, session_id=session_id)
if receipt is None:
return {"active": False, "task_id": None}
status = self._poll_receipt(receipt, grant=grant)
@@ -769,17 +576,10 @@ class PeerRunsHTTPClient:
"execution_generation": receipt["execution_generation"],
"status": status.get("status"),
"run_id": status.get("run_id"),
"approval": status.get("approval"),
}
"approval": status.get("approval")}
def approve_receipt(
self,
*,
task_id: str,
execution_generation: int,
request_id: str,
choice: str,
grant: str,
self, *, task_id: str, execution_generation: int, request_id: str, choice: str, grant: str
) -> Mapping[str, Any] | None:
"""Resolve approval for the exact durable remote run."""
record = self._receipt(task_id, execution_generation)
@@ -790,74 +590,43 @@ class PeerRunsHTTPClient:
raise PeerRunsHTTPError("an exact approval request_id is required")
self._require_room_grant(grant)
return self._post_run_action(
record, "approval", body={"choice": choice, "request_id": request_id}, grant=grant
)
record, "approval", body={"choice": choice, "request_id": request_id}, grant=grant)
def _post_run_action(
self, record: Mapping[str, Any], action: str, *, body: dict[str, Any], grant: str
) -> dict[str, Any]:
run_id = str(record["run_id"])
result = self._request(
f"/v1/runs/{urllib.parse.quote(run_id, safe='')}/{action}",
method="POST",
body=body,
room_grant=grant,
)
self._status_cache.pop(run_id, None)
_run_path(record, action), method="POST", body=body, room_grant=grant)
self._status_cache.pop(str(record["run_id"]), None)
return result
def stop(
self,
*,
dispatch: Mapping[str, Any],
grant: str,
) -> Mapping[str, Any] | None:
def stop(self, *, dispatch: Mapping[str, Any], grant: str) -> Mapping[str, Any] | None:
checked = HostedMemberDispatch.from_mapping(dispatch)
self._bind_dispatch_scope(checked)
return self.stop_receipt(
task_id=checked.task_id,
execution_generation=checked.execution_generation,
grant=grant,
)
task_id=checked.task_id, execution_generation=checked.execution_generation, grant=grant)
def stop_receipt(
self,
*,
task_id: str,
execution_generation: int,
grant: str,
) -> Mapping[str, Any] | None:
self, *, task_id: str, execution_generation: int, grant: str) -> Mapping[str, Any] | None:
"""Stop the exact durable remote run after a home restart."""
record = self._receipt(task_id, execution_generation)
if record is None:
return None
result = self._post_run_action(
record, "stop", body={}, grant=self._require_room_grant(grant)
)
record, "stop", body={}, grant=self._require_room_grant(grant))
if self._run_is_terminal(result):
self._terminal_receipts.add((str(task_id), int(execution_generation)))
return result
def issue_invitation(
self,
*,
room_id: str,
home_install_id: str,
authority_gateway_id: str,
authority_epoch: int,
member_id: str,
grant_id: str,
ttl_seconds: float = 3600,
status_ttl_seconds: float | None = None,
) -> Mapping[str, Any]:
self, *, room_id: str, home_install_id: str, authority_gateway_id: str,
authority_epoch: int, member_id: str, grant_id: str, ttl_seconds: float = 3600,
status_ttl_seconds: float | None = None) -> Mapping[str, Any]:
"""Ask the target gateway to mint a scoped room-member grant."""
if not self.api_key:
raise PeerRunsHTTPError(
"issuing an invitation requires the target gateway API key"
)
raise PeerRunsHTTPError("issuing an invitation requires the target gateway API key")
return self._request(
"/v1/room-members/invitations",
method="POST",
"/v1/room-members/invitations", method="POST",
body={
"room_id": room_id,
"home_install_id": home_install_id,
@@ -869,23 +638,15 @@ class PeerRunsHTTPClient:
**(
{"status_ttl_seconds": status_ttl_seconds}
if status_ttl_seconds is not None
else {}
),
},
)
else {})})
def refresh_grant(
self,
*,
grant: str,
ttl_seconds: float = 24 * 60 * 60,
capability_digest: str | None = None,
execution_policy_digest: str | None = None,
self, *, grant: str, ttl_seconds: float = 24 * 60 * 60,
capability_digest: str | None = None, execution_policy_digest: str | None = None,
) -> Mapping[str, Any]:
"""Renew dispatch access only while its frozen authority is unchanged."""
refreshed = self._scoped_post(
"/v1/room-members/grants/refresh", grant, body={"ttl_seconds": ttl_seconds}
)
"/v1/room-members/grants/refresh", grant, body={"ttl_seconds": ttl_seconds})
replacement = str(refreshed.get("grant") or "")
if not replacement:
raise PeerRunsHTTPError("peer returned no refreshed room grant")
@@ -894,9 +655,7 @@ class PeerRunsHTTPClient:
probe = self.probe(grant=replacement)
error = digest_reauthorization_error(
GatewayRoomCatalog.from_mapping(probe.get("catalog")),
capability_digest=capability_digest,
execution_policy_digest=execution_policy_digest,
)
capability_digest=capability_digest, execution_policy_digest=execution_policy_digest)
if error is not None:
raise error
return {**refreshed, "catalog": probe.get("catalog")}
@@ -907,12 +666,12 @@ class PeerRunsHTTPClient:
def probe(self, *, grant: str) -> Mapping[str, Any]:
"""Verify gateway reachability and the live scoped capability catalog."""
self._require_room_grant(grant)
return self._request("/v1/room-members/capabilities", room_grant=grant)
return self._request(
"/v1/room-members/capabilities", room_grant=self._require_room_grant(grant))
def _scoped_post(self, path: str, grant: str, *, body: dict[str, Any]) -> dict[str, Any]:
self._require_room_grant(grant)
return self._request(path, method="POST", body=body, room_grant=grant)
return self._request(
path, method="POST", body=body, room_grant=self._require_room_grant(grant))
@staticmethod
def _require_room_grant(grant: str) -> str:
tui_gateway/hosted_room_peer_transport.py
+32 -94
View File
@@ -17,11 +17,7 @@ from typing import Any, Protocol
from gateway.hosted_room_driver import TaskIdentity
from gateway.hosted_room_peer import HostedMemberDispatch, PROTOCOL_VERSION
from tui_gateway.hosted_room_driver import (
ROOM_SESSION_SOURCE,
HostedRoomBinding,
InternalSessionRPC,
room_session_title,
)
ROOM_SESSION_SOURCE, HostedRoomBinding, InternalSessionRPC, room_session_title)
class HostedRoomPeerClient(Protocol):
@@ -30,13 +26,7 @@ class HostedRoomPeerClient(Protocol):
def bind_room_scope(self, **scope: Any) -> None: ...
def prepare(
self,
*,
room_id: str,
profile: str,
source: str,
grant: str,
create: bool,
self, *, room_id: str, profile: str, source: str, grant: str, create: bool,
expected_session_id: str | None = None,
) -> Mapping[str, Any] | None: ...
@@ -71,12 +61,8 @@ class FailoverHostedRoomPeerClient:
"""Try alternate links without changing target or logical task identity."""
def __init__(
self,
candidates: Sequence[RoomLinkCandidate],
*,
reprobe_interval_seconds: float = 60,
clock: Callable[[], float] = time.monotonic,
) -> None:
self, candidates: Sequence[RoomLinkCandidate], *, reprobe_interval_seconds: float = 60,
clock: Callable[[], float] = time.monotonic) -> None:
if not candidates:
raise ValueError("at least one RoomLink candidate is required")
if len({candidate.target_install_id for candidate in candidates}) != 1:
@@ -158,17 +144,9 @@ class PeerMemberRoute:
def build_member_dispatch(
*,
binding: HostedRoomBinding,
route: PeerMemberRoute,
room_id: str,
task_id: str,
target_profile: str,
execution_generation: int,
source_event_seq: int,
prompt: str,
trace_id: str,
) -> HostedMemberDispatch:
*, binding: HostedRoomBinding, route: PeerMemberRoute, room_id: str, task_id: str,
target_profile: str, execution_generation: int, source_event_seq: int, prompt: str,
trace_id: str) -> HostedMemberDispatch:
"""Build the fully fenced member dispatch shared by submit and recovery."""
return HostedMemberDispatch.from_mapping({
"protocol_version": PROTOCOL_VERSION,
@@ -187,23 +165,16 @@ def build_member_dispatch(
"prompt_digest": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
"capability_digest": route.capability_digest,
"execution_policy_digest": route.execution_policy_digest,
"trace_id": trace_id,
})
"trace_id": trace_id})
class PeerHostedRoomTransport(InternalSessionRPC):
"""Translate runtime session operations into recipient-validated peer RPC."""
def __init__(
self,
*,
binding: HostedRoomBinding,
route: PeerMemberRoute,
client: HostedRoomPeerClient,
source_event_seq: int = 1,
task_id: str | None = None,
execution_generation: int | None = None,
) -> None:
self, *, binding: HostedRoomBinding, route: PeerMemberRoute,
client: HostedRoomPeerClient, source_event_seq: int = 1, task_id: str | None = None,
execution_generation: int | None = None) -> None:
self.binding = binding
self.route = route
self.client = client
@@ -217,14 +188,10 @@ class PeerHostedRoomTransport(InternalSessionRPC):
bind_scope = getattr(self.client, "bind_room_scope", None)
if callable(bind_scope):
bind_scope(
room_id=self.binding.room_id,
home_install_id=self.route.home_install_id,
authority_gateway_id=self.binding.gateway_id,
authority_epoch=self.binding.authority_epoch,
member_id=self.route.member_id,
target_install_id=self.route.target_install_id,
target_profile=self.route.target_profile,
)
room_id=binding.room_id, home_install_id=route.home_install_id,
authority_gateway_id=binding.gateway_id, authority_epoch=binding.authority_epoch,
member_id=route.member_id, target_install_id=route.target_install_id,
target_profile=route.target_profile)
def _validate_coordinates(self, *, profile: str, source: str, title: str | None = None) -> None:
if source != ROOM_SESSION_SOURCE:
@@ -234,15 +201,13 @@ class PeerHostedRoomTransport(InternalSessionRPC):
if title is not None and title != room_session_title(self.binding.room_id):
raise ValueError("peer room transport title does not match room identity")
def _scoped(self, **extra: Any) -> dict[str, Any]:
"""Room id + grant keyword arguments shared by every scoped client call."""
return {"room_id": self.binding.room_id, "grant": self.route.grant, **extra}
def _prepare(self, *, profile: str, source: str, create: bool, **extra):
return self.client.prepare(
room_id=self.binding.room_id,
profile=profile,
source=source,
grant=self.route.grant,
create=create,
**extra,
)
**self._scoped(profile=profile, source=source, create=create, **extra))
def resolve_exact(self, *, profile: str, title: str, source: str) -> Mapping[str, Any] | None:
self._validate_coordinates(profile=profile, source=source, title=title)
@@ -259,38 +224,24 @@ class PeerHostedRoomTransport(InternalSessionRPC):
def resume(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]:
self._validate_coordinates(profile=profile, source=source)
session = self._prepare(
profile=profile, source=source, create=False, expected_session_id=session_id
)
profile=profile, source=source, create=False, expected_session_id=session_id)
if session is None:
raise RuntimeError("peer room session is unavailable")
self._session_id = session_id
return session
def submit(
self,
*,
profile: str,
session_id: str,
prompt: str,
source: str,
task: TaskIdentity,
execution_generation: int,
on_terminal: Callable[[Mapping[str, Any]], None],
self, *, profile: str, session_id: str, prompt: str, source: str, task: TaskIdentity,
execution_generation: int, on_terminal: Callable[[Mapping[str, Any]], None],
) -> Mapping[str, Any]:
self._validate_coordinates(profile=profile, source=source)
if self._session_id not in {None, session_id}:
raise ValueError("peer room session changed during admission")
dispatch = build_member_dispatch(
binding=self.binding,
route=self.route,
room_id=task.room_id,
task_id=task.task_id,
target_profile=profile,
execution_generation=execution_generation,
source_event_seq=self.source_event_seq,
prompt=prompt,
trace_id=self.route.trace_id or f"trace-{uuid.uuid4().hex}",
)
binding=self.binding, route=self.route, room_id=task.room_id, task_id=task.task_id,
target_profile=profile, execution_generation=execution_generation,
source_event_seq=self.source_event_seq, prompt=prompt,
trace_id=self.route.trace_id or f"trace-{uuid.uuid4().hex}")
self._dispatch = dispatch
self._session_id = session_id
result = self.client.dispatch(dispatch=dispatch.as_mapping(), grant=self.route.grant)
@@ -300,21 +251,11 @@ class PeerHostedRoomTransport(InternalSessionRPC):
def history(self, *, profile: str, session_id: str, source: str) -> Sequence[Mapping[str, Any]]:
self._validate_coordinates(profile=profile, source=source)
return self.client.history(
room_id=self.binding.room_id,
profile=profile,
session_id=session_id,
grant=self.route.grant,
)
return self.client.history(**self._scoped(profile=profile, session_id=session_id))
def info(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]:
self._validate_coordinates(profile=profile, source=source)
return self.client.status(
room_id=self.binding.room_id,
profile=profile,
session_id=session_id,
grant=self.route.grant,
)
return self.client.status(**self._scoped(profile=profile, session_id=session_id))
def interrupt(
self, *, profile: str, session_id: str, source: str, expected_task_id: str
@@ -325,14 +266,11 @@ class PeerHostedRoomTransport(InternalSessionRPC):
if (
self.task_id != expected_task_id
or not self.execution_generation
or not hasattr(self.client, "stop_receipt")
):
or not hasattr(self.client, "stop_receipt")):
return None
return self.client.stop_receipt(
task_id=expected_task_id,
execution_generation=self.execution_generation,
grant=self.route.grant,
)
task_id=expected_task_id, execution_generation=self.execution_generation,
grant=self.route.grant)
if dispatch.task_id != expected_task_id:
return None
return self.client.stop(dispatch=dispatch.as_mapping(), grant=self.route.grant)
tui_gateway/hosted_room_server_rpc.py
+18 -37
View File
@@ -36,17 +36,14 @@ class HostedRoomServerRPC:
self._ids = itertools.count(1)
def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
handler = self.server._methods[method]
envelope = handler(f"hosted-room-{next(self._ids)}", params)
envelope = self.server._methods[method](f"hosted-room-{next(self._ids)}", params)
if not isinstance(envelope, dict):
envelope = {}
error = envelope.get("error")
if isinstance(error, dict):
raise HostedRoomSessionError(
method,
int(error.get("code") or 5000),
str(error.get("message") or "gateway rejected the request"),
)
method, int(error.get("code") or 5000),
str(error.get("message") or "gateway rejected the request"))
result = envelope.get("result")
if not isinstance(result, dict):
raise HostedRoomSessionError(method, 5000, "gateway returned no result")
@@ -55,16 +52,14 @@ class HostedRoomServerRPC:
def resolve_exact(self, *, profile: str, title: str, source: str) -> Mapping[str, Any] | None:
del source
result = self._call(
"session.list", {"profile": profile, "title": title, "include_hidden": True}
)
"session.list", {"profile": profile, "title": title, "include_hidden": True})
rows = result.get("sessions")
if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
return None
row = rows[0]
return {
"session_id": row.get("resolved_id") or row.get("id"),
"title": row.get("title") or title,
}
"title": row.get("title") or title}
def create(self, *, profile: str, title: str, source: str) -> Mapping[str, Any]:
return self._call(
@@ -76,26 +71,16 @@ class HostedRoomServerRPC:
"hidden": True,
"room_plumbing": True,
"follow_profile_config": True,
"close_on_disconnect": False,
},
)
"close_on_disconnect": False})
def resume(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]:
return self._call(
"session.resume",
{"profile": profile, "session_id": session_id, "omit_messages": True, "source": source},
)
{"profile": profile, "session_id": session_id, "omit_messages": True, "source": source})
def submit(
self,
*,
profile: str,
session_id: str,
prompt: str,
source: str,
task: state.TaskIdentity,
execution_generation: int,
on_terminal: Callable[[Mapping[str, Any]], None],
self, *, profile: str, session_id: str, prompt: str, source: str, task: state.TaskIdentity,
execution_generation: int, on_terminal: Callable[[Mapping[str, Any]], None],
) -> Mapping[str, Any]:
try:
return self._call(
@@ -110,11 +95,8 @@ class HostedRoomServerRPC:
"task_id": task.task_id,
"thread_id": task.thread_id,
"turn_id": task.turn_id,
"execution_generation": execution_generation,
},
"_hosted_terminal_callback": on_terminal,
},
)
"execution_generation": execution_generation},
"_hosted_terminal_callback": on_terminal})
except HostedRoomSessionError as exc:
# In-process prompt.submit error envelopes are returned before the
# background turn is admitted. Preserve that proof so the driver
@@ -150,14 +132,12 @@ class HostedRoomServerRPC:
task = record.get("_hosted_room_task")
result = {
"active": bool(record.get("running")),
"task_id": task.get("task_id") if isinstance(task, dict) else None,
}
"task_id": task.get("task_id") if isinstance(task, dict) else None}
pending_reader = getattr(self.server, "_pending_approval_request_payload", None)
pending = (
pending_reader(str(record.get("session_key") or ""))
if callable(pending_reader)
else None
)
else None)
if pending:
result["status"] = "waiting_for_approval"
result["pending_approval"] = pending
@@ -167,8 +147,7 @@ class HostedRoomServerRPC:
"""Resolve one exact local room approval without broad policy changes."""
return self._call(
"approval.respond",
{"session_id": session_id, "request_id": request_id, "choice": choice, "all": False},
)
{"session_id": session_id, "request_id": request_id, "choice": choice, "all": False})
def interrupt(
self, *, profile: str, session_id: str, source: str, expected_task_id: str
@@ -176,5 +155,7 @@ class HostedRoomServerRPC:
del source
return self._call(
"session.interrupt",
{"profile": profile, "session_id": session_id, "expected_hosted_task_id": expected_task_id},
)
{
"profile": profile,
"session_id": session_id,
"expected_hosted_task_id": expected_task_id})