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

refactor(agent/relay): compact relay/transport docstrings and comments (keep every invariant); reflow boolean spans; drop no-op helpers

Browse Source
This commit is contained in:
Teknium
2026-09-02 18:53:45 -07:00
parent a560fd840c
commit be3a659b14
6 changed files with 130 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
agent/relay_llm.py
View File

@@ -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

195
agent/relay_runtime.py
View File

@@ -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

10
agent/relay_tools.py
View File

@@ -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:

7
agent/transports/__init__.py
View File

@@ -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}")

33
agent/transports/anthropic.py
View File

@@ -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

7
agent/transports/base.py
View File

@@ -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