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

refactor(agent/adapters): simplify plugin_llm, backend_identity, stream hooks (-447 LOC)

Browse Source 
PluginLlm's four public entry points share _gate/_finish/_host_kwargs; drop
dead classify_failure_scope/_REASON_SCOPES (and their tests) and unify the
three _norm_* helpers; should_skip_candidate routes through a scope predicate
table. Injected caller kwargs, audit dicts and log lines unchanged.
This commit is contained in:
Teknium
2026-09-02 11:21:11 -07:00
parent 1db2452272
commit 63abd4d174
5 changed files with 431 additions and 901 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
agent/backend_identity.py
View File

@@ -1,32 +1,27 @@
"""Single owner for backend identity and failure-scoped skip decisions.
Every fallback / dedup / skip / quarantine decision in Hermes ultimately asks
one question: **"is this candidate the same backend as the one that failed,
along the axis that failure invalidated?"** Before this module, that
question was re-implemented inline at six call sites across four subsystems,
each comparing whatever string was locally convenient (provider label,
provider+model, base_url+model, ...). Each incident fixed one site while the
others kept the bug: #22548 (same-shim aliases), #70893 (xai-oauth vs xai —
same host, distinct credential), #59561 (aux chain skipped sibling models),
#72468 (aux main-model safety net, same bug three weeks later), #62984 /
#54250 / #57584 (dedup ignoring base_url strands multi-endpoint pools).
Every fallback / dedup / skip / quarantine decision asks one question: **"is
this candidate the same backend as the one that failed, along the axis that
failure invalidated?"** Answering it inline at each call site (comparing
whatever string was locally convenient) repeatedly reintroduced the same bugs:
same-shim aliases treated as distinct, sibling models skipped for one model's
timeout, dedup ignoring ``base_url`` and stranding multi-endpoint pools.
The root insight: "provider" conflates three independent identity axes, and
each failure class invalidates a different one:
"provider" conflates three independent identity axes, each invalidated by a
different failure class:
* **credential surface** — auth 401 / payment 402 kill everything sharing the
credential (every model, every host reached with that key/token).
* **endpoint** — DNS failure / connection refused kill everything behind the
URL, regardless of model or credential.
* **model deployment** — timeout / overload / rate limit / model-incompatible
kill ONE model's deployment. A sibling model behind the same URL is an
independent deployment (real incident: aux ``glm-5.2`` hung and timed out
while main ``macaron-v1-venti`` on the identical endpoint was serving
448K-token turns).
kill ONE model's deployment. A sibling model behind the same URL is an
independent deployment (one model hung while another on the identical
endpoint kept serving).
Call sites should build :class:`BackendIdentity` values, classify the failure
with :func:`classify_failure_scope`, and ask :func:`should_skip_candidate`.
Do not re-implement any comparison inline — extend THIS module instead.
Call sites build :class:`BackendIdentity` values and ask
:func:`should_skip_candidate`. Do not re-implement any comparison inline —
extend THIS module instead.
"""
from __future__ import annotations
@@ -45,53 +40,22 @@ class FailureScope(Enum):
#: Timeout, overload/429, connection blip, model-incompatible, invalid
#: response: evidence against ONE model deployment only.
MODEL = "model"
#: Auth 401 / payment 402: evidence against the shared credential —
#: every model reached with it is equally dead.
#: Auth 401 / payment 402: evidence against the shared credential.
CREDENTIAL = "credential"
#: DNS / connection-refused / unreachable host: evidence against the
#: endpoint — every model behind the URL is equally dead.
#: DNS / connection-refused / unreachable host: evidence against the endpoint.
ENDPOINT = "endpoint"
#: Reason strings already used by auxiliary_client's except-chain, mapped to
#: scopes. Unknown reasons default to MODEL — the least-invalidating scope —
#: so an unrecognized failure never over-skips viable candidates.
_REASON_SCOPES = {
"auth error": FailureScope.CREDENTIAL,
"payment error": FailureScope.CREDENTIAL,
"rate limit": FailureScope.MODEL,
"model incompatible with route": FailureScope.MODEL,
"invalid provider response": FailureScope.MODEL,
"connection error": FailureScope.MODEL,
"timeout": FailureScope.MODEL,
}
def classify_failure_scope(reason: Optional[str]) -> FailureScope:
"""Map a human-readable failure reason to the identity axis it kills."""
return _REASON_SCOPES.get((reason or "").strip().lower(), FailureScope.MODEL)
def _norm_provider(value: Optional[str]) -> str:
def _norm(value: Optional[str]) -> str:
return (value or "").strip().lower()
def _norm_model(value: Optional[str]) -> str:
return (value or "").strip().lower()
def _norm_base_url(value: Optional[str]) -> str:
return (value or "").strip().rstrip("/").lower()
@dataclass(frozen=True)
class BackendIdentity:
"""Normalized identity of one (provider, model, endpoint) deployment.
Empty fields mean "unknown" — comparisons treat an unknown axis as
non-distinguishing (it can neither prove sameness nor difference on its
own; the remaining axes decide).
"""
Empty fields mean "unknown" — an unknown axis can neither prove sameness
nor difference on its own; the remaining axes decide."""
provider: str = ""
model: str = ""
@@ -105,9 +69,9 @@ class BackendIdentity:
base_url: Optional[str] = None,
) -> "BackendIdentity":
return cls(
provider=_norm_provider(provider),
model=_norm_model(model),
base_url=_norm_base_url(base_url),
provider=_norm(provider),
model=_norm(model),
base_url=_norm(base_url).rstrip("/"),
)
@@ -116,9 +80,8 @@ def _both_first_class(a: BackendIdentity, b: BackendIdentity) -> bool:
Two different registry providers have distinct credential surfaces even
when they share an inference host (xai-oauth vs xai, openai-codex vs
openai-api) — #70893. Custom/shim aliases are NOT in the registry, so
two aliases pointing at one URL still count as the same backend (#22548).
"""
openai-api). Custom/shim aliases are NOT in the registry, so two aliases
pointing at one URL still count as the same backend."""
if not a.provider or not b.provider or a.provider == b.provider:
return False
try:
@@ -132,58 +95,44 @@ def _both_first_class(a: BackendIdentity, b: BackendIdentity) -> bool:
def same_credential_surface(a: BackendIdentity, b: BackendIdentity) -> bool:
"""Do two identities share the credential a 401/402 just invalidated?
Conservative on purpose: an unprovable axis must answer "different"
(try the candidate — worst case one wasted RTT) rather than "same"
(skip — worst case stranded failover). Two distinct custom labels at
one URL may carry different per-entry api_keys, so a shared URL alone
never proves a shared credential; it is only used as a weak signal
when a provider label is missing entirely.
"""
Conservative on purpose: an unprovable axis answers "different" (one wasted
RTT) rather than "same" (stranded failover). Same label = same configured
credential; different labels = different credential config (custom entries
can each carry their own api_key, so a shared URL alone never proves a
shared credential — it is only a weak signal when a label is missing)."""
if a.provider and b.provider:
# Same label = same configured credential. Different labels =
# different credential config (first-class registry providers
# explicitly so — #70893; custom entries can each carry their own
# api_key, so sameness is unprovable and we must not skip).
return a.provider == b.provider
# Provider unknown on a side: same explicit URL is the best signal left.
return bool(a.base_url and a.base_url == b.base_url)
def same_endpoint(a: BackendIdentity, b: BackendIdentity) -> bool:
"""Do two identities sit behind the endpoint that just went unreachable?"""
"""Do two identities sit behind the endpoint that just went unreachable?
An unknown base_url inherits the provider default, so a shared provider
label implies the same default endpoint."""
if a.base_url and b.base_url:
return a.base_url == b.base_url
# An unknown base_url inherits the provider default → same provider
# label implies the same default endpoint.
return bool(a.provider and a.provider == b.provider)
def same_deployment(a: BackendIdentity, b: BackendIdentity) -> bool:
"""Are these the exact same model deployment (the thing a timeout kills)?
Provider+model must match; the base_url axis distinguishes only when BOTH
sides carry an explicit URL (#62984: same provider+model on two different
explicit URLs is two deployments — a pool). A side with an unknown URL
inherits the provider default and cannot prove difference.
"""
Provider+model must match; base_url distinguishes only when BOTH sides carry
an explicit URL (same provider+model on two explicit URLs is a pool, not a
dup). Different labels with the same URL + model are still one deployment
(same-host shim aliases) — unless both labels are first-class registry
providers."""
if not (a.provider and b.provider and a.provider == b.provider):
# Same-host different-label shims: same URL + same model IS the same
# deployment even when the alias labels differ (#22548) — unless both
# labels are first-class registry providers (#70893).
if (
return bool(
a.base_url
and a.base_url == b.base_url
and a.model
and a.model == b.model
and not _both_first_class(a, b)
):
return True
return False
)
if not (a.model and b.model and a.model == b.model):
return False
if a.base_url and b.base_url and a.base_url != b.base_url:
return False # distinct explicit endpoints — a pool, not a dup
return True
return not (a.base_url and b.base_url and a.base_url != b.base_url)
def should_skip_candidate(
@@ -192,11 +141,8 @@ def should_skip_candidate(
scope: FailureScope = FailureScope.MODEL,
) -> bool:
"""THE skip predicate: would trying ``candidate`` just repeat the failure?
True when the candidate is the same backend as ``failed`` along the axis
``scope`` says the failure invalidated. Every fallback/dedup/skip site
must call this instead of comparing labels inline.
"""
True when it is the same backend as ``failed`` along the axis ``scope``
invalidated. Every fallback/dedup/skip site must call this."""
if scope is FailureScope.CREDENTIAL:
return same_credential_surface(candidate, failed)
if scope is FailureScope.ENDPOINT:

