From 63abd4d17437b912f1c99d7058ed57555d58d89c Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 11:21:11 -0700 Subject: [PATCH] refactor(agent/adapters): simplify plugin_llm, backend_identity, stream hooks (-447 LOC) 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. --- agent/backend_identity.py | 138 ++-- agent/plugin_llm.py | 1033 ++++++++------------------ agent/plugin_stream_hooks.py | 95 ++- agent/provider_projection.py | 43 +- tests/agent/test_backend_identity.py | 23 - 5 files changed, 431 insertions(+), 901 deletions(-) diff --git a/agent/backend_identity.py b/agent/backend_identity.py index 7a7e9efb6b..502e87ec2a 100644 --- a/agent/backend_identity.py +++ b/agent/backend_identity.py @@ -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: diff --git a/agent/plugin_llm.py b/agent/plugin_llm.py index e0cc8ccca5..c0ea2d8528 100644 --- a/agent/plugin_llm.py +++ b/agent/plugin_llm.py @@ -2,39 +2,16 @@ Plugin LLM facade — host-owned LLM access for trusted plugins. ============================================================== -Plugins built on Hermes Agent often need to make their own LLM calls -out-of-band — a hook that rewrites a tool error before the user sees -it, a gateway adapter that translates inbound text, a slash command -that summarises a paste, a scheduled job that scores yesterday's -activity into a single line on a status board. +Plugins that need their own out-of-band model call (rewrite a tool error, +translate inbound text, summarise a paste, score a scheduled job) get +``ctx.llm`` on :class:`~hermes_cli.plugins.PluginContext`: ``complete`` / +``complete_structured`` (text + image inputs, JSON schema validation) and their +async siblings ``acomplete`` / ``acomplete_structured``. -Today the only stable plugin surfaces extend an existing Hermes -subsystem: ``register_tool``, ``register_platform``, -``register_memory_provider``, etc. None of those help when the -plugin's job is to make its own model call. This module is the -supported lane for that case. - -The plugin gets ``ctx.llm`` exposed on its -:class:`~hermes_cli.plugins.PluginContext`: - -* ``complete(messages, ...)`` — chat completion against the user's - active model + auth. -* ``complete_structured(instructions=..., input=[...], json_schema=...)`` - — bounded structured inference with optional image inputs, JSON - schema validation, and parsed JSON output. -* async siblings ``acomplete()`` / ``acomplete_structured()`` for - plugins running on asyncio loops (gateway adapters, hooks). - -Provider/model/agent_id/profile are explicit keyword arguments — no -embedded slugs, no shorthands. This mirrors Hermes' main config -shape (``model.provider`` + ``model.model``) so plugin authors who -already understand the host config don't have to learn anything new. - -The host owns provider routing, auth resolution, timeouts, and -fallback. The plugin never sees raw OAuth tokens or API keys. All -override knobs (``provider=``, ``model=``, ``agent_id=``, -``profile=``) are gated behind explicit per-plugin trust flags in -``config.yaml``:: +Provider/model/agent_id/profile are explicit keyword arguments mirroring the +host config shape (``model.provider`` + ``model.model``) — no embedded slugs. +The host owns routing, auth, timeouts, and fallback; the plugin never sees raw +tokens or keys. Every override knob is gated by per-plugin trust flags:: plugins: entries: @@ -48,22 +25,15 @@ override knobs (``provider=``, ``model=``, ``agent_id=``, allow_profile_override: false allow_task_override: false # borrow the host's built-in aux tasks -Untrusted plugins still get the default surface — they just can't -steer provider, model, agent, or auth-profile selection. The trust -gate is fail-closed: a missing config block means "no overrides," -not "anything goes." +The gate is fail-closed: a missing config block means "no overrides". -The ``task=`` kwarg on the ``complete``/``acomplete`` family routes a -call through a plugin-registered auxiliary model slot +``task=`` routes a call through a plugin-registered auxiliary model slot (``ctx.register_auxiliary_task``). A plugin may always name a slot it -registered itself; ``allow_task_override`` additionally lets it route -through the host's *built-in* auxiliary tasks. A foreign or unknown key -is rejected loudly (error + logged warning), never silently downgraded -to the main model. +registered itself; ``allow_task_override`` additionally lets it use the host's +*built-in* auxiliary tasks. A foreign or unknown key is rejected loudly +(error + logged warning), never silently downgraded to the main model. -Backed by :func:`agent.auxiliary_client.call_llm`, which already -handles every provider, fallback chain, and per-task override Hermes -supports. +Backed by :func:`agent.auxiliary_client.call_llm`. """ from __future__ import annotations @@ -93,13 +63,8 @@ class PluginLlmTextInput: @dataclass class PluginLlmImageInput: - """Image block in a structured input list. - - Either ``data`` (raw bytes) or ``url`` (http(s) or data: URL) must be - provided. ``mime_type`` defaults to ``image/png`` when ``data`` is - used and is required for non-PNG bytes to render correctly across - providers. - """ + """Image block. Provide ``data`` (raw bytes) or ``url`` (http(s)/data: URL). + ``mime_type`` is required for non-PNG bytes to render across providers.""" data: Optional[bytes] = None url: Optional[str] = None @@ -109,21 +74,16 @@ class PluginLlmImageInput: PluginLlmInput = Union[PluginLlmTextInput, PluginLlmImageInput, Dict[str, Any]] -"""A single structured input block. - -Plugins may pass either the dataclasses above or plain dicts with the -same shape — dicts are normalized internally. Dict shape:: - - {"type": "text", "text": "..."} - {"type": "image", "data": , "mime_type": "image/png", "file_name": "receipt.png"} - {"type": "image", "url": "https://..."} -""" +"""A single structured input block: one of the dataclasses above or a plain dict +of the same shape (``{"type": "text", "text": ...}`` / +``{"type": "image", "data": , "mime_type": ..., "file_name": ...}`` / +``{"type": "image", "url": ...}``).""" @dataclass class PluginLlmUsage: - """Token + cost usage for a completion. All fields optional — providers - differ on what they return. ``cost_usd`` is the host's best estimate.""" + """Token + cost usage. All fields optional — providers differ on what they + return. ``cost_usd`` is the host's best estimate.""" input_tokens: int = 0 output_tokens: int = 0 @@ -149,10 +109,9 @@ class PluginLlmCompleteResult: class PluginLlmStructuredResult: """Result of :meth:`PluginLlm.complete_structured`. - ``parsed`` is set only when ``json_mode=True`` or ``json_schema`` is - provided AND the response was valid JSON. ``content_type`` is - ``"json"`` in that case, ``"text"`` otherwise (e.g. the model - refused or the response wasn't requested as JSON).""" + ``parsed`` is set only when JSON output was requested (``json_mode`` or + ``json_schema``) AND the response was valid JSON; ``content_type`` is then + ``"json"``, otherwise ``"text"``.""" text: str provider: str @@ -182,44 +141,42 @@ class _TrustPolicy: allow_any_model: bool = False # True when allowed_models == ["*"] allow_agent_id_override: bool = False allow_profile_override: bool = False - # Gates routing a call through a *built-in* auxiliary task slot via - # ``ctx.llm.complete(task=...)``. A plugin may always route through a - # slot it registered itself; this flag additionally lets it borrow the - # host's built-in aux tasks. Off by default (fail-closed). + # Lets ``complete(task=...)`` borrow the host's *built-in* aux task slots. + # Slots the plugin registered itself are always allowed. Fail-closed. allow_task_override: bool = False +# The ``allow_*_override`` config keys; each is a same-named ``_TrustPolicy`` field. +_OVERRIDE_FLAGS = ( + "allow_provider_override", + "allow_model_override", + "allow_agent_id_override", + "allow_profile_override", + "allow_task_override", +) + + def _normalize_ref(raw: str) -> str: """Lower-case + strip whitespace. Used for allowlist matching.""" return (raw or "").strip().lower() def _coerce_allowlist(raw: Any) -> tuple[Optional[frozenset], bool]: - """Coerce a YAML list into ``(frozenset_or_None, allow_any)``. - - ``["*"]`` (or any list containing ``"*"``) → ``(frozenset(), True)``. - Any other list → ``(frozenset({...}), False)``. - Missing / non-list → ``(None, False)`` meaning "no allowlist." - """ + """YAML list → ``(frozenset_or_None, allow_any)``. A ``"*"`` entry sets + ``allow_any``; missing / non-list → ``(None, False)`` = no allowlist.""" if not isinstance(raw, list): return None, False normalized = [_normalize_ref(item) for item in raw if isinstance(item, str)] allow_any = "*" in normalized cleaned = {item for item in normalized if item and item != "*"} - if allow_any and not cleaned: - return frozenset(), True - if cleaned: - return frozenset(cleaned), allow_any - return frozenset(), allow_any + return frozenset(cleaned), allow_any def _resolve_trust_policy(plugin_id: str) -> _TrustPolicy: """Read ``plugins.entries..llm`` from config.yaml. - Missing config → fully restrictive policy (default deny on every - override). The policy is resolved per-call rather than cached so - config edits take effect without restarting the agent. - """ + Missing config → fully restrictive policy. Resolved per call (not cached) + so config edits take effect without restarting the agent.""" if not plugin_id: return _TrustPolicy(plugin_id="") @@ -229,35 +186,21 @@ def _resolve_trust_policy(plugin_id: str) -> _TrustPolicy: except Exception: # pragma: no cover — config IO failure return _TrustPolicy(plugin_id=plugin_id) - plugins_cfg = config.get("plugins") - if not isinstance(plugins_cfg, dict): - return _TrustPolicy(plugin_id=plugin_id) - entries = plugins_cfg.get("entries") - if not isinstance(entries, dict): - return _TrustPolicy(plugin_id=plugin_id) - entry = entries.get(plugin_id) - if not isinstance(entry, dict): - return _TrustPolicy(plugin_id=plugin_id) - llm_cfg = entry.get("llm") + llm_cfg: Any = config.get("plugins") + for key in ("entries", plugin_id, "llm"): + llm_cfg = llm_cfg.get(key) if isinstance(llm_cfg, dict) else None if not isinstance(llm_cfg, dict): return _TrustPolicy(plugin_id=plugin_id) allowed_models, allow_any_model = _coerce_allowlist(llm_cfg.get("allowed_models")) - allowed_providers, allow_any_provider = _coerce_allowlist( - llm_cfg.get("allowed_providers") - ) - + allowed_providers, allow_any_provider = _coerce_allowlist(llm_cfg.get("allowed_providers")) return _TrustPolicy( plugin_id=plugin_id, - allow_provider_override=bool(llm_cfg.get("allow_provider_override", False)), allowed_providers=allowed_providers, allow_any_provider=allow_any_provider, - allow_model_override=bool(llm_cfg.get("allow_model_override", False)), allowed_models=allowed_models, allow_any_model=allow_any_model, - allow_agent_id_override=bool(llm_cfg.get("allow_agent_id_override", False)), - allow_profile_override=bool(llm_cfg.get("allow_profile_override", False)), - allow_task_override=bool(llm_cfg.get("allow_task_override", False)), + **{name: bool(llm_cfg.get(name, False)) for name in _OVERRIDE_FLAGS}, ) @@ -265,108 +208,68 @@ class PluginLlmTrustError(PermissionError): """Raised when a plugin attempts an LLM override without trust.""" -def _check_overrides( - policy: _TrustPolicy, - *, - requested_provider: Optional[str], - requested_model: Optional[str], - requested_agent_id: Optional[str], - requested_profile: Optional[str], -) -> tuple[Optional[str], Optional[str], Optional[str], Optional[str]]: - """Apply the trust gate. Returns the validated overrides as - ``(provider, model, agent_id, profile)`` or raises - :class:`PluginLlmTrustError`. +def _denied(plugin_id: str, what: str, flag: str) -> PluginLlmTrustError: + """Uniform "flag not set" trust error.""" + return PluginLlmTrustError( + f"Plugin {plugin_id!r} cannot {what} " + f"(set plugins.entries.{plugin_id}.llm.{flag} to true to allow)." + ) - Each override (``provider``, ``model``, ``agent_id``, ``profile``) - is independently gated. ``provider`` and ``model`` each have an - optional allowlist via ``allowed_providers`` / ``allowed_models``. - """ - final_provider: Optional[str] = None - final_model: Optional[str] = None - final_profile: Optional[str] = None - if requested_provider: - if not policy.allow_provider_override: - raise PluginLlmTrustError( - f"Plugin {policy.plugin_id!r} cannot override the provider " - f"(set plugins.entries.{policy.plugin_id}.llm.allow_provider_override " - f"to true to allow)." - ) - normalized = _normalize_ref(requested_provider) - if ( - not policy.allow_any_provider - and policy.allowed_providers is not None - and normalized not in policy.allowed_providers - ): - raise PluginLlmTrustError( - f"Plugin {policy.plugin_id!r} provider override " - f"{requested_provider!r} is not in plugins.entries." - f"{policy.plugin_id}.llm.allowed_providers." - ) - final_provider = requested_provider.strip() - - if requested_model: - if not policy.allow_model_override: - raise PluginLlmTrustError( - f"Plugin {policy.plugin_id!r} cannot override the model " - f"(set plugins.entries.{policy.plugin_id}.llm.allow_model_override " - f"to true to allow)." - ) - normalized = _normalize_ref(requested_model) - if ( - not policy.allow_any_model - and policy.allowed_models is not None - and normalized not in policy.allowed_models - ): - raise PluginLlmTrustError( - f"Plugin {policy.plugin_id!r} model override " - f"{requested_model!r} is not in plugins.entries." - f"{policy.plugin_id}.llm.allowed_models." - ) - final_model = requested_model.strip() - - if requested_agent_id and not policy.allow_agent_id_override: +def _gate_ref_override(policy: _TrustPolicy, kind: str, requested: str) -> str: + """Gate a ``provider`` / ``model`` override: trust flag, then optional + allowlist. Returns the stripped value or raises.""" + if not getattr(policy, f"allow_{kind}_override"): + raise _denied(policy.plugin_id, f"override the {kind}", f"allow_{kind}_override") + allowed = getattr(policy, f"allowed_{kind}s") + if not getattr(policy, f"allow_any_{kind}") and allowed is not None and _normalize_ref(requested) not in allowed: raise PluginLlmTrustError( - f"Plugin {policy.plugin_id!r} cannot run completions against a " - f"non-default agent id (set plugins.entries.{policy.plugin_id}." - f"llm.allow_agent_id_override to true to allow)." + f"Plugin {policy.plugin_id!r} {kind} override " + f"{requested!r} is not in plugins.entries." + f"{policy.plugin_id}.llm.allowed_{kind}s." ) + return requested.strip() - if requested_profile: - if not policy.allow_profile_override: - raise PluginLlmTrustError( - f"Plugin {policy.plugin_id!r} cannot override the auth profile " - f"(set plugins.entries.{policy.plugin_id}.llm.allow_profile_override " - f"to true to allow)." - ) - final_profile = requested_profile.strip() +# Overrides gated by a bare trust flag (no allowlist): ``kind`` -> denial wording. +_FLAG_ONLY_OVERRIDES = { + "agent_id": "run completions against a non-default agent id", + "profile": "override the auth profile", +} + + +def _check_overrides( + policy: _TrustPolicy, *, requested_provider: Optional[str], requested_model: Optional[str], + requested_agent_id: Optional[str], requested_profile: Optional[str], +) -> tuple[Optional[str], Optional[str], Optional[str], Optional[str]]: + """Apply the trust gate; each override is gated independently, in the order + provider, model, agent_id, profile. Returns ``(provider, model, agent_id, + profile)`` (agent_id unstripped) or raises :class:`PluginLlmTrustError`.""" + final_provider = _gate_ref_override(policy, "provider", requested_provider) if requested_provider else None + final_model = _gate_ref_override(policy, "model", requested_model) if requested_model else None + for kind, requested in (("agent_id", requested_agent_id), ("profile", requested_profile)): + if requested and not getattr(policy, f"allow_{kind}_override"): + raise _denied(policy.plugin_id, _FLAG_ONLY_OVERRIDES[kind], f"allow_{kind}_override") + final_profile = requested_profile.strip() if requested_profile else None return final_provider, final_model, requested_agent_id, final_profile def _resolve_task_ownership(plugin_id: str) -> tuple[frozenset, frozenset]: """Return ``(owned_keys, builtin_keys)`` for the task trust gate. - ``owned_keys`` are auxiliary-task keys ``plugin_id`` registered itself - via ``ctx.register_auxiliary_task``; ``builtin_keys`` are the host's - reserved auxiliary tasks. Both imports are lazy so plugin discovery - doesn't hit a circular import at module load. A registry that can't be - read yields empty sets, which fails the gate closed (unknown → rejected). - - Ownership matches on the same canonical id ``ctx.llm`` is bound to - (``manifest.key or manifest.name``); ``register_auxiliary_task`` stores - that same id as the entry's ``plugin`` owner. - """ + Imports are lazy (circular import at plugin discovery). An unreadable + registry yields empty sets, failing the gate closed. Ownership matches on + the canonical id ``ctx.llm`` is bound to (``manifest.key or manifest.name``), + which is what ``register_auxiliary_task`` stores as the entry's ``plugin``.""" owned: set = set() builtin: set = set() try: from hermes_cli.plugins import get_plugin_auxiliary_tasks - for entry in get_plugin_auxiliary_tasks(): - if entry.get("plugin") == plugin_id: - key = entry.get("key") - if isinstance(key, str) and key: - owned.add(key) + owned = { + e.get("key") for e in get_plugin_auxiliary_tasks() + if e.get("plugin") == plugin_id and isinstance(e.get("key"), str) and e.get("key") + } except Exception: # pragma: no cover — registry unavailable pass try: @@ -384,24 +287,14 @@ def _check_task( plugin_id: str, requested_task: Optional[str], ) -> Optional[str]: - """Validate a plugin's requested auxiliary ``task`` routing key. + """Validate a plugin's requested auxiliary ``task`` key. - Returns the normalized key to route through, or ``None`` for the - default main-model path. Resolution: - - * unset / ``""`` / ``"auto"`` → ``None`` (today's behavior, byte-for-byte). + * unset / ``""`` / ``"auto"`` → ``None`` (main-model path). * a key the plugin registered itself → allowed. - * a built-in auxiliary key → allowed only when - ``plugins.entries..llm.allow_task_override`` is true. - * anything else (foreign or unknown) → **rejected loudly**. - - A foreign/unknown key raises :class:`PluginLlmTrustError` and logs a - warning naming the offending plugin and key. It is deliberately *not* - silently downgraded to ``auto``: silent fallback masks the - misconfiguration and could route the call to the main model the user - may have steered elsewhere on purpose (round-2 design correction, - tracked in #64182 / #64174). - """ + * a built-in key → allowed only with ``allow_task_override``. + * anything else → raises + logs a warning. Never silently downgraded to + ``auto``: that would mask the misconfiguration and could route to a main + model the user steered elsewhere on purpose.""" if not requested_task: return None task = requested_task.strip() @@ -409,28 +302,18 @@ def _check_task( return None owned, builtin = _resolve_task_ownership(plugin_id) - if task in owned: + if task in owned or (task in builtin and policy.allow_task_override): return task if task in builtin: - if policy.allow_task_override: - return task logger.warning( "plugin_llm task routing denied: plugin %r requested built-in " "auxiliary task %r without plugins.entries.%s.llm.allow_task_override", - plugin_id, - task, - plugin_id, - ) - raise PluginLlmTrustError( - f"Plugin {plugin_id!r} cannot route through the built-in auxiliary " - f"task {task!r} (set plugins.entries.{plugin_id}.llm." - f"allow_task_override to true to allow)." + plugin_id, task, plugin_id, ) + raise _denied(plugin_id, f"route through the built-in auxiliary task {task!r}", "allow_task_override") logger.warning( - "plugin_llm task routing denied: plugin %r requested auxiliary task %r " - "it did not register", - plugin_id, - task, + "plugin_llm task routing denied: plugin %r requested auxiliary task %r it did not register", + plugin_id, task, ) raise PluginLlmTrustError( f"Plugin {plugin_id!r} cannot route through auxiliary task {task!r} — a " @@ -446,40 +329,44 @@ def _check_task( def _normalize_input_block(block: PluginLlmInput) -> Dict[str, Any]: - """Coerce a structured input block to a plain dict the message - builder understands. Unknown shapes raise ``ValueError``.""" + """Coerce a structured input block to a plain dict. Unknown shapes raise ``ValueError``.""" if isinstance(block, PluginLlmTextInput): return {"type": "text", "text": block.text} if isinstance(block, PluginLlmImageInput): - d: Dict[str, Any] = { - "type": "image", - "mime_type": block.mime_type, - "file_name": block.file_name, - } + d: Dict[str, Any] = {"type": "image", "mime_type": block.mime_type, "file_name": block.file_name} if block.data is not None: d["data"] = block.data if block.url: d["url"] = block.url return d - if isinstance(block, dict): - kind = block.get("type") - if kind == "text": - text = block.get("text") - if not isinstance(text, str): - raise ValueError("text input block requires 'text' string") - return {"type": "text", "text": text} - if kind == "image": - if "data" not in block and not block.get("url"): - raise ValueError("image input block requires 'data' bytes or 'url'") - return { - "type": "image", - "data": block.get("data"), - "url": block.get("url"), - "mime_type": block.get("mime_type") or "image/png", - "file_name": block.get("file_name") or "", - } - raise ValueError(f"Unknown input block type: {kind!r}") - raise ValueError(f"Unsupported input block: {type(block).__name__}") + if not isinstance(block, dict): + raise ValueError(f"Unsupported input block: {type(block).__name__}") + kind = block.get("type") + if kind == "text": + text = block.get("text") + if not isinstance(text, str): + raise ValueError("text input block requires 'text' string") + return {"type": "text", "text": text} + if kind == "image": + if "data" not in block and not block.get("url"): + raise ValueError("image input block requires 'data' bytes or 'url'") + return { + "type": "image", "data": block.get("data"), "url": block.get("url"), + "mime_type": block.get("mime_type") or "image/png", "file_name": block.get("file_name") or "", + } + raise ValueError(f"Unknown input block type: {kind!r}") + + +def _image_part(norm: Dict[str, Any]) -> Dict[str, Any]: + """Normalized image block → OpenAI ``image_url`` part (data: URL for bytes).""" + url = norm.get("url") + if not url: + data = norm.get("data") or b"" + if not isinstance(data, (bytes, bytearray)): + raise ValueError("image input 'data' must be bytes") + b64 = base64.b64encode(data).decode("ascii") + url = f"data:{norm.get('mime_type') or 'image/png'};base64,{b64}" + return {"type": "image_url", "image_url": {"url": url}} def _build_structured_messages( @@ -491,17 +378,12 @@ def _build_structured_messages( schema_name: Optional[str], system_prompt: Optional[str], ) -> List[Dict[str, Any]]: - """Build the OpenAI-style messages list for a structured call. - - The instructions become the first text part of the user message, - followed by an optional ``Schema name: `` hint and an optional - JSON-only directive when JSON output is requested. Image inputs are - encoded as ``image_url`` parts. - """ + """Build OpenAI-style messages for a structured call: optional system + message (prompt + JSON-only directive), then a user message whose first + text part is the instructions (+ schema name / JSON schema) followed by + the input blocks.""" messages: List[Dict[str, Any]] = [] - sys_parts: List[str] = [] - if system_prompt: - sys_parts.append(system_prompt.strip()) + sys_parts: List[str] = [system_prompt.strip()] if system_prompt else [] if json_mode or json_schema is not None: sys_parts.append( "Respond with a single JSON object that matches the requested shape. " @@ -510,7 +392,6 @@ def _build_structured_messages( if sys_parts: messages.append({"role": "system", "content": "\n\n".join(sys_parts)}) - user_parts: List[Dict[str, Any]] = [] header = instructions.strip() if schema_name: header = f"{header}\n\nSchema name: {schema_name}" @@ -520,29 +401,11 @@ def _build_structured_messages( except (TypeError, ValueError): schema_text = str(json_schema) header = f"{header}\n\nJSON schema:\n{schema_text}" - user_parts.append({"type": "text", "text": header}) + user_parts: List[Dict[str, Any]] = [{"type": "text", "text": header}] for block in inputs: - norm = _normalize_input_block(block) - if norm["type"] == "text": - user_parts.append({"type": "text", "text": norm["text"]}) - elif norm["type"] == "image": - if norm.get("url"): - user_parts.append({ - "type": "image_url", - "image_url": {"url": norm["url"]}, - }) - else: - data = norm.get("data") or b"" - if not isinstance(data, (bytes, bytearray)): - raise ValueError("image input 'data' must be bytes") - b64 = base64.b64encode(data).decode("ascii") - mime = norm.get("mime_type") or "image/png" - user_parts.append({ - "type": "image_url", - "image_url": {"url": f"data:{mime};base64,{b64}"}, - }) - + norm = _normalize_input_block(block) # always "text" or "image" + user_parts.append({"type": "text", "text": norm["text"]} if norm["type"] == "text" else _image_part(norm)) messages.append({"role": "user", "content": user_parts}) return messages @@ -556,23 +419,19 @@ _FENCE_RE = re.compile(r"```(?:json)?\s*(.+?)```", re.DOTALL | re.IGNORECASE) def _strip_code_fences(text: str) -> str: - """Pull the first fenced code block out of ``text`` if any. Returns - ``text`` unchanged when no fence is present.""" + """Return the first fenced code block's body, or the stripped text when unfenced.""" match = _FENCE_RE.search(text) - if match: - return match.group(1).strip() - return text.strip() + return match.group(1).strip() if match else text.strip() def _parse_structured_text( *, text: str, json_mode: bool, json_schema: Optional[Any] ) -> tuple[Optional[Any], str]: - """Return ``(parsed, content_type)``. ``content_type`` is ``"json"`` - when parsing succeeded and (when a schema was given) validation - passed; ``"text"`` otherwise.""" - if not (json_mode or json_schema is not None): - return None, "text" - if not text: + """Return ``(parsed, content_type)``: ``"json"`` when parsing (and schema + validation, if a schema was given) succeeded, ``"text"`` otherwise. + Schema violations raise ``ValueError``; a missing ``jsonschema`` package + skips validation with a debug log.""" + if not (json_mode or json_schema is not None) or not text: return None, "text" try: @@ -585,7 +444,6 @@ def _parse_structured_text( import jsonschema # type: ignore[import-untyped] jsonschema.validate(parsed, json_schema) except ImportError: - # jsonschema is optional; skip strict validation when absent. logger.debug("jsonschema unavailable; skipping schema validation") except jsonschema.ValidationError as exc: # type: ignore[attr-defined] raise ValueError( @@ -596,16 +454,14 @@ def _parse_structured_text( # --------------------------------------------------------------------------- -# Usage extraction +# Response extraction # --------------------------------------------------------------------------- def _extract_usage(response: Any) -> PluginLlmUsage: - """Pull token usage out of an OpenAI-shaped response object. - - Tolerant of provider differences — Anthropic via the auxiliary - adapter exposes ``usage.prompt_tokens`` / ``usage.completion_tokens``; - direct OpenAI also exposes ``cache_read_input_tokens``.""" + """Pull token usage out of an OpenAI-shaped response, tolerating provider + naming differences (Anthropic via the aux adapter: ``prompt_tokens`` / + ``completion_tokens``; direct OpenAI adds ``cache_read_input_tokens``).""" usage = PluginLlmUsage() raw = getattr(response, "usage", None) if raw is None: @@ -631,26 +487,30 @@ def _extract_usage(response: Any) -> PluginLlmUsage: def _extract_text(response: Any) -> str: """Pull the assistant text out of an OpenAI-shaped response object.""" try: - msg = response.choices[0].message - content = getattr(msg, "content", None) + content = getattr(response.choices[0].message, "content", None) if isinstance(content, str): return content if isinstance(content, list): - parts: List[str] = [] - for part in content: - if isinstance(part, dict): - if part.get("type") == "text" and isinstance(part.get("text"), str): - parts.append(part["text"]) - else: - txt = getattr(part, "text", None) - if isinstance(txt, str): - parts.append(txt) - return "".join(parts) + texts = ( + (part.get("text") if part.get("type") == "text" else None) + if isinstance(part, dict) else getattr(part, "text", None) + for part in content + ) + return "".join(t for t in texts if isinstance(t, str)) except (AttributeError, IndexError, TypeError): pass return "" +def _main_config_value(reader: str, default: str) -> str: + """Read the current main provider/model via ``agent.auxiliary_client``.""" + try: + import agent.auxiliary_client as ac + return (getattr(ac, reader)() or "").strip() or default + except Exception: # pragma: no cover — defensive + return default + + def _resolve_attribution( *, provider_override: Optional[str], @@ -660,52 +520,16 @@ def _resolve_attribution( ) -> tuple[str, str]: """Decide what to record as ``result.provider`` / ``result.model``. - Precedence: - - 1. Explicit overrides win — if the plugin asked for ``provider="x"`` - or ``model="y"``, that's what we record (it's what the call - actually targeted). - 2. ``response.model`` (if present) overrides the recorded model - string. Providers post-resolution often return a slightly - different model id than the request (e.g. ``gpt-4o`` → - ``gpt-4o-2024-08-06``); the plugin's audit log should reflect - what actually ran. - 3. The route selected by ``auxiliary_client`` supplies the provider/model - when no override or response model is available. - 4. Otherwise the current main provider/model is used. - 5. If everything above is empty, fall back to ``"auto"`` / - ``"default"`` so the result object has non-empty strings. - """ + Provider: route selected by ``auxiliary_client`` > explicit override > + current main provider > ``"auto"``. Model: ``response.model`` (providers + return the canonical id that actually ran, e.g. ``gpt-4o-2024-08-06``) > + route > override > current main model > ``"default"``.""" route_info = route_info or {} - route_provider = route_info.get("provider") - route_model = route_info.get("model") - - if route_provider: - provider = route_provider - elif provider_override: - provider = provider_override - else: - try: - from agent.auxiliary_client import _read_main_provider - provider = (_read_main_provider() or "").strip() or "auto" - except Exception: # pragma: no cover — defensive - provider = "auto" - + provider = route_info.get("provider") or provider_override or _main_config_value("_read_main_provider", "auto") response_model = getattr(response, "model", None) if isinstance(response_model, str) and response_model.strip(): - model = response_model.strip() - elif route_model: - model = route_model - elif model_override: - model = model_override - else: - try: - from agent.auxiliary_client import _read_main_model - model = (_read_main_model() or "").strip() or "default" - except Exception: # pragma: no cover — defensive - model = "default" - - return provider, model + return provider, response_model.strip() + return provider, route_info.get("model") or model_override or _main_config_value("_read_main_model", "default") # --------------------------------------------------------------------------- @@ -713,14 +537,44 @@ def _resolve_attribution( # --------------------------------------------------------------------------- +def _json_response_format(*, json_mode: bool, json_schema: Optional[Any]) -> Optional[Dict[str, Any]]: + """``extra_body.response_format`` for the request; falls back to + ``json_object`` without a schema so schema-blind providers still get a hint.""" + if json_schema is not None: + schema = {"name": "plugin_structured_output", "schema": json_schema, "strict": False} + return {"response_format": {"type": "json_schema", "json_schema": schema}} + if json_mode: + return {"response_format": {"type": "json_object"}} + return None + + +def _structured_spec( + name: str, instructions: str, input: Sequence[PluginLlmInput], system_prompt: Optional[str], + json_mode: bool, json_schema: Optional[Any], schema_name: Optional[str], +) -> Dict[str, Any]: + """Argument check for the structured methods (runs before the trust gate); + returns the spec ``_gate`` / ``_finish`` consume.""" + if not instructions or not instructions.strip(): + raise ValueError(f"{name} requires non-empty instructions") + if not input: + raise ValueError(f"{name} requires at least one input block") + return dict( + instructions=instructions, inputs=list(input), system_prompt=system_prompt, + json_mode=json_mode, json_schema=json_schema, schema_name=schema_name, + ) + + class PluginLlm: """Host-owned LLM access for one trusted plugin. - Instances are constructed by :class:`hermes_cli.plugins.PluginContext` - and exposed as ``ctx.llm``. Plugins should not instantiate this - directly — the constructor binds plugin identity for trust-gate - enforcement. - """ + Constructed by :class:`hermes_cli.plugins.PluginContext` and exposed as + ``ctx.llm``; the constructor binds plugin identity for trust enforcement, + so plugins should not instantiate it directly. + + Every public method is ``_gate`` (trust checks → call kwargs) → + ``_invoke_*`` (host ``call_llm`` or injected caller) → ``_finish`` + (result + audit log); the sync/async and plain/structured variants differ + only in which pieces they pass through.""" def __init__( self, @@ -753,60 +607,12 @@ class PluginLlm: ) -> PluginLlmCompleteResult: """Run a host-owned chat completion against the user's active model. - ``messages`` is the standard OpenAI shape. ``provider``, - ``model``, ``agent_id``, and ``profile`` follow the same - explicit shape as the host's main config (``model.provider`` - + ``model.model``). Each is independently gated by - ``plugins.entries..llm.allow_*_override`` (see module - docstring). - - ``task`` optionally routes the call through a plugin-registered - auxiliary model slot (``ctx.register_auxiliary_task``): unset or - ``"auto"`` keeps today's main-model behavior, a slot the plugin - registered itself resolves through ``auxiliary.`` config, - and a foreign/unknown key is rejected (see :func:`_check_task`). - """ - policy = self._policy_loader(self._plugin_id) - eff_task = _check_task(policy, plugin_id=self._plugin_id, requested_task=task) - eff_provider, eff_model, eff_agent, eff_profile = _check_overrides( - policy, - requested_provider=provider, - requested_model=model, - requested_agent_id=agent_id, - requested_profile=profile, - ) - real_provider, real_model, response = self._invoke_sync( - messages=messages, - provider_override=eff_provider, - model_override=eff_model, - profile_override=eff_profile, - temperature=temperature, - max_tokens=max_tokens, - timeout=timeout, - task=eff_task, - ) - text = _extract_text(response) - usage = _extract_usage(response) - result = PluginLlmCompleteResult( - text=text, - provider=real_provider, - model=real_model, - agent_id=eff_agent or "default", - usage=usage, - audit={ - "plugin_id": self._plugin_id, - "purpose": purpose or "", - "profile": eff_profile or "", - "task": eff_task or "", - }, - ) - logger.info( - "plugin_llm.complete plugin=%s provider=%s model=%s task=%s " - "purpose=%s tokens=%d", - self._plugin_id, real_provider, real_model, eff_task or "", - purpose or "", usage.total_tokens, - ) - return result + ``messages`` is the standard OpenAI shape. ``provider``/``model``/ + ``agent_id``/``profile`` are each gated by + ``plugins.entries..llm.allow_*_override``. ``task`` routes through + a plugin-registered auxiliary slot (see :func:`_check_task`).""" + agent, kw = self._gate(provider, model, agent_id, profile, task, messages, temperature, max_tokens, timeout) + return self._finish("complete", agent, kw, self._invoke_sync(kw), purpose) def complete_structured( self, @@ -829,83 +635,13 @@ class PluginLlm: ) -> PluginLlmStructuredResult: """Run a bounded host-owned structured completion. - ``input`` accepts text and image blocks (see - :class:`PluginLlmTextInput` / :class:`PluginLlmImageInput`). When - ``json_mode=True`` or ``json_schema`` is provided, the response - is parsed and (if a schema is given) validated; the parsed value - is returned in :attr:`PluginLlmStructuredResult.parsed`. - - Validation requires the optional ``jsonschema`` package. When it - isn't installed, JSON mode still works but schema enforcement is - skipped with a debug log. - - ``task`` routes through a plugin-registered auxiliary slot (see - :meth:`complete`). - """ - if not instructions or not instructions.strip(): - raise ValueError("complete_structured requires non-empty instructions") - if not input: - raise ValueError("complete_structured requires at least one input block") - - policy = self._policy_loader(self._plugin_id) - eff_task = _check_task(policy, plugin_id=self._plugin_id, requested_task=task) - eff_provider, eff_model, eff_agent, eff_profile = _check_overrides( - policy, - requested_provider=provider, - requested_model=model, - requested_agent_id=agent_id, - requested_profile=profile, - ) - - messages = _build_structured_messages( - instructions=instructions, - inputs=list(input), - json_mode=json_mode, - json_schema=json_schema, - schema_name=schema_name, - system_prompt=system_prompt, - ) - extra_body = self._json_response_format(json_mode=json_mode, json_schema=json_schema) - - real_provider, real_model, response = self._invoke_sync( - messages=messages, - provider_override=eff_provider, - model_override=eff_model, - profile_override=eff_profile, - temperature=temperature, - max_tokens=max_tokens, - timeout=timeout, - extra_body=extra_body, - task=eff_task, - ) - text = _extract_text(response) - usage = _extract_usage(response) - parsed, content_type = _parse_structured_text( - text=text, json_mode=json_mode, json_schema=json_schema - ) - result = PluginLlmStructuredResult( - text=text, - provider=real_provider, - model=real_model, - agent_id=eff_agent or "default", - usage=usage, - parsed=parsed, - content_type=content_type, - audit={ - "plugin_id": self._plugin_id, - "purpose": purpose or "", - "profile": eff_profile or "", - "schema_name": schema_name or "", - "task": eff_task or "", - }, - ) - logger.info( - "plugin_llm.complete_structured plugin=%s provider=%s model=%s " - "task=%s purpose=%s content_type=%s tokens=%d", - self._plugin_id, real_provider, real_model, eff_task or "", - purpose or "", content_type, usage.total_tokens, - ) - return result + ``input`` accepts text and image blocks. With ``json_mode=True`` or a + ``json_schema`` the response is parsed (and validated when the optional + ``jsonschema`` package is installed) into ``result.parsed``. + ``task`` routes as in :meth:`complete`.""" + spec = _structured_spec("complete_structured", instructions, input, system_prompt, json_mode, json_schema, schema_name) + agent, kw = self._gate(provider, model, agent_id, profile, task, None, temperature, max_tokens, timeout, spec) + return self._finish("complete_structured", agent, kw, self._invoke_sync(kw), purpose, spec) # -- public async API --------------------------------------------------- @@ -924,47 +660,8 @@ class PluginLlm: task: Optional[str] = None, ) -> PluginLlmCompleteResult: """Async sibling of :meth:`complete`.""" - policy = self._policy_loader(self._plugin_id) - eff_task = _check_task(policy, plugin_id=self._plugin_id, requested_task=task) - eff_provider, eff_model, eff_agent, eff_profile = _check_overrides( - policy, - requested_provider=provider, - requested_model=model, - requested_agent_id=agent_id, - requested_profile=profile, - ) - real_provider, real_model, response = await self._invoke_async( - messages=messages, - provider_override=eff_provider, - model_override=eff_model, - profile_override=eff_profile, - temperature=temperature, - max_tokens=max_tokens, - timeout=timeout, - task=eff_task, - ) - text = _extract_text(response) - usage = _extract_usage(response) - result = PluginLlmCompleteResult( - text=text, - provider=real_provider, - model=real_model, - agent_id=eff_agent or "default", - usage=usage, - audit={ - "plugin_id": self._plugin_id, - "purpose": purpose or "", - "profile": eff_profile or "", - "task": eff_task or "", - }, - ) - logger.info( - "plugin_llm.acomplete plugin=%s provider=%s model=%s task=%s " - "purpose=%s tokens=%d", - self._plugin_id, real_provider, real_model, eff_task or "", - purpose or "", usage.total_tokens, - ) - return result + agent, kw = self._gate(provider, model, agent_id, profile, task, messages, temperature, max_tokens, timeout) + return self._finish("acomplete", agent, kw, await self._invoke_async(kw), purpose) async def acomplete_structured( self, @@ -986,30 +683,43 @@ class PluginLlm: task: Optional[str] = None, ) -> PluginLlmStructuredResult: """Async sibling of :meth:`complete_structured`.""" - if not instructions or not instructions.strip(): - raise ValueError("acomplete_structured requires non-empty instructions") - if not input: - raise ValueError("acomplete_structured requires at least one input block") + spec = _structured_spec("acomplete_structured", instructions, input, system_prompt, json_mode, json_schema, schema_name) + agent, kw = self._gate(provider, model, agent_id, profile, task, None, temperature, max_tokens, timeout, spec) + return self._finish("acomplete_structured", agent, kw, await self._invoke_async(kw), purpose, spec) + # -- shared core -------------------------------------------------------- + + def _gate( + self, + provider: Optional[str], + model: Optional[str], + agent_id: Optional[str], + profile: Optional[str], + task: Optional[str], + messages: Optional[List[Dict[str, Any]]], + temperature: Optional[float], + max_tokens: Optional[int], + timeout: Optional[float], + spec: Optional[Dict[str, Any]] = None, + ) -> tuple[Optional[str], Dict[str, Any]]: + """Run the trust gate (task first, then overrides), then — for a + structured ``spec`` — build messages/response_format (input-shape errors + surface only after trust passes). Returns the effective agent id + (result-only) and the call kwargs handed to ``_invoke_*`` / an injected + caller, in the documented order: messages, provider_override, + model_override, profile_override, temperature, max_tokens, timeout, + extra_body, task.""" policy = self._policy_loader(self._plugin_id) eff_task = _check_task(policy, plugin_id=self._plugin_id, requested_task=task) eff_provider, eff_model, eff_agent, eff_profile = _check_overrides( - policy, - requested_provider=provider, - requested_model=model, - requested_agent_id=agent_id, - requested_profile=profile, + policy, requested_provider=provider, requested_model=model, + requested_agent_id=agent_id, requested_profile=profile, ) - messages = _build_structured_messages( - instructions=instructions, - inputs=list(input), - json_mode=json_mode, - json_schema=json_schema, - schema_name=schema_name, - system_prompt=system_prompt, - ) - extra_body = self._json_response_format(json_mode=json_mode, json_schema=json_schema) - real_provider, real_model, response = await self._invoke_async( + extra_body = None + if spec is not None: + messages = _build_structured_messages(**spec) + extra_body = _json_response_format(json_mode=spec["json_mode"], json_schema=spec["json_schema"]) + return eff_agent, dict( messages=messages, provider_override=eff_provider, model_override=eff_model, @@ -1020,163 +730,86 @@ class PluginLlm: extra_body=extra_body, task=eff_task, ) + + def _finish( + self, + name: str, + agent_id: Optional[str], + kw: Dict[str, Any], + invoked: tuple[str, str, Any], + purpose: Optional[str], + spec: Optional[Dict[str, Any]] = None, + ) -> Any: + """Build the result object + audit dict and emit the INFO audit line.""" + real_provider, real_model, response = invoked text = _extract_text(response) usage = _extract_usage(response) - parsed, content_type = _parse_structured_text( - text=text, json_mode=json_mode, json_schema=json_schema - ) - result = PluginLlmStructuredResult( - text=text, - provider=real_provider, - model=real_model, - agent_id=eff_agent or "default", - usage=usage, - parsed=parsed, - content_type=content_type, - audit={ - "plugin_id": self._plugin_id, - "purpose": purpose or "", - "profile": eff_profile or "", - "schema_name": schema_name or "", - "task": eff_task or "", - }, - ) - logger.info( - "plugin_llm.acomplete_structured plugin=%s provider=%s model=%s " - "task=%s purpose=%s content_type=%s tokens=%d", - self._plugin_id, real_provider, real_model, eff_task or "", - purpose or "", content_type, usage.total_tokens, - ) - return result + eff_task = kw["task"] or "" + audit: Dict[str, Any] = {"plugin_id": self._plugin_id, "purpose": purpose or "", "profile": kw["profile_override"] or ""} + fields: Dict[str, Any] = dict(text=text, provider=real_provider, model=real_model, agent_id=agent_id or "default", usage=usage) + fmt = f"plugin_llm.{name} plugin=%s provider=%s model=%s task=%s purpose=%s " + log_args = [self._plugin_id, real_provider, real_model, eff_task, purpose or ""] + cls: Any = PluginLlmCompleteResult + if spec is not None: + parsed, content_type = _parse_structured_text(text=text, json_mode=spec["json_mode"], json_schema=spec["json_schema"]) + audit["schema_name"] = spec["schema_name"] or "" + fields.update(parsed=parsed, content_type=content_type) + fmt += "content_type=%s " + log_args.append(content_type) + cls = PluginLlmStructuredResult + audit["task"] = eff_task + logger.info(fmt + "tokens=%d", *log_args, usage.total_tokens) + return cls(**fields, audit=audit) - # -- internals --------------------------------------------------------- + # -- host invocation --------------------------------------------------- @staticmethod - def _json_response_format( - *, json_mode: bool, json_schema: Optional[Any] - ) -> Optional[Dict[str, Any]]: - """Build the ``extra_body.response_format`` payload for the - provider request. Falls back to ``json_object`` when no schema - is given so providers that ignore json_schema still get a hint.""" - if json_schema is not None: - return { - "response_format": { - "type": "json_schema", - "json_schema": { - "name": "plugin_structured_output", - "schema": json_schema, - "strict": False, - }, - } - } - if json_mode: - return {"response_format": {"type": "json_object"}} - return None + def _host_kwargs(kw: Dict[str, Any]) -> tuple[Dict[str, Any], Optional[Dict[str, str]]]: + """Translate call kwargs into ``call_llm`` kwargs. The auth profile + rides in ``extra_body.metadata.auth_profile``; ``route_info`` is only + requested when routing through a task slot.""" + merged_extra = dict(kw["extra_body"] or {}) + if kw["profile_override"]: + merged_extra.setdefault("metadata", {})["auth_profile"] = kw["profile_override"] + route_info: Optional[Dict[str, str]] = {} if kw["task"] else None + return dict( + task=kw["task"], + provider=kw["provider_override"], + model=kw["model_override"], + messages=kw["messages"], + temperature=kw["temperature"], + max_tokens=kw["max_tokens"], + timeout=kw["timeout"], + extra_body=merged_extra or None, + route_info=route_info, + ), route_info - def _invoke_sync( - self, - *, - messages: List[Dict[str, Any]], - provider_override: Optional[str], - model_override: Optional[str], - profile_override: Optional[str], - temperature: Optional[float], - max_tokens: Optional[int], - timeout: Optional[float], - extra_body: Optional[Dict[str, Any]] = None, - task: Optional[str] = None, - ) -> tuple[str, str, Any]: - """Invoke the host's ``call_llm``. Lazy-imports - ``agent.auxiliary_client`` to avoid circular deps at plugin - discovery time. + @staticmethod + def _attributed(kw: Dict[str, Any], response: Any, route_info: Optional[Dict[str, str]]) -> tuple[str, str, Any]: + provider, model = _resolve_attribution( + provider_override=kw["provider_override"], model_override=kw["model_override"], + response=response, route_info=route_info, + ) + return provider, model, response - ``task`` (already trust-checked by the caller) routes through the - matching ``auxiliary.`` slot; ``None`` keeps the main model. - """ + def _invoke_sync(self, kw: Dict[str, Any]) -> tuple[str, str, Any]: + """Invoke the host's ``call_llm`` (lazy import: circular deps at plugin + discovery) and return ``(provider, model, response)``. ``task`` is + already trust-checked; ``None`` keeps the main model. An injected + ``sync_caller`` replaces the whole path and receives the call kwargs.""" if self._sync_caller is not None: - return self._sync_caller( - messages=messages, - provider_override=provider_override, - model_override=model_override, - profile_override=profile_override, - temperature=temperature, - max_tokens=max_tokens, - timeout=timeout, - extra_body=extra_body, - task=task, - ) + return self._sync_caller(**kw) from agent.auxiliary_client import call_llm - merged_extra = dict(extra_body or {}) - if profile_override: - merged_extra.setdefault("metadata", {})["auth_profile"] = profile_override - route_info: Optional[Dict[str, str]] = {} if task else None - response = call_llm( - task=task, - provider=provider_override, - model=model_override, - messages=messages, - temperature=temperature, - max_tokens=max_tokens, - timeout=timeout, - extra_body=merged_extra or None, - route_info=route_info, - ) - provider, model = _resolve_attribution( - provider_override=provider_override, - model_override=model_override, - response=response, - route_info=route_info, - ) - return provider, model, response + call_kw, route_info = self._host_kwargs(kw) + return self._attributed(kw, call_llm(**call_kw), route_info) - async def _invoke_async( - self, - *, - messages: List[Dict[str, Any]], - provider_override: Optional[str], - model_override: Optional[str], - profile_override: Optional[str], - temperature: Optional[float], - max_tokens: Optional[int], - timeout: Optional[float], - extra_body: Optional[Dict[str, Any]] = None, - task: Optional[str] = None, - ) -> tuple[str, str, Any]: + async def _invoke_async(self, kw: Dict[str, Any]) -> tuple[str, str, Any]: + """Async sibling of :meth:`_invoke_sync` (``async_call_llm`` / ``async_caller``).""" if self._async_caller is not None: - return await self._async_caller( - messages=messages, - provider_override=provider_override, - model_override=model_override, - profile_override=profile_override, - temperature=temperature, - max_tokens=max_tokens, - timeout=timeout, - extra_body=extra_body, - task=task, - ) + return await self._async_caller(**kw) from agent.auxiliary_client import async_call_llm - merged_extra = dict(extra_body or {}) - if profile_override: - merged_extra.setdefault("metadata", {})["auth_profile"] = profile_override - route_info: Optional[Dict[str, str]] = {} if task else None - response = await async_call_llm( - task=task, - provider=provider_override, - model=model_override, - messages=messages, - temperature=temperature, - max_tokens=max_tokens, - timeout=timeout, - extra_body=merged_extra or None, - route_info=route_info, - ) - provider, model = _resolve_attribution( - provider_override=provider_override, - model_override=model_override, - response=response, - route_info=route_info, - ) - return provider, model, response + call_kw, route_info = self._host_kwargs(kw) + return self._attributed(kw, await async_call_llm(**call_kw), route_info) # --------------------------------------------------------------------------- @@ -1191,17 +824,9 @@ def make_plugin_llm_for_test( sync_caller: Optional[Callable[..., Any]] = None, async_caller: Optional[Callable[..., Awaitable[Any]]] = None, ) -> PluginLlm: - """Construct a :class:`PluginLlm` with an injected policy and caller. - - Used by unit tests that don't want to round-trip through config.yaml - or hit a real provider. Not part of the public plugin API. - """ - return PluginLlm( - plugin_id=plugin_id, - policy_loader=lambda _pid: policy, - sync_caller=sync_caller, - async_caller=async_caller, - ) + """:class:`PluginLlm` with an injected policy and caller (no config.yaml, + no provider). Not part of the public plugin API.""" + return PluginLlm(plugin_id=plugin_id, policy_loader=lambda _pid: policy, sync_caller=sync_caller, async_caller=async_caller) __all__ = [ diff --git a/agent/plugin_stream_hooks.py b/agent/plugin_stream_hooks.py index a5c62e1d51..7910fc7491 100644 --- a/agent/plugin_stream_hooks.py +++ b/agent/plugin_stream_hooks.py @@ -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 diff --git a/agent/provider_projection.py b/agent/provider_projection.py index 6e28d9a9df..b3788941e5 100644 --- a/agent/provider_projection.py +++ b/agent/provider_projection.py @@ -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: diff --git a/tests/agent/test_backend_identity.py b/tests/agent/test_backend_identity.py index ca396a194f..4cb6360975 100644 --- a/tests/agent/test_backend_identity.py +++ b/tests/agent/test_backend_identity.py @@ -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