From 7b6f60b4f9a0103cc99f11caed2e343849e3d288 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:46:51 -0700 Subject: [PATCH] =?UTF-8?q?refactor(gateway/platforms):=20runs/room=20?= =?UTF-8?q?=E2=80=94=20fold=20=5Femit=5Fto=5Fstream=20into=20=5Fmark=5Frun?= =?UTF-8?q?=5Fevent,=20unify=20room-grant=20error=20builder,=20inline=20sw?= =?UTF-8?q?eep=20filters,=20tighten=20layouts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gateway/platforms/api_server_room_dispatch.py | 3 +- gateway/platforms/api_server_room_grants.py | 50 +++----- .../platforms/api_server_run_idempotency.py | 13 +- gateway/platforms/api_server_runs.py | 116 +++++++----------- 4 files changed, 64 insertions(+), 118 deletions(-) diff --git a/gateway/platforms/api_server_room_dispatch.py b/gateway/platforms/api_server_room_dispatch.py index 38f56a8da8..5c755a3ceb 100644 --- a/gateway/platforms/api_server_room_dispatch.py +++ b/gateway/platforms/api_server_room_dispatch.py @@ -63,8 +63,7 @@ def _room_dispatch_error(exc: Exception, *, _openai_error) -> "web.Response": async def _normalize_room_dispatch( self, request: "web.Request", body: Any, *, _api_server) -> tuple[Any, "web.Response | None"]: """Validate and normalize a scoped RoomLink dispatch request.""" - _openai_error = _api_server._openai_error - room_token = self._room_grant_token(request) + _openai_error, room_token = _api_server._openai_error, self._room_grant_token(request) if not room_token: return body, None if not isinstance(body, dict) or set(body) - {"input", "hosted_room_dispatch"}: diff --git a/gateway/platforms/api_server_room_grants.py b/gateway/platforms/api_server_room_grants.py index 7cf55ec459..1c2eab3e34 100644 --- a/gateway/platforms/api_server_room_grants.py +++ b/gateway/platforms/api_server_room_grants.py @@ -2,7 +2,7 @@ import time import uuid -from typing import Any +from typing import Any, Optional try: from aiohttp import web @@ -25,18 +25,13 @@ def _require_unchanged_execution_policy(claims: dict[str, Any], execution_policy raise RoomGrantReauthorizationRequired("room execution policy changed") -def _invalid_room_grant(_openai_error) -> "web.Response": - return _json_error( - _openai_error, "Room authorization is invalid or expired.", - err_type="gateway_auth_error", code="invalid_room_grant", status=401) - - -def _room_grant_error_response(exc: Exception, *, _openai_error) -> "web.Response": - if not isinstance(exc, RoomGrantReauthorizationRequired): - return _invalid_room_grant(_openai_error) - return _json_error( - _openai_error, "Room authorization needs to be renewed.", - err_type="gateway_auth_error", code="room_reauthorization_required", status=403) +def _room_grant_error_response(exc: Optional[Exception] = None, *, _openai_error) -> "web.Response": + """401 invalid grant, or 403 reauthorization-required for a revoked/superseded grant.""" + if isinstance(exc, RoomGrantReauthorizationRequired): + message, code, status = "Room authorization needs to be renewed.", "room_reauthorization_required", 403 + else: + message, code, status = "Room authorization is invalid or expired.", "invalid_room_grant", 401 + return _json_error(_openai_error, message, err_type="gateway_auth_error", code=code, status=status) def _hard_expiry(claims: dict[str, Any]) -> float: @@ -57,8 +52,7 @@ def _local_target(claims: dict[str, Any] | None, _api_request_profile) -> tuple[ from gateway import hosted_rooms profile = _api_request_profile.get() or "default" installation_id = hosted_rooms.local_authority_gateway_id() - if claims is not None and ( - claims["target_profile"] != profile or claims["target_install_id"] != installation_id): + if claims is not None and (claims["target_profile"], claims["target_install_id"]) != (profile, installation_id): raise ValueError("room grant target does not match this profile") return profile, installation_id @@ -85,11 +79,8 @@ def _http_routes(self) -> list[tuple[str, str, Any]]: def _room_grant_token(request: "web.Request") -> str: - authorization = str(request.headers.get("Authorization") or "") - scheme, separator, token = authorization.partition(" ") - if not separator or scheme.lower() != "hermesroom": - return "" - return token.strip() + scheme, separator, token = str(request.headers.get("Authorization") or "").partition(" ") + return token.strip() if separator and scheme.lower() == "hermesroom" else "" def _room_grant_secret(self) -> bytes: @@ -156,13 +147,9 @@ async def _handle_room_member_invitation( except Exception as exc: return _json_error(_openai_error, str(exc), code="invalid_room_invitation", status=400) return web.json_response({ - "object": "hermes.room_member.invitation", - "grant": token, - "target_profile": profile, - "catalog": catalog, - "expires_at": float(claims["expires_at"]), - "status_expires_at": float(claims["status_expires_at"]), - }, status=201) + "object": "hermes.room_member.invitation", "grant": token, "target_profile": profile, + "catalog": catalog, "expires_at": float(claims["expires_at"]), + "status_expires_at": float(claims["status_expires_at"])}, status=201) async def _handle_room_member_capabilities( @@ -214,11 +201,8 @@ async def _handle_room_member_grant_refresh( except Exception as exc: return _room_grant_error_response(exc, _openai_error=_openai_error) return web.json_response({ - "object": "hermes.room_member.grant", - "grant": token, - "expires_at": now + dispatch_ttl, - "status_expires_at": hard_expiry, - "execution_policy": execution_policy}) + "object": "hermes.room_member.grant", "grant": token, "expires_at": now + dispatch_ttl, + "status_expires_at": hard_expiry, "execution_policy": execution_policy}) async def _handle_room_member_grant_revoke( @@ -240,5 +224,5 @@ async def _handle_room_member_grant_revoke( hosted_rooms.revoke_room_grant_scope( hosted_rooms.default_db_path(), claims=claims, expires_at=_hard_expiry(claims)) except Exception: - return _invalid_room_grant(_openai_error) + return _room_grant_error_response(_openai_error=_openai_error) return web.json_response({"object": "hermes.room_member.grant.revocation", "revoked": True}) diff --git a/gateway/platforms/api_server_run_idempotency.py b/gateway/platforms/api_server_run_idempotency.py index 92271c24f4..2d47e78a25 100644 --- a/gateway/platforms/api_server_run_idempotency.py +++ b/gateway/platforms/api_server_run_idempotency.py @@ -60,7 +60,6 @@ class RunIdempotencyStore: def durable(self) -> bool: """Whether reservations survive this process.""" return self._db_path is not None - def __init__(self, db_path: str = None): if db_path is None: try: @@ -106,9 +105,7 @@ class RunIdempotencyStore: self._tighten_permissions() def _tighten_permissions(self) -> None: - if not self._db_path: - return - for suffix in ("", "-wal", "-shm"): + for suffix in ("", "-wal", "-shm") if self._db_path else (): candidate = Path(self._db_path + suffix) try: if candidate.exists(): @@ -146,9 +143,8 @@ class RunIdempotencyStore: "scope,idempotency_key,fingerprint,run_id,status_json," "owner_pid,owner_started,retention_until,created_at,updated_at" ") VALUES(?,?,?,?,?,?,?,?,?,?)", - ( - scope, key, fingerprint, run_id, encoded, - int(owner_pid or 0), int(owner_started or 0), retention_until, now, now)) + (scope, key, fingerprint, run_id, encoded, int(owner_pid or 0), int(owner_started or 0), + retention_until, now, now)) self._conn.commit() return "created", _record(run_id, encoded, owner_pid, owner_started, now) | {"status": status} @@ -182,8 +178,7 @@ class RunIdempotencyStore: terminal = False if terminal: self._conn.execute( - "DELETE FROM run_idempotency WHERE scope=? AND idempotency_key=?", - (stale_scope, stale_key)) + "DELETE FROM run_idempotency WHERE scope=? AND idempotency_key=?", (stale_scope, stale_key)) def status_for_run(self, scope: str, run_id: str, *, retention_until: float = 0) -> dict[str, Any] | None: """Load one durable run status inside its authenticated scope.""" diff --git a/gateway/platforms/api_server_runs.py b/gateway/platforms/api_server_runs.py index d5050adba2..b7dcfbb3b7 100644 --- a/gateway/platforms/api_server_runs.py +++ b/gateway/platforms/api_server_runs.py @@ -24,15 +24,13 @@ from gateway.platforms.api_server_run_idempotency import TERMINAL_STATUSES logger = logging.getLogger("gateway.platforms.api_server") _ROOM_RETENTION_REQUEST_KEY = ( - RequestKey("hermes.room_run_retention_until", float) - if RequestKey is not None + RequestKey("hermes.room_run_retention_until", float) if RequestKey is not None else "hermes.room_run_retention_until") # Forwarded subagent lifecycle fields; free-text ones are secret-redacted. _SUBAGENT_EVENT_KEYS = ( - "goal", "task_count", "task_index", "subagent_id", "child_session_id", - "delegation_id", "parent_id", "depth", "model", "tool_count", "status", - "summary", "duration_seconds", "input_tokens", "output_tokens", - "reasoning_tokens", "api_calls", "cost_usd", "files_read", "files_written", + "goal", "task_count", "task_index", "subagent_id", "child_session_id", "delegation_id", "parent_id", + "depth", "model", "tool_count", "status", "summary", "duration_seconds", "input_tokens", + "output_tokens", "reasoning_tokens", "api_calls", "cost_usd", "files_read", "files_written", "output_tail") _SUBAGENT_TEXT_KEYS = ("goal", "summary", "output_tail") # Terminal usage payload: (wire key, agent attribute), in wire order. @@ -115,11 +113,9 @@ def _idempotency_capabilities(self, *, store_type) -> dict[str, Any]: def _close_run_state(self) -> None: - store = getattr(self, "_run_idempotency_store", None) - if store is None: - return try: - store.close() + if getattr(self, "_run_idempotency_store", None) is not None: + self._run_idempotency_store.close() except Exception: logger.debug("Failed to close run idempotency store for %s", self.name, exc_info=True) @@ -154,13 +150,11 @@ def _make_run_event_callback(self, run_id: str, loop: "asyncio.AbstractEventLoop def _push(event: Dict[str, Any]) -> None: self._set_run_status( - run_id, self._run_statuses.get(run_id, {}).get("status", "running"), last_event=event.get("event") - ) + run_id, self._run_statuses.get(run_id, {}).get("status", "running"), last_event=event.get("event")) q = self._run_streams.get(run_id) - if q is None: - return - with suppress(Exception): - loop.call_soon_threadsafe(q.put_nowait, event) + if q is not None: + with suppress(Exception): + loop.call_soon_threadsafe(q.put_nowait, event) def _callback(event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs): # _thinking / subagent.tool / subagent_progress are deliberately dropped (UI noise); @@ -174,12 +168,10 @@ def _make_run_event_callback(self, run_id: str, loop: "asyncio.AbstractEventLoop event["preview"] = redact_sensitive_text(str(preview), force=True) for key in _SUBAGENT_EVENT_KEYS: value = kwargs.get(key) - if value is None: - continue - # Free text may carry child tool output: force secret redaction on this public stream. - if key in _SUBAGENT_TEXT_KEYS and isinstance(value, str): - value = redact_sensitive_text(value, force=True) - event[key] = value + if value is not None: + # Free text may carry child tool output: force secret redaction on this public stream. + redact = key in _SUBAGENT_TEXT_KEYS and isinstance(value, str) + event[key] = redact_sensitive_text(value, force=True) if redact else value _push(event) return _callback @@ -265,10 +257,9 @@ def _resolve_conversation_history( if raw_history: if not isinstance(raw_history, list): return [], instructions, None, _json_error( - _openai_error, "'conversation_history' must be an array of message objects", status=400 - ) + _openai_error, "'conversation_history' must be an array of message objects", status=400) for i, entry in enumerate(raw_history): - if not isinstance(entry, dict) or "role" not in entry or "content" not in entry: + if not isinstance(entry, dict) or {"role", "content"} - set(entry): return [], instructions, None, _json_error( _openai_error, f"conversation_history[{i}] must have 'role' and 'content' fields", status=400) @@ -288,9 +279,8 @@ def _resolve_conversation_history( if isinstance(msg, dict) and msg.get("role") and msg.get("content"): content = msg["content"] if isinstance(content, list): # flatten multi-part content blocks to text - content = " ".join( - part.get("text", "") for part in content - if isinstance(part, dict) and part.get("type") == "text") + content = " ".join(p.get("text", "") for p in content + if isinstance(p, dict) and p.get("type") == "text") conversation_history.append({"role": msg["role"], "content": str(content)}) return conversation_history, instructions, stored_session_id, None @@ -335,8 +325,7 @@ class _RunLaunch: @property def approval_session_key(self) -> str: - # Approval queues are isolated per run: session ids are conversation scopes, not - # authorization namespaces, so resolving one run's approval must not unblock another's. + # Isolated per run: session ids are conversation scopes, not authorization namespaces. return self.run_id def put_event(self, event: Optional[Dict]) -> None: @@ -345,26 +334,17 @@ class _RunLaunch: self.queue.put_nowait(event) -def _optional_dict(body: Any, key: str) -> Optional[dict]: - value = body.get(key) if isinstance(body, dict) else None - return value if isinstance(value, dict) else None - - def _forget_run(self, run_id: str, *tables) -> None: """Drop *run_id* from the given run-keyed dicts/sets, then release its owner stamp.""" for table in tables: - if isinstance(table, set): - table.discard(run_id) - else: - table.pop(run_id, None) + (table.discard if isinstance(table, set) else lambda k: table.pop(k, None))(run_id) self._release_run_owner_if_forgotten(run_id) def _retire_live_run(self, run_id: str) -> None: """Retire agent/task/approval control state once the executor-backed task is done.""" - _forget_run( - self, run_id, self._active_run_agents, self._active_run_tasks, - self._run_approval_sessions, self._stopping_run_ids) + _forget_run(self, run_id, self._active_run_agents, self._active_run_tasks, self._run_approval_sessions, + self._stopping_run_ids) def _drop_run_transport(self, run_id: str) -> None: @@ -385,8 +365,10 @@ async def _handle_runs(self, request: "web.Request", *, _api_server) -> "web.Res body, room_error = await self._normalize_room_dispatch(request, body) if room_error is not None: return room_error - room_dispatch = _optional_dict(body, "hosted_room_dispatch") - room_execution_policy = _optional_dict(body, "_room_execution_policy") + room_dispatch, room_execution_policy = ( + v if isinstance(v, dict) else None for v in ( + (body.get("hosted_room_dispatch"), body.get("_room_execution_policy")) + if isinstance(body, dict) else (None, None))) idempotency_key = request.headers.get("Idempotency-Key", "").strip() if len(idempotency_key) > 255 or any(ord(ch) < 33 or ord(ch) > 126 for ch in idempotency_key): return _json_error( @@ -443,10 +425,8 @@ async def _handle_runs(self, request: "web.Request", *, _api_server) -> "web.Res # An explicit or chained session owns its routing key and is never rebound to the header. _declared_selected = not session_id and bool(gateway_session_key) session_id = session_id or self._declared_conversation_session(gateway_session_key) or run_id - q: "asyncio.Queue[Optional[Dict]]" = asyncio.Queue() - created_at = time.time() - self._run_streams[run_id] = q - self._run_streams_created[run_id] = created_at + q = self._run_streams[run_id] = asyncio.Queue() + created_at = self._run_streams_created[run_id] = time.time() self._run_approval_sessions[run_id] = run_id # approval session key (see _RunLaunch) initial_status = self._set_run_status( run_id, "queued", created_at=created_at, session_id=session_id, model=body.get("model", self._model_name)) @@ -465,22 +445,16 @@ async def _handle_runs(self, request: "web.Request", *, _api_server) -> "web.Res self, run_id, q, session_id, gateway_session_key, _declared_selected, user_message, conversation_history, agent_kwargs=dict( - ephemeral_system_prompt=instructions, session_id=session_id, - gateway_session_key=gateway_session_key, - requested_model=agent_overrides.get("requested_model"), - requested_provider=agent_overrides.get("requested_provider"), - model_options=agent_overrides.get("model_options"), route=route, - room_dispatch=room_dispatch, room_execution_policy=room_execution_policy), + ephemeral_system_prompt=instructions, session_id=session_id, gateway_session_key=gateway_session_key, + route=route, room_dispatch=room_dispatch, room_execution_policy=room_execution_policy, + **{k: agent_overrides.get(k) for k in ("requested_model", "requested_provider", "model_options")}), request_profile=_api_server._api_request_profile.get(), browser_control_principal=_api_server._api_request_browser_control_principal.get(), browser_control_transport_family=_api_server._api_request_browser_control_transport_family.get()) self._activate_admitted_request() - task = asyncio.create_task(_execute_run(self, launch, _api_server=_api_server)) - self._active_run_tasks[run_id] = task - try: + task = self._active_run_tasks[run_id] = asyncio.create_task(_execute_run(self, launch, _api_server=_api_server)) + with suppress(TypeError): self._background_tasks.add(task) # tracked for shutdown drain - except TypeError: - pass if hasattr(task, "add_done_callback"): task.add_done_callback(self._background_tasks.discard) return _accepted_response(run_id, "started", gateway_session_key, replayed=False) @@ -597,8 +571,7 @@ async def _execute_run(self, run: _RunLaunch, *, _api_server) -> None: _finish("failed", error=_redact_api_error_text(result.get("error") or "agent run failed")) else: # Undelivered steer text rides on the terminal event/status for client replay. - pending_steer = result.get("pending_steer") - extra = {"pending_steer": pending_steer} if pending_steer else {} + extra = {"pending_steer": result["pending_steer"]} if result.get("pending_steer") else {} _finish("completed", extra, output=result.get("final_response", ""), usage=usage) except asyncio.CancelledError: _finish("cancelled") @@ -646,7 +619,7 @@ def _request_owns_run(self, request: "web.Request", run_id: str) -> bool: return self._run_idempotency_store.owns_run(scope, run_id) -def _load_owned_run(self, request: "web.Request", *, _api_server, permission: Optional[str], active_fallback: bool): +def _load_owned_run(self, request, *, _api_server, permission: Optional[str], active_fallback: bool): """Authenticate (*permission* -> room-grant aware; ``None`` -> API key only) and resolve ``(run_id, status, agent, task, error)``; *active_fallback* reports a live in-process run without pollable status as ``running`` instead of 404.""" @@ -713,17 +686,13 @@ async def _handle_run_events(self, request: "web.Request", *, _api_server) -> "w return response -def _emit_to_stream(self, run_id: str, event: Dict[str, Any]) -> None: +def _mark_run_event(self, run_id: str, name: str, **fields: Any) -> None: + """Record a control-plane event on the run status and (best effort) its SSE stream.""" + self._set_run_status(run_id, "running", last_event=name) q = self._run_streams.get(run_id) if q is not None: with suppress(Exception): - q.put_nowait(event) - - -def _mark_run_event(self, run_id: str, name: str, **fields: Any) -> None: - """Record a control-plane event on the run status and its SSE stream.""" - self._set_run_status(run_id, "running", last_event=name) - _emit_to_stream(self, run_id, _run_event(run_id, name, **fields)) + q.put_nowait(_run_event(run_id, name, **fields)) _APPROVAL_CHOICE_ALIASES = {"approve": "once", "approved": "once", "allow": "once"} @@ -849,10 +818,9 @@ def _sweep_orphaned_runs_once(self, now: Optional[float] = None) -> None: """Expire old SSE buffers without treating transport age as run age.""" if now is None: now = time.time() - stale = [ - run_id for run_id, created_at in list(self._run_streams_created.items()) - if now - created_at > self._RUN_STREAM_TTL and run_id not in self._run_stream_subscribers] - for run_id in stale: + for run_id, created_at in list(self._run_streams_created.items()): + if now - created_at <= self._RUN_STREAM_TTL or run_id in self._run_stream_subscribers: + continue logger.debug("[api_server] sweeping expired run transport %s", run_id) task = self._active_run_tasks.get(run_id) # Transport TTL bounds buffering; live control state survives until the task returns.