1033
agent/plugin_llm.py
View File

File diff suppressed because it is too large Load Diff

95
agent/plugin_stream_hooks.py
View File

@@ -1,4 +1,10 @@
"""Asynchronous per-consumer plugin observers for streaming LLM output."""
"""Asynchronous per-consumer plugin observers for streaming LLM output.
Each registered hook callback gets its own bounded queue + daemon worker thread
so plugin code never runs inline on the token path. Queues drop the oldest
pending event when full; dispatchers for callbacks that are no longer
registered are stopped lazily on the next lookup.
"""
from __future__ import annotations
@@ -32,6 +38,24 @@ def _callback_name(callback: Callable[..., Any]) -> str:
return getattr(callback, "__name__", repr(callback))
def _put_drop_oldest(events: "queue.Queue[Any]", item: Any) -> bool:
"""put_nowait; on a full queue evict the oldest pending event and retry once."""
try:
events.put_nowait(item)
return True
except queue.Full:
try:
events.get_nowait()
events.task_done()
except queue.Empty:
pass
try:
events.put_nowait(item)
return True
except queue.Full:
return False
def _worker(dispatcher: _ConsumerDispatcher) -> None:
while True:
item = dispatcher.events.get()
@@ -44,10 +68,7 @@ def _worker(dispatcher: _ConsumerDispatcher) -> None:
dispatcher.callback(**payload)
except Exception as exc:
logger.warning(
"Hook '%s' callback %s raised: %s",
dispatcher.hook_name,
_callback_name(dispatcher.callback),
exc,
"Hook '%s' callback %s raised: %s", dispatcher.hook_name, _callback_name(dispatcher.callback), exc
)
finally:
dispatcher.events.task_done()
@@ -64,54 +85,36 @@ def _registered_callbacks(hook_name: str) -> tuple[Callable[..., Any], ...]:
def _stop_dispatcher(dispatcher: _ConsumerDispatcher, timeout: float = 1.0) -> None:
try:
dispatcher.events.put_nowait(_STOP)
except queue.Full:
try:
dispatcher.events.get_nowait()
dispatcher.events.task_done()
except queue.Empty:
pass
try:
dispatcher.events.put_nowait(_STOP)
except queue.Full:
pass
_put_drop_oldest(dispatcher.events, _STOP)
if dispatcher.thread is not None:
dispatcher.thread.join(timeout=timeout)
def _start_dispatcher(hook_name: str, callback: Callable[..., Any]) -> _ConsumerDispatcher:
dispatcher = _ConsumerDispatcher(hook_name=hook_name, callback=callback, events=queue.Queue(maxsize=_QUEUE_SIZE))
dispatcher.thread = threading.Thread(
target=_worker, args=(dispatcher,), daemon=True, name=f"plugin-stream-hook:{hook_name}"
)
dispatcher.thread.start()
return dispatcher
def _dispatchers_for(hook_name: str) -> list[_ConsumerDispatcher]:
"""Live dispatcher per registered callback (restarting dead workers); stale
ones for unregistered callbacks are stopped outside the lock."""
callbacks = _registered_callbacks(hook_name)
if not callbacks:
return []
callback_ids = {id(callback) for callback in callbacks}
stale: list[_ConsumerDispatcher] = []
ready: list[_ConsumerDispatcher] = []
with _dispatcher_lock:
for key, dispatcher in list(_dispatchers.items()):
key_hook_name, callback_id = key
if key_hook_name == hook_name and callback_id not in callback_ids:
stale.append(_dispatchers.pop(key))
stale = [_dispatchers.pop(key) for key in list(_dispatchers) if key[0] == hook_name and key[1] not in callback_ids]
for callback in callbacks:
key = (hook_name, id(callback))
dispatcher = _dispatchers.get(key)
if dispatcher is None or dispatcher.thread is None or not dispatcher.thread.is_alive():
events: "queue.Queue[dict[str, Any] | object]" = queue.Queue(maxsize=_QUEUE_SIZE)
dispatcher = _ConsumerDispatcher(
hook_name=hook_name,
callback=callback,
events=events,
)
dispatcher.thread = threading.Thread(
target=_worker,
args=(dispatcher,),
daemon=True,
name=f"plugin-stream-hook:{hook_name}",
)
dispatcher.thread.start()
_dispatchers[key] = dispatcher
dispatcher = _dispatchers[key] = _start_dispatcher(hook_name, callback)
ready.append(dispatcher)
for dispatcher in stale:
@@ -124,24 +127,12 @@ def enqueue_plugin_stream_hook(hook_name: str, **payload: Any) -> bool:
queued = False
item = dict(payload)
for dispatcher in _dispatchers_for(hook_name):
try:
dispatcher.events.put_nowait(item)
if _put_drop_oldest(dispatcher.events, item):
queued = True
continue
except queue.Full:
try:
dispatcher.events.get_nowait()
dispatcher.events.task_done()
except queue.Empty:
pass
try:
dispatcher.events.put_nowait(item)
queued = True
except queue.Full:
else:
logger.debug(
"plugin stream hook queue full after drop-oldest: %s callback=%s",
hook_name,
_callback_name(dispatcher.callback),
hook_name, _callback_name(dispatcher.callback),
)
return queued

