agent/moa_loop.py
Rama main · réplica de lectura del repositorio original.
"""Mixture-of-Agents runtime helpers for /moa turns.
The slash command marks one user turn as MoA-enabled; the normal agent loop still
owns tool calling and turn termination, while this module gathers reference-model
context before each model iteration.
"""
from __future__ import annotations
import contextlib
import functools
import hashlib
import json
import logging
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor, wait as _futures_wait
from dataclasses import KW_ONLY, dataclass, replace
from types import SimpleNamespace
from typing import Any
from agent.auxiliary_client import call_llm
from agent.message_content import flatten_message_text
from agent.message_metadata import DATA_USE
from agent.moa_alternation import destination_key, is_role_alternation_rejection, merge_same_role_messages
from agent.transports import get_transport
from agent.usage_pricing import CanonicalUsage
logger = logging.getLogger(__name__)
# Privacy filter (moa.privacy_filter: '' | display | full): PII classes agent.redact
# leaves alone. The phone pattern requires explicit delimiters so line numbers,
# dates, times, SHAs, IPs and versions never match.
# Advisor (reference) outputs can echo PII from the conversation — emails, phone numbers, credentials pasted
# by the user — into surfaces the user may not expect: the labelled reference blocks rendered in the UI,
# saved MoA trace files, and (in `full` mode) the guidance block injected into the aggregator prompt (issue
# #59959). Secret/credential shapes (API-key prefixes, JWTs, private keys, DB connection strings, E.164
# phone numbers) are handled by the repo's central redactor, ``agent.redact .redact_sensitive_text`` — the
# MoA filter never re-implements those. The two patterns below cover the PII classes the central redactor
# deliberately leaves alone for log/tool output (emails and formatted phone numbers). Pattern safety:
# advisory text is frequently code-review-shaped — line numbers, timestamps, git SHAs, IDs, IP addresses. A
# bare 10-digit match would mangle all of those, so the phone pattern requires clearly delimited formatting:
# a parenthesized area code and/or explicit `-`/`.` separators between groups ((555) 123-4567, 555-123-4567,
# 555.123.4567, +1 555-123-4567). Undelimited digit runs (5551234567), dates (2026-07-12), times (12:34:56),
# hex IDs, and dotted quads never match. International numbers in E.164 form (+14155551234) are already
# masked by the central redactor.
_MOA_EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")
_MOA_PHONE_RE = re.compile(
r"(?<![\w.+-])" # no leading word char / dot / + / - (kills IPs, IDs, versions)
r"(?:\+?1[ .-])?" # optional NA country code
r"(?:\(\d{3}\)[ .-]?|\d{3}[.-])" # delimited area code: (555) or 555- / 555.
r"\d{3}[.-]\d{4}" # exchange-subscriber with explicit separator
r"(?![\w-])" # no trailing word char / hyphen
)
def _redact_reference_text(text: Any) -> Any:
"""Redact secrets (central redactor) then MoA PII patterns.
force=True: the privacy filter is its own opt-in, independent of the global
log-redaction toggle. code_file=True: keeps the ENV/JSON assignment heuristics
(which mangle source snippets) off advisory prose/code.
"""
if not isinstance(text, str) or not text:
return text
from agent.redact import redact_sensitive_text
text = redact_sensitive_text(text, force=True, code_file=True)
text = _MOA_EMAIL_RE.sub("[redacted email]", text)
return _MOA_PHONE_RE.sub("[redacted phone]", text)
def _moa_privacy_mode(moa_raw: Any) -> str:
"""Normalized privacy-filter mode from a raw ``moa`` config."""
from hermes_cli.moa_config import coerce_privacy_filter
raw = moa_raw if isinstance(moa_raw, dict) else {}
return coerce_privacy_filter(raw.get("privacy_filter"))
def _redact_reference_outputs(reference_outputs: list[tuple[str, str, Any]]) -> list[tuple[str, str, Any]]:
"""Redact advisor text in reference-output tuples; accounting slot untouched."""
return [(label, _redact_reference_text(text), acct) for label, text, acct in reference_outputs]
def _redact_message_content(content: Any) -> Any:
"""Redact a message's content: a string, or the text parts of a content-part list."""
if isinstance(content, str):
return _redact_reference_text(content)
if isinstance(content, list):
return [
{**p, "text": _redact_reference_text(p.get("text"))}
if isinstance(p, dict) and isinstance(p.get("text"), str)
else p
for p in content
]
return content
def _redact_trace_messages(messages: Any) -> Any:
"""Redact message copies for trace persistence (string or content-part lists)."""
if not isinstance(messages, list):
return messages
return [
{**m, "content": _redact_message_content(m.get("content"))} if isinstance(m, dict) else m
for m in messages
]
def _redact_trace_accounting(acct: Any) -> Any:
"""Copy a ``_RefAccounting`` with its trace text (messages/output) redacted."""
if not isinstance(acct, _RefAccounting):
return acct
return replace(acct, messages=_redact_trace_messages(acct.messages), output=_redact_reference_text(acct.output))
# Cold-start caches: preset and per-(provider, model) runtime are immutable for a turn.
# A MoA preset switch used to re-resolve the full config + preset + every slot's provider runtime on EACH
# create() call (once per tool-loop iteration), serially before the parallel fan-out could start — adding
# 5-30s of "frozen" latency on complex presets (#66793).
_preset_cache_lock = threading.Lock()
_preset_cache: dict[tuple, Any] = {}
def _resolve_preset_cached(preset_name: str) -> tuple[dict[str, Any], Any]:
"""``(preset, raw moa config)``; the resolved preset is cached per config file signature
(skips resolve_moa_preset's full validation of the moa block on every create())."""
from hermes_cli.config import get_config_path, load_config
from hermes_cli.moa_config import resolve_moa_preset
from hermes_cli.config_backend import config_cache_version_or_none
from agent.safe_worker_policy import worker_config_snapshot
# A safe worker's config is its frozen snapshot: never open the profile file (uncached there).
cfg_stamp = None if worker_config_snapshot() is not None else config_cache_version_or_none(get_config_path())
moa_raw = load_config().get("moa") or {}
key = (cfg_stamp, preset_name)
with _preset_cache_lock:
preset = _preset_cache.get(key) if cfg_stamp is not None else None
if preset is None:
preset = resolve_moa_preset(moa_raw, preset_name)
if cfg_stamp is not None:
with _preset_cache_lock:
_preset_cache.clear() # one live config stamp at a time
_preset_cache[key] = preset
return preset, moa_raw
_runtime_cache_lock = threading.Lock()
_runtime_cache: dict[tuple[str, str, str], tuple[float, dict[str, Any]]] = {}
# Short TTL so rotated keys / base_url edits are picked up within 5 minutes.
_RUNTIME_CACHE_TTL_SECONDS = 300.0
# Cap on concurrent reference calls (guards pathologically large presets).
_MAX_REFERENCE_WORKERS = 8
@dataclass(slots=True)
class _RefAccounting:
"""Per-reference usage, cost and full trace (third slot of a reference-output tuple).
Cost is priced at the advisor's OWN rate and summed in dollars (advisors may run
on a different model than the aggregator). Trace fields are only populated when
tracing is on.
"""
usage: Any
cost_usd: Any = None
cost_status: str | None = None
cost_source: str | None = None
_: KW_ONLY
messages: Any = None
output: str | None = None
model: str | None = None
provider: str | None = None
temperature: Any = None
# Per-tool-result char budget for the advisory view: tool CALLS are kept in full,
# tool RESULTS are head+tail previewed. The aggregator always gets the full transcript.
_REFERENCE_TOOL_RESULT_BUDGET = 4000
# Reference system prompt: without this framing a reference assumes it is the acting
# agent and refuses ("I can't access repositories") or tries to call tools.
_REFERENCE_SYSTEM_PROMPT = (
"You are a reference advisor in a Mixture of Agents (MoA) process. You are "
"NOT the acting agent and you do NOT execute anything: you cannot call "
"tools, run commands, browse, or access files, repositories, or URLs, and "
"you should not try to or apologize for being unable to. A separate "
"aggregator/orchestrator model holds those capabilities and will take the "
"actual actions.\n\n"
"CRITICAL: You must NEVER claim or imply that you have executed a command, "
"downloaded a file, accessed a URL, or performed any action. You can only "
"analyze and advise based on the conversation context. Examples of what to "
"avoid:\n"
"- Bad: \"I ran curl and got 404.\"\n"
"- Bad: \"I downloaded the file successfully.\"\n"
"- Bad: \"I checked the repository and found...\"\n"
"- Good: \"Based on the error pattern, a curl request to that URL would likely return 404.\"\n"
"- Good: \"The conversation suggests downloading this file may help.\"\n"
"- Good: \"From the context, checking the repository would reveal...\"\n\n"
"The conversation below is the current state of a task handled by that "
"acting agent. Your job is to give your most intelligent analysis of that "
"state: understand the goal, reason about the problem, and advise on what "
"to do next. Surface the best approach, concrete next steps and tool-use "
"strategy, likely pitfalls and risks, and anything the acting agent may "
"have missed or gotten wrong. Assume any referenced files, URLs, or "
"systems exist and reason about them from the context given rather than "
"asking for access.\n\n"
"Respond with your advice directly — no preamble, no disclaimers about "
"tools or access. Advise in prose: never emit a tool call or a JSON "
"tool-call object, because the aggregator replays what looks like one. "
"Your response is private guidance handed to the "
"aggregator, not an answer shown to the user. NEVER claim to have executed "
"anything."
)
def _slot_label(slot: dict[str, Any]) -> str:
label = f"{(slot.get('provider') or '').strip()}:{(slot.get('model') or '').strip()}"
effort = str(slot.get("reasoning_effort") or "").strip()
return f"{label}[reasoning={effort}]" if effort else label
def _slot_reasoning_config(slot: dict[str, Any]) -> dict[str, Any] | None:
"""Translate optional per-MoA-slot reasoning_effort into runtime config."""
try:
from hermes_constants import parse_reasoning_effort
return parse_reasoning_effort(slot.get("reasoning_effort"))
except Exception: # pragma: no cover - bad config must not break MoA
return None
def _aggregator_reasoning_config(aggregator: dict[str, Any]) -> dict[str, Any] | None:
"""Aggregator reasoning config: slot > per-model > global (shared chokepoint).
References deliberately do NOT fall back: inheriting a global ``xhigh`` into
every advisor would multiply cost.
"""
cfg = _slot_reasoning_config(aggregator)
if cfg is not None:
return cfg
try:
from hermes_cli.config import load_config
from hermes_constants import resolve_reasoning_config
return resolve_reasoning_config(load_config() or {}, str(aggregator.get("model") or ""))
except Exception: # pragma: no cover - bad config must not break MoA
return None
def _slot_runtime(slot: dict[str, Any]) -> dict[str, Any]:
"""Slot → ``call_llm`` kwargs with the provider's real api_mode/base_url/api_key.
Cached per (profile home, provider, model) with a short TTL. Falls back to bare provider/model
on error — never cached, or a transient error would pin bare kwargs for a TTL.
"""
provider = str(slot.get("provider") or "").strip()
model = str(slot.get("model") or "").strip()
# hermes_home_key() in the key: the resolved api_key/base_url are per-profile, and under a
# multiplex gateway two profiles can share (provider, model) with different accounts.
from hermes_constants import hermes_home_key
cache_key = (hermes_home_key(), provider, model)
now = time.monotonic()
with _runtime_cache_lock:
entry = _runtime_cache.get(cache_key)
if entry is not None and now - entry[0] < _RUNTIME_CACHE_TTL_SECONDS:
return entry[1]
out: dict[str, Any] = {"provider": provider, "model": model}
try:
from hermes_cli.runtime_provider import resolve_runtime_provider
rt = resolve_runtime_provider(requested=provider, target_model=model)
out.update({k: rt[k] for k in ("base_url", "api_key", "api_mode") if rt.get(k)})
overrides = rt.get("request_overrides")
extra_body = overrides.get("extra_body") if isinstance(overrides, dict) else None
if isinstance(extra_body, dict) and extra_body:
out["extra_body"] = dict(extra_body)
except Exception as exc:
logger.warning("MoA slot %s: provider '%s' could not be resolved (%s); calling with bare provider/model",
_slot_label(slot), provider, exc)
return out
with _runtime_cache_lock:
_runtime_cache[cache_key] = (now, out)
return out
def _merge_slot_extra_body(slot_extra_body: Any, caller_extra_body: Any) -> Any:
"""Merge slot defaults with a caller override (caller wins) for ``call_llm``."""
if not (isinstance(slot_extra_body, dict) and slot_extra_body):
return caller_extra_body
if isinstance(caller_extra_body, dict):
return {**slot_extra_body, **caller_extra_body}
return caller_extra_body or dict(slot_extra_body)
def _agent_cache_opts(agent: Any) -> tuple[Any, Any]:
"""The live agent's ``(_cache_disabled, _cache_ttl)``; ``(None, None)`` without an agent."""
return (None, None) if agent is None else (getattr(agent, "_cache_disabled", None), getattr(agent, "_cache_ttl", None))
def _with_cache_disabled(runtime: dict[str, Any], cache_disabled: Any) -> dict[str, Any]:
"""Pin the live agent's cache disable onto a runtime snapshot (None is a no-op)."""
return runtime if cache_disabled is None else {**runtime, "_cache_disabled": cache_disabled}
def _reference_liveness_hook(agent: Any, label: str) -> Any:
"""Thread-local aux progress hook bridging advisor stream chunks to the agent's
activity clock.
The MoA fan-out runs in worker threads outside the main agent loop, so without this
bridge nothing stamps the turn's activity clock while advisors stream — the
turn-liveness watchdog then reads the clock frozen at the last main-loop stamp and
force-aborts healthy, still-streaming advisor turns at ``agent.turn_liveness.timeout_s``
(default 600s) even though the aux stream layer deliberately permits
``max(600, 4 × timeout)``. ``call_llm`` preserves a thread-local hook across its
dispatch/protected-daemon paths and ticks it per substantive stream chunk, so each
chunk re-arms the watchdog. A failed touch must never break the advisor call.
"""
touch = getattr(agent, "_touch_activity", None)
def _touch_on_progress() -> None:
if callable(touch):
touch(f"MoA reference {label}: stream progress")
return _touch_on_progress
def _touch_fanout_progress(agent: Any, description: str) -> None:
"""Best-effort activity stamp for fan-out milestones; never raises (display of
progress must not decide turn liveness on its own, it only supports it)."""
touch = getattr(agent, "_touch_activity", None)
if callable(touch):
with contextlib.suppress(Exception):
touch(description)
def _maybe_apply_moa_cache_control(
messages: list[dict[str, Any]], runtime: dict[str, Any], *, cache_disabled: bool | None = None,
cache_ttl: str | None = None,
) -> list[dict[str, Any]]:
"""Apply cache_control to an advisor/aggregator request when its route honors it.
Same policy/marker helpers as the main loop; MoA has no static prefix so the
legacy system-and-3 fallback is used. ``cache_disabled`` is stamped onto the
stub so ``cache_ttl: off`` is honored; ``cache_ttl`` is clamped per destination.
Returns the messages unchanged on any error.
``cache_disabled`` (or the live config when omitted) is stamped onto the policy stub so
``prompt_caching.cache_ttl: off`` is not bypassed by the blank-agent pattern (#76085).
"""
try:
from agent.agent_runtime_helpers import anthropic_prompt_cache_policy, blank_cache_policy_stub
from agent.prompt_caching import (
apply_anthropic_cache_control, effective_cache_ttl, envelope_tool_part_cache_markers_supported,
)
# Explicit kwarg > runtime snapshot (threaded from the live agent) > config.
if cache_disabled is None and "_cache_disabled" in runtime:
cache_disabled = runtime.get("_cache_disabled")
provider, model, base_url = runtime.get("provider") or "", runtime.get("model") or "", runtime.get("base_url") or ""
# blank_cache_policy_stub is the only sanctioned stub (carries _cache_disabled).
should_cache, native_layout = anthropic_prompt_cache_policy(
blank_cache_policy_stub(cache_disabled), provider=provider, base_url=base_url,
api_mode=runtime.get("api_mode") or "", model=model,
)
if not should_cache:
return messages
# None → "5m"; cache-disabled routes already returned above. Envelope routes
# reject part-level markers in tool_result.content[].
return apply_anthropic_cache_control(
messages, cache_ttl=effective_cache_ttl(cache_ttl, provider=provider, model=model),
native_anthropic=native_layout,
tool_part_markers=envelope_tool_part_cache_markers_supported(provider, base_url),
)
except Exception as exc: # pragma: no cover - decoration must never break a call
logger.debug("MoA cache_control decoration skipped: %s", exc)
return messages
def _price_reference_response(
response: Any, slot: dict[str, Any], runtime: dict[str, Any]
) -> tuple[Any, Any, str | None, str | None]:
"""Normalize a reference's usage with the slot's OWN provider/api_mode and price it
at its own rate (hence fan-out cost is summed in dollars). Never raises."""
from agent.usage_pricing import estimate_usage_cost, normalize_usage
usage = CanonicalUsage()
raw_usage = getattr(response, "usage", None)
if raw_usage:
with contextlib.suppress(Exception): # pragma: no cover - defensive
usage = normalize_usage(raw_usage, provider=runtime.get("provider"), api_mode=runtime.get("api_mode"))
try:
cost = estimate_usage_cost(
slot.get("model") or "", usage, provider=runtime.get("provider"),
base_url=runtime.get("base_url"), api_key=runtime.get("api_key"),
)
return usage, cost.amount_usd, cost.status, cost.source
except Exception: # pragma: no cover - defensive
return usage, None, None, None
def _slot_request_headers(runtime: dict[str, Any], agent: Any = None) -> dict[str, str] | None:
"""User-initiated request headers of the slot's provider, attributed at that provider's own endpoint
(MoA attributes by slot provider, whatever account host the slot resolved to). Asked like every
per-request hook (``route_answer``: memoized, one failure policy for the agent's session)."""
from agent.auxiliary_client import _normalize_aux_provider
from agent.provider_route_hooks import route_answer
from providers import get_provider_profile
provider = _normalize_aux_provider(str(runtime.get("provider") or ""))
profile = get_provider_profile(provider) if provider else None
if profile is None:
return None
return route_answer("request_headers", provider=provider, base_url=profile.base_url or "",
model=str(runtime.get("model") or ""), user_initiated=True, scope=agent)
def _run_reference(
slot: dict[str, Any], ref_messages: list[dict[str, Any]], *, temperature: float | None = None,
max_tokens: int | None = None, reference_timeout: float | None = None, context_length_cache: Any = None,
cache_disabled: bool | None = None, cache_ttl: str | None = None, agent: Any = None,
) -> tuple[str, str, Any]:
"""Call one reference model; return ``(label, text, accounting)``. Never raises:
a failed reference becomes a labelled ``[failed: …]`` note. Runs in a thread pool."""
label = _slot_label(slot)
runtime = _slot_runtime(slot)
trace_fields = {"model": slot.get("model"), "provider": runtime.get("provider") or slot.get("provider"), "temperature": temperature}
# The advisory view already stripped the agent's system prompt; this is the only one.
messages = [{"role": "system", "content": _REFERENCE_SYSTEM_PROMPT}, *ref_messages]
try:
# Trim to THIS model's window (advisors may be smaller than the aggregator); the
# advisory view is append-only across iterations, so cache_control lets
# iteration N+1 replay N's cached prefix.
# Reference models may have a smaller window than the aggregator (e.g. kimi-k2.7-code @ 262K
# advising a glm-5.2 @ 1M conversation); without this trim the provider returns a hard HTTP 400
# which the except below silently converts to a [failed: …] note (issue #60345). Estimated AFTER the
# advisory system prompt is prepended so its tokens count against the budget too.
trimmed = _trim_messages_for_reference(
messages, slot, runtime, reserve_output_tokens=max_tokens, context_length_cache=context_length_cache,
)
trimmed = _maybe_apply_moa_cache_control(trimmed, _with_cache_disabled(runtime, cache_disabled), cache_ttl=cache_ttl)
# Providers may gate premium models on request attribution; MoA fan-out serves the
# user's current turn, so mirror the main agent's user-initiated request headers.
initiator_headers = _slot_request_headers(runtime, agent)
if agent is not None:
from agent.auxiliary_client import aux_progress_hook
with aux_progress_hook(_reference_liveness_hook(agent, label)):
response = call_llm(
task="moa_reference", messages=trimmed, temperature=temperature,
max_tokens=max_tokens,
timeout=reference_timeout, reasoning_config=_slot_reasoning_config(slot),
extra_headers=initiator_headers, **runtime,
)
else:
response = call_llm(
task="moa_reference", messages=trimmed, temperature=temperature,
max_tokens=max_tokens,
timeout=reference_timeout, reasoning_config=_slot_reasoning_config(slot),
extra_headers=initiator_headers, **runtime,
)
output_text = _extract_text(response) or "(empty response)"
acct = _RefAccounting(*_price_reference_response(response, slot, runtime), messages=trimmed, output=output_text, **trace_fields)
return label, output_text, acct
except Exception as exc:
logger.warning("MoA reference model %s failed: %s", label, exc)
note = f"[failed: {exc}]"
return label, note, _RefAccounting(CanonicalUsage(), messages=messages, output=note, **trace_fields)
# Output headroom reserved in the reference window when reference_max_tokens is unset.
_REFERENCE_DEFAULT_OUTPUT_RESERVE = 8192
# Estimator slack: estimate_messages_tokens_rough is a rough chars/4 heuristic.
_REFERENCE_TRIM_SAFETY_FRACTION = 0.10
def _reference_context_length(slot: dict[str, Any], runtime: dict[str, Any], cache: Any) -> int | None:
"""Context window for a slot, memoized in ``cache`` per (provider, model) when given.
Failures are cached too so a flaky metadata source is not re-probed per reference.
"""
from agent.model_metadata import get_model_context_length
model = str(slot.get("model") or "")
provider = str(runtime.get("provider") or slot.get("provider") or "")
key = (provider, model)
has_cache = isinstance(cache, dict)
if has_cache and key in cache:
return cache[key]
try:
context_length = get_model_context_length(
model=model, base_url=str(runtime.get("base_url") or ""), api_key=str(runtime.get("api_key") or ""), provider=provider,
)
except Exception:
logger.debug("MoA reference context-length resolution failed for %s", _slot_label(slot))
context_length = None
if has_cache:
cache[key] = context_length
return context_length
def _trim_messages_for_reference(
messages: list[dict[str, Any]], slot: dict[str, str], runtime: dict[str, Any], *,
reserve_output_tokens: int | None = None, context_length_cache: Any = None,
) -> list[dict[str, Any]]:
"""Trim an advisory request to fit a reference model's context window.
Budget = window − ``reserve_output_tokens`` (or a default) − a safety fraction.
Drops the OLDEST frames after the system prompt, always keeping a user-first
body and the trailing user turn plus one preceding turn (even if still over
budget). ``context_length_cache`` memoizes the window per (provider, model);
unresolvable windows leave messages unchanged.
Reference models may have a smaller context window than the aggregator or the main conversation. Without
this trim, a reference whose window is exceeded gets a hard HTTP 400 from the provider, which
``_run_reference``'s try/except silently converts to a ``[failed: …]`` note — the MoA turn silently
degrades to fewer references (issue #60345).
"""
if not messages or not slot.get("model"):
return messages
from agent.model_metadata import estimate_messages_tokens_rough
context_length = _reference_context_length(slot, runtime, context_length_cache)
if not isinstance(context_length, int) or context_length <= 0:
return messages
reserve = reserve_output_tokens if isinstance(reserve_output_tokens, int) and reserve_output_tokens > 0 else _REFERENCE_DEFAULT_OUTPUT_RESERVE
budget = int(context_length * (1.0 - _REFERENCE_TRIM_SAFETY_FRACTION)) - reserve
estimated = estimate_messages_tokens_rough(messages)
if budget <= 0 or estimated <= budget:
return messages
has_system = bool(messages) and messages[0].get("role") == "system"
head_count = 1 if has_system else 0
# estimate_messages_tokens_rough is a pure sum of per-message weights,
# so weigh each message once and track a running total while popping
# instead of re-estimating head + body after every pop. The naive form
# paid one full memo walk per dropped frame (quadratic on long
# histories). The pop sequence is unchanged, so the trim result is
# identical to the naive loop.
weights = [estimate_messages_tokens_rough([m]) for m in messages]
total = sum(weights)
# Keep the trailing user turn plus at least one preceding turn.
start = head_count
remaining = len(messages) - head_count
while remaining > 2 and total > budget:
total -= weights[start]
start += 1
remaining -= 1
# Preserve the user-first invariant: never leave the advisory
# conversation starting on an assistant turn after a pop.
while remaining > 2 and messages[start].get("role") == "assistant":
total -= weights[start]
start += 1
remaining -= 1
# The loop can stop with two frames left where the first is an
# assistant turn — enforce user-first even then (a lone trailing user
# turn is a valid request; an assistant-first one is not).
while remaining > 1 and messages[start].get("role") == "assistant":
total -= weights[start]
start += 1
remaining -= 1
trimmed = messages[:head_count] + messages[start:]
dropped = len(messages) - len(trimmed)
if dropped:
logger.info(
"MoA reference %s: estimated %d tokens exceeds budget %d "
"(window %d, output reserve %d); dropped %d oldest message(s).",
_slot_label(slot), estimated, budget, context_length, reserve, dropped,
)
return trimmed
_REFERENCE_POLL_INTERVAL_S = 5.0
_REFERENCE_INTERRUPT_SETTLE_S = 0.05
# Sentinel for a reference aborted by user interrupt; the facade must never cache it.
_INTERRUPTED_REFERENCE_NOTE = "[skipped: interrupted by user]"
def _placeholder_output(slot: dict[str, Any], note: str) -> tuple[str, str, Any]:
"""A reference-output tuple for a slot that was not (fully) run: zero accounting."""
return _slot_label(slot), note, _RefAccounting(CanonicalUsage())
def _settle_interrupted(
futures: dict[Any, int], results: list, reference_models: list[dict[str, Any]], late_accounting_sink: Any,
) -> None:
"""Fill every unfinished slot after a user interrupt: cancel never-dispatched
futures (nothing billed), keep real output of ones that just finished, and hand
running ones (cannot be killed, WILL bill) to ``late_accounting_sink``."""
for future, idx in futures.items():
if results[idx] is not None:
continue
# #38922: a slow confirmation does NOT necessarily mean the send failed — but we must distinguish
# two cases via future.cancel()'s return value: cancel() == False -> the coroutine was already
# running on the gateway loop when the timeout fired; the request is in flight on the wire and
# cannot be un-sent. Re-sending via standalone would be a guaranteed DUPLICATE, so treat it as
# delivered (assume-delivered). cancel() == True -> the scheduled callback never started executing
# (loop wedged/backlogged for the full 60s), so nothing was sent. We MUST fall through to the
# standalone path or the message is silently dropped (worse than a duplicate).
cancelled = future.cancel()
if not cancelled and future.done():
results[idx] = future.result()
continue
results[idx] = _placeholder_output(reference_models[idx], _INTERRUPTED_REFERENCE_NOTE)
if not cancelled and late_accounting_sink is not None:
future.add_done_callback(functools.partial(_record_late, late_accounting_sink, results[idx][0]))
def _record_late(sink: Any, label: str, future: Any) -> None:
"""Done-callback for an abandoned reference future: forward its real accounting."""
try:
sink(label, future.result()[2])
except Exception: # pragma: no cover - defensive
logger.debug("MoA: late accounting sink failed for %s", label)
def _run_references_parallel(
reference_models: list[dict[str, Any]], ref_messages: list[dict[str, Any]], *,
temperature: float | None = None, max_tokens: int | None = None, progress_callback: Any = None,
reference_timeout: float | None = None, agent: Any = None, late_accounting_sink: Any = None,
) -> list[tuple[str, str, Any]]:
"""Fan out all reference models in parallel; ``(label, text, _RefAccounting)`` per
slot in ``reference_models`` order.
``provider == "moa"`` slots are skipped with a note (recursion guard).
``progress_callback(refs_done, refs_total, label)`` fires per completion. With
*agent*, the wait polls every ``_REFERENCE_POLL_INTERVAL_S`` so a user interrupt
can abort it; in-flight calls cannot be killed and bill via ``late_accounting_sink``.
"""
if not reference_models:
return []
total = len(reference_models)
results: list[tuple[str, str, Any] | None] = [None] * total
futures: dict[Any, int] = {}
# Propagate the turn's contextvars (approval callbacks, Nous conversation tag).
from tools.thread_context import propagate_context_to_thread
completed = 0
executor = ThreadPoolExecutor(max_workers=min(_MAX_REFERENCE_WORKERS, total))
interrupted = False
# Shared per-fan-out context-length cache (dict get/set is GIL-atomic).
ctx_len_cache: dict[tuple[str, str], int | None] = {}
cache_disabled, cache_ttl = _agent_cache_opts(agent)
def collect(done: set[Any]) -> None:
nonlocal completed
for future in done:
idx = futures[future]
results[idx] = future.result()
completed += 1
# Each completion is fan-out progress the main loop cannot see: the
# fan-out blocks it, so stamp the turn's activity clock here (non-streaming
# advisors otherwise tick nothing between dispatch and completion).
_touch_fanout_progress(agent, f"MoA: {completed} of {total} references complete")
if progress_callback is not None:
try:
progress_callback(completed, total, _slot_label(reference_models[idx]))
except Exception as exc: # pragma: no cover - display must never break
logger.debug("MoA progress_callback failed: %s", exc)
try:
for idx, slot in enumerate(reference_models):
if slot.get("provider") == "moa":
results[idx] = _placeholder_output(slot, "[skipped: MoA presets cannot recursively reference MoA]")
continue
futures[
executor.submit(
propagate_context_to_thread(_run_reference),
slot,
ref_messages,
temperature=temperature,
max_tokens=max_tokens,
reference_timeout=reference_timeout,
context_length_cache=ctx_len_cache,
cache_disabled=cache_disabled,
cache_ttl=cache_ttl,
agent=agent,
)
] = idx
# Collect every reference (no early exit except a user interrupt).
pending = set(futures)
while pending:
done, pending = _futures_wait(pending, timeout=_REFERENCE_POLL_INTERVAL_S)
collect(done)
if pending and agent is not None and getattr(agent, "_interrupt_requested", False):
# A worker can raise the interrupt immediately before publishing
# its own result. Give concurrently-finishing work one scheduler
# turn so completed output is not replaced by an interrupt note.
done, pending = _futures_wait(pending, timeout=_REFERENCE_INTERRUPT_SETTLE_S)
collect(done)
interrupted = True
_settle_interrupted(futures, results, reference_models, late_accounting_sink)
break
finally:
executor.shutdown(wait=not interrupted, cancel_futures=interrupted)
return [r for r in results if r is not None]
def _truncate_tool_result(text: str, budget: int = _REFERENCE_TOOL_RESULT_BUDGET) -> str:
"""Head+tail preview of a tool result with an ``[... N chars omitted ...]`` marker."""
if not text or len(text) <= budget:
return text
half = budget // 2
omitted = len(text) - 2 * half
return f"{text[:half]}\n[... {omitted} chars omitted ...]\n{text[-half:]}"
def _field(obj: Any, name: str) -> Any:
"""Read ``name`` from a dict or an attribute object (tool calls arrive as either)."""
if obj is None:
return None
return obj.get(name) if isinstance(obj, dict) else getattr(obj, name, None)
def _render_tool_calls(tool_calls: Any) -> str:
"""Render an assistant turn's tool_calls as ``[called tool: name(args)]`` lines.
Tolerates dict- and SimpleNamespace-shaped entries (and nested ``function``).
"""
lines: list[str] = []
for tc in tool_calls or []:
fn = _field(tc, "function")
name = _field(fn, "name") or _field(tc, "name") or "tool"
fn_args = _field(fn, "arguments")
if isinstance(fn_args, str) or fn_args is None:
args_text = fn_args or ""
else:
try:
args_text = json.dumps(fn_args, ensure_ascii=False)
except Exception:
args_text = str(fn_args)
lines.append(f"[called tool: {name}({args_text})]" if args_text else f"[called tool: {name}]")
return "\n".join(lines)
# Cached guidance (user_turn / off-cadence every_n fanout) is reused on later iterations of the
# same turn, where it predates the tool results the acting model now sees. Without this line the
# block reads as fresh instruction and an advisor's suggested tool call gets replayed after it
# already ran.
_STALE_GUIDANCE_NOTE = (
"This guidance was produced earlier in this turn, before the tool results below it. "
"Check the transcript before acting on it: a step it suggests may already have run, and "
"repeating a completed tool call is never the next step.\n"
)
_ADVISORY_INSTRUCTION = (
"[The conversation above is the current state of the task. Give your "
"most intelligent judgement: what is going on, what should happen next, "
"what risks or mistakes you see, and how the acting agent should "
"proceed.]"
)
def _tool_activity_since_last_user(messages: list[dict[str, Any]]) -> bool:
"""Whether the acting model has already called tools since the last real user turn.
Guidance cached from the start of the turn predates those results, so replaying a
tool call it suggests can repeat work the transcript already shows as done."""
for msg in reversed(messages):
role = msg.get("role")
if role == "user":
return False
if role == "tool" or (role == "assistant" and msg.get("tool_calls")):
return True
return False
def _reference_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Build the advisory (reference-model) view of the conversation.
Plain user/assistant TEXT turns only: system prompt dropped, tool_calls rendered
inline, tool results folded into the preceding assistant turn as previews (no
tool-role messages / tool_calls arrays, so strict providers do not 400). Always
ends on a ``user`` turn (Anthropic treats a trailing assistant turn as prefill)
by APPENDING a synthetic request. The aggregator always gets the full transcript.
"""
rendered: list[dict[str, Any]] = []
last_user_content: str | None = None
for msg in messages:
role = msg.get("role")
content = msg.get("content")
# Decorated (cache_control parts) and undecorated transcripts must yield a
# byte-identical view so the advisory prefix stays cache-stable.
text = flatten_message_text(content)
if role == "user":
if not text.strip() and isinstance(content, list) and content:
# Image-only turn: strict providers reject empty user messages and
# skipping would break alternation.
text = "[user sent non-text content (e.g. an image attachment)]"
if not text.strip():
continue # genuinely empty user turn: strict providers 400 on it
last_user_content = text
rendered.append({"role": "user", "content": text})
elif role == "assistant":
parts = [text.strip()] if text.strip() else []
calls_text = _render_tool_calls(msg.get("tool_calls"))
if calls_text:
parts.append(calls_text)
if parts: # empty assistant turns carry nothing advisory
rendered.append({"role": "assistant", "content": "\n".join(parts)})
elif role == "tool":
# Fold the tool result into the preceding assistant turn as text (a leading
# tool result with no assistant turn opens one).
block = f"[tool result: {_truncate_tool_result(text)}]"
if rendered and rendered[-1].get("role") == "assistant":
rendered[-1]["content"] = rendered[-1]["content"] + "\n" + block
else:
rendered.append({"role": "assistant", "content": block})
# The preview is still the raw result, so the folded turn keeps its data-use requirement.
if msg.get(DATA_USE):
rendered[-1][DATA_USE] = sorted({*rendered[-1].get(DATA_USE, ()), *msg[DATA_USE]})
# system and any other role are ignored.
# Anthropic rejects trailing assistant prefill: end on a synthetic user request.
if rendered and rendered[-1].get("role") == "assistant":
rendered.append({"role": "user", "content": _ADVISORY_INSTRUCTION})
if not rendered:
# Nothing rendered: fall back to the latest user turn.
if last_user_content is not None:
return [{"role": "user", "content": last_user_content}]
for msg in reversed(messages):
fallback_text = flatten_message_text(msg.get("content")) if msg.get("role") == "user" else ""
if fallback_text.strip():
return [{"role": "user", "content": fallback_text}]
return rendered
def _extract_text(response: Any) -> str:
"""Assistant text of a completed response: transport-normalized, else ``choices[0]``."""
with contextlib.suppress(Exception):
text = (get_transport("chat_completions").normalize_response(response).content or "").strip()
if text:
return text
try:
message = response.choices[0].message
content = message.get("content") if isinstance(message, dict) else getattr(message, "content", message)
if not isinstance(content, str):
content = str(content) if content else ""
return content.strip()
except Exception:
return ""
def _preset_temperature(preset: dict[str, Any], key: str) -> float | None:
"""Read an optional preset temperature; None (absent/empty/null) = provider default."""
value = preset.get(key)
if value is None or (isinstance(value, str) and not value.strip()):
return None
try:
return float(value)
except (TypeError, ValueError):
logger.warning("ignoring non-numeric %s=%r in MoA preset", key, value)
return None
def _hash_messages(msgs: list[dict[str, Any]]) -> str:
return hashlib.sha256("\u0000".join(f"{m.get('role')}:{m.get('content')}" for m in msgs).encode("utf-8", "replace")).hexdigest()
def _is_failed_reference(text: str) -> bool:
"""Whether a reference output is a ``[failed: …]`` / ``[skipped: …]`` sentinel."""
return text.lstrip().lower().startswith(("[failed:", "[skipped:"))
def _join_reference_outputs(outputs: list[tuple[str, str, Any]], degraded: str = "") -> str:
"""Render numbered reference blocks for the aggregator, appending any degraded notice."""
joined = "\n\n".join(f"Reference {idx} — {label}:\n{text}" for idx, (label, text, _acct) in enumerate(outputs, start=1))
if degraded:
joined = f"{joined}\n\n{degraded}" if joined else degraded
return joined
def _sum_reference_accounting(outputs: list[tuple[str, str, Any]]) -> tuple[Any, Any]:
"""Sum fan-out usage AND cost in dollars (each advisor priced at its OWN rate)."""
usage = CanonicalUsage()
cost: Any = None
for _lbl, _txt, acct in outputs:
if isinstance(acct, _RefAccounting):
if isinstance(acct.usage, CanonicalUsage):
usage = usage + acct.usage
if acct.cost_usd is not None:
cost = (cost or 0) + acct.cost_usd
return usage, cost
def _degraded_notice(failed_labels: list[str], policy: str) -> str:
if not failed_labels or policy.strip().lower() == "silent":
return ""
return f"[Reference models unavailable: {', '.join(failed_labels)}]"
def _slot_labels(slots: list[dict[str, Any]]) -> str:
return ", ".join(_slot_label(slot) for slot in slots)
def _guidance_inputs(
reference_outputs: list[tuple[str, str, Any]], privacy_full: bool, policy: str,
) -> tuple[list[tuple[str, str, Any]], str, bool]:
"""``(advisor outputs for the aggregator, degraded notice, all_failed)``.
'full' privacy mode redacts advisor text reaching the aggregator (applied to a
per-call copy — caches hold raw text). Failed refs are filtered out first.
"""
successful = [o for o in reference_outputs if not _is_failed_reference(o[1])]
failed_labels = [label for label, text, _acct in reference_outputs if _is_failed_reference(text)]
agg_refs = _redact_reference_outputs(successful) if privacy_full else successful
return agg_refs, _degraded_notice(failed_labels, policy), bool(reference_outputs) and not successful
def aggregate_moa_context(
*, user_prompt: str, api_messages: list[dict[str, Any]], reference_models: list[dict[str, Any]],
aggregator: dict[str, Any], temperature: float | None = None, aggregator_temperature: float | None = None,
reference_max_tokens: int | None = None, reference_timeout: float | None = None,
degraded_reference_policy: str = "loud", agent: Any = None,
) -> str:
"""Run configured reference models and synthesize their advice (one-shot /moa).
Failures become model-specific notes instead of aborting the loop.
``reference_max_tokens`` caps ONLY the fan-out (capping the aggregator truncated
long syntheses). ``agent`` makes the fan-out interruptible.
``reference_max_tokens`` applies ONLY to the reference fan-out — the aggregator's own synthesis call is
not given an advisor budget. Omission uses provider-specific defaults; native protocols may
still require an internal wire limit.
A hardcoded cap on the aggregator call previously truncated long aggregator syntheses (#53580) — passing
``reference_max_tokens`` to both calls here would silently reintroduce that regression.
"""
reference_models = [slot for slot in reference_models if slot.get("enabled", True)]
reference_outputs = _run_references_parallel(
reference_models, _reference_messages(api_messages), temperature=temperature,
max_tokens=reference_max_tokens, reference_timeout=reference_timeout, agent=agent,
)
privacy_full = False
try:
from hermes_cli.config import load_config as _load_config
privacy_full = _moa_privacy_mode((_load_config() or {}).get("moa")) == "full"
except Exception: # pragma: no cover - privacy filter must never break a turn
logger.debug("MoA privacy filter check failed", exc_info=True)
agg_refs, degraded, all_failed = _guidance_inputs(reference_outputs, privacy_full, degraded_reference_policy)
joined = _join_reference_outputs(agg_refs, degraded)
if all_failed:
# Skip the aggregator: synthesizing over nothing can block for the full provider timeout.
logger.warning("MoA: all %d reference(s) failed — skipping aggregator synthesis", len(reference_outputs))
return (
"[Mixture of Agents context — all reference models failed. "
"Proceeding without aggregated guidance.]\n"
f"References: {_slot_labels(reference_models)}\n\n"
f"{degraded or '[Reference models unavailable]'}"
)
synth_prompt = (
"You are the aggregator in a Mixture of Agents process. Synthesize the "
"reference responses into concise, actionable guidance for the main "
"Hermes agent. Focus on next steps, tool-use strategy, risks, and any "
"disagreements. Do not answer the user directly unless that is all that "
"is needed; produce context the main agent should use in its normal loop.\n\n"
f"Original user prompt:\n{user_prompt}\n\n"
f"Reference responses:\n{joined}"
)
agg_label = _slot_label(aggregator)
agg_runtime = _slot_runtime(aggregator)
_touch_fanout_progress(
agent, f"MoA: synthesizing aggregator guidance ({agg_label})"
)
try:
# Undecorated: the prompt embeds this call's reference outputs, so no later call
# reads it back and a marker would only pay a cache write (is_one_shot_request).
synthesis = _extract_text(call_llm(
task="moa_aggregator", messages=[{"role": "user", "content": synth_prompt}], temperature=aggregator_temperature,
reasoning_config=_aggregator_reasoning_config(aggregator), **agg_runtime,
))
except Exception as exc:
logger.warning("MoA aggregator model %s failed: %s", agg_label, exc)
synthesis = ""
return (
"[Mixture of Agents context — use this as private guidance for the "
"normal Hermes agent loop. You may call tools, continue reasoning, or "
"finish normally.]\n"
f"Aggregator: {agg_label}\n"
f"References: {_slot_labels(reference_models)}\n\n"
f"{(synthesis or joined).strip()}"
)
def _completed_response_as_stream_chunk(response: Any) -> Any:
"""Adapt a completed response into one delta stream chunk (facade boundary only)."""
choices = getattr(response, "choices", None)
first_choice = choices[0] if isinstance(choices, (list, tuple)) and choices else None
message = getattr(first_choice, "message", None)
raw_tool_calls = getattr(message, "tool_calls", None)
tool_call_deltas = None
if isinstance(raw_tool_calls, (list, tuple)) and raw_tool_calls:
tool_call_deltas = [
SimpleNamespace(
index=getattr(tc, "index", index), id=getattr(tc, "id", None), type=getattr(tc, "type", None) or "function",
function=SimpleNamespace(
name=getattr(getattr(tc, "function", None), "name", None),
arguments=getattr(getattr(tc, "function", None), "arguments", None),
),
)
for index, tc in enumerate(raw_tool_calls)
]
delta = SimpleNamespace(
content=getattr(message, "content", None), tool_calls=tool_call_deltas,
**{k: getattr(message, k, None) for k in ("reasoning_content", "reasoning", "reasoning_details")},
)
choice = SimpleNamespace(
index=getattr(first_choice, "index", 0), delta=delta, finish_reason=getattr(first_choice, "finish_reason", None) or "stop",
)
return SimpleNamespace(
id=getattr(response, "id", None), model=getattr(response, "model", None), choices=[choice], usage=getattr(response, "usage", None),
)
def _attach_reference_guidance(agg_messages: list[dict[str, Any]], guidance: str) -> None:
"""Attach the per-turn reference block as its OWN trailing user message.
The block varies per turn; appending keeps ``[system][task][tool-history]``
cache-stable. It is never merged into a trailing user turn: iteration 1 of a
tool loop ends on ``user(task)``, and a merged ``user(task + guidance)`` byte-differs
from the ``user(task)`` every later iteration replays, so the provider prefix cache
collapsed to the system prompt on iteration 2 of every turn (#112358). Converters
that require strict alternation (Anthropic Messages, Converse, native Gemini) merge
the two user turns as SEPARATE content blocks, so the task block stays byte-stable
there too; on the OpenAI-compatible wire the request ends ``user(task), user(guidance)``,
which a chat template that enforces strict user/assistant alternation rejects.
"""
agg_messages.append({"role": "user", "content": guidance})
def peel_reference_guidance(messages: list[dict[str, Any]], guidance: Any) -> list[dict[str, Any]]:
"""Exact inverse of ``_attach_reference_guidance`` (plain string, or its cache-decorated
single-text-part form), so a cache breakpoint never lands on the turn-varying guidance.
Inputs are not mutated."""
if not guidance or not messages:
return messages
guidance_text = str(guidance)
last = messages[-1]
if not isinstance(last, dict) or last.get("role") != "user":
return messages
content = last.get("content")
if content == guidance_text:
return list(messages[:-1])
if isinstance(content, list) and len(content) == 1:
part = content[0]
if isinstance(part, dict) and part.get("type", "text") == "text" and (part.get("text") or "") == guidance_text:
return list(messages[:-1])
return messages
class MoAChatCompletions:
"""OpenAI-chat-compatible facade where the aggregator is the acting model.
``reference_callback(event, **kwargs)`` is an optional best-effort display hook
(events: ``moa.reference``, ``moa.progress``, ``moa.phase``, ``moa.aggregating``;
kwargs per ``_RELAY_EVENTS``). ``agent`` is the owning AIAgent; it lets the
fan-out check ``_interrupt_requested``.
"""
def __init__(self, preset_name: str, reference_callback: Any = None, agent: Any = None):
self.preset_name = preset_name or "default"
self.reference_callback = reference_callback
self._agent = agent
# Reference cache keyed on the advisory-view signature (HIT = no re-run, no re-emit).
self._ref_cache_key: tuple | None = None
self._ref_cache_outputs: list[tuple[str, str, Any]] = []
# Fan-out spend awaiting consume_reference_usage (nothing deposited on a HIT so
# spend counts once); the lock guards late-accounting callbacks on worker threads.
self._pending_reference_usage: Any = CanonicalUsage()
self._pending_reference_cost: Any = None
self._accounting_lock = threading.Lock()
# Real aggregator slot so cost accounting prices the acting turn at its model.
self.last_aggregator_slot: Any = None
# Full-turn trace parts from a cache-MISS create(), flushed by consume_and_save_trace.
self._pending_trace: Any = None
# Per-advisor metrics for observability hooks; NOT consumed (post_api_request
# fires on a different branch than consume_and_save_trace).
self._last_reference_metrics: Any = None
# every_n cadence state, scoped to one USER TURN so iteration 1 is on-cadence.
self._fanout_iteration_count = 0
self._fanout_turn_sig: str | None = None
self._fanout_last_state_sig: str | None = None
self._privacy_mode: str = "" # normalized moa.privacy_filter, refreshed per create()
# Destinations (route, model) that 400'd on adjacent same-role messages this session:
# their aggregator requests are pre-merged; every other destination keeps the split,
# cache-stable shape (agent/moa_alternation.py).
self._merge_same_role_destinations: set[tuple[str, str]] = set()
def consume_reference_usage(self) -> tuple[Any, Any]:
"""Pop pending fan-out ``(CanonicalUsage, cost_usd_or_None)`` and reset both
(so a streaming retry re-entering accounting cannot double-count)."""
with self._accounting_lock:
usage = self._pending_reference_usage or CanonicalUsage()
cost = self._pending_reference_cost
self._pending_reference_usage = CanonicalUsage()
self._pending_reference_cost = None
return usage, cost
def last_reference_metrics(self) -> Any:
"""Per-advisor metrics from the most recent fan-out, or None (read-only)."""
return self._last_reference_metrics
def _record_late_reference_accounting(self, label: str, accounting: Any) -> None:
"""Done-callback for abandoned (still billing) futures: fold their real spend in."""
if not isinstance(accounting, _RefAccounting):
return
self._fold_pending_accounting(*_sum_reference_accounting([(label, "", accounting)]))
logger.debug("MoA: recorded late accounting for interrupted reference %s", label)
def _fold_pending_accounting(self, usage: Any, cost: Any) -> None:
"""Add (never overwrite) fan-out spend so late interrupted-reference deposits survive."""
with self._accounting_lock:
self._pending_reference_usage = (self._pending_reference_usage or CanonicalUsage()) + usage
if cost is not None:
self._pending_reference_cost = (self._pending_reference_cost or 0) + cost
def consume_and_save_trace(self, session_id: Any = None, aggregator_output_fallback: Any = None) -> None:
"""Flush the pending full-turn trace to disk (no-op when nothing is pending).
``aggregator_output_fallback`` is the caller's resolved acting text for the
streaming path (not capturable at ``create()`` time). Never raises.
"""
pending = self._pending_trace
self._pending_trace = None
if not pending or "aggregator_input_messages" not in pending:
return
try:
from agent.moa_trace import save_moa_turn
agg_slot = pending.get("aggregator_slot") or {}
# Inline capture (non-streaming) beats the caller's streamed text.
agg_output = pending.get("aggregator_output")
if agg_output is None and aggregator_output_fallback:
agg_output = aggregator_output_fallback
save_moa_turn(
session_id=session_id, preset_name=pending.get("preset", ""),
reference_outputs=pending.get("reference_outputs", []),
aggregator_label=pending.get("aggregator_label", ""), aggregator_model=agg_slot.get("model"),
aggregator_provider=agg_slot.get("provider"), aggregator_temperature=pending.get("aggregator_temperature"),
aggregator_input_messages=pending.get("aggregator_input_messages"),
aggregator_output=agg_output, aggregator_streamed=bool(pending.get("aggregator_streamed")),
)
except Exception as exc: # pragma: no cover - tracing must never break a turn
logger.debug("MoA trace flush failed: %s", exc)
def _emit(self, event: str, **kwargs: Any) -> None:
if self.reference_callback is None:
return
try:
self.reference_callback(event, **kwargs)
except Exception as exc: # pragma: no cover - display must never break the turn
logger.debug("MoA reference_callback failed for %s: %s", event, exc)
def prepare(self, messages: list[dict[str, Any]]) -> dict[str, Any]:
"""Run the advisor fan-out and return the exact aggregator request, which the
loop measures before its compression gate and hands back to ``create()``."""
return self.create(messages=messages, _moa_prepare_only=True)
def rebase_prepared_request(self, prepared: dict[str, Any], messages: list[dict[str, Any]]) -> dict[str, Any]:
"""Re-attach already-generated guidance to a rebuilt (compressed) transcript."""
guidance = prepared.get("guidance")
agg_messages = [dict(message) for message in messages]
if guidance:
_attach_reference_guidance(agg_messages, str(guidance))
return {**prepared, "messages": agg_messages}
def _plan_aggregator_cache(
self, agg_messages: list[dict[str, Any]], tools: Any, guidance: Any, agg_runtime: dict[str, Any],
) -> tuple[list[dict[str, Any]], Any]:
"""Cache-breakpoint the aggregator request for its destination.
Guidance is peeled before planning and re-attached after so a breakpoint never
lands on the turn-varying block. Any error → undecorated request (warning, not
debug: this is the aggregator's ONLY decoration path).
"""
try:
from agent.agent_runtime_helpers import plan_cache_sections_for_destination
planning_messages = peel_reference_guidance(agg_messages, str(guidance)) if guidance else agg_messages
# Tri-state cache_disabled: facades built via __new__ have no _agent; forcing
# False would suppress the planner's config fallback.
# plan_cache_sections_for_destination never mutates its inputs and always returns request-local
# copies, so the prepared state stays canonical. Tri-state: only pass a bool when a live agent
# snapshot exists. See #76085.
_agent = getattr(self, "_agent", None)
cache_disabled, cache_ttl = _agent_cache_opts(_agent)
# Agent TTL + stable system prefix so MoA does not regress 1h → 5m.
agg_messages, tools = plan_cache_sections_for_destination(
planning_messages, tools, provider=agg_runtime.get("provider") or "",
base_url=agg_runtime.get("base_url") or "", api_mode=agg_runtime.get("api_mode") or "",
model=agg_runtime.get("model") or "", cache_disabled=cache_disabled, cache_ttl=cache_ttl,
static_system_prefix=getattr(_agent, "_cached_system_prompt_static", None),
)
if guidance:
_attach_reference_guidance(agg_messages, str(guidance))
except Exception as exc: # pragma: no cover - cache planning must not block MoA
logger.warning(
"MoA aggregator cache plan failed — sending undecorated "
"request (cache misses expected): %s", exc,
)
return agg_messages, tools
def _call_prepared_aggregator(self, prepared: dict[str, Any], api_kwargs: dict[str, Any]) -> Any:
"""Send an already prepared MoA aggregator request exactly once."""
aggregator = prepared["aggregator"]
if aggregator.get("provider") == "moa":
raise RuntimeError("MoA aggregator cannot be another MoA preset")
agg_runtime = _slot_runtime(aggregator)
agg_messages, tools = self._plan_aggregator_cache(
prepared["messages"], api_kwargs.get("tools"), prepared.get("guidance"), agg_runtime
)
trace = self._pending_trace
if trace is not None:
trace["aggregator_label"] = _slot_label(aggregator)
# stream=True returns the RAW token stream (consumer reassembles + retries);
# the non-streaming path forwards no stream/stream_options/timeout. The
# consumer's stream-read timeout must govern the aggregator stream.
stream = bool(api_kwargs.get("stream"))
stream_kwargs: dict[str, Any] = {}
if stream:
stream_kwargs = {"stream": True, "stream_options": api_kwargs.get("stream_options") or {"include_usage": True}}
if api_kwargs.get("timeout") is not None:
stream_kwargs["timeout"] = api_kwargs["timeout"]
# Pop the runtime's extra_body so the explicit kwarg never collides with **agg_runtime.
agg_extra_body = _merge_slot_extra_body(agg_runtime.pop("extra_body", None), api_kwargs.get("extra_body"))
destination = destination_key(agg_runtime)
# Facades built via __new__ (tests, swaps) have no __init__ state.
remembered = getattr(self, "_merge_same_role_destinations", None)
if remembered is None:
remembered = self._merge_same_role_destinations = set()
merged = destination in remembered
if merged:
agg_messages = merge_same_role_messages(agg_messages)
send = functools.partial(
call_llm, task="moa_aggregator", temperature=prepared["aggregator_temperature"],
max_tokens=api_kwargs.get("max_tokens"), tools=tools, extra_body=agg_extra_body,
reasoning_config=_aggregator_reasoning_config(aggregator), # same policy as direct create()
**stream_kwargs, **agg_runtime,
)
try:
agg_response = send(messages=agg_messages)
except Exception as exc:
# Strict-alternation template rejected ``user(task), user(guidance)``: merge the pair for
# THIS destination only and retry once; remember it so later iterations pre-merge.
retry_messages = None if merged else merge_same_role_messages(agg_messages)
if retry_messages is None or retry_messages is agg_messages or not is_role_alternation_rejection(exc, agg_runtime):
raise
remembered.add(destination)
logger.warning(
"MoA aggregator %s rejected adjacent same-role messages — merging them for this "
"destination for the rest of the session and retrying once: %.200s", _slot_label(aggregator), exc,
)
agg_messages = retry_messages
agg_response = send(messages=agg_messages)
if trace is not None:
# Trace the exact aggregator INPUT as sent (persisted copy redacted; live input raw).
trace["aggregator_input_messages"] = (
_redact_trace_messages([dict(m) for m in agg_messages]) if getattr(self, "_privacy_mode", "") else agg_messages
)
# Streaming output lands as the turn's assistant message; the trace marks it.
trace["aggregator_streamed"] = stream
output = None
if not stream:
with contextlib.suppress(Exception): # pragma: no cover - defensive
output = _extract_text(agg_response)
trace["aggregator_output"] = output
if stream and hasattr(agg_response, "choices"):
# Some adapters (openai-codex Responses) return a completed response even
# when streaming was requested; hand the loop a one-chunk iterator.
return iter((_completed_response_as_stream_chunk(agg_response),))
return agg_response
def _fanout_cache_key(
self, preset: dict[str, Any], ref_messages: list[dict[str, Any]], reference_models: list[dict[str, Any]],
) -> tuple:
"""Turn-scoped reference cache key per the preset's fan-out cadence.
"user_turn" (default) hashes only the prefix up to the LAST USER message, so
later tool iterations are HITs. "per_iteration" re-runs whenever the advisory
view changes. "every_n:<N>": iteration 1 of a turn, then every Nth; in-between
iterations return the pinned last on-cadence key (HIT: no calls, no re-emit).
"""
# "user_turn" (default — cheapest cadence, #67199): advisors run ONCE per user turn; subsequent tool
# iterations reuse that turn's advice and the aggregator acts alone (the original MoA shape:
# synthesize at the start, then let the acting model work). Implemented by hashing only the prefix
# up to the LAST USER message so mid-turn growth doesn't change the signature — iteration 2+ becomes
# a cache HIT. "per_iteration": advisors re-run whenever the advisory view changes — i.e. every tool
# iteration, since the view grows with each tool result; advice tracks live task state at the cost
# of multiplying advisor latency/spend by tool-loop depth. "every_n:<N>" (N >= 2): the middle ground
# (issue #63393 — advisor fan-out multiplies latency/cost by the tool-iteration count). Advisors run
# on iteration 1 of a user turn and then every Nth tool iteration; the iterations in between REUSE
# the cached guidance from the last on-cadence run (same mechanism as user_turn's cache HIT — the
# aggregator still gets advice every iteration, it's just not refreshed against the very latest tool
# results). The iteration counter is scoped per user turn and resets on a new user message, so every
# turn starts with fresh advice.
fanout_mode = str(preset.get("fanout") or "user_turn").strip().lower()
every_n = 0
if fanout_mode.startswith("every_n:"):
with contextlib.suppress(TypeError, ValueError):
every_n = int(fanout_mode.split(":", 1)[1])
if every_n < 2:
fanout_mode = "per_iteration" # every_n:1 IS per-iteration (mirrors _coerce_fanout)
sig_messages = turn_prefix = ref_messages
if fanout_mode == "user_turn" or every_n >= 2:
# Last REAL user message: the synthetic _ADVISORY_INSTRUCTION marker must not
# count or the prefix would grow (and re-sign) every iteration.
last_user = next(
(i for i in range(len(ref_messages) - 1, -1, -1)
if ref_messages[i].get("role") == "user" and ref_messages[i].get("content") != _ADVISORY_INSTRUCTION),
None,
)
if last_user is not None:
turn_prefix = ref_messages[: last_user + 1]
if fanout_mode == "user_turn":
sig_messages = turn_prefix
# every_n bookkeeping: advance the counter only when the advisory STATE changed
# (a streaming retry must not consume a cadence slot); reset on a new turn prefix.
if every_n >= 2:
turn_sig = _hash_messages(turn_prefix)
if turn_sig != self._fanout_turn_sig:
self._fanout_turn_sig = turn_sig
self._fanout_iteration_count = 0
self._fanout_last_state_sig = None
state_sig = _hash_messages(ref_messages)
if state_sig != self._fanout_last_state_sig:
self._fanout_last_state_sig = state_sig
self._fanout_iteration_count += 1
# Iteration 1 is on-cadence; then every Nth iteration after it.
if (self._fanout_iteration_count - 1) % every_n != 0 and self._ref_cache_outputs:
return self._ref_cache_key
return (self.preset_name, _hash_messages(sig_messages), tuple(_slot_label(s) for s in reference_models))
def _run_fanout(
self, preset: dict[str, Any], ref_messages: list[dict[str, Any]], reference_models: list[dict[str, Any]],
aggregator: dict[str, Any], aggregator_temperature: Any, cache_key: tuple,
) -> list[tuple[str, str, Any]]:
"""Cache-MISS path of ``create``: run the advisors, account, trace and emit.
A preset MAY cap ADVISOR output (dominant MoA latency); the acting aggregator
is never capped. None timeout = inherit auxiliary.moa_reference.timeout.
"""
raw_reference_timeout = preset.get("reference_timeout")
reference_outputs = _run_references_parallel(
reference_models, ref_messages, temperature=_preset_temperature(preset, "reference_temperature"),
progress_callback=lambda done, total, label: self._emit("moa.progress", refs_done=done, refs_total=total, label=label),
reference_timeout=float(raw_reference_timeout) if raw_reference_timeout else None,
agent=self._agent, late_accounting_sink=self._record_late_reference_accounting,
)
# An interrupted fan-out is a partial snapshot: never cache it (a HIT would
# replay placeholder notes every iteration).
interrupted = any(text == _INTERRUPTED_REFERENCE_NOTE for _lbl, text, _acct in reference_outputs)
self._ref_cache_key = None if interrupted else cache_key
self._ref_cache_outputs = [] if interrupted else list(reference_outputs)
# Fold advisor spend into accounting exactly once per turn.
self._fold_pending_accounting(*_sum_reference_accounting(reference_outputs))
# Stash the fan-out for trace persistence (aggregator parts filled in later).
# Traces are persisted, so ANY active privacy mode redacts them.
privacy_mode = self._privacy_mode
trace_refs = [
(label, _redact_reference_text(text), _redact_trace_accounting(acct)) for label, text, acct in reference_outputs
] if privacy_mode else list(reference_outputs)
self._pending_trace = {
"preset": self.preset_name, "reference_outputs": trace_refs,
"aggregator_slot": aggregator, "aggregator_temperature": aggregator_temperature,
}
# Derived from the privacy-redacted trace_refs.
try:
from agent.moa_trace import slot_metrics
self._last_reference_metrics = [slot_metrics(acct, label, output=text) for label, text, acct in trace_refs]
except Exception as exc: # pragma: no cover - never break a turn
logger.debug("MoA reference metrics render failed: %s", exc)
self._last_reference_metrics = None
# Surface each answer BEFORE the aggregator acts; the cache keeps RAW text.
ref_count = len(reference_outputs)
for idx, (label, text, _accounting) in enumerate(reference_outputs, start=1):
self._emit(
"moa.reference", index=idx, count=ref_count, label=label,
text=_redact_reference_text(text) if privacy_mode else text,
)
if ref_count:
# Phase transition: fan-out complete, aggregator about to act.
agg_label = _slot_label(aggregator)
self._emit("moa.phase", phase="aggregator", refs_done=ref_count, refs_total=ref_count, aggregator=agg_label)
self._emit("moa.aggregating", aggregator=agg_label, ref_count=ref_count)
return reference_outputs
def _build_guidance(
self, reference_outputs: list[tuple[str, str, Any]], aggregator: dict[str, Any], degraded_reference_policy: str,
stale: bool = False,
) -> str | None:
"""Render the reference block attached to the aggregator prompt (None = nothing)."""
agg_refs, degraded, all_failed = _guidance_inputs(
reference_outputs, self._privacy_mode == "full", degraded_reference_policy
)
header = (
"[Mixture of Agents reference context]\n"
f"Preset: {self.preset_name}\n"
f"Aggregator/acting model: {_slot_label(aggregator)}\n"
)
if all_failed:
# Every reference failed: the aggregator acts alone (loud policy → notice).
logger.warning(
"MoA: all %d reference(s) failed — acting aggregator-alone "
"without reference guidance",
len(reference_outputs),
)
if degraded:
return (
f"{header}\n"
"All reference models failed this turn — no advisory "
"guidance is available. Act on your own judgment.\n\n"
f"{degraded}"
)
return None
if agg_refs or degraded:
return (
f"{header}"
f"References: {', '.join(label for label, _, _ in agg_refs)}\n\n"
"Use the reference responses below as private context. You are the aggregator and acting model: "
"answer the user directly or call tools as needed.\n"
f"{_STALE_GUIDANCE_NOTE if stale else ''}\n"
f"{_join_reference_outputs(agg_refs, degraded)}"
)
return None
def create(self, **api_kwargs: Any) -> Any:
prepared_request = api_kwargs.pop("_moa_prepared_request", None)
if prepared_request is not None:
if not isinstance(prepared_request, dict):
raise TypeError("_moa_prepared_request must be a dict")
return self._call_prepared_aggregator(prepared_request, api_kwargs)
preset, moa_raw = _resolve_preset_cached(self.preset_name)
# Remembered on self so _call_prepared_aggregator redacts the trace consistently.
self._privacy_mode = _moa_privacy_mode(moa_raw)
messages = list(api_kwargs.get("messages") or [])
# A disabled preset = "use the aggregator directly".
reference_models = [
slot for slot in (preset.get("reference_models") or []) if slot.get("enabled", True)
] if preset.get("enabled", True) else []
aggregator = preset.get("aggregator") or {}
# The MoA path's virtual model/provider have no pricing entry; expose the real slot.
self.last_aggregator_slot = dict(aggregator) if aggregator else None
# None = provider default (see _preset_temperature); the acting agent's own
# temperature applies to the aggregator (the acting model).
aggregator_temperature = _preset_temperature(preset, "aggregator_temperature")
if aggregator_temperature is None and api_kwargs.get("temperature") is not None:
aggregator_temperature = api_kwargs.get("temperature")
ref_messages = _reference_messages(messages)
cache_key = self._fanout_cache_key(preset, ref_messages, reference_models)
cache_hit = bool(cache_key == self._ref_cache_key and self._ref_cache_outputs)
if cache_hit:
# HIT: already ran and accounted. Do NOT zero pending totals (a late
# interrupted reference may have deposited) and no trace (not a new turn).
reference_outputs = list(self._ref_cache_outputs)
self._pending_trace = None
else:
reference_outputs = self._run_fanout(preset, ref_messages, reference_models, aggregator, aggregator_temperature, cache_key)
agg_messages = [dict(m) for m in messages]
guidance = self._build_guidance(
reference_outputs, aggregator, str(preset.get("degraded_reference_policy") or "loud"),
stale=cache_hit and _tool_activity_since_last_user(messages),
)
if guidance:
_attach_reference_guidance(agg_messages, guidance)
prepared_request = {
"messages": agg_messages, "guidance": guidance, "aggregator": aggregator,
"aggregator_temperature": aggregator_temperature,
}
if api_kwargs.pop("_moa_prepare_only", False):
return prepared_request
return self._call_prepared_aggregator(prepared_request, api_kwargs)
class MoAClient:
"""OpenAI-client-shaped wrapper: ``client.chat.completions`` is a ``MoAChatCompletions``;
the accounting/trace surface below is delegated to that facade."""
def __init__(self, preset_name: str, reference_callback: Any = None, agent: Any = None):
self.chat = type("_MoAChat", (), {})()
self.chat.completions = MoAChatCompletions(preset_name, reference_callback=reference_callback, agent=agent)
def consume_reference_usage(self) -> Any:
return self.chat.completions.consume_reference_usage()
@property
def last_aggregator_slot(self) -> Any:
return getattr(self.chat.completions, "last_aggregator_slot", None)
def consume_and_save_trace(self, session_id: Any = None, aggregator_output_fallback: Any = None) -> None:
return self.chat.completions.consume_and_save_trace(session_id, aggregator_output_fallback=aggregator_output_fallback)
def last_reference_metrics(self) -> Any:
return self.chat.completions.last_reference_metrics()
# Relay table: event -> (primary kwarg, secondary kwarg or None, {cb kwarg: emit kwarg}).
# The callback signature is ``cb(event, label, text, None, **moa_*)``.
_RELAY_EVENTS: dict[str, tuple[str, str | None, dict[str, str]]] = {
"moa.reference": ("label", "text", {"moa_index": "index", "moa_count": "count"}),
"moa.progress": ("label", None, {"moa_refs_done": "refs_done", "moa_refs_total": "refs_total"}),
"moa.phase": ("aggregator", None, {"moa_phase": "phase", "moa_refs_done": "refs_done", "moa_refs_total": "refs_total"}),
"moa.aggregating": ("aggregator", None, {"moa_ref_count": "ref_count"}),
}
def build_moa_facade(agent, preset_name: Any = None) -> MoAClient:
"""Single construction point for ``MoAClient``: a bare ``MoAClient(preset)`` would
drop the ``reference_callback`` relay and silence display events for the session.
The relay reads ``agent.tool_progress_callback`` at emit time."""
def _moa_reference_relay(event: str, **kwargs: Any) -> None:
cb = getattr(agent, "tool_progress_callback", None)
spec = _RELAY_EVENTS.get(event)
if cb is None or spec is None:
return
primary, secondary, extra_map = spec
with contextlib.suppress(Exception):
cb(
event, str(kwargs.get(primary) or ""), str(kwargs.get(secondary) or "") if secondary else None, None,
**{out: kwargs.get(src) for out, src in extra_map.items()},
)
resolved_preset = preset_name
if resolved_preset is None and getattr(agent, "provider", None) == "moa":
resolved_preset = getattr(agent, "model", None)
resolved_preset = str(resolved_preset or "default")
try:
from hermes_cli.config import load_config
from hermes_cli.moa_config import normalize_moa_config
moa_cfg = normalize_moa_config(load_config().get("moa") or {})
if resolved_preset not in (moa_cfg.get("presets") or {}):
resolved_preset = moa_cfg.get("default_preset") or "default"
except Exception:
resolved_preset = "default"
# ``agent`` lets the fan-out wait be aborted on a user interrupt.
return MoAClient(resolved_preset, reference_callback=_moa_reference_relay, agent=agent)
def bind_moa_runtime(agent, preset_name: Any, api_key: Any = None) -> None:
"""Make ``agent`` act as the MoA preset: pin the virtual runtime fields and install the facade.
Every site that puts an agent onto ``provider: moa`` (init, ``/model`` switch, fallback
activation) must pin the same fields — the facade speaks only chat.completions, has no HTTP
endpoint and no OpenAI client kwargs — or the next dispatch/rebuild reaches a real wire with a
virtual identity (``moa://local`` 404, or the preset name sent as a model id).
"""
agent.model = str(preset_name or "default")
agent.provider = agent.requested_provider = "moa"
agent.api_mode = "chat_completions"
agent.api_key = api_key or "moa-virtual-provider"
agent.base_url = "moa://local"
agent._client_kwargs = {}
agent.client = build_moa_facade(agent, agent.model)