diff --git a/agent/relay_llm.py b/agent/relay_llm.py index cb90d9c564..5d89941cd4 100644 --- a/agent/relay_llm.py +++ b/agent/relay_llm.py @@ -142,16 +142,12 @@ class _ManagedAttempt: """Re-raise the provider's own error, or recover a completed provider result. Must be called from the ``except`` handling ``exc`` (bare ``raise``).""" callback_error = self.raw_response.get("error") - if ( - callback_error is not None - and relay_runtime._is_relay_wrapped_callback_error(exc, callback_error) - ): + if (callback_error is not None and relay_runtime._is_relay_wrapped_callback_error(exc, callback_error)): raise callback_error if (not isinstance(exc, Exception) or callback_error is not None or "value" not in self.raw_response): raise logger.warning( - "NeMo Relay LLM post-processing failed after provider success; " - "returning the provider response", + "NeMo Relay LLM post-processing failed after provider success; returning the provider response", exc_info=True, ) if not defer_logical_completion: @@ -192,9 +188,7 @@ async def execute_async( if attempt is None: return await callback(request) try: - managed = await attempt.run_managed( - attempt.runtime.relay.llm.execute, partial(attempt.invoke_async, callback) - ) + managed = await attempt.run_managed(attempt.runtime.relay.llm.execute, partial(attempt.invoke_async, callback)) except BaseException as exc: return attempt.resolve_failure(exc, defer_logical_completion) return attempt.result(managed, defer_logical_completion) @@ -440,13 +434,10 @@ class ManagedLlmStream(Iterator[Any]): def _recoverable_relay_failure(self, exc: BaseException) -> bool: """Relay post-processing failed after the provider already succeeded.""" - recoverable = ( - isinstance(exc, Exception) and self._provider_completed and self._callback_error is None - ) + recoverable = (isinstance(exc, Exception) and self._provider_completed and self._callback_error is None) if recoverable: logger.warning( - "NeMo Relay stream post-processing failed after provider success; " - "preserving the provider result", + "NeMo Relay stream post-processing failed after provider success; preserving the provider result", exc_info=True, ) return recoverable @@ -491,10 +482,7 @@ class ManagedLlmStream(Iterator[Any]): raise StopIteration from None except BaseException as exc: callback_error = self._callback_error - if ( - callback_error is not None - and relay_runtime._is_relay_wrapped_callback_error(exc, callback_error) - ): + if (callback_error is not None and relay_runtime._is_relay_wrapped_callback_error(exc, callback_error)): self._close(logical_outcome="failed") raise callback_error if self._recoverable_relay_failure(exc): @@ -595,9 +583,7 @@ class ManagedLlmStream(Iterator[Any]): stream = ManagedLlmStream -_ANTHROPIC_APPEND_DELTAS = { - "text_delta": "text", "thinking_delta": "thinking", "signature_delta": "signature" -} +_ANTHROPIC_APPEND_DELTAS = {"text_delta": "text", "thinking_delta": "thinking", "signature_delta": "signature"} class AnthropicStreamAccumulator: @@ -637,9 +623,7 @@ class AnthropicStreamAccumulator: if field is not None: block[field] = str(block.get(field) or "") + str(delta.get(field) or "") elif delta_type == "input_json_delta": - block["_partial_json"] = str(block.pop("_partial_json", "")) + str( - delta.get("partial_json") or "" - ) + block["_partial_json"] = str(block.pop("_partial_json", "")) + str(delta.get("partial_json") or "") elif delta_type == "citations_delta" and "citation" in delta: block.setdefault("citations", []).append(delta["citation"]) @@ -705,9 +689,8 @@ def _logical_parent( def _complete_logical( - logical: _LogicalCall | None, *, outcome: str, model_name: str | None = None, - provider_name: str | None = None, response_model_name: str | None = None, - operation_lease: relay_runtime.RelayOperationLease | None = None, + logical: _LogicalCall | None, *, outcome: str, model_name: str | None = None, provider_name: str | None = None, + response_model_name: str | None = None, operation_lease: relay_runtime.RelayOperationLease | None = None, ) -> None: if logical is None: return diff --git a/agent/relay_runtime.py b/agent/relay_runtime.py index 48445ea989..90969928cb 100644 --- a/agent/relay_runtime.py +++ b/agent/relay_runtime.py @@ -55,9 +55,7 @@ def _scope_op_executor(): with _SCOPE_OP_EXECUTOR_LOCK: if _SCOPE_OP_EXECUTOR is None: from tools.daemon_pool import DaemonThreadPoolExecutor - _SCOPE_OP_EXECUTOR = DaemonThreadPoolExecutor( - max_workers=8, thread_name_prefix="relay-scope-op" - ) + _SCOPE_OP_EXECUTOR = DaemonThreadPoolExecutor(max_workers=8, thread_name_prefix="relay-scope-op") return _SCOPE_OP_EXECUTOR @@ -85,9 +83,7 @@ def _run_on_daemon_thread( return outcome.get("result") -def pop_relay_scope( - relay: Any, handle: Any, *, output: Any = None, metadata: Any = None, timestamp: Any = None -) -> Any: +def pop_relay_scope(relay: Any, handle: Any, *, output: Any = None, metadata: Any = None, timestamp: Any = None) -> 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.""" pop = relay.scope.pop @@ -198,9 +194,7 @@ class RelayOperationLease: self._lock = threading.Lock() self._runtime: RelayRuntime | None = runtime - def run_in_session( - self, session: RelaySession, callback: Callable[..., Any], *args: Any, **kwargs: Any - ) -> Any: + def run_in_session(self, session: RelaySession, callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: """Run cleanup while this lease still owns the runtime lifetime.""" with self._lock: runtime = self._runtime @@ -298,9 +292,7 @@ class _ProcessRelayPluginConfiguration: relay.plugin.initialize_with_dynamic_plugins(plugin_config, dynamic_plugins) ) if activation is None: - raise RuntimeError( - "NeMo Relay dynamic plugin initialization returned no activation handle" - ) + raise RuntimeError("NeMo Relay dynamic plugin initialization returned no activation handle") self._activation = activation except Exception as exc: raise RuntimeError("Hermes Relay dynamic plugin activation failed") from exc @@ -309,9 +301,7 @@ class _ProcessRelayPluginConfiguration: _resolve_plugin_awaitable(relay.plugin.initialize(plugin_config)) return True - def _remember( - self, owner_id: int, state: _RelayPluginConfigurationState - ) -> _RelayPluginConfigurationState: + def _remember(self, owner_id: int, state: _RelayPluginConfigurationState) -> _RelayPluginConfigurationState: """Retain one process decision for all concurrently hosted profiles.""" self._owners.add(owner_id) self._state = state @@ -430,8 +420,7 @@ class RelayRuntime: return context.run(*args, input={}, **push_kwargs) def _open_session_scope( - self, session: RelaySession, scope_metadata: dict[str, Any], *, resolve_parent: bool, - **push_kwargs: Any, + self, session: RelaySession, scope_metadata: dict[str, Any], *, resolve_parent: bool, **push_kwargs: Any, ) -> None: """Push a fresh session scope for ``session`` and record its handle + context. Subagents parent under their spawning turn/session handle; ``resolve_parent`` @@ -446,9 +435,7 @@ class RelayRuntime: parent_handle = parent.handle scope_metadata["nemo_relay_scope_role"] = "subagent" context = contextvars.Context() - session.handle = self._push_session_scope( - context, handle=parent_handle, metadata=scope_metadata, **push_kwargs - ) + session.handle = self._push_session_scope(context, handle=parent_handle, metadata=scope_metadata, **push_kwargs) session.context = context def ensure_session( @@ -495,26 +482,22 @@ class RelayRuntime: session.rotate_pending = False try: self.run_in_session( - session, self.relay.scope.pop, old_handle, - output={"hermes.session.segment_reason": reason}, + session, self.relay.scope.pop, old_handle, output={"hermes.session.segment_reason": reason}, metadata=runtime_metadata(self.runtime_id), timeout=_SCOPE_OP_TIMEOUT, ) except Exception: logger.warning( - "Hermes Relay segment close failed (session=%s segment=%d); " - "abandoning the old segment span", + "Hermes Relay segment close failed (session=%s segment=%d); abandoning the old segment span", session.session_id, session.segment - 1, exc_info=True, ) scope_metadata = runtime_metadata( - self.runtime_id, - **{"hermes.session.segment": session.segment, "hermes.session.segment_reason": reason}, + self.runtime_id, **{"hermes.session.segment": session.segment, "hermes.session.segment_reason": reason}, ) try: self._open_session_scope(session, scope_metadata, resolve_parent=False) except Exception: logger.warning( - "Hermes Relay segment open failed (session=%s segment=%d); " - "keeping the prior scope handle", + "Hermes Relay segment open failed (session=%s segment=%d); keeping the prior scope handle", session.session_id, session.segment, exc_info=True, ) @@ -672,9 +655,7 @@ class RelayRuntime: self._begin_operation() return RelayOperationLease(self) - def apply_tool_request_intercepts( - self, *, session_id: str, tool_name: str, args: dict[str, Any] - ) -> dict[str, Any]: + def apply_tool_request_intercepts(self, *, session_id: str, tool_name: str, args: dict[str, Any]) -> dict[str, Any]: """Apply Relay request rewriting before Hermes authorizes a tool call.""" if not self.managed_execution_enabled(): return args @@ -688,8 +669,7 @@ class RelayRuntime: return result if isinstance(result, dict) else args def _pop_with_drain( - self, handle: Any, *, output: dict[str, Any], metadata: dict[str, Any], session_root: Any, - drain_limit: int, + self, handle: Any, *, output: dict[str, Any], metadata: dict[str, Any], session_root: Any, drain_limit: int, ) -> 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`` @@ -721,9 +701,8 @@ class RelayRuntime: return retry_exc def _close_scope_handle( - self, session: RelaySession, handle: Any, *, output: dict[str, Any] | None = None, - allow_closing: bool = False, failure_label: str = "scope close failed", - drain_limit: int = 32, operation_already_held: bool = False, + self, session: RelaySession, handle: Any, *, output: dict[str, Any] | None = None, allow_closing: bool = False, + failure_label: str = "scope close failed", drain_limit: int = 32, operation_already_held: bool = False, ) -> str | None: """Pop ``handle``, draining orphaned children in the same session context. Relay scopes are strict LIFO; empty-stream retries + interrupt can abandon a @@ -830,9 +809,7 @@ class NoopRelayRuntime: profile_key: str reason: str - def apply_tool_request_intercepts( - self, *, session_id: str, tool_name: str, args: dict[str, Any] - ) -> dict[str, Any]: + def apply_tool_request_intercepts(self, *, session_id: str, tool_name: str, args: dict[str, Any]) -> dict[str, Any]: return args @staticmethod @@ -976,9 +953,7 @@ class RelaySessionCoordinator: self._active_turns_lock = threading.RLock() self._active_turns: dict[tuple[str, str], set[int]] = {} - def register_session_initializer( - self, name: str, callback: Callable[[RelayRuntime, dict[str, Any]], None] - ) -> None: + def register_session_initializer(self, name: str, callback: Callable[[RelayRuntime, dict[str, Any]], None]) -> None: """Register idempotent profile/session preparation before scope creation.""" with self._initializer_lock: self._session_initializers[name] = callback @@ -993,8 +968,7 @@ class RelaySessionCoordinator: logger.warning("Hermes Relay session initializer failed: %s", name, exc_info=True) def acquire_conversation( - self, *, profile_key: str, session_id: str, platform: str, parent_session_id: str = "", - model: str = "", + self, *, profile_key: str, session_id: str, platform: str, parent_session_id: str = "", model: str = "", ) -> ConversationLease: host = self.registry.for_profile(profile_key) or NoopRelayRuntime( profile_key, "Relay host creation was disabled" @@ -1050,9 +1024,7 @@ class RelaySessionCoordinator: 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"} - ), + metadata=runtime_metadata(host.runtime_id, **{"hermes.execution_surface": lease.platform or "unknown"}), timeout=_SCOPE_OP_TIMEOUT, ) turn._previous_turn = _CURRENT_TURN.get() @@ -1124,9 +1096,7 @@ class RelaySessionCoordinator: 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 = "" - ) -> None: + def notify_session_compacted(self, *, profile_key: str, session_id: str, old_session_id: str = "") -> None: """React to a completed compaction, per compaction mode. 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. @@ -1139,9 +1109,7 @@ class RelaySessionCoordinator: profile_key, session_id, old_session_id, ) - def _notify_session_compacted_unguarded( - self, profile_key: str, session_id: str, old_session_id: str - ) -> None: + 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) @@ -1307,14 +1275,10 @@ def _is_relay_wrapped_callback_error(relay_error: BaseException, callback_error: return False callback_type = callback_error.__class__ type_names = { - callback_type.__name__, callback_type.__qualname__, - f"{callback_type.__module__}.{callback_type.__qualname__}", + callback_type.__name__, callback_type.__qualname__, f"{callback_type.__module__}.{callback_type.__qualname__}", } message = str(relay_error) - return any( - message.startswith(f"internal error: {type_name}: {callback_error}") - for type_name in type_names - ) + return any(message.startswith(f"internal error: {type_name}: {callback_error}") for type_name in type_names) def get_runtime(*, create: bool = True, profile_key: str | None = None) -> RelayRuntime | None: diff --git a/agent/relay_tools.py b/agent/relay_tools.py index ae227a7f3f..68f16f92d6 100644 --- a/agent/relay_tools.py +++ b/agent/relay_tools.py @@ -54,15 +54,11 @@ def execute( ) ) except BaseException as exc: - if ( - callback_error is not None - and relay_runtime._is_relay_wrapped_callback_error(exc, callback_error) - ): + if (callback_error is not None and relay_runtime._is_relay_wrapped_callback_error(exc, callback_error)): raise callback_error if isinstance(exc, Exception) and callback_error is None and "value" in raw_result: logger.warning( - "NeMo Relay tool post-processing failed after dispatch success; " - "returning the Hermes tool result", + "NeMo Relay tool post-processing failed after dispatch success; returning the Hermes tool result", exc_info=True, ) return raw_result["value"], observed_args diff --git a/agent/transports/anthropic.py b/agent/transports/anthropic.py index 49360a91da..b00469d3af 100644 --- a/agent/transports/anthropic.py +++ b/agent/transports/anthropic.py @@ -25,9 +25,8 @@ def _unprefix_oauth_tool_name(name: str) -> str: # build_kwargs params forwarded to build_anthropic_kwargs, with the defaults applied when absent. _BUILD_KWARG_DEFAULTS = { - "max_tokens": 16384, "reasoning_config": None, "tool_choice": None, "is_oauth": False, - "preserve_dots": False, "context_length": None, "base_url": None, "fast_mode": False, - "drop_context_1m_beta": False, + "max_tokens": 16384, "reasoning_config": None, "tool_choice": None, "is_oauth": False, "preserve_dots": False, + "context_length": None, "base_url": None, "fast_mode": False, "drop_context_1m_beta": False, } @@ -54,8 +53,7 @@ class AnthropicTransport(ProviderTransport): return convert_tools_to_anthropic(tools) def build_kwargs( - self, model: str, messages: List[Dict[str, Any]], - tools: Optional[List[Dict[str, Any]]] = None, **params, + self, model: str, messages: List[Dict[str, Any]], tools: Optional[List[Dict[str, Any]]] = None, **params, ) -> Dict[str, Any]: """Build Anthropic messages.create() kwargs (converts messages and tools internally).""" from agent.anthropic_adapter import build_anthropic_kwargs