43
agent/provider_projection.py
View File

@@ -1,31 +1,23 @@
"""Fold an agent-as-provider's own activity back into Hermes' turn state.
Most providers are models: they ask Hermes to run a tool and Hermes runs it, so
the transcript and the loop's counters see every tool iteration. Some providers
are *agents* — an ACP CLI reached through a client shim, or the codex
app-server, which takes an analogous path in ``agent/codex_runtime.py``. They
execute their own read/edit/execute tools inside their own session, and by the
time Hermes sees the response that work is already done.
Some providers are *agents* (an ACP CLI behind a client shim; the codex
app-server takes an analogous path in ``agent/codex_runtime.py``): they run
their own tools inside their own session, so by the time Hermes sees the
response that work is done. Those calls must never come back as pending
``tool_calls`` (Hermes would re-run finished work), but two subsystems go blind
if they are merely summarised into ``reasoning``:
Those calls must never come back as pending ``tool_calls`` — Hermes would re-run
finished work. But two subsystems go blind if they are merely summarised into
the ``reasoning`` field:
* the **self-improvement loop**, which replays ``messages`` to distil memories
and skills;
* the **skill-review nudge**, whose ``_iters_since_skill`` counter only moves on
Hermes tool iterations.
* the **self-improvement loop**, which distils memories and skills by replaying
``messages`` — a one-line activity feed teaches it nothing;
* the **skill-review nudge**, whose counter (``_iters_since_skill``) only moves
on Hermes tool iterations, of which there are none.
So the provider client hands both back on the completion object and this helper
applies them: ``hermes_projected_messages`` (already-completed
``assistant(tool_calls=[…])`` + ``tool(result)`` history rows) and
``hermes_provider_tool_iterations`` (how many tool iterations happened inside
the provider). Clients that set neither are unaffected, which is every ordinary
OpenAI-compatible provider.
The splice is append-only and rows go through ``append_message`` like every
other live-transcript append, so they carry a timestamp and persist the same way
the codex projection path's rows do.
So the client hands both back on the completion object —
``hermes_projected_messages`` (completed ``assistant(tool_calls=[…])`` +
``tool(result)`` rows) and ``hermes_provider_tool_iterations`` — and this helper
applies them. Ordinary OpenAI-compatible clients set neither and are unaffected.
The splice is append-only through ``append_message`` so rows carry a timestamp
and persist like any other live-transcript append.
"""
from __future__ import annotations
@@ -59,9 +51,8 @@ def splice_provider_projection(
getattr(agent, "provider", "?"),
)
raw_iters = getattr(response, "hermes_provider_tool_iterations", 0)
try:
iterations = int(raw_iters or 0)
iterations = int(getattr(response, "hermes_provider_tool_iterations", 0) or 0)
except (TypeError, ValueError):
iterations = 0
if iterations > 0:

