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

Merge simp/r3-06-C into simp/r3-06

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:08:11 -07:00
parent 1b2835b4a1 7c08619130
commit 24aeb15600
10 changed files with 794 additions and 1432 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/moa_loop.py
+529 -829
View File
File diff suppressed because it is too large Load Diff
agent/moa_trace.py
+16 -33
View File
@@ -1,12 +1,9 @@
"""Full MoA turn trace persistence (opt-in via config ``moa.save_traces``).
When enabled, every Mixture-of-Agents turn that runs the reference fan-out (a
cache MISS in ``MoAChatCompletions.create``) appends one JSON line to
``<hermes_home>/moa-traces/<session_id>.jsonl``: the exact messages each
reference received, each reference's full output, and the exact aggregator input
plus its output when available — what every model saw, said, and cost.
Side-channel only: never enters the ``messages`` table, history or replay
Every MoA turn that runs the reference fan-out (a cache MISS in
``MoAChatCompletions.create``) appends one JSON line to
``<hermes_home>/moa-traces/<session_id>.jsonl``: what every model saw, said, and
cost. Side-channel only: never enters the ``messages`` table, history or replay
(references are advisory side-calls whose rows would corrupt role alternation).
Off by default; when off the only overhead is the config read.
"""
@@ -30,9 +27,8 @@ def _traces_enabled_and_dir() -> Optional[Path]:
call (once per cache-MISS turn); ``moa.trace_dir`` overrides the default."""
try:
from hermes_cli.config import load_config
moa_cfg = (load_config() or {}).get("moa") or {}
except Exception: # pragma: no cover - defensive: never break a turn over tracing
except Exception: # pragma: no cover - never break a turn over tracing
return None
if not moa_cfg.get("save_traces"):
return None
@@ -54,14 +50,12 @@ _COST_FIELDS = ("cost_usd", "cost_status", "cost_source")
def _slot_trace(acct: Any, label: str) -> dict[str, Any]:
"""Render one reference's _RefAccounting into a full trace dict, including
the FULL input messages and output (not the truncated display preview)."""
"""One reference's _RefAccounting as a full trace dict, including the FULL
input messages and output (not the truncated display preview)."""
usage = getattr(acct, "usage", None)
return {
"label": label,
**{f: getattr(acct, f, None) for f in _ACCT_FIELDS},
"input_messages": getattr(acct, "messages", None),
"output": getattr(acct, "output", None),
"label": label, **{f: getattr(acct, f, None) for f in _ACCT_FIELDS},
"input_messages": getattr(acct, "messages", None), "output": getattr(acct, "output", None),
"usage": {f: getattr(usage, f, 0) for f in _USAGE_FIELDS} if usage is not None else {},
**{f: getattr(acct, f, None) for f in _COST_FIELDS},
}
@@ -79,16 +73,9 @@ def slot_metrics(acct: Any, label: str, output: Any = None) -> dict[str, Any]:
def save_moa_turn(
*,
session_id: Optional[str],
preset_name: str,
reference_outputs: list[tuple[str, str, Any]],
aggregator_label: str,
aggregator_model: Optional[str],
aggregator_provider: Optional[str],
aggregator_temperature: Any,
aggregator_input_messages: Any,
aggregator_output: Optional[str],
*, session_id: Optional[str], preset_name: str, reference_outputs: list[tuple[str, str, Any]],
aggregator_label: str, aggregator_model: Optional[str], aggregator_provider: Optional[str],
aggregator_temperature: Any, aggregator_input_messages: Any, aggregator_output: Optional[str],
aggregator_streamed: bool,
) -> None:
"""Append one full MoA turn record to the session's trace JSONL, if enabled.
@@ -116,14 +103,10 @@ def save_moa_turn(
"preset": preset_name,
"references": [_slot_trace(acct, label) for label, _text, acct in reference_outputs],
"aggregator": {
"label": aggregator_label,
"model": aggregator_model,
"provider": aggregator_provider,
"temperature": aggregator_temperature,
"input_messages": aggregator_input_messages,
"output": aggregator_output,
"streamed": aggregator_streamed,
"output_location": output_location,
"label": aggregator_label, "model": aggregator_model,
"provider": aggregator_provider, "temperature": aggregator_temperature,
"input_messages": aggregator_input_messages, "output": aggregator_output,
"streamed": aggregator_streamed, "output_location": output_location,
},
}
with path.open("a", encoding="utf-8") as f:
agent/moonshot_schema.py
+30 -46
View File
@@ -1,16 +1,11 @@
"""Helpers for translating OpenAI-style tool schemas to Moonshot's schema subset.
"""Translate OpenAI-style tool schemas to Moonshot's (Kimi) stricter JSON Schema subset.
Moonshot (Kimi) accepts a stricter subset of JSON Schema than OpenAI tool
calling; violations fail with HTTP 400 "tools.function.parameters is not a
valid moonshot flavored json schema". Rules applied here:
1. Every property schema must carry a ``type`` (JSON Schema allows omitting it).
2. With ``anyOf``, ``type`` belongs on the children, never the parent.
3. Enum arrays under scalar types may not contain null / empty string.
4. Every object schema must carry a ``required`` array, even an empty one.
The ``#/definitions/...`` → ``#/$defs/...`` rewrite for draft-07 refs lives in
``tools/mcp_tool._normalize_mcp_input_schema`` so it applies to all providers.
Violations fail with HTTP 400 "tools.function.parameters is not a valid moonshot
flavored json schema". Rules: (1) every property schema carries a ``type``;
(2) with ``anyOf``, ``type`` belongs on the children, never the parent; (3) enum
arrays under scalar types may not contain null / empty string; (4) every object
schema carries a ``required`` array, even an empty one. The ``#/definitions/`` →
``#/$defs/`` rewrite lives in ``tools/mcp_tool`` so it applies to all providers.
"""
from __future__ import annotations
@@ -26,6 +21,10 @@ _SCHEMA_LIST_KEYS = frozenset({"anyOf", "oneOf", "allOf", "prefixItems"})
# Values are a single nested schema (additionalProperties may also be a bool).
_SCHEMA_NODE_KEYS = frozenset({"items", "contains", "not", "additionalProperties", "propertyNames"})
_SCALAR_TYPES = frozenset({"string", "integer", "number", "boolean"})
# bool before int: bool is an int subclass.
_ENUM_SAMPLE_TYPES = ((bool, "boolean"), (int, "integer"), (float, "number"))
def _empty_object_schema() -> Dict[str, Any]:
return {"type": "object", "properties": {}, "required": []}
@@ -42,9 +41,9 @@ def _repair_schema(node: Any) -> Any:
for key, value in node.items():
if key in _SCHEMA_MAP_KEYS and isinstance(value, dict):
repaired[key] = {sub_key: _repair_schema(sub_val) for sub_key, sub_val in value.items()}
elif key in _SCHEMA_LIST_KEYS and isinstance(value, list):
repaired[key] = [_repair_schema(v) for v in value]
elif key in _SCHEMA_NODE_KEYS and isinstance(value, dict):
elif (key in _SCHEMA_LIST_KEYS and isinstance(value, list)) or (
key in _SCHEMA_NODE_KEYS and isinstance(value, dict)
):
repaired[key] = _repair_schema(value)
else:
repaired[key] = value
@@ -60,9 +59,7 @@ def _repair_schema(node: Any) -> Any:
if len(non_null) > 1:
repaired["anyOf"] = non_null
return repaired
merge = {k: v for k, v in repaired.items() if k != "anyOf"}
merge.update(non_null[0])
repaired = merge
repaired = {**{k: v for k, v in repaired.items() if k != "anyOf"}, **non_null[0]}
# Moonshot also rejects the non-standard ``nullable`` keyword.
repaired.pop("nullable", None)
@@ -73,7 +70,7 @@ def _repair_schema(node: Any) -> Any:
repaired = _fill_missing_type(repaired)
# Rule 3: drop null/"" enum values under scalar types; drop an emptied enum.
if isinstance(repaired.get("enum"), list) and repaired.get("type") in {"string", "integer", "number", "boolean"}:
if isinstance(repaired.get("enum"), list) and repaired.get("type") in _SCALAR_TYPES:
cleaned = [v for v in repaired["enum"] if v is not None and v != ""]
if cleaned:
repaired["enum"] = cleaned
@@ -88,9 +85,8 @@ def _repair_schema(node: Any) -> Any:
def _ensure_required_array(node: Dict[str, Any]) -> Dict[str, Any]:
"""Guarantee an object schema carries a ``required`` list (Moonshot 400s
otherwise), pruning names that don't exist in ``properties`` — Moonshot
also rejects dangling names. Mutates and returns ``node``."""
"""Guarantee an object schema carries a ``required`` list, pruning names not in
``properties`` (Moonshot also rejects dangling names). Mutates and returns ``node``."""
props = node.get("properties")
req = node.get("required")
if isinstance(req, list):
@@ -102,7 +98,7 @@ def _ensure_required_array(node: Dict[str, Any]) -> Dict[str, Any]:
def _fill_missing_type(node: Dict[str, Any]) -> Dict[str, Any]:
"""Infer a reasonable ``type`` if this schema node has none.
"""Infer a ``type`` if this schema node has none.
A type list collapses to its first concrete member; otherwise
``properties``/``required``/``additionalProperties`` → object,
@@ -111,10 +107,7 @@ def _fill_missing_type(node: Dict[str, Any]) -> Dict[str, Any]:
"""
node_type = node.get("type")
if isinstance(node_type, list):
concrete = next(
(t for t in node_type if isinstance(t, str) and t not in {"", "null"}),
"string",
)
concrete = next((t for t in node_type if isinstance(t, str) and t not in {"", "null"}), "string")
return {**node, "type": concrete}
if "type" in node and node_type not in {None, ""}:
return node
@@ -124,24 +117,20 @@ def _fill_missing_type(node: Dict[str, Any]) -> Dict[str, Any]:
elif "items" in node or "prefixItems" in node:
inferred = "array"
elif isinstance(node.get("enum"), list) and node["enum"]:
sample = node["enum"][0] # bool before int: bool is an int subclass
scalar_types = ((bool, "boolean"), (int, "integer"), (float, "number"))
inferred = next((t for cls, t in scalar_types if isinstance(sample, cls)), "string")
sample = node["enum"][0]
inferred = next((t for cls, t in _ENUM_SAMPLE_TYPES if isinstance(sample, cls)), "string")
else:
inferred = "string"
return {**node, "type": inferred}
def sanitize_moonshot_tool_parameters(parameters: Any) -> Dict[str, Any]:
"""Return a deep-copied, Moonshot-compatible object schema; input is not mutated."""
"""Deep-copied, Moonshot-compatible object schema; input is not mutated."""
if not isinstance(parameters, dict):
return _empty_object_schema()
repaired = _repair_schema(copy.deepcopy(parameters))
if not isinstance(repaired, dict):
return _empty_object_schema()
# Top-level must be an object schema.
repaired["type"] = "object"
repaired.setdefault("properties", {})
@@ -155,22 +144,17 @@ def sanitize_moonshot_tools(tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]
"""
if not tools:
return tools
sanitized: List[Dict[str, Any]] = []
any_change = False
for tool in tools:
fn = tool.get("function") if isinstance(tool, dict) else None
if not isinstance(fn, dict):
sanitized.append(tool)
continue
params = fn.get("parameters")
repaired = sanitize_moonshot_tool_parameters(params)
if repaired is not params:
any_change = True
sanitized.append({**tool, "function": {**fn, "parameters": repaired}})
else:
sanitized.append(tool)
if isinstance(fn, dict):
params = fn.get("parameters")
repaired = sanitize_moonshot_tool_parameters(params)
if repaired is not params:
any_change = True
tool = {**tool, "function": {**fn, "parameters": repaired}}
sanitized.append(tool)
return sanitized if any_change else tools
agent/nous_rate_guard.py
+15 -56
View File
@@ -1,9 +1,9 @@
"""Cross-session rate limit guard for Nous Portal.
Writes rate limit state to a shared file so all sessions (CLI, gateway,
cron, auxiliary) can check whether Nous Portal is currently rate-limited
before making requests. Without it each 429 fans out into up to 9 calls per
turn (3 SDK retries x 3 Hermes retries), all counted against RPH.
Writes rate limit state to a shared file so all sessions (CLI, gateway, cron,
auxiliary) can check whether Nous Portal is currently rate-limited before making
requests. Without it each 429 fans out into up to 9 calls per turn (3 SDK
retries x 3 Hermes retries), all counted against RPH.
"""
from __future__ import annotations
@@ -12,17 +12,11 @@ import contextlib
import json
import logging
import os
import tempfile
import time
from typing import Any, Mapping, Optional
from utils import atomic_replace
from utils import atomic_write_text
from agent.rate_limit_tracker import (
_BUCKET_TAGS,
_fmt_seconds,
_safe_float,
_safe_int,
has_rate_limit_headers,
lower_headers,
_BUCKET_TAGS, _fmt_seconds, _safe_float, _safe_int, has_rate_limit_headers, lower_headers,
)
logger = logging.getLogger(__name__)
@@ -35,7 +29,7 @@ format_remaining = _fmt_seconds
def _state_path() -> str:
"""Return the path to the Nous rate limit state file."""
"""Path to the Nous rate limit state file."""
try:
from hermes_constants import get_hermes_home
base = get_hermes_home()
@@ -47,11 +41,7 @@ def _state_path() -> str:
def _parse_reset_seconds(headers: Optional[Mapping[str, str]]) -> Optional[float]:
"""Best reset estimate (seconds from now) from hourly, per-minute, then retry-after headers."""
lowered = lower_headers(headers)
for key in (
"x-ratelimit-reset-requests-1h",
"x-ratelimit-reset-requests",
"retry-after",
):
for key in ("x-ratelimit-reset-requests-1h", "x-ratelimit-reset-requests", "retry-after"):
val = _safe_float(lowered.get(key), 0.0)
if val > 0:
return val
@@ -59,9 +49,7 @@ def _parse_reset_seconds(headers: Optional[Mapping[str, str]]) -> Optional[float
def record_nous_rate_limit(
*,
headers: Optional[Mapping[str, str]] = None,
error_context: Optional[dict[str, Any]] = None,
*, headers: Optional[Mapping[str, str]] = None, error_context: Optional[dict[str, Any]] = None,
default_cooldown: float = 300.0,
) -> None:
"""Record that Nous Portal is rate-limited in the shared state file.
@@ -71,44 +59,20 @@ def record_nous_rate_limit(
"""
now = time.time()
reset_at = None
header_seconds = _parse_reset_seconds(headers)
if header_seconds is not None:
reset_at = now + header_seconds
if reset_at is None and isinstance(error_context, dict):
ctx_reset = error_context.get("reset_at")
if isinstance(ctx_reset, (int, float)) and ctx_reset > now:
reset_at = float(ctx_reset)
if reset_at is None:
reset_at = now + default_cooldown
path = _state_path()
state = {"reset_at": reset_at, "recorded_at": now, "reset_seconds": reset_at - now}
try:
state_dir = os.path.dirname(path)
os.makedirs(state_dir, exist_ok=True)
state = {
"reset_at": reset_at,
"recorded_at": now,
"reset_seconds": reset_at - now,
}
fd, tmp_path = tempfile.mkstemp(dir=state_dir, suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(state, f)
atomic_replace(tmp_path, path)
except Exception:
with contextlib.suppress(OSError):
os.unlink(tmp_path)
raise
logger.info(
"Nous rate limit recorded: resets in %.0fs (at %.0f)",
reset_at - now, reset_at,
)
atomic_write_text(_state_path(), json.dumps(state))
logger.info("Nous rate limit recorded: resets in %.0fs (at %.0f)", reset_at - now, reset_at)
except Exception as exc:
logger.debug("Failed to write Nous rate limit state: %s", exc)
@@ -150,9 +114,7 @@ def _is_exhausted(remaining: Optional[int], reset: Optional[float]) -> bool:
def is_genuine_nous_rate_limit(
*,
headers: Optional[Mapping[str, str]] = None,
last_known_state: Optional[Any] = None,
*, headers: Optional[Mapping[str, str]] = None, last_known_state: Optional[Any] = None,
) -> bool:
"""Decide whether a 429 from Nous Portal is a real account rate limit.
@@ -170,11 +132,10 @@ def is_genuine_nous_rate_limit(
def _parse_buckets_from_headers(
headers: Optional[Mapping[str, str]],
) -> dict[str, tuple[Optional[int], Optional[float]]]:
"""Extract (remaining, reset_seconds) per bucket from x-ratelimit-* headers ({} if none)."""
"""(remaining, reset_seconds) per bucket from x-ratelimit-* headers ({} if none)."""
lowered = lower_headers(headers)
if not has_rate_limit_headers(lowered):
return {}
result: dict[str, tuple[Optional[int], Optional[float]]] = {}
for _attr, tag in _BUCKET_TAGS:
remaining = _safe_int(lowered.get(f"x-ratelimit-remaining-{tag}"), None)
@@ -184,9 +145,7 @@ def _parse_buckets_from_headers(
return result
def _has_exhausted_bucket(
buckets: Mapping[str, tuple[Optional[int], Optional[float]]],
) -> bool:
def _has_exhausted_bucket(buckets: Mapping[str, tuple[Optional[int], Optional[float]]]) -> bool:
return any(_is_exhausted(remaining, reset) for remaining, reset in buckets.values())
agent/plugin_llm.py
+138 -316
View File
@@ -1,39 +1,12 @@
"""
Plugin LLM facade — host-owned LLM access for trusted plugins.
==============================================================
"""Plugin LLM facade — host-owned LLM access for trusted plugins (``ctx.llm``).
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``.
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:
my-plugin:
llm:
allow_provider_override: true
allow_model_override: true
allowed_providers: [openrouter, anthropic] # optional
allowed_models: [openai/gpt-4o-mini] # optional
allow_agent_id_override: false
allow_profile_override: false
allow_task_override: false # borrow the host's built-in aux tasks
The gate is fail-closed: a missing config block means "no overrides".
``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 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`.
``complete`` / ``complete_structured`` (text + image inputs, JSON schema validation)
and their async siblings. Provider/model/agent_id/profile are explicit keyword
arguments mirroring the host config shape; the host owns routing, auth, timeouts
and fallback, so the plugin never sees raw tokens or keys. Every override knob is
gated by the per-plugin ``plugins.entries.<id>.llm.allow_*_override`` trust flags
(fail-closed: a missing block means "no overrides"). Backed by
:func:`agent.auxiliary_client.call_llm`.
"""
from __future__ import annotations
@@ -48,9 +21,7 @@ from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, Uni
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Public dataclasses
# ---------------------------------------------------------------------------
# -- public dataclasses -------------------------------------------------------
@dataclass
@@ -63,7 +34,7 @@ class PluginLlmTextInput:
@dataclass
class PluginLlmImageInput:
"""Image block. Provide ``data`` (raw bytes) or ``url`` (http(s)/data: URL).
"""Image block: ``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
@@ -74,16 +45,12 @@ class PluginLlmImageInput:
PluginLlmInput = Union[PluginLlmTextInput, PluginLlmImageInput, Dict[str, Any]]
"""A single structured input block: one of the dataclasses above or a plain dict
of the same shape (``{"type": "text", "text": ...}`` /
``{"type": "image", "data": <bytes>, "mime_type": ..., "file_name": ...}`` /
``{"type": "image", "url": ...}``)."""
"""One structured input block: a dataclass above or a plain dict of the same shape."""
@dataclass
class PluginLlmUsage:
"""Token + cost usage. All fields optional — providers differ on what they
return. ``cost_usd`` is the host's best estimate."""
"""Token + cost usage; every field optional. ``cost_usd`` is the host's best estimate."""
input_tokens: int = 0
output_tokens: int = 0
@@ -109,9 +76,8 @@ class PluginLlmCompleteResult:
class PluginLlmStructuredResult:
"""Result of :meth:`PluginLlm.complete_structured`.
``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"``."""
``parsed`` is set only when JSON output was requested AND the response was
valid JSON; ``content_type`` is then ``"json"``, otherwise ``"text"``."""
text: str
provider: str
@@ -123,9 +89,7 @@ class PluginLlmStructuredResult:
audit: Dict[str, Any] = field(default_factory=dict)
# ---------------------------------------------------------------------------
# Trust gate
# ---------------------------------------------------------------------------
# -- trust gate ---------------------------------------------------------------
@dataclass(frozen=True)
@@ -148,16 +112,12 @@ class _TrustPolicy:
# 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",
"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()
@@ -167,16 +127,12 @@ def _coerce_allowlist(raw: Any) -> tuple[Optional[frozenset], bool]:
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 != "*"}
return frozenset(cleaned), allow_any
return frozenset(item for item in normalized if item and item != "*"), "*" in normalized
def _resolve_trust_policy(plugin_id: str) -> _TrustPolicy:
"""Read ``plugins.entries.<plugin_id>.llm`` from config.yaml.
Missing config → fully restrictive policy. Resolved per call (not cached)
so config edits take effect without restarting the agent."""
"""Read ``plugins.entries.<plugin_id>.llm`` from config.yaml (missing → fully
restrictive). Resolved per call so config edits apply without a restart."""
if not plugin_id:
return _TrustPolicy(plugin_id="")
@@ -195,11 +151,8 @@ def _resolve_trust_policy(plugin_id: str) -> _TrustPolicy:
allowed_models, allow_any_model = _coerce_allowlist(llm_cfg.get("allowed_models"))
allowed_providers, allow_any_provider = _coerce_allowlist(llm_cfg.get("allowed_providers"))
return _TrustPolicy(
plugin_id=plugin_id,
allowed_providers=allowed_providers,
allow_any_provider=allow_any_provider,
allowed_models=allowed_models,
allow_any_model=allow_any_model,
plugin_id=plugin_id, allowed_providers=allowed_providers, allow_any_provider=allow_any_provider,
allowed_models=allowed_models, allow_any_model=allow_any_model,
**{name: bool(llm_cfg.get(name, False)) for name in _OVERRIDE_FLAGS},
)
@@ -209,7 +162,6 @@ class PluginLlmTrustError(PermissionError):
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)."
@@ -217,8 +169,7 @@ def _denied(plugin_id: str, what: str, flag: str) -> PluginLlmTrustError:
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."""
"""Gate a ``provider`` / ``model`` override: trust flag, then optional allowlist."""
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")
@@ -233,8 +184,7 @@ def _gate_ref_override(policy: _TrustPolicy, kind: str, requested: str) -> str:
# 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",
"agent_id": "run completions against a non-default agent id", "profile": "override the auth profile",
}
@@ -242,9 +192,8 @@ 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`."""
"""Gate each override independently, in the order provider, model, agent_id,
profile. Returns ``(provider, model, agent_id, profile)`` (agent_id unstripped)."""
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)):
@@ -255,17 +204,16 @@ def _check_overrides(
def _resolve_task_ownership(plugin_id: str) -> tuple[frozenset, frozenset]:
"""Return ``(owned_keys, builtin_keys)`` for the task trust gate.
"""``(owned_keys, builtin_keys)`` for the task trust gate.
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``."""
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
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")
@@ -274,30 +222,21 @@ def _resolve_task_ownership(plugin_id: str) -> tuple[frozenset, frozenset]:
pass
try:
from hermes_cli.main import _AUX_TASKS
builtin = {k for k, _name, _desc in _AUX_TASKS}
except Exception: # pragma: no cover — main import failure
pass
return frozenset(owned), frozenset(builtin)
def _check_task(
policy: _TrustPolicy,
*,
plugin_id: str,
requested_task: Optional[str],
) -> Optional[str]:
def _check_task(policy: _TrustPolicy, *, plugin_id: str, requested_task: Optional[str]) -> Optional[str]:
"""Validate a plugin's requested auxiliary ``task`` key.
* unset / ``""`` / ``"auto"`` → ``None`` (main-model path).
* a key the plugin registered itself → allowed.
* 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()
unset / ``""`` / ``"auto"`` → ``None`` (main-model path); a key the plugin
registered itself → allowed; a built-in key → only with ``allow_task_override``;
anything else raises + logs. Never silently downgraded to ``auto``: that would
mask the misconfiguration and could route to a main model the user steered
elsewhere on purpose."""
task = (requested_task or "").strip()
if not task or task.lower() == "auto":
return None
@@ -323,9 +262,7 @@ def _check_task(
)
# ---------------------------------------------------------------------------
# Input normalization
# ---------------------------------------------------------------------------
# -- input normalization ------------------------------------------------------
def _normalize_input_block(block: PluginLlmInput) -> Dict[str, Any]:
@@ -370,18 +307,12 @@ def _image_part(norm: Dict[str, Any]) -> Dict[str, Any]:
def _build_structured_messages(
*,
instructions: str,
inputs: Sequence[PluginLlmInput],
json_mode: bool,
json_schema: Optional[Any],
schema_name: Optional[str],
system_prompt: Optional[str],
*, instructions: str, inputs: Sequence[PluginLlmInput], json_mode: bool,
json_schema: Optional[Any], schema_name: Optional[str], system_prompt: Optional[str],
) -> List[Dict[str, Any]]:
"""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."""
"""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] = [system_prompt.strip()] if system_prompt else []
if json_mode or json_schema is not None:
@@ -402,7 +333,6 @@ def _build_structured_messages(
schema_text = str(json_schema)
header = f"{header}\n\nJSON schema:\n{schema_text}"
user_parts: List[Dict[str, Any]] = [{"type": "text", "text": header}]
for block in inputs:
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))
@@ -410,16 +340,14 @@ def _build_structured_messages(
return messages
# ---------------------------------------------------------------------------
# JSON parsing
# ---------------------------------------------------------------------------
# -- JSON parsing / response extraction --------------------------------------
_FENCE_RE = re.compile(r"```(?:json)?\s*(.+?)```", re.DOTALL | re.IGNORECASE)
def _strip_code_fences(text: str) -> str:
"""Return the first fenced code block's body, or the stripped text when unfenced."""
"""The first fenced code block's body, or the stripped text when unfenced."""
match = _FENCE_RE.search(text)
return match.group(1).strip() if match else text.strip()
@@ -427,10 +355,9 @@ def _strip_code_fences(text: str) -> str:
def _parse_structured_text(
*, text: str, json_mode: bool, json_schema: Optional[Any]
) -> tuple[Optional[Any], str]:
"""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."""
"""``(parsed, content_type)``: ``"json"`` when parsing (and schema validation, if
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"
@@ -446,46 +373,41 @@ def _parse_structured_text(
except ImportError:
logger.debug("jsonschema unavailable; skipping schema validation")
except jsonschema.ValidationError as exc: # type: ignore[attr-defined]
raise ValueError(
f"Plugin LLM structured output did not match schema: {exc.message}"
) from exc
raise ValueError(f"Plugin LLM structured output did not match schema: {exc.message}") from exc
return parsed, "json"
# ---------------------------------------------------------------------------
# Response extraction
# ---------------------------------------------------------------------------
def _extract_usage(response: Any) -> PluginLlmUsage:
"""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()
"""Token usage from an OpenAI-shaped response, tolerating provider naming
(``prompt_tokens``/``completion_tokens`` vs ``input_tokens``/``output_tokens``,
``cache_read_input_tokens`` vs ``cache_read_tokens``)."""
raw = getattr(response, "usage", None)
if raw is None:
return usage
return PluginLlmUsage()
def _g(name: str) -> int:
v = getattr(raw, name, None)
if v is None and isinstance(raw, dict):
v = raw.get(name)
try:
return int(v) if v is not None else 0
except (TypeError, ValueError):
return 0
def _g(*names: str) -> int:
for name in names:
v = getattr(raw, name, None)
if v is None and isinstance(raw, dict):
v = raw.get(name)
try:
if v is not None and int(v):
return int(v)
except (TypeError, ValueError):
pass
return 0
usage.input_tokens = _g("prompt_tokens") or _g("input_tokens")
usage.output_tokens = _g("completion_tokens") or _g("output_tokens")
usage.total_tokens = _g("total_tokens") or (usage.input_tokens + usage.output_tokens)
usage.cache_read_tokens = _g("cache_read_input_tokens") or _g("cache_read_tokens")
usage.cache_write_tokens = _g("cache_creation_input_tokens") or _g("cache_write_tokens")
return usage
inp, out = _g("prompt_tokens", "input_tokens"), _g("completion_tokens", "output_tokens")
return PluginLlmUsage(
input_tokens=inp, output_tokens=out, total_tokens=_g("total_tokens") or (inp + out),
cache_read_tokens=_g("cache_read_input_tokens", "cache_read_tokens"),
cache_write_tokens=_g("cache_creation_input_tokens", "cache_write_tokens"),
)
def _extract_text(response: Any) -> str:
"""Pull the assistant text out of an OpenAI-shaped response object."""
"""Assistant text of an OpenAI-shaped response (string or text-part list content)."""
try:
content = getattr(response.choices[0].message, "content", None)
if isinstance(content, str):
@@ -512,18 +434,14 @@ def _main_config_value(reader: str, default: str) -> str:
def _resolve_attribution(
*,
provider_override: Optional[str],
model_override: Optional[str],
response: Any,
*, provider_override: Optional[str], model_override: Optional[str], response: Any,
route_info: Optional[Dict[str, str]] = None,
) -> tuple[str, str]:
"""Decide what to record as ``result.provider`` / ``result.model``.
"""``(provider, model)`` to record on the result.
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"``."""
Provider: route selected by ``auxiliary_client`` > explicit override > current
main provider > ``"auto"``. Model: ``response.model`` (the canonical id that
actually ran) > route > override > current main model > ``"default"``."""
route_info = route_info or {}
provider = route_info.get("provider") or provider_override or _main_config_value("_read_main_provider", "auto")
response_model = getattr(response, "model", None)
@@ -532,14 +450,12 @@ def _resolve_attribution(
return provider, route_info.get("model") or model_override or _main_config_value("_read_main_model", "default")
# ---------------------------------------------------------------------------
# PluginLlm facade
# ---------------------------------------------------------------------------
# -- PluginLlm facade ---------------------------------------------------------
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."""
"""``extra_body.response_format``; 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}}
@@ -552,8 +468,7 @@ 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."""
"""Argument check for the structured methods (runs before the trust gate)."""
if not instructions or not instructions.strip():
raise ValueError(f"{name} requires non-empty instructions")
if not input:
@@ -568,19 +483,13 @@ class PluginLlm:
"""Host-owned LLM access for one trusted plugin.
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."""
``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)."""
def __init__(
self,
*,
plugin_id: str,
policy_loader: Optional[Callable[[str], _TrustPolicy]] = None,
self, *, plugin_id: str, policy_loader: Optional[Callable[[str], _TrustPolicy]] = None,
sync_caller: Optional[Callable[..., Any]] = None,
async_caller: Optional[Callable[..., Awaitable[Any]]] = None,
) -> None:
@@ -589,126 +498,72 @@ class PluginLlm:
self._sync_caller = sync_caller
self._async_caller = async_caller
# -- public sync API ----------------------------------------------------
# -- public API -----------------------------------------------------------
def complete(
self,
messages: List[Dict[str, Any]],
*,
provider: Optional[str] = None,
model: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
timeout: Optional[float] = None,
agent_id: Optional[str] = None,
profile: Optional[str] = None,
purpose: Optional[str] = None,
task: Optional[str] = None,
self, messages: List[Dict[str, Any]], *, provider: Optional[str] = None, model: Optional[str] = None,
temperature: Optional[float] = None, max_tokens: Optional[int] = None,
timeout: Optional[float] = None, agent_id: Optional[str] = None, profile: Optional[str] = None,
purpose: Optional[str] = None, task: Optional[str] = None,
) -> PluginLlmCompleteResult:
"""Run a host-owned chat completion against the user's active model.
``messages`` is the standard OpenAI shape. ``provider``/``model``/
``agent_id``/``profile`` are each gated by
``plugins.entries.<id>.llm.allow_*_override``. ``task`` routes through
a plugin-registered auxiliary slot (see :func:`_check_task`)."""
``provider``/``model``/``agent_id``/``profile`` are each gated by
``plugins.entries.<id>.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,
*,
instructions: str,
input: Sequence[PluginLlmInput],
json_schema: Optional[Any] = None,
json_mode: bool = False,
schema_name: Optional[str] = None,
system_prompt: Optional[str] = None,
provider: Optional[str] = None,
model: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
timeout: Optional[float] = None,
agent_id: Optional[str] = None,
profile: Optional[str] = None,
purpose: Optional[str] = None,
task: Optional[str] = None,
self, *, instructions: str, input: Sequence[PluginLlmInput], json_schema: Optional[Any] = None,
json_mode: bool = False, schema_name: Optional[str] = None, system_prompt: Optional[str] = None,
provider: Optional[str] = None, model: Optional[str] = None, temperature: Optional[float] = None,
max_tokens: Optional[int] = None, timeout: Optional[float] = None, agent_id: Optional[str] = None,
profile: Optional[str] = None, purpose: Optional[str] = None, task: Optional[str] = None,
) -> PluginLlmStructuredResult:
"""Run a bounded host-owned structured completion.
``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`."""
``jsonschema`` package is installed) into ``result.parsed``."""
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 ---------------------------------------------------
async def acomplete(
self,
messages: List[Dict[str, Any]],
*,
provider: Optional[str] = None,
model: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
timeout: Optional[float] = None,
agent_id: Optional[str] = None,
profile: Optional[str] = None,
purpose: Optional[str] = None,
task: Optional[str] = None,
self, messages: List[Dict[str, Any]], *, provider: Optional[str] = None, model: Optional[str] = None,
temperature: Optional[float] = None, max_tokens: Optional[int] = None,
timeout: Optional[float] = None, agent_id: Optional[str] = None, profile: Optional[str] = None,
purpose: Optional[str] = None, task: Optional[str] = None,
) -> PluginLlmCompleteResult:
"""Async sibling of :meth:`complete`."""
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,
*,
instructions: str,
input: Sequence[PluginLlmInput],
json_schema: Optional[Any] = None,
json_mode: bool = False,
schema_name: Optional[str] = None,
system_prompt: Optional[str] = None,
provider: Optional[str] = None,
model: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
timeout: Optional[float] = None,
agent_id: Optional[str] = None,
profile: Optional[str] = None,
purpose: Optional[str] = None,
task: Optional[str] = None,
self, *, instructions: str, input: Sequence[PluginLlmInput], json_schema: Optional[Any] = None,
json_mode: bool = False, schema_name: Optional[str] = None, system_prompt: Optional[str] = None,
provider: Optional[str] = None, model: Optional[str] = None, temperature: Optional[float] = None,
max_tokens: Optional[int] = None, timeout: Optional[float] = None, agent_id: Optional[str] = None,
profile: Optional[str] = None, purpose: Optional[str] = None, task: Optional[str] = None,
) -> PluginLlmStructuredResult:
"""Async sibling of :meth:`complete_structured`."""
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 --------------------------------------------------------
# -- 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,
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."""
"""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 and the call kwargs, 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(
@@ -720,25 +575,14 @@ class PluginLlm:
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,
profile_override=eff_profile,
temperature=temperature,
max_tokens=max_tokens,
timeout=timeout,
extra_body=extra_body,
task=eff_task,
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,
)
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,
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
@@ -761,27 +605,21 @@ class PluginLlm:
logger.info(fmt + "tokens=%d", *log_args, usage.total_tokens)
return cls(**fields, audit=audit)
# -- host invocation ---------------------------------------------------
# -- host invocation ------------------------------------------------------
@staticmethod
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."""
"""Call kwargs → ``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,
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
@staticmethod
@@ -793,10 +631,9 @@ class PluginLlm:
return provider, model, response
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."""
"""Host ``call_llm`` (lazy import: circular deps at plugin discovery) →
``(provider, model, response)``. An injected ``sync_caller`` replaces the
whole path and receives the call kwargs."""
if self._sync_caller is not None:
return self._sync_caller(**kw)
from agent.auxiliary_client import call_llm
@@ -812,31 +649,16 @@ class PluginLlm:
return self._attributed(kw, await async_call_llm(**call_kw), route_info)
# ---------------------------------------------------------------------------
# Test helpers
# ---------------------------------------------------------------------------
def make_plugin_llm_for_test(
*,
plugin_id: str,
policy: _TrustPolicy,
sync_caller: Optional[Callable[..., Any]] = None,
*, plugin_id: str, policy: _TrustPolicy, sync_caller: Optional[Callable[..., Any]] = None,
async_caller: Optional[Callable[..., Awaitable[Any]]] = None,
) -> PluginLlm:
""":class:`PluginLlm` with an injected policy and caller (no config.yaml,
no provider). Not part of the public plugin API."""
""":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__ = [
"PluginLlm",
"PluginLlmTextInput",
"PluginLlmImageInput",
"PluginLlmInput",
"PluginLlmUsage",
"PluginLlmCompleteResult",
"PluginLlmStructuredResult",
"PluginLlmTrustError",
"make_plugin_llm_for_test",
"PluginLlm", "PluginLlmTextInput", "PluginLlmImageInput", "PluginLlmInput", "PluginLlmUsage",
"PluginLlmCompleteResult", "PluginLlmStructuredResult", "PluginLlmTrustError", "make_plugin_llm_for_test",
]
agent/plugin_stream_hooks.py
-2
View File
@@ -77,7 +77,6 @@ def _worker(dispatcher: _ConsumerDispatcher) -> None:
def _registered_callbacks(hook_name: str) -> tuple[Callable[..., Any], ...]:
try:
from hermes_cli import plugins
return plugins.iter_hook_callbacks(hook_name)
except Exception:
logger.debug("plugin stream hook callback lookup failed: %s", hook_name, exc_info=True)
@@ -149,7 +148,6 @@ def stream_reasoning_deltas_enabled() -> bool:
"""Return True only when the user opted plugins into reasoning deltas."""
try:
from hermes_cli import config as config_mod
config = config_mod.load_config()
return bool(config_mod.cfg_get(config, "plugins", "stream_reasoning_deltas", default=False))
except Exception:
agent/provider_base.py
+12 -23
View File
@@ -1,12 +1,10 @@
"""Shared base classes for the pluggable-backend provider ABCs.
Every tool-provider ABC (browser, TTS, image/video gen, transcription, web
search, terminal env) shares the same identity + ``hermes tools`` picker
surface; the previous per-ABC copies of these defaults were byte-identical.
Concrete ABCs subclass :class:`ProviderBase` (or :class:`CatalogProviderBase`
when the backend also exposes a model catalog and is available by default) and
add only their domain methods. Plugins keep subclassing the concrete ABC, so
``isinstance`` checks and abstract-method sets are unchanged.
Every tool-provider ABC shares the same identity + ``hermes tools`` picker surface.
Concrete ABCs subclass :class:`ProviderBase` (or :class:`CatalogProviderBase` when
the backend also exposes a model catalog and is available by default) and add only
their domain methods. Plugins keep subclassing the concrete ABC, so ``isinstance``
checks and abstract-method sets are unchanged.
"""
from __future__ import annotations
@@ -36,16 +34,10 @@ class ProviderBase(abc.ABC):
"""Provider row for the ``hermes tools`` picker.
Shape: ``{"name", "badge", "tag", "env_vars": [{"key", "prompt", "url"}, ...]}``
(browser providers may add ``"post_setup"``). Default: a minimal entry
derived from ``display_name`` with no env vars — override to expose API
key prompts and badges.
(browser providers may add ``"post_setup"``). Override to expose API key
prompts and badges.
"""
return {
"name": self.display_name,
"badge": "",
"tag": "",
"env_vars": [],
}
return {"name": self.display_name, "badge": "", "tag": "", "env_vars": []}
class CatalogProviderBase(ProviderBase):
@@ -59,19 +51,16 @@ class CatalogProviderBase(ProviderBase):
def is_available(self) -> bool:
"""True when this provider can service calls (API key present, SDK importable).
Default True. Must NOT raise and must NOT make network calls — the picker
and ``hermes setup`` call it on every paint.
Must NOT raise and must NOT make network calls — the picker and
``hermes setup`` call it on every paint.
"""
return True
def list_models(self) -> List[Dict[str, Any]]:
"""Model catalog entries (``{"id": ..., "display": ...}`` plus optional
provider-specific keys). Default: empty (no user-selectable models)."""
"""Model catalog entries (``{"id": ..., "display": ...}`` + provider-specific keys)."""
return []
def default_model(self) -> Optional[str]:
"""Id of the first catalog entry, or None when the catalog is empty."""
models = self.list_models()
if models:
return models[0].get("id")
return None
return models[0].get("id") if models else None
agent/provider_media.py
+18 -34
View File
@@ -1,10 +1,10 @@
"""Shared ``$HERMES_HOME/cache/<kind>/`` materialisation helpers for the
image/video generation provider ABCs.
"""``$HERMES_HOME/cache/<kind>/`` materialisation helpers for the image/video
generation provider ABCs.
Several backends (xAI, OpenAI, DeepInfra, FAL) return *ephemeral* delivery URLs
that expire before a downstream consumer (Telegram ``send_photo``, browser
fetch) can resolve them, so providers materialise the bytes locally at
tool-completion time. Filenames are ``<prefix>_<YYYYMMDD_HHMMSS>_<uuid8>.<ext>``.
Several backends return *ephemeral* delivery URLs that expire before a downstream
consumer (Telegram ``send_photo``, browser fetch) can resolve them, so providers
materialise the bytes locally at tool-completion time. Filenames are
``<prefix>_<YYYYMMDD_HHMMSS>_<uuid8>.<ext>``.
"""
from __future__ import annotations
@@ -19,7 +19,6 @@ from typing import Dict, Tuple
def cache_dir(kind: str) -> Path:
"""Return ``$HERMES_HOME/cache/<kind>/``, creating parents as needed."""
from hermes_constants import get_hermes_home
path = get_hermes_home() / "cache" / kind
path.mkdir(parents=True, exist_ok=True)
return path
@@ -44,30 +43,19 @@ def save_b64(kind: str, b64_data: str, *, prefix: str, extension: str) -> Path:
def save_url(
kind: str,
url: str,
*,
prefix: str,
timeout: float,
max_bytes: int,
chunk_size: int,
content_types: Dict[str, str],
url_extensions: Tuple[str, ...],
default_extension: str,
label: str,
empty_error: str,
kind: str, url: str, *, prefix: str, timeout: float, max_bytes: int, chunk_size: int,
content_types: Dict[str, str], url_extensions: Tuple[str, ...], default_extension: str,
label: str, empty_error: str,
) -> Path:
"""Stream-download *url* into the cache with a size cap.
The extension comes from the response ``Content-Type`` (a small explicit
table — never inherit a type that points at HTML/JSON from a degenerate
response), then the URL suffix (some CDNs return
``application/octet-stream``), then *default_extension*. Raises on any
network / HTTP / oversize / empty error so callers can fall back to the bare
URL; a partial file is never left behind.
The extension comes from the response ``Content-Type`` (an explicit table —
never inherit a type pointing at HTML/JSON from a degenerate response), then
the URL suffix (some CDNs return ``application/octet-stream``), then
*default_extension*. Raises on any network / HTTP / oversize / empty error so
callers can fall back to the bare URL; a partial file is never left behind.
"""
import requests
response = requests.get(url, timeout=timeout, stream=True)
response.raise_for_status()
@@ -75,15 +63,11 @@ def save_url(
extension = content_types.get(content_type)
if extension is None:
url_path = url.split("?", 1)[0].lower()
for ext in url_extensions:
if url_path.endswith(f".{ext}"):
extension = "jpg" if ext == "jpeg" else ext
break
if extension is None:
extension = default_extension
extension = next(
("jpg" if ext == "jpeg" else ext for ext in url_extensions if url_path.endswith(f".{ext}")),
default_extension,
)
path = cache_path(kind, prefix, extension)
bytes_written = 0
with path.open("wb") as fh:
for chunk in response.iter_content(chunk_size=chunk_size):
agent/provider_projection.py
+10 -24
View File
@@ -1,23 +1,13 @@
"""Fold an agent-as-provider's own activity back into Hermes' turn state.
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``:
* 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.
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.
Agent providers (ACP CLI shims, the codex app-server) run their own tools, so that
work must never come back as pending ``tool_calls`` (Hermes would re-run it) — but
the self-improvement loop (replays ``messages``) and the skill-review nudge
(``_iters_since_skill`` counter) go blind if it is merely summarised into
``reasoning``. The client hands back ``hermes_projected_messages`` (completed
assistant/tool rows) and ``hermes_provider_tool_iterations`` on the completion
object; this helper applies them append-only via ``append_message`` (timestamped,
persisted). Ordinary OpenAI-compatible clients set neither and are unaffected.
"""
from __future__ import annotations
@@ -32,9 +22,7 @@ logger = logging.getLogger(__name__)
__all__ = ["splice_provider_projection"]
def splice_provider_projection(
agent: Any, response: Any, messages: list[dict[str, Any]]
) -> int:
def splice_provider_projection(agent: Any, response: Any, messages: list[dict[str, Any]]) -> int:
"""Append the provider's projected history rows and tick the nudge counter.
Returns the number of rows spliced. Tolerates absent/garbage attributes so a
@@ -46,9 +34,7 @@ def splice_provider_projection(
append_message(messages, row)
if rows:
logger.debug(
"spliced %d provider-projected transcript row(s) from %s",
len(rows),
getattr(agent, "provider", "?"),
"spliced %d provider-projected transcript row(s) from %s", len(rows), getattr(agent, "provider", "?"),
)
try:
agent/provider_registry.py
+26 -69
View File
@@ -1,14 +1,12 @@
"""Shared engine behind the ``agent.*_registry`` provider registries.
Every pluggable-backend registry (browser, TTS, image/video gen, transcription,
web search, terminal env) has the same shape: a global name->provider map plus
per-profile *scoped* maps (multiplexed gateways), a lock, registration with
re-registration logging, and the snapshot/restore pair that
:mod:`hermes_cli.plugins` uses to unwind a plugin's registrations. Each
``*_registry`` module instantiates one :class:`ProviderRegistry` and re-exports
its bound methods under the historical module-level names via
:meth:`ProviderRegistry.export`, so call sites, ``patch("agent.x_registry.get_provider")``
targets, and the ``_providers`` / ``_scoped_providers`` / ``_lock`` test hooks are unchanged.
Every pluggable-backend registry has the same shape: a global name->provider map
plus per-profile *scoped* maps (multiplexed gateways), a lock, registration with
re-registration logging, and the snapshot/restore pair :mod:`hermes_cli.plugins`
uses to unwind a plugin. Each ``*_registry`` module instantiates one
:class:`ProviderRegistry` and re-exports its bound methods under the historical
module-level names via :meth:`ProviderRegistry.export`, so ``patch("agent.x_registry.get_provider")``
targets and the ``_providers`` / ``_scoped_providers`` / ``_lock`` test hooks are unchanged.
"""
from __future__ import annotations
@@ -22,10 +20,6 @@ from hermes_constants import hermes_home_key
P = TypeVar("P")
def strip_key(name: str) -> str:
return name.strip()
def lower_key(name: str) -> str:
return name.strip().lower()
@@ -33,24 +27,16 @@ def lower_key(name: str) -> str:
class ProviderRegistry(Generic[P]):
"""Global + per-scope provider map with plugin snapshot/restore support.
Args:
label: Human label used in log/error strings (``"Browser"``, ``"TTS"``).
provider_cls: ABC every registered instance must satisfy (TypeError otherwise).
logger: The owning module's logger, so record names stay per-registry.
normalize: Key normalizer — ``strip_key`` or ``lower_key`` (case-insensitive
registries mirror how their dispatcher normalizes the configured name).
builtin_names: Reserved names owned by in-tree implementations; a collision
calls ``on_builtin_collision(key)`` and, if that returns, skips registration.
``normalize`` is ``str.strip`` or ``lower_key`` (case-insensitive registries mirror
how their dispatcher normalizes the configured name). ``builtin_names`` are reserved
for in-tree implementations; a collision calls ``on_builtin_collision(key)`` and, if
that returns, skips registration. ``logger`` is the owning module's so record names
stay per-registry.
"""
def __init__(
self,
*,
label: str,
provider_cls: type,
logger: logging.Logger,
normalize: Callable[[str], str] = strip_key,
builtin_names: FrozenSet[str] = frozenset(),
self, *, label: str, provider_cls: type, logger: logging.Logger,
normalize: Callable[[str], str] = str.strip, builtin_names: FrozenSet[str] = frozenset(),
on_builtin_collision: Optional[Callable[[str], None]] = None,
) -> None:
self.label = label
@@ -67,8 +53,6 @@ class ProviderRegistry(Generic[P]):
# "TTS provider" but "Registered browser provider": acronyms keep their case.
self._log_label = label if label.isupper() else label[0].lower() + label[1:]
# -- internal helpers (caller holds the lock) ---------------------------
def _target(self, scope: Optional[str], *, create: bool) -> Dict[str, P]:
if scope is None:
return self._providers
@@ -82,8 +66,6 @@ class ProviderRegistry(Generic[P]):
else:
self._scoped_generations[scope] = self._scoped_generations.get(scope, 0) + 1
# -- registration -------------------------------------------------------
def register(self, provider: P, *, scope: Optional[str] = None) -> None:
"""Register a provider; same-name re-registration overwrites (hot reload)."""
if not isinstance(provider, self.provider_cls):
@@ -107,17 +89,13 @@ class ProviderRegistry(Generic[P]):
self._bump(scope)
if existing is not None:
self.logger.debug(
f"{self.label} provider '%s' re-registered (was %r)",
key, type(existing).__name__,
f"{self.label} provider '%s' re-registered (was %r)", key, type(existing).__name__,
)
else:
self.logger.debug(
f"Registered {self._log_label} provider '%s' (%s)",
key, type(provider).__name__,
f"Registered {self._log_label} provider '%s' (%s)", key, type(provider).__name__,
)
# -- lookup ---------------------------------------------------------------
def merged(self, scope: Optional[str] = None) -> Dict[str, P]:
"""Global map overlaid with the active profile's scoped map (a copy)."""
with self._lock:
@@ -146,8 +124,6 @@ class ProviderRegistry(Generic[P]):
with self._lock:
return self._generation, self._scoped_generations.get(active_scope, 0)
# -- plugin unload support (hermes_cli.plugins) -----------------------------
def snapshot_registration(self, name: str, *, scope: Optional[str] = None) -> Optional[P]:
"""Exact-slot lookup (no global fallback) used to detect plugin ownership."""
with self._lock:
@@ -180,35 +156,19 @@ class ProviderRegistry(Generic[P]):
self._generation += 1
def export(self, namespace: Dict[str, Any]) -> None:
"""Bind the historical module-level API into a ``*_registry`` module.
Installs ``register_provider``/``list_providers``/``get_provider``/
``snapshot_registration``/``restore_registration``/``registry_generation``/
``_reset_for_tests`` plus the ``_providers``/``_scoped_providers``/``_lock``
test hooks, so ``patch("agent.x_registry.get_provider")`` and direct
``_providers`` manipulation in tests keep working unchanged.
"""
"""Bind the historical module-level API (+ ``_providers``/``_scoped_providers``/
``_lock`` test hooks) into a ``*_registry`` module namespace."""
namespace.update(
_providers=self._providers,
_scoped_providers=self._scoped_providers,
_lock=self._lock,
register_provider=self.register,
list_providers=self.list_providers,
get_provider=self.get_provider,
snapshot_registration=self.snapshot_registration,
_providers=self._providers, _scoped_providers=self._scoped_providers, _lock=self._lock,
register_provider=self.register, list_providers=self.list_providers,
get_provider=self.get_provider, snapshot_registration=self.snapshot_registration,
restore_registration=self.restore_registration,
registry_generation=self.registry_generation,
_reset_for_tests=self.reset_for_tests,
registry_generation=self.registry_generation, _reset_for_tests=self.reset_for_tests,
)
def is_available_safe(
provider: Any,
logger: logging.Logger,
fmt: str,
*,
level: int = logging.DEBUG,
exc_info: bool = False,
provider: Any, logger: logging.Logger, fmt: str, *, level: int = logging.DEBUG, exc_info: bool = False,
) -> bool:
"""``bool(provider.is_available())`` that treats a raising provider as unavailable."""
try:
@@ -224,19 +184,16 @@ def configured_provider_name(section: str, logger: logging.Logger) -> Optional[s
configured: Optional[str] = None
try:
from hermes_cli.config import load_config_readonly
cfg = load_config_readonly()
block = cfg.get(section) if isinstance(cfg, dict) else None
if isinstance(block, dict):
raw = block.get("provider")
if isinstance(raw, str) and raw.strip():
configured = raw.strip()
raw = block.get("provider") if isinstance(block, dict) else None
if isinstance(raw, str) and raw.strip():
configured = raw.strip()
except Exception as exc:
logger.debug("Could not read %s.provider from config: %s", section, exc)
if configured:
try:
from tools.tool_backend_helpers import NOUS_MANAGED_PROVIDER
if configured.lower() == NOUS_MANAGED_PROVIDER:
configured = "fal"
except Exception: # pragma: no cover — helpers are in-repo