From 2ed41353988615423dd2ef08520375bd85b70579 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:12:37 -0700 Subject: [PATCH] refactor(agent/relay_runtime): _warn_on_error fail-open helper for telemetry paths; conversation/compaction bodies lifted out of try blocks --- agent/relay_llm.py | 25 +++--- agent/relay_runtime.py | 174 +++++++++++++++++++++-------------------- 2 files changed, 99 insertions(+), 100 deletions(-) diff --git a/agent/relay_llm.py b/agent/relay_llm.py index 84552a0af8..7b3152e4ef 100644 --- a/agent/relay_llm.py +++ b/agent/relay_llm.py @@ -8,6 +8,7 @@ import inspect import json import logging from collections.abc import Callable, Iterator +from functools import partial from types import SimpleNamespace from typing import Any @@ -51,10 +52,12 @@ class _ManagedAttempt: @classmethod def resolve( - cls, session_id: str, request: dict[str, Any], metadata: dict[str, Any] | None, *, - name: str, model_name: str, + cls, session_id: str | None, request: dict[str, Any], metadata: dict[str, Any] | None, + *, name: str, model_name: str, ) -> "_ManagedAttempt | None": """Return the managed attempt for ``session_id``, or None to run unmanaged.""" + if session_id is None: + return None runtime, session, parent = relay_runtime.resolve_execution_context(session_id) if runtime is None or session is None or not runtime.managed_execution_enabled(): return None @@ -184,9 +187,7 @@ def execute( if attempt is None: return callback(request) - def invoke(next_request: Any) -> Any: - return attempt.invoke(callback, next_request) - + invoke = partial(attempt.invoke, callback) try: managed = _run_awaitable(attempt.run_managed(attempt.runtime.relay.llm.execute, invoke)) except BaseException as exc: @@ -206,9 +207,7 @@ async def execute_async( if attempt is None: return await callback(request) - async def invoke(next_request: Any) -> Any: - return await attempt.invoke_async(callback, next_request) - + invoke = partial(attempt.invoke_async, callback) try: managed = await attempt.run_managed(attempt.runtime.relay.llm.execute, invoke) except BaseException as exc: @@ -227,11 +226,8 @@ def execute_current( model_name: str, metadata: dict[str, Any] | None = None, defer_logical_completion: bool = False, ) -> Any: """Run a provider attempt under the inherited Hermes turn when present.""" - session_id = _current_session_id() - if session_id is None: - return callback(request) return execute( - request, callback, session_id=session_id, name=name, model_name=model_name, + request, callback, session_id=_current_session_id(), name=name, model_name=model_name, metadata=metadata, defer_logical_completion=defer_logical_completion, ) @@ -241,11 +237,8 @@ async def execute_current_async( model_name: str, metadata: dict[str, Any] | None = None, defer_logical_completion: bool = False, ) -> Any: """Run an async provider attempt under the inherited turn when present.""" - session_id = _current_session_id() - if session_id is None: - return await callback(request) return await execute_async( - request, callback, session_id=session_id, name=name, model_name=model_name, + request, callback, session_id=_current_session_id(), name=name, model_name=model_name, metadata=metadata, defer_logical_completion=defer_logical_completion, ) diff --git a/agent/relay_runtime.py b/agent/relay_runtime.py index 8afc910fc0..f135b35f10 100644 --- a/agent/relay_runtime.py +++ b/agent/relay_runtime.py @@ -334,10 +334,6 @@ class _ProcessRelayPluginConfiguration: self._state = state return state - def _reset_if_cleared(self) -> None: - if self._clear_active(): - self._state = _RelayPluginConfigurationState.UNINITIALIZED - def release(self, owner: Any) -> None: """Release one host and clear Relay after the final host exits.""" with self._lock: @@ -350,13 +346,13 @@ class _ProcessRelayPluginConfiguration: """Clear process-global state left by directly constructed test hosts.""" with self._lock: self._owners.clear() - self._reset_if_cleared() + self.retry_pending_cleanup() def retry_pending_cleanup(self) -> None: """Retry a failed final cleanup without disrupting live owners.""" with self._lock: - if not self._owners: - self._reset_if_cleared() + if not self._owners and self._clear_active(): + self._state = _RelayPluginConfigurationState.UNINITIALIZED def _clear_active(self) -> bool: relay = self._relay @@ -983,9 +979,8 @@ class ConversationLease: def live_runtime(self) -> RelayRuntime | None: """Return the real Relay host when this lease owns an open session.""" - if isinstance(self.host, RelayRuntime) and self.session is not None: - return self.host - return None + host = self.host + return host if isinstance(host, RelayRuntime) and self.session is not None else None @dataclass @@ -1035,6 +1030,15 @@ class managed_callback_guard: _MANAGED_CALLBACK_DEPTH.reset(self._token) +def _warn_on_error(what: str, callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + """Run fail-open telemetry work: log ``Hermes Relay failed`` and return None on error.""" + try: + return callback(*args, **kwargs) + except Exception: + logger.warning("Hermes Relay %s failed", what, exc_info=True) + return None + + def _flag_open_session(session: RelaySession, flag: str) -> None: """Set a pending-rotation/close flag unless the session is already closing.""" with session.lock: @@ -1077,26 +1081,32 @@ class RelaySessionCoordinator: ) session = None if isinstance(host, RelayRuntime): - try: - self._prepare_session(host, { - "profile_key": profile_key, "session_id": session_id, "platform": platform, - "parent_session_id": parent_session_id, "model": model, - }) - metadata = {"hermes.execution_surface": platform or "unknown"} - if parent_session_id and parent_session_id != session_id: - session = host.register_subagent( - {"parent_session_id": parent_session_id, "child_session_id": session_id}, - metadata=metadata, - ) - else: - session = host.ensure_session({"session_id": session_id}, metadata=metadata) - except Exception: - logger.warning("Hermes Relay conversation initialization failed", exc_info=True) + session = _warn_on_error( + "conversation initialization", self._open_conversation_session, host, + profile_key=profile_key, session_id=session_id, platform=platform, + parent_session_id=parent_session_id, model=model, + ) return ConversationLease( profile_key=profile_key, session_id=session_id, platform=platform, host=host, session=session, parent_session_id=parent_session_id, ) + def _open_conversation_session( + self, host: RelayRuntime, *, profile_key: str, session_id: str, platform: str, + parent_session_id: str, model: str, + ) -> RelaySession | None: + self._prepare_session(host, { + "profile_key": profile_key, "session_id": session_id, "platform": platform, + "parent_session_id": parent_session_id, "model": model, + }) + metadata = {"hermes.execution_surface": platform or "unknown"} + if parent_session_id and parent_session_id != session_id: + return host.register_subagent( + {"parent_session_id": parent_session_id, "child_session_id": session_id}, + metadata=metadata, + ) + return host.ensure_session({"session_id": session_id}, metadata=metadata) + def begin_turn( self, lease: ConversationLease, *, turn_id: str, task_id: str ) -> RelayTurnContext: @@ -1110,7 +1120,7 @@ class RelaySessionCoordinator: # sibling scopes whose completion order is not LIFO. turn.relay_enabled = False logger.warning( - "Skipping Relay instrumentation for concurrent Hermes turn " "%s in session %s", + "Skipping Relay instrumentation for concurrent Hermes turn %s in session %s", turn_id, lease.session_id, ) else: @@ -1120,25 +1130,15 @@ class RelaySessionCoordinator: if host is not None: # Segment rotation happens HERE — the only point with no live turn scope on # the stack, so the session scope can close/reopen without breaking LIFO. - try: - self._maybe_rotate_segment(host, lease.session) - except Exception: - logger.warning("Hermes Relay segment rotation failed", exc_info=True) - try: - turn.handle = host.run_in_session( - lease.session, - host.relay.scope.push, - TURN_SCOPE, - host.relay.ScopeType.Function, - handle=lease.session.handle, - input={}, - metadata=runtime_metadata( - host.runtime_id, **{"hermes.execution_surface": lease.platform or "unknown"} - ), - timeout=_SCOPE_OP_TIMEOUT, - ) - except Exception: - logger.warning("Hermes Relay turn initialization failed", exc_info=True) + _warn_on_error("segment rotation", self._maybe_rotate_segment, host, lease.session) + turn.handle = _warn_on_error( + "turn initialization", host.run_in_session, lease.session, host.relay.scope.push, + TURN_SCOPE, host.relay.ScopeType.Function, handle=lease.session.handle, input={}, + metadata=runtime_metadata( + host.runtime_id, **{"hermes.execution_surface": lease.platform or "unknown"} + ), + timeout=_SCOPE_OP_TIMEOUT, + ) turn._previous_turn = _CURRENT_TURN.get() _CURRENT_TURN.set(turn) return turn @@ -1175,11 +1175,10 @@ class RelaySessionCoordinator: # Delegated agents own one turn: close their conversation while the # active-turn guard is held so a parent timeout fallback cannot race it. if lease.parent_session_id and isinstance(lease.host, RelayRuntime): - lease.host.unregister_subagent({"child_session_id": lease.session_id}) - except Exception: - logger.warning( - "Hermes Relay child conversation finalization failed", exc_info=True - ) + _warn_on_error( + "child conversation finalization", lease.host.unregister_subagent, + {"child_session_id": lease.session_id}, + ) finally: self._unregister_active_turn(turn) self._reset_turn_context(turn) @@ -1206,18 +1205,19 @@ class RelaySessionCoordinator: turn (closing then breaks LIFO). The last live turn consumes it here after its own scope popped and it left the active-turn table. """ - try: - host = lease.live_runtime() - if host is None: - return - with lease.session.lock: - pending = lease.session.close_pending and not lease.session.closing - if pending and not self.has_active_turn( - profile_key=lease.profile_key, session_id=lease.session_id - ): - host.close_session({"session_id": lease.session_id}) - except Exception: # noqa: BLE001 - telemetry must never block end_turn - logger.warning("Hermes Relay deferred session close failed", exc_info=True) + # Telemetry must never block end_turn. + _warn_on_error("deferred session close", self._consume_deferred_close_unguarded, lease) + + def _consume_deferred_close_unguarded(self, lease: ConversationLease) -> None: + host = lease.live_runtime() + if host is None: + return + with lease.session.lock: + pending = lease.session.close_pending and not lease.session.closing + if pending and not self.has_active_turn( + profile_key=lease.profile_key, session_id=lease.session_id + ): + host.close_session({"session_id": lease.session_id}) def notify_session_compacted( self, *, profile_key: str, session_id: str, old_session_id: str = "" @@ -1230,29 +1230,35 @@ class RelaySessionCoordinator: the OLD session now or its scope stays an unexported orphan. Unknown sessions and disabled config are silent no-ops. """ - try: - if not _segments_config()["on_compaction"]: - return - host = self.registry.for_profile(profile_key) - if not isinstance(host, RelayRuntime): - return - if old_session_id and old_session_id != session_id: - # A LIVE turn on the old session: closing now would pop under it (LIFO). - with host._sessions_lock: - old_session = host._sessions.get(old_session_id) - if old_session is not None and self.has_active_turn( - profile_key=profile_key, session_id=old_session_id - ): - _flag_open_session(old_session, "close_pending") - else: - host.close_session({"session_id": old_session_id}) - return + # Telemetry must never block compaction. + _warn_on_error( + "compaction notification", self._notify_session_compacted_unguarded, + profile_key, session_id, old_session_id, + ) + + def _notify_session_compacted_unguarded( + self, profile_key: str, session_id: str, old_session_id: str + ) -> None: + if not _segments_config()["on_compaction"]: + return + host = self.registry.for_profile(profile_key) + if not isinstance(host, RelayRuntime): + return + if old_session_id and old_session_id != session_id: + # A LIVE turn on the old session: closing now would pop under it (LIFO). with host._sessions_lock: - session = host._sessions.get(session_id) - if session is not None: - _flag_open_session(session, "rotate_pending") - except Exception: # noqa: BLE001 - telemetry must never block compaction - logger.warning("Hermes Relay compaction notification failed", exc_info=True) + old_session = host._sessions.get(old_session_id) + if old_session is not None and self.has_active_turn( + profile_key=profile_key, session_id=old_session_id + ): + _flag_open_session(old_session, "close_pending") + else: + host.close_session({"session_id": old_session_id}) + return + with host._sessions_lock: + session = host._sessions.get(session_id) + if session is not None: + _flag_open_session(session, "rotate_pending") def has_active_turn(self, *, profile_key: str, session_id: str) -> bool: """Return whether a turn is still running for one profile/session."""