23
tests/agent/test_backend_identity.py
View File

@@ -11,7 +11,6 @@ from unittest.mock import patch
from agent.backend_identity import (
BackendIdentity,
FailureScope,
classify_failure_scope,
same_credential_surface,
same_deployment,
same_endpoint,
@@ -23,28 +22,6 @@ def _id(provider="", model="", base_url=""):
return BackendIdentity.build(provider=provider, model=model, base_url=base_url)
class TestClassifyFailureScope:
def test_auth_and_payment_are_credential_scoped(self):
assert classify_failure_scope("auth error") is FailureScope.CREDENTIAL
assert classify_failure_scope("payment error") is FailureScope.CREDENTIAL
def test_model_scoped_reasons(self):
for reason in (
"rate limit",
"timeout",
"connection error",
"model incompatible with route",
"invalid provider response",
):
assert classify_failure_scope(reason) is FailureScope.MODEL, reason
def test_unknown_reason_defaults_to_least_invalidating_scope(self):
"""Never over-skip on a reason string we don't recognize."""
assert classify_failure_scope("weird new error") is FailureScope.MODEL
assert classify_failure_scope(None) is FailureScope.MODEL
assert classify_failure_scope("") is FailureScope.MODEL
class TestSameDeployment:
def test_incident_59561_72468_sibling_model_same_provider_is_different(self):
"""aux glm-5.2 timing out says nothing about main macaron on the