From be3a659b14de92852567bdae164800a3feec133c Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:53:45 -0700 Subject: [PATCH] refactor(agent/relay): compact relay/transport docstrings and comments (keep every invariant); reflow boolean spans; drop no-op helpers --- agent/relay_llm.py | 79 ++++++-------- agent/relay_runtime.py | 195 ++++++++++++++-------------------- agent/relay_tools.py | 10 +- agent/transports/__init__.py | 7 +- agent/transports/anthropic.py | 33 ++---- agent/transports/base.py | 7 +- 6 files changed, 130 insertions(+), 201 deletions(-) diff --git a/agent/relay_llm.py b/agent/relay_llm.py index 38cb33ec8b..84552a0af8 100644 --- a/agent/relay_llm.py +++ b/agent/relay_llm.py @@ -94,9 +94,8 @@ class _ManagedAttempt: def run_callback(self, callback: Callable[..., Any], *args: Any) -> Any: """Run a Hermes callback in a fresh copy of the captured context. - Relay can invoke callbacks while another one still owns the captured - Context, hence the copy. Nested relay calls inside a managed provider - callback must run unmanaged — see relay_runtime.managed_callback_guard. + Relay can invoke callbacks while another still owns the captured Context (hence the + copy); nested relay calls run unmanaged — see relay_runtime.managed_callback_guard. """ def guarded() -> Any: with relay_runtime.managed_callback_guard(): @@ -123,8 +122,7 @@ class _ManagedAttempt: final_request = self.provider_request(next_request) async def call_provider() -> Any: - # Nested relay calls inside a managed provider callback must - # run unmanaged — see relay_runtime.managed_callback_guard. + # Nested relay calls run unmanaged — see relay_runtime.managed_callback_guard. with relay_runtime.managed_callback_guard(): return await callback(final_request) @@ -153,8 +151,7 @@ class _ManagedAttempt: ): raise callback_error if ( - not isinstance(exc, Exception) - or callback_error is not None + not isinstance(exc, Exception) or callback_error is not None or "value" not in self.raw_response ): raise @@ -269,22 +266,19 @@ def stream_current( ) -> Any: """Run a provider stream under the inherited Hermes turn when present. - With ``completed_response_predicate`` set, a factory that ignores - ``stream=True`` and returns a complete response is unwrapped and returned - directly (pre-Relay ``call_llm(stream=True)`` behavior); otherwise it - would stay trapped as ``final_response`` on the inner ManagedLlmStream. - Detecting that shape starts the lazy managed pipeline: a genuine first - chunk is buffered, but provider latency and pre-first-yield errors may - surface before this function returns. + With ``completed_response_predicate`` set, a factory that ignores ``stream=True`` and + returns a complete response is unwrapped and returned directly (pre-Relay behavior) + instead of staying trapped as ``final_response``. Detecting that primes the lazy + pipeline: a genuine first chunk is buffered, but provider latency and pre-first-yield + errors may surface before this returns. """ session_id = _current_session_id() if session_id is None: return stream_factory(request) if _has_running_event_loop(): - # Managed provider callbacks run on the Relay session's event loop; a - # nested ManagedLlmStream would be iterated synchronously on that same - # loop thread, which asyncio forbids. The outer managed stream already - # tracks the enclosing attempt and traps a completed response itself. + # We are on the Relay session's loop (inside a managed callback): a nested + # ManagedLlmStream would be iterated synchronously on that loop, which asyncio + # forbids. The outer managed stream already tracks this attempt. return stream_factory(request) managed = stream( request, stream_factory, session_id=session_id, name=name, model_name=model_name, @@ -292,12 +286,11 @@ def stream_current( completed_response_predicate=completed_response_predicate, ) if completed_response_predicate is not None: - # Relay may defer the provider callback until the first pull; prime - # once so a completed response surfaces. A real first chunk is buffered. + # Relay may defer the provider callback until the first pull; prime once so a + # completed response surfaces (a real first chunk is buffered). managed._prime_completed_response() - completed = getattr(managed, "final_response", None) - if completed is not None: - return completed + if managed.final_response is not None: + return managed.final_response return managed @@ -429,9 +422,8 @@ class ManagedLlmStream(Iterator[Any]): run_callback(self._on_chunk, _jsonable(chunk)) def relay_finalizer() -> Any: - # Relay can invoke the finalizer while unwinding a provider-stream - # failure; keep that original error instead of a secondary - # "missing terminal response" error. + # Relay may call this while unwinding a provider-stream failure; keep the + # original error instead of a secondary "missing terminal response". if self._callback_error is not None: return None try: @@ -793,8 +785,8 @@ def _complete_logical( output=output, metadata=relay_runtime.runtime_metadata(lease.host.runtime_id), ) except Exception: - # The provider result is authoritative. Retain the handle so turn - # finalization can retry cleanup without changing that result. + # Provider result is authoritative; retain the handle so turn finalization + # can retry cleanup. logger.warning("Hermes Relay logical LLM finalization failed", exc_info=True) return with turn.logical_llm_lock: @@ -840,9 +832,9 @@ def _provider_request( if codec_baseline_body is not None and not _json_equal(content, relay_request_body): baseline = codec_baseline_body intercepted = _provider_request_body(content, metadata) - # Typed codecs may not represent provider-specific fields. Overlay only - # values that changed from the codec-facing baseline so unrelated - # intercepts cannot delete or normalize unknown provider arguments. + # Typed codecs may not represent provider-specific fields: overlay only values + # that changed from the codec-facing baseline so unrelated intercepts cannot + # delete or normalize unknown provider arguments. for key in baseline.keys() | intercepted.keys(): if key not in intercepted: final.pop(key, None) @@ -863,9 +855,8 @@ def _provider_request( def _codex_codec_tools(body: dict[str, Any]) -> None: - # The Responses SDK accepts ``tools=None`` as "no tools" while Relay's - # typed codec expects an array or an absent field; normalize only the - # codec-facing copy (the original request is restored when unchanged). + # The Responses SDK accepts ``tools=None`` as "no tools" while Relay's typed codec + # wants an array or an absent field; only the codec-facing copy is normalized. if body.get("tools") is None: body.pop("tools", None) elif isinstance(body.get("tools"), list): @@ -885,8 +876,7 @@ def _chat_codec_tools(body: dict[str, Any]) -> None: if isinstance(tools, list): body["tools"] = [ {"type": "function", **tool} - if isinstance(tool, dict) and "function" in tool and "type" not in tool - else tool + if isinstance(tool, dict) and "function" in tool and "type" not in tool else tool for tool in tools ] @@ -899,8 +889,7 @@ _CODEC_TOOL_NORMALIZERS = { def _relay_request_body(request: dict[str, Any], metadata: dict[str, Any] | None) -> dict[str, Any]: body = _jsonable_dict(request) - # ``timeout`` configures the provider SDK client, not a wire protocol: - # keep it on the original callback request, never on Relay intercepts. + # ``timeout`` configures the SDK client, not the wire: never expose it to intercepts. body.pop("timeout", None) normalize = _CODEC_TOOL_NORMALIZERS.get(_api_mode(metadata)) if normalize is not None: @@ -924,10 +913,8 @@ def _restore_provider_message_extensions( original_message, final_message, baseline_message, intercepted_message = messages for key in _PROVIDER_MESSAGE_EXTENSION_KEYS: if ( - key in original_message - and key not in baseline_message - and key not in intercepted_message - and key not in final_message + key in original_message and key not in baseline_message + and key not in intercepted_message and key not in final_message ): final_message[key] = original_message[key] @@ -969,8 +956,7 @@ def _provider_request_body( body["tools"] = [ {"type": "function", **dict(tool["function"])} if isinstance(tool, dict) - and tool.get("type") == "function" - and isinstance(tool.get("function"), dict) + and tool.get("type") == "function" and isinstance(tool.get("function"), dict) else tool for tool in tools ] @@ -996,12 +982,11 @@ def _jsonable(value: Any) -> Any: model_dump = getattr(type(value), "model_dump", None) if callable(model_dump): try: - # warnings=False: pydantic warns on generic-union SDK stream events - # and that warning would leak to the user's terminal mid-response. + # warnings=False: pydantic's generic-union warning would leak to the terminal + # mid-response; TypeError = duck-typed model_dump without pydantic's signature. try: return _jsonable(value.model_dump(mode="json", warnings=False)) except TypeError: - # Duck-typed model_dump without pydantic's signature. return _jsonable(value.model_dump()) except Exception: pass diff --git a/agent/relay_runtime.py b/agent/relay_runtime.py index bf965bd87e..8afc910fc0 100644 --- a/agent/relay_runtime.py +++ b/agent/relay_runtime.py @@ -34,8 +34,8 @@ RUNTIME_INSTANCE_KEY = "hermes.relay.runtime_instance" RELAY_PLUGINS_EXECUTION_CONSUMER = "hermes.nemo_relay.plugins" _PROFILE_KEY_CACHE: dict[str, str] = {} -# Bound for native scope lifecycle ops (push/pop/flush) gating turn/session completion. -# Healthy ops take microseconds; a wedged pipeline costs one lost span, never a blocked agent. +# Bound for native scope ops gating turn/session completion: a wedged pipeline costs one +# lost span, never a blocked agent. _SCOPE_OP_TIMEOUT = 10.0 _SCOPE_OP_EXECUTOR: Any = None @@ -48,11 +48,10 @@ def runtime_metadata(runtime_id: str, **extra: Any) -> dict[str, Any]: def _scope_op_executor(): - """Shared daemon executor for bounded native scope operations. + """Shared daemon executor for bounded native scope ops. - Daemon workers (tools.daemon_pool) so a wedged call abandoned at timeout - cannot block interpreter exit. ``Future.result(timeout=...)`` bounds callers - even when every worker is wedged, so exhaustion degrades to fast timeouts. + Daemon workers so a wedged call abandoned at timeout cannot block interpreter exit; + ``Future.result(timeout=...)`` still bounds callers when every worker is wedged. """ global _SCOPE_OP_EXECUTOR if _SCOPE_OP_EXECUTOR is None: @@ -71,8 +70,8 @@ def _run_on_daemon_thread( ) -> Any: """Run ``fn`` on a fresh daemon thread; re-raise its error or return its result. - With ``timeout`` the join is bounded and a still-running worker is abandoned - with ``TimeoutError`` — a daemon thread cannot block interpreter exit. + With ``timeout`` a still-running worker is abandoned with ``TimeoutError`` (daemon: + cannot block interpreter exit). """ result: list[Any] = [] error: list[BaseException] = [] @@ -98,8 +97,7 @@ def pop_relay_scope( ) -> Any: """Pop a Relay scope, forwarding only the kwargs the live binding accepts. - ``scope.pop`` gained ``metadata`` in nemo-relay 0.4+; older wheels raise - TypeError on it, which would wedge turn/session close. + ``scope.pop`` gained ``metadata`` in nemo-relay 0.4+; older wheels raise TypeError. """ pop = relay.scope.pop kwargs = { @@ -118,8 +116,8 @@ def pop_relay_scope( def _current_top(relay: Any) -> Any: """Return the current top-of-stack scope handle, or None.""" - # Prefer ``scope.get_handle()``: ``get_scope_stack()`` may return a native - # ScopeStack object that ``scope.pop`` rejects, so never treat it as a handle. + # Prefer scope.get_handle(): get_scope_stack() may return a native ScopeStack + # object that scope.pop rejects, so never treat it as a handle. get_handle = getattr(getattr(relay, "scope", None), "get_handle", None) if callable(get_handle): try: @@ -127,11 +125,8 @@ def _current_top(relay: Any) -> Any: except Exception: pass top = relay.get_scope_stack() - # Some builds return the live stack (list), others the top handle directly - # (including tuple handles from test fakes): only unwrap real lists. - if isinstance(top, list): - return top[-1] if top else None - return top + # Some builds return the live stack (list), others the top handle: only unwrap real lists. + return (top[-1] if top else None) if isinstance(top, list) else top def _same_handle(a: Any, b: Any) -> bool: @@ -166,20 +161,18 @@ class RelaySession: closing: bool = False handle: Any = None context: contextvars.Context | None = None - # Session-span segmentation (continuous sessions): rotation closes the - # current session scope and pushes segment N+1 at a turn boundary. + # Session-span segmentation: rotation closes the current session scope and pushes + # segment N+1 at a turn boundary (the only LIFO-safe point). segment: int = 0 # index of the CURRENT session scope (0 = first) segment_turns: int = 0 # turns completed within the current segment rotate_pending: bool = False # set by compaction; consumed at next begin_turn - # A rotating compaction landed while a turn was live on THIS session; closing - # now would pop the session scope under the live turn (LIFO violation), so - # end_turn consumes this and closes the session. + # Rotating compaction landed while a turn was live here; closing now would pop the + # session scope under the live turn, so end_turn consumes this instead. close_pending: bool = False -# Segmentation config (gateway.telemetry.session_segments), cached at first read. -# Both defaults OFF => rotation never fires and the scope lifecycle is identical -# to the pre-segmentation behavior. +# gateway.telemetry.session_segments, cached at first read. Both defaults OFF => +# rotation never fires and the scope lifecycle is unchanged. _SEGMENTS_CONFIG: dict[str, Any] | None = None _SEGMENTS_CONFIG_LOCK = threading.Lock() @@ -329,8 +322,7 @@ class _ProcessRelayPluginConfiguration: except Exception as exc: raise RuntimeError("Hermes Relay dynamic plugin activation failed") from exc if self._activation is None: - # Hermes only enters Relay's initialization path after an - # explicit opt-in; Relay owns any subsequent ambient layering. + # Reached only after explicit opt-in; Relay owns any ambient layering. _resolve_plugin_awaitable(relay.plugin.initialize(plugin_config)) return True @@ -443,25 +435,18 @@ class RelayRuntime: with self._execution_consumers_lock: return bool(self._execution_consumers) - def _subagent_parent_handle(self, session: RelaySession) -> Any: - with self._sessions_lock: - return self._subagent_parent_handles.get(session.session_id) - def _push_session_scope( self, context: contextvars.Context, *, exit_fallback: bool = False, **push_kwargs: Any ) -> Any: """Push a SESSION_SCOPE Agent scope inside ``context``, bounded by ``_SCOPE_OP_TIMEOUT``. - ``exit_fallback``: at interpreter shutdown the executor refuses new futures - (RuntimeError); push synchronously instead, since no agent turn waits at exit. + ``exit_fallback``: at interpreter shutdown the executor refuses new futures; push + synchronously instead (no agent turn waits at exit). """ args = (self.relay.scope.push, SESSION_SCOPE, self.relay.ScopeType.Agent) try: - return ( - _scope_op_executor() - .submit(context.run, *args, input={}, **push_kwargs) - .result(timeout=_SCOPE_OP_TIMEOUT) - ) + future = _scope_op_executor().submit(context.run, *args, input={}, **push_kwargs) + return future.result(timeout=_SCOPE_OP_TIMEOUT) except RuntimeError: if not exit_fallback: raise @@ -473,12 +458,13 @@ class RelayRuntime: ) -> None: """Push a fresh session scope for ``session`` and record its handle + context. - Subagent sessions parent under their spawning turn/session handle; - ``resolve_parent`` creates the parent session when its handle is unknown. + Subagents parent under their spawning turn/session handle; ``resolve_parent`` + creates the parent session when its handle is unknown. """ parent_handle = None if session.parent_session_id: - parent_handle = self._subagent_parent_handle(session) + with self._sessions_lock: + parent_handle = self._subagent_parent_handles.get(session.session_id) if parent_handle is None and resolve_parent: parent = self.ensure_session({"session_id": session.parent_session_id}) if parent is not None: @@ -524,18 +510,15 @@ class RelayRuntime: def rotate_session_scope(self, session: RelaySession, *, reason: str) -> None: """Close the current session scope and open the next segment. - Called ONLY at a turn boundary (before the turn scope pushes): the scope - stack is LIFO and rotating under a live child would close a parent out - of order. Both native calls are bounded by ``_SCOPE_OP_TIMEOUT``, and - segment bookkeeping advances even when a native call fails so a degraded - rotation cannot retry on every turn. + Called ONLY at a turn boundary: the stack is LIFO and rotating under a live child + would close a parent out of order. Bookkeeping advances even when a native call + fails so a degraded rotation cannot retry on every turn. """ with session.lock: if session.closing or session.handle is None: return old_handle = session.handle - # Advance bookkeeping FIRST: a failed native call must not leave - # rotate_pending set (tight rotation loop on every turn). + # Bookkeeping FIRST: a failed native call must not leave rotate_pending set. session.segment += 1 session.segment_turns = 0 session.rotate_pending = False @@ -583,11 +566,8 @@ class RelayRuntime: parent_handle = None if parent is None else parent.handle turn = active_turn(parent_session_id) if ( - turn is not None - and not turn.closed - and turn.handle is not None - and turn.lease.host is self - and turn.lease.session is not None + turn is not None and not turn.closed and turn.handle is not None + and turn.lease.host is self and turn.lease.session is not None and turn.lease.session.session_id == parent_session_id ): parent_handle = turn.handle @@ -631,8 +611,7 @@ class RelayRuntime: if session.context is None or session.handle is None: raise RuntimeError("Hermes Relay session context is unavailable") relay_context = session.context.copy() - # A copy permits a helper called by an existing Relay callback to - # re-enter the same logical session without re-entering Context. + # A copy lets a helper inside a Relay callback re-enter the session's Context. context = contextvars.copy_context() for variable, value in relay_context.items(): context.run(variable.set, value) @@ -644,12 +623,9 @@ class RelayRuntime: ) -> Any: """Run a Relay operation against a session's isolated scope stack. - ``timeout`` (seconds) bounds the native call on a shared daemon - executor; ``TimeoutError`` propagates on breach. ``None`` keeps the - synchronous behavior. Scope lifecycle ops that gate turn/session - completion pass ``_SCOPE_OP_TIMEOUT``: the native ``scope.pop`` is - unbounded, and a wedged pipeline must cost at most one span, never the - agent (the abandoned daemon worker cannot block process exit). + ``timeout`` bounds the native call on the daemon executor (``TimeoutError`` on + breach); ``None`` runs synchronously. Lifecycle ops gating turn/session completion + pass ``_SCOPE_OP_TIMEOUT``: a wedged pipeline must cost one span, never the agent. """ self._begin_operation() try: @@ -675,9 +651,8 @@ class RelayRuntime: try: future = _scope_op_executor().submit(context.run, invoke) except RuntimeError: - # Interpreter shutdown: the executor refuses new futures, but the - # atexit close path must still flush — and still bounded, since a - # wedged native call must not block process exit. + # Interpreter shutdown: the executor refuses new futures, but the atexit close + # path must still flush — still bounded so a wedged call cannot block exit. return _run_on_daemon_thread( lambda: context.run(invoke), name="relay-scope-op-exit", @@ -770,8 +745,8 @@ class RelayRuntime: ) -> BaseException | None: """Pop ``handle``; if that fails, drain orphans above it and retry once. - Returns the retry's error (None on success). Must run inside ONE - ``run_in_session`` callback so ContextVar stack views stay consistent. + Returns the retry's error (None on success). Must run inside ONE ``run_in_session`` + callback so ContextVar stack views stay consistent. """ try: pop_relay_scope(self.relay, handle, output=output, metadata=metadata) @@ -785,8 +760,7 @@ class RelayRuntime: break # Never pop the session root while draining for a nested handle. if ( - session_root is not None - and _same_handle(top, session_root) + session_root is not None and _same_handle(top, session_root) and handle is not session_root ): break @@ -816,10 +790,9 @@ class RelayRuntime: ) -> str | None: """Pop ``handle``, draining orphaned children in the same session context. - Relay scopes are strict LIFO; empty-stream retries + interrupt can - abandon a physical LLM scope above TURN/SESSION. The whole drain+close - is bounded like the direct pops it replaced: a wedged native pipeline - must never block turn/session completion. Returns a failure string. + Relay scopes are strict LIFO; empty-stream retries + interrupt can abandon a + physical LLM scope above TURN/SESSION. Drain+close is bounded so a wedged pipeline + never blocks turn/session completion. Returns a failure string or None. """ if handle is None: return None @@ -865,9 +838,8 @@ class RelayRuntime: session, session.handle, output={}, allow_closing=True, failure_label="session scope close failed", operation_already_held=True, ) - # Subscriber flushing is process-wide and may wait for publications - # owned by other sessions; final plugin teardown flushes once after all - # tracked operations drain. Flushing here can deadlock an asyncio loop. + # No subscriber flush here: it is process-wide, may wait on other sessions' + # publications and can deadlock an asyncio loop; final plugin teardown flushes once. with self._sessions_lock: if self._sessions.get(session_id) is session: self._sessions.pop(session_id, None) @@ -941,12 +913,11 @@ class NoopRelayRuntime: def apply_tool_request_intercepts( self, *, session_id: str, tool_name: str, args: dict[str, Any] ) -> dict[str, Any]: - del session_id, tool_name return args @staticmethod def retain_managed_execution(consumer: str) -> None: - del consumer + pass release_managed_execution = retain_managed_execution @@ -1038,13 +1009,12 @@ _CURRENT_TURN: contextvars.ContextVar[RelayTurnContext | None] = contextvars.Con "hermes_relay_turn", default=None ) -# Depth of managed Relay callbacks on the current logical call path (>0 while the -# native pipeline is mid-dispatch of a Hermes tool/LLM callback). Nested managed -# execution there is structurally broken: the native pipeline binds its Futures to -# the OUTER call's event loop, which is blocked inside the synchronous callback -# ("attached to a different loop" at best, deadlock or "Event loop is closed" at -# worst), so resolve_execution_context() bypasses Relay while set. A ContextVar so -# the marker follows contextvars.copy_context() into worker threads / per-thread loops. +# Depth of managed Relay callbacks on the current call path (>0 while the native pipeline +# is mid-dispatch of a Hermes tool/LLM callback). Nested managed execution there is +# structurally broken: the pipeline binds its Futures to the OUTER call's event loop, which +# is blocked inside the synchronous callback (wrong loop / deadlock / "Event loop is +# closed"), so resolve_execution_context() bypasses Relay while set. A ContextVar so the +# marker follows copy_context() into worker threads / per-thread loops. _MANAGED_CALLBACK_DEPTH: contextvars.ContextVar[int] = contextvars.ContextVar( "hermes_relay_managed_callback_depth", default=0 ) @@ -1053,9 +1023,8 @@ _MANAGED_CALLBACK_DEPTH: contextvars.ContextVar[int] = contextvars.ContextVar( class managed_callback_guard: """Mark the current context as inside a managed Relay callback. - Wrap the ``invoke()`` callbacks handed to the native pipeline; everything - they transitively call (including work forwarded via copy_context()) sees - the marker and runs unmanaged. + Wrap the ``invoke()`` callbacks handed to the native pipeline; everything they + transitively call (incl. work forwarded via copy_context()) runs unmanaged. """ def __enter__(self) -> "managed_callback_guard": @@ -1137,8 +1106,8 @@ class RelaySessionCoordinator: key = (lease.profile_key, lease.session_id) with self._active_turns_lock: if self._active_turns.get(key): - # A Relay session owns one physical scope stack; concurrent turns - # would create sibling scopes whose completion order is not LIFO. + # One physical scope stack per session; concurrent turns would create + # 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", @@ -1149,9 +1118,8 @@ class RelaySessionCoordinator: turn._active_registered = True host = lease.live_runtime() if turn.relay_enabled else None if host is not None: - # Segment rotation (pending compaction flag or max_turns cap) happens - # HERE — the only point with no live turn scope on the session's - # stack, so the session scope can close/reopen without breaking LIFO. + # 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: @@ -1204,9 +1172,8 @@ class RelaySessionCoordinator: except Exception: # noqa: BLE001 - accounting must never block pass try: - # Delegated agents own one turn: close their conversation - # while the active-turn guard is still held so a parent - # timeout fallback cannot race this terminal boundary. + # 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: @@ -1235,11 +1202,9 @@ class RelaySessionCoordinator: def _consume_deferred_close(self, lease: Any) -> None: """Close a session whose rotating-compaction close was deferred. - ``notify_session_compacted`` sets ``close_pending`` when the old session - still had a live turn (closing then would break LIFO). The turn that was - live consumes it here, after its own scope popped and it left the - active-turn table; if another turn is still live, that turn's end_turn - consumes it instead. + ``notify_session_compacted`` sets ``close_pending`` when the old session had a live + 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() @@ -1259,13 +1224,11 @@ class RelaySessionCoordinator: ) -> None: """React to a completed compaction, per compaction mode. - In-place compaction (``old_session_id`` empty or equal): flag the session - for rotation at its next turn boundary — never rotate immediately, since - a compaction can finish while a turn is live and rotating under it would - break LIFO; ``begin_turn`` consumes the flag. Rotating compaction (ids - differ): the next turn gets a fresh session under the new id, so close - the OLD session now or its scope stays an unexported orphan. Unknown - sessions and disabled config are silent no-ops. + In-place (``old_session_id`` empty/equal): flag rotation for the next turn boundary + — never rotate immediately, a turn may be live and rotating under it breaks LIFO. + Rotating (ids differ): the next turn gets a fresh session under the new id, so close + 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"]: @@ -1274,8 +1237,7 @@ class RelaySessionCoordinator: if not isinstance(host, RelayRuntime): return if old_session_id and old_session_id != session_id: - # If a turn is still LIVE on the old session, closing now would - # pop the session scope under it (LIFO) — defer to its end_turn. + # 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( @@ -1333,9 +1295,8 @@ class RelaySessionCoordinator: if failure is None: continue with turn.logical_llm_lock: - # Relay scopes are stack-owned: if the newest remaining handle - # cannot close even after orphan drain, older ones cannot close - # safely either — retain the unclosed prefix for diagnostics. + # Stack-owned scopes: if the newest handle cannot close even after orphan + # drain, older ones cannot either — retain the unclosed prefix. for pending_request_id, pending_handle in logical_calls[: index + 1]: turn.logical_llm_calls.setdefault(pending_request_id, pending_handle) logger.warning("Hermes Relay logical LLM finalization failed: %s", failure) @@ -1403,9 +1364,8 @@ def resolve_execution_context( ) -> tuple[RelayRuntime | None, RelaySession | None, Any]: """Resolve one active turn/session parent for managed Relay execution.""" if _MANAGED_CALLBACK_DEPTH.get() > 0: - # Inside a managed Relay callback: nested managed execution is impossible - # (see _MANAGED_CALLBACK_DEPTH). Run unmanaged; the outer scope still - # records the tool-level event for observability. + # Nested managed execution is impossible (see _MANAGED_CALLBACK_DEPTH); the + # outer scope still records the tool-level event. return None, None, None inherited_turn = current_turn() if inherited_turn is not None and (not inherited_turn.relay_enabled or inherited_turn.closed): @@ -1415,9 +1375,8 @@ def resolve_execution_context( if host is not None: session = turn.lease.session return host, session, turn.handle or session.handle - # Managed-execution consumers create and retain the profile host before - # reaching an out-of-turn adapter; never initialize Relay for the default - # no-consumer path. + # Consumers retain the profile host before reaching an out-of-turn adapter; never + # initialize Relay for the default no-consumer path. runtime = get_runtime(create=False) if runtime is None or not runtime.managed_execution_enabled(): return None, None, None diff --git a/agent/relay_tools.py b/agent/relay_tools.py index 99f464684d..d4e7f13a75 100644 --- a/agent/relay_tools.py +++ b/agent/relay_tools.py @@ -32,10 +32,9 @@ def execute( observed_args = next_args if isinstance(next_args, dict) else args def guarded(final_args: dict[str, Any]) -> Any: - # Everything the tool transitively calls (including auxiliary LLM - # calls it forwards to worker threads) must bypass managed Relay - # execution — the native pipeline's Futures bind to THIS loop, - # which is blocked until the tool returns (#77244). + # Everything the tool transitively calls (incl. auxiliary LLM calls on worker + # threads) must bypass managed Relay: the pipeline's Futures bind to THIS loop, + # which is blocked until the tool returns. with relay_runtime.managed_callback_guard(): return callback(final_args) @@ -87,8 +86,7 @@ def _jsonable(value: Any) -> Any: model_dump = getattr(value, "model_dump", None) if callable(model_dump): try: - # warnings=False: suppress pydantic's serializer UserWarnings on - # generic-union SDK models; they would leak to the CLI mid-turn. + # warnings=False: pydantic's generic-union warning would leak to the CLI mid-turn. try: return _jsonable(model_dump(mode="json", warnings=False)) except TypeError: diff --git a/agent/transports/__init__.py b/agent/transports/__init__.py index 23557be795..8164e2b319 100644 --- a/agent/transports/__init__.py +++ b/agent/transports/__init__.py @@ -4,6 +4,8 @@ result = transport.normalize_response(raw_response) """ +import importlib + from agent.transports.types import ( # noqa: F401 NormalizedResponse, ToolCall, @@ -28,8 +30,7 @@ def get_transport(api_mode: str): _discover_transports() cls = _REGISTRY.get(api_mode) if cls is None: - # A directly-imported transport module leaves the registry partially - # populated; discover on misses so import order can't hide a valid api_mode. + # A directly-imported transport leaves the registry partial; rediscover on misses. _discover_transports() cls = _REGISTRY.get(api_mode) return None if cls is None else cls() @@ -39,8 +40,6 @@ def _discover_transports() -> None: """Import all transport modules to trigger auto-registration.""" global _discovered _discovered = True - import importlib - for name in _TRANSPORT_MODULES: try: importlib.import_module(f"agent.transports.{name}") diff --git a/agent/transports/anthropic.py b/agent/transports/anthropic.py index e3814f8ab3..f234115240 100644 --- a/agent/transports/anthropic.py +++ b/agent/transports/anthropic.py @@ -1,8 +1,4 @@ -"""Anthropic Messages API transport. - -Delegates format conversion to agent/anthropic_adapter.py; owns normalization, -not client lifecycle. -""" +"""Anthropic Messages API transport: conversion via agent/anthropic_adapter.py, normalization here.""" from typing import Any, Dict, List, Optional @@ -15,11 +11,9 @@ _MCP_PREFIX = "mcp__" def _unprefix_oauth_tool_name(name: str) -> str: """Reverse the OAuth-wire ``mcp__`` prefix back to the registered tool name. - Two originals map onto one wire name (``mcp__read_file`` <- ``read_file``; - ``mcp__linear_get_issue`` <- ``mcp_linear_get_issue``), so resolve by registry - lookup, never rewriting a name that already resolves natively (GH-25255). - OAuth wire aliases (e.g. chat_history_lookup -> session_search) are checked - LAST so a real tool registered under the wire name still wins. + Two originals map onto one wire name (``read_file`` / ``mcp_linear_get_issue``), so + resolve by registry lookup, never rewriting a name that already resolves natively. + OAuth wire aliases are checked LAST so a real tool under the wire name still wins. """ from agent.anthropic_adapter import _OAUTH_TOOL_NAME_REVERSE_ALIASES from tools.registry import registry as _tool_registry @@ -80,18 +74,15 @@ class AnthropicTransport(ProviderTransport): strip_tool_prefix = kwargs.get("strip_tool_prefix", False) text_parts, reasoning_parts, reasoning_details, tool_calls = [], [], [], [] - # Anthropic signs each thinking block against the blocks that PRECEDE it. - # When thinking interleaves with tool_use, the parallel reasoning_details + - # tool_calls lists lose that ordering and replay -> HTTP 400 "thinking ... - # blocks cannot be modified". Keep the exact sequence for the adapter. + # Anthropic signs each thinking block against the blocks PRECEDING it; when thinking + # interleaves with tool_use the parallel lists lose that order and replay -> HTTP 400. ordered_blocks = [] for block in response.content: block_dict = _to_plain_data(block) clean_block = None if isinstance(block_dict, dict): - # Sanitize at capture so output-only SDK fields never persist to - # state.db and leak back as request input on replay (HTTP 400). + # Sanitize at capture so output-only SDK fields never persist and replay (400). clean_block = _sanitize_replay_block(block_dict) if clean_block is not None: ordered_blocks.append(clean_block) @@ -100,7 +91,7 @@ class AnthropicTransport(ProviderTransport): elif block.type in ("thinking", "redacted_thinking"): if block.type == "thinking": reasoning_parts.append(block.thinking) - # Prefer the sanitized block (replayed on the non-ordered path); raw only if sanitize dropped it. + # Sanitized block preferred; raw only if sanitize dropped it. if isinstance(clean_block, dict): reasoning_details.append(clean_block) elif isinstance(block_dict, dict): @@ -114,8 +105,7 @@ class AnthropicTransport(ProviderTransport): provider_data = {} if reasoning_details: provider_data["reasoning_details"] = reasoning_details - # Carry the ordered channel only for the one shape the parallel lists - # reconstruct wrongly: signed thinking interleaved with tool_use. + # Ordered channel only for the shape the parallel lists reconstruct wrongly. _has_signed_thinking = any( isinstance(b, dict) and b.get("type") in ("thinking", "redacted_thinking") and (b.get("signature") or b.get("data")) for b in ordered_blocks @@ -131,9 +121,8 @@ class AnthropicTransport(ProviderTransport): ) def validate_response(self, response: Any) -> bool: - """Structural check. An empty content list is legitimate for ``end_turn`` (nothing to add - after a tool turn) and ``refusal`` (Claude 4.5+ declines with empty content); treating - either as invalid would retry a completed/deterministic response forever.""" + """Structural check; empty content is legitimate for ``end_turn``/``refusal`` (retrying + either would loop forever).""" content_blocks = getattr(response, "content", None) if response is not None else None if not isinstance(content_blocks, list): return False diff --git a/agent/transports/base.py b/agent/transports/base.py index e56b97ed03..07b8a88ac3 100644 --- a/agent/transports/base.py +++ b/agent/transports/base.py @@ -1,9 +1,8 @@ """Abstract base for provider transports. -A transport owns the data path for one api_mode: - convert_messages -> convert_tools -> build_kwargs -> normalize_response -It does NOT own client construction, streaming, credential refresh, prompt -caching, interrupt handling, or retry logic — those stay on AIAgent. +A transport owns one api_mode's data path (convert_messages -> convert_tools -> build_kwargs +-> normalize_response), NOT client construction, streaming, credentials, caching, interrupts +or retries — those stay on AIAgent. """ from abc import ABC, abstractmethod