From 850f576f3d3abb7ab45ed633fc5b838d8d5c671e Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 23 Jul 2026 12:48:49 -0700 Subject: [PATCH] feat(moa): add every_n fanout cadence with cached-guidance reuse Extends the fanout enum with 'every_n:' (N >= 2): advisors run on the first iteration of each user turn and every Nth tool iteration after it; off-cadence iterations REUSE the cached guidance from the last on-cadence run via the same cache mechanism the user_turn fanout uses, so the aggregator still gets advice on every step. The cadence counter is scoped per user turn (resets on a new user message) and only advances when the advisory state actually changes, so streaming retries never consume a cadence slot. Mapping form {mode: every_n, n: N} normalizes to the canonical string. Unknown/degenerate values fall back to per_iteration. Addresses issue #63393 (advisor fan-out multiplies turn latency/cost by the tool-iteration count). Redesigned from PR #63448: the submitted shape skipped references entirely on off-cadence iterations (aggregator ran advice-less); this version keeps the last advice in play, credited for the idea and cadence framing. Config-gated, default-off (default fanout remains per_iteration). Co-authored-by: webtecnica <75556242+webtecnica@users.noreply.github.com> --- agent/moa_loop.py | 276 +++++++++++++++++- hermes_cli/moa_config.py | 64 +++- tests/hermes_cli/test_moa_config.py | 71 +++++ tests/run_agent/test_moa_fanout_cadence.py | 222 ++++++++++++++ .../user-guide/features/mixture-of-agents.md | 60 ++++ 5 files changed, 675 insertions(+), 18 deletions(-) create mode 100644 tests/run_agent/test_moa_fanout_cadence.py diff --git a/agent/moa_loop.py b/agent/moa_loop.py index 8522750cd04b..3eceeb945bac 100644 --- a/agent/moa_loop.py +++ b/agent/moa_loop.py @@ -10,6 +10,7 @@ from __future__ import annotations import hashlib import logging +import re from concurrent.futures import ThreadPoolExecutor from typing import Any @@ -19,6 +20,136 @@ from agent.transports import get_transport logger = logging.getLogger(__name__) +# --- MoA privacy filter (config: moa.privacy_filter — '' | display | full) --- +# +# 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"(? Any: + """Redact secrets + PII from one advisor/reference text surface. + + Centralized secret shapes first (force=True: the MoA privacy filter is + its own explicit opt-in, independent of the global log-redaction toggle; + code_file=True: advisory text is prose/code, so the ENV/JSON assignment + heuristics that mangle source snippets stay off), then the MoA-specific + email/formatted-phone patterns. Non-string inputs pass through unchanged. + """ + 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) + text = _MOA_PHONE_RE.sub("[redacted phone]", text) + return text + + +def _moa_privacy_mode(moa_raw: Any) -> str: + """Resolve the 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]]: + """Return reference-output tuples with their advisor text redacted. + + The ``_RefAccounting`` third slot is left as-is — accounting fields carry + no advisor text; the full-output/input trace fields are redacted + separately at trace-stash time (see create()) so the LIVE cache keeps raw + accounting objects untouched. + """ + return [ + (label, _redact_reference_text(text), acct) + for label, text, acct in reference_outputs + ] + + +def _redact_trace_messages(messages: Any) -> Any: + """Redact message copies destined for trace persistence. + + Handles both string content and structured content-part lists (e.g. + cache_control-decorated text parts). Unknown shapes pass through. + """ + if not isinstance(messages, list): + return messages + out: list[Any] = [] + for m in messages: + if not isinstance(m, dict): + out.append(m) + continue + content = m.get("content") + if isinstance(content, str): + out.append({**m, "content": _redact_reference_text(content)}) + elif isinstance(content, list): + out.append( + { + **m, + "content": [ + {**p, "text": _redact_reference_text(p.get("text"))} + if isinstance(p, dict) and isinstance(p.get("text"), str) + else p + for p in content + ], + } + ) + else: + out.append(m) + return out + + +def _redact_trace_accounting(acct: Any) -> Any: + """Return a copy of a ``_RefAccounting`` with its trace text redacted. + + Traces persist the advisor's FULL input messages and output to disk, so a + privacy-filtered run must not write raw PII there. Usage/cost fields are + copied verbatim (numbers, no text). Non-accounting objects pass through. + """ + if not isinstance(acct, _RefAccounting): + return acct + return _RefAccounting( + acct.usage, + acct.cost_usd, + acct.cost_status, + acct.cost_source, + messages=_redact_trace_messages(acct.messages), + output=_redact_reference_text(acct.output), + model=acct.model, + provider=acct.provider, + temperature=acct.temperature, + ) + + + # Upper bound on concurrent reference-model calls. References are independent # advisory calls (no tools, no inter-dependence), so we fan them out the same # way delegate_task runs a batch: all in flight at once, results collected when @@ -742,6 +873,18 @@ def aggregate_moa_context( max_tokens=reference_max_tokens, ) + # 'full' privacy mode (moa.privacy_filter) also covers this one-shot /moa + # synthesis path: advisor text is redacted before it reaches the + # synthesizing aggregator. 'display' does not apply here — this path has + # no user-visible reference blocks or trace records of its own. + try: + from hermes_cli.config import load_config as _load_config + + if _moa_privacy_mode((_load_config() or {}).get("moa")) == "full": + reference_outputs = _redact_reference_outputs(reference_outputs) + except Exception: # pragma: no cover - privacy filter must never break a turn + logger.debug("MoA privacy filter check failed", exc_info=True) + joined = "\n\n".join( f"Reference {idx} — {label}:\n{text}" for idx, (label, text, _usage) in enumerate(reference_outputs, start=1) @@ -877,6 +1020,18 @@ class MoAChatCompletions: # caller to stitch in the live session_id + resolved aggregator output # and flush to the trace file (only when moa.save_traces is on). self._pending_trace: Any = None + # every_n fan-out cadence state. The iteration counter is scoped to a + # single USER TURN (not the facade lifetime): it counts create() calls + # since the last new user message and resets whenever the user-turn + # signature changes, so cadence position never leaks across turns — + # iteration 1 of every turn is always on-cadence (fresh advice for a + # fresh request). See the fanout handling in create(). + self._fanout_iteration_count = 0 + self._fanout_turn_sig: str | None = None + self._fanout_last_state_sig: str | None = None + # Normalized moa.privacy_filter mode for the current turn ('' | + # 'display' | 'full'), refreshed from config on every create(). + self._privacy_mode: str = "" def consume_reference_usage(self) -> tuple[Any, Any]: """Pop pending reference-fan-out usage + cost, resetting both to empty. @@ -992,9 +1147,17 @@ class MoAChatCompletions: extra_body: Any = agg_kwargs.get("extra_body") # Record the exact aggregator INPUT (incl. the injected reference # context) into the pending trace so a trace captures what the - # aggregator actually saw, not a reconstruction. + # aggregator actually saw, not a reconstruction. Traces are a + # persisted surface: when the privacy filter is active, the stored + # COPY is redacted ('display' mode's live aggregator input stays raw — + # only the on-disk record is filtered; 'full' mode's input is already + # redacted upstream, so this is a near no-op there). if self._pending_trace is not None: - self._pending_trace["aggregator_input_messages"] = agg_messages + self._pending_trace["aggregator_input_messages"] = ( + _redact_trace_messages([dict(m) for m in agg_messages]) + if getattr(self, "_privacy_mode", "") + else agg_messages + ) self._pending_trace["aggregator_label"] = _slot_label(aggregator) # The aggregator is the acting model. Resolve its slot to the provider's # real runtime (base_url/api_key/api_mode) and call it through the same @@ -1070,7 +1233,15 @@ class MoAChatCompletions: from hermes_cli.config import load_config from hermes_cli.moa_config import resolve_moa_preset - preset = resolve_moa_preset(load_config().get("moa") or {}, self.preset_name) + _moa_raw = load_config().get("moa") or {} + preset = resolve_moa_preset(_moa_raw, self.preset_name) + # Privacy filter mode: '' (off, default) | 'display' | 'full'. See + # coerce_privacy_filter / the pattern block at the top of this module. + # Remembered on self so _call_prepared_aggregator (which may run on a + # later prepared-request call without re-reading config) redacts the + # trace's aggregator input consistently with this turn's fan-out. + privacy_mode = _moa_privacy_mode(_moa_raw) + self._privacy_mode = privacy_mode messages = list(api_kwargs.get("messages") or []) reference_models = preset.get("reference_models") or [] aggregator = preset.get("aggregator") or {} @@ -1121,9 +1292,29 @@ class MoAChatCompletions: # 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. + # "every_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 "per_iteration").strip().lower() + every_n = 0 + if fanout_mode.startswith("every_n:"): + try: + every_n = int(fanout_mode.split(":", 1)[1]) + except (TypeError, ValueError): + every_n = 0 + if every_n < 2: + # Unparseable / degenerate cadence degrades to the default, + # mirroring _coerce_fanout's tolerant-read contract. + fanout_mode = "per_iteration" sig_messages = ref_messages - if fanout_mode == "user_turn": + turn_prefix = ref_messages + if fanout_mode in ("user_turn",) or every_n >= 2: # Find the last REAL user message. The advisory view appends a # synthetic user marker (_ADVISORY_INSTRUCTION) when it ends on an # assistant turn — i.e. on every tool iteration after the first — @@ -1138,19 +1329,51 @@ class MoAChatCompletions: last_user_idx = _i break if last_user_idx is not None: - sig_messages = ref_messages[: last_user_idx + 1] + turn_prefix = ref_messages[: last_user_idx + 1] + if fanout_mode == "user_turn": + sig_messages = turn_prefix + + 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() + + # every_n cadence bookkeeping: advance the per-turn iteration counter + # only when the advisory STATE actually advanced (a redundant create() + # with identical state — e.g. a streaming retry — must not consume a + # cadence slot), and reset it whenever the user-turn prefix changes. + _every_n_reuse = False + 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. + _on_cadence = (self._fanout_iteration_count - 1) % every_n == 0 + _every_n_reuse = not _on_cadence and bool(self._ref_cache_outputs) # Turn-scoped cache: only run + display references when the advisory # view changed (i.e. a new user turn). Within one turn the agent loop # calls create() once per tool iteration; in user_turn mode the # signature is stable across those iterations (prefix hash above), so # the fan-out runs once per user turn and iterations reuse the advice. - _sig = hashlib.sha256( - "\u0000".join( - f"{m.get('role')}:{m.get('content')}" for m in sig_messages - ).encode("utf-8", "replace") - ).hexdigest() + _sig = _hash_messages(sig_messages) _cache_key = (self.preset_name, _sig, tuple(_slot_label(s) for s in reference_models)) + if _every_n_reuse: + # Off-cadence every_n iteration: pin the key to the last + # on-cadence run so the lookup below is a HIT and its guidance is + # reused (no advisor calls, no double accounting, no re-emit) — + # exactly the user_turn cache-HIT path. When the cache is empty + # (defensive; a new turn resets the counter to on-cadence) the + # flag above stays False and the references run normally. + _cache_key = self._ref_cache_key _refs_from_cache = _cache_key == self._ref_cache_key and bool(self._ref_cache_outputs) if _refs_from_cache: @@ -1197,9 +1420,19 @@ class MoAChatCompletions: # built; the aggregator OUTPUT is stitched in by the caller # (consume_and_save_trace) once the response resolves — the caller # holds the live session_id and the resolved aggregator response. + # Traces are a persisted, user-readable surface, so ANY active + # privacy mode ('display' or 'full') redacts the advisor text and + # the full per-advisor input/output carried by _RefAccounting. + if privacy_mode: + _trace_refs = [ + (label, _redact_reference_text(text), _redact_trace_accounting(acct)) + for label, text, acct in reference_outputs + ] + else: + _trace_refs = list(reference_outputs) self._pending_trace = { "preset": self.preset_name, - "reference_outputs": list(reference_outputs), + "reference_outputs": _trace_refs, "aggregator_slot": aggregator, "aggregator_temperature": aggregator_temperature, } @@ -1209,7 +1442,10 @@ class MoAChatCompletions: # actually ran them). The user sees one labelled block per # reference (rendered like a thinking block) so the MoA process is # visible rather than a silent pause. Best-effort: never blocks the - # turn. + # turn. Reference blocks are a user-visible surface: both privacy + # modes redact them (the cache keeps the RAW text — redaction + # always happens at the consuming surface, so a mid-session mode + # change never leaks or double-redacts). _ref_count = len(reference_outputs) for _idx, (_label, _text, _usage) in enumerate(reference_outputs, start=1): self._emit( @@ -1217,7 +1453,7 @@ class MoAChatCompletions: index=_idx, count=_ref_count, label=_label, - text=_text, + text=_redact_reference_text(_text) if privacy_mode else _text, ) if _ref_count: self._emit( @@ -1229,15 +1465,25 @@ class MoAChatCompletions: guidance: str | None = None agg_messages = [dict(m) for m in messages] if reference_outputs: + # 'full' privacy mode: redact the advisor text that reaches the + # AGGREGATOR too (issue #59959's literal ask). 'display' leaves + # the aggregator input raw so synthesis quality is unaffected. + # The redaction is applied to a per-call copy — the cache always + # holds raw advisor text (see the emit comment above). + _agg_refs = ( + _redact_reference_outputs(reference_outputs) + if privacy_mode == "full" + else reference_outputs + ) joined = "\n\n".join( f"Reference {idx} — {label}:\n{text}" - for idx, (label, text, _usage) in enumerate(reference_outputs, start=1) + for idx, (label, text, _usage) in enumerate(_agg_refs, start=1) ) guidance = ( "[Mixture of Agents reference context]\n" f"Preset: {self.preset_name}\n" f"Aggregator/acting model: {_slot_label(aggregator)}\n" - f"References: {', '.join(label for label, _, _ in reference_outputs)}\n\n" + 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\n" f"{joined}" diff --git a/hermes_cli/moa_config.py b/hermes_cli/moa_config.py index 9353587fe722..cfa6826fe1bd 100644 --- a/hermes_cli/moa_config.py +++ b/hermes_cli/moa_config.py @@ -68,9 +68,59 @@ def _coerce_int_or_none(value: Any) -> int | None: def _coerce_fanout(value: Any) -> str: - """Normalize the fan-out cadence; unknown values fall back to default.""" + """Normalize the fan-out cadence; unknown values fall back to default. + + Canonical values are the strings ``per_iteration``, ``user_turn``, and + ``every_n:`` (N >= 2). The ``every_n`` cadence also accepts the mapping + form ``{mode: every_n, n: N}`` from hand-edited YAML and normalizes it to + the canonical string, so the rest of the pipeline (presets, flattened + view, runtime) only ever sees one shape. ``every_n:1`` means "run every + iteration" and collapses to ``per_iteration``; anything unparseable falls + back to ``per_iteration`` (the tolerant-read contract of this module). + """ + if isinstance(value, dict): + # Mapping form: {mode: every_n, n: 3}. Non-every_n mapping modes fall + # through to the string path below (e.g. {mode: user_turn}). + mode = str(value.get("mode") or "").strip().lower() + if mode == "every_n": + n = _coerce_int(value.get("n"), 0) + return f"every_n:{n}" if n >= 2 else "per_iteration" + value = mode mode = str(value or "").strip().lower() - return mode if mode in {"per_iteration", "user_turn"} else "per_iteration" + if mode in {"per_iteration", "user_turn"}: + return mode + if mode.startswith("every_n"): + _, sep, rest = mode.partition(":") + n = _coerce_int(rest.strip(), 0) if sep else 0 + if n >= 2: + return f"every_n:{n}" + return "per_iteration" + + +def coerce_privacy_filter(value: Any) -> str: + """Normalize ``moa.privacy_filter`` to '' (off), 'display', or 'full'. + + - ``''`` (empty string): filter off — the default. ``false``/``None``/ + unknown values land here so a hand-edited config degrades to prior + behavior (tolerant-read contract). + - ``'display'``: redact user-visible surfaces only — the reference blocks + shown in the UI and the saved MoA trace records. The aggregator still + sees raw advisor text, so answer quality is unaffected. + - ``'full'``: additionally redact the advisor text injected into the + aggregator prompt (issue #59959's literal ask). A hand-edited boolean + ``true`` maps here because the issue framed the toggle as "redact + before passing to the aggregator". + """ + if value is True: + return "full" + if value is None or value is False: + return "" + mode = str(value).strip().lower() + if mode in {"display", "full"}: + return mode + if mode in {"true", "on", "yes", "1"}: + return "full" + return "" def _clean_reasoning_effort(value: Any) -> str | None: @@ -246,7 +296,11 @@ def _normalize_preset(raw: Any) -> dict[str, Any]: # iteration, so advice tracks live task state. "user_turn" runs the # advisors ONCE per user turn (the original MoA shape): the # aggregator gets their upfront plan-level advice, then acts alone - # for the rest of the tool loop. + # for the rest of the tool loop. "every_n:" (N >= 2) is the middle + # ground: advisors run on the first iteration of each user turn and + # every Nth tool iteration after it; in-between iterations reuse the + # cached guidance from the last advisor run. Also accepts the mapping + # form {mode: every_n, n: N}, normalized to the canonical string. "fanout": _coerce_fanout(raw.get("fanout")), } @@ -296,6 +350,10 @@ def normalize_moa_config(raw: Any) -> dict[str, Any]: "reference_max_tokens": active.get("reference_max_tokens"), "fanout": active.get("fanout", "per_iteration"), "enabled": active["enabled"], + # MoA-level (not per-preset) toggles ride at the top level alongside + # save_traces. privacy_filter: '' (off, default) | 'display' | 'full' + # — see coerce_privacy_filter for the semantics of each mode. + "privacy_filter": coerce_privacy_filter(raw.get("privacy_filter")), } diff --git a/tests/hermes_cli/test_moa_config.py b/tests/hermes_cli/test_moa_config.py index 2b4ae54c6274..ee16f873050e 100644 --- a/tests/hermes_cli/test_moa_config.py +++ b/tests/hermes_cli/test_moa_config.py @@ -583,3 +583,74 @@ def test_slot_max_tokens_absent_by_default(): ) ref = cfg["presets"]["p"]["reference_models"][0] assert "max_tokens" not in ref + + +# --- fanout cadence normalization (every_n) --- + + +def test_fanout_defaults_to_per_iteration(): + cfg = normalize_moa_config({}) + assert cfg["fanout"] == "per_iteration" + + +def test_fanout_every_n_string_form_normalized(): + cfg = normalize_moa_config({"fanout": "every_n:3"}) + assert cfg["fanout"] == "every_n:3" + assert cfg["presets"][DEFAULT_MOA_PRESET_NAME]["fanout"] == "every_n:3" + + +def test_fanout_every_n_mapping_form_normalized_to_string(): + cfg = normalize_moa_config({"fanout": {"mode": "every_n", "n": 4}}) + assert cfg["fanout"] == "every_n:4" + + +def test_fanout_every_n_degenerate_n_falls_back(): + # n=1 means "every iteration" — that IS per_iteration; n=0 / negative / + # garbage must never produce a broken cadence string. + assert normalize_moa_config({"fanout": "every_n:1"})["fanout"] == "per_iteration" + assert normalize_moa_config({"fanout": "every_n:0"})["fanout"] == "per_iteration" + assert normalize_moa_config({"fanout": "every_n:-2"})["fanout"] == "per_iteration" + assert normalize_moa_config({"fanout": "every_n:x"})["fanout"] == "per_iteration" + assert normalize_moa_config({"fanout": "every_n"})["fanout"] == "per_iteration" + assert normalize_moa_config({"fanout": {"mode": "every_n"}})["fanout"] == "per_iteration" + + +def test_fanout_every_n_round_trips_through_normalize(): + once = normalize_moa_config({"fanout": "every_n:3"}) + twice = normalize_moa_config(once) + assert twice["fanout"] == "every_n:3" + assert twice["presets"][DEFAULT_MOA_PRESET_NAME]["fanout"] == "every_n:3" + + +def test_fanout_mapping_user_turn_mode_accepted(): + cfg = normalize_moa_config({"fanout": {"mode": "user_turn"}}) + assert cfg["fanout"] == "user_turn" + + +# --- privacy_filter normalization --- + + +def test_privacy_filter_defaults_off(): + cfg = normalize_moa_config({}) + assert cfg["privacy_filter"] == "" + + +def test_privacy_filter_modes_normalized(): + from hermes_cli.moa_config import coerce_privacy_filter + + assert coerce_privacy_filter("display") == "display" + assert coerce_privacy_filter("FULL") == "full" + assert coerce_privacy_filter(True) == "full" # legacy boolean → issue #59959 ask + assert coerce_privacy_filter("true") == "full" + assert coerce_privacy_filter(False) == "" + assert coerce_privacy_filter(None) == "" + assert coerce_privacy_filter("bogus") == "" + assert coerce_privacy_filter("off") == "" + + +def test_privacy_filter_round_trips_through_normalize(): + once = normalize_moa_config({"privacy_filter": "display"}) + assert once["privacy_filter"] == "display" + assert normalize_moa_config(once)["privacy_filter"] == "display" + full = normalize_moa_config({"privacy_filter": "full"}) + assert normalize_moa_config(full)["privacy_filter"] == "full" diff --git a/tests/run_agent/test_moa_fanout_cadence.py b/tests/run_agent/test_moa_fanout_cadence.py new file mode 100644 index 000000000000..09cae419ee38 --- /dev/null +++ b/tests/run_agent/test_moa_fanout_cadence.py @@ -0,0 +1,222 @@ +"""every_n fanout cadence: advisors refresh every Nth tool iteration and +off-cadence iterations reuse the cached guidance from the last on-cadence run. + +Redesigned from PR #63448's intent (issue #63393 — advisor fan-out multiplies +turn latency/cost by the tool-iteration count). Unlike the submitted shape +(which dropped references entirely on off-cadence iterations), off-cadence +iterations here still feed the aggregator the LAST advisor guidance via the +same cache-reuse mechanism the user_turn fanout uses. +""" + +from types import SimpleNamespace + + +def _response(content="done", *, tool_calls=None): + message = SimpleNamespace(content=content, tool_calls=tool_calls or []) + choice = SimpleNamespace(message=message, finish_reason="stop") + return SimpleNamespace(choices=[choice], usage=None, model="fake-model") + + +def _cadence_config(home, fanout="every_n:3"): + home.mkdir() + (home / "config.yaml").write_text( + f""" +moa: + default_preset: review + presets: + review: + fanout: "{fanout}" + reference_models: + - provider: openai-codex + model: gpt-5.5 + aggregator: + provider: openrouter + model: anthropic/claude-opus-4.8 +""".strip(), + encoding="utf-8", + ) + + +def _install_fake_llm(monkeypatch, ref_runs): + def fake_call_llm(**kwargs): + if kwargs["task"] == "moa_reference": + ref_runs.append(kwargs["model"]) + return _response(f"advice #{len(ref_runs)}") + return _response("acted") + + monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm) + + +def _iteration_messages(base, iterations): + """Yield message lists simulating a growing tool loop: the base user turn, + then one new (assistant tool_call, tool result) pair per iteration.""" + msgs = list(base) + yield list(msgs) + for i in range(1, iterations): + msgs = msgs + [ + { + "role": "assistant", + "content": "", + "tool_calls": [ + {"id": f"c{i}", "function": {"name": "f", "arguments": "{}"}} + ], + }, + {"role": "tool", "tool_call_id": f"c{i}", "content": f"result {i}"}, + ] + yield list(msgs) + + +def test_every_n_cadence_runs_references_every_nth_iteration(monkeypatch, tmp_path): + """With every_n:3, references run on iterations 1 and 4 of a 6-iteration + tool loop (1 on-cadence, then every 3rd), not on all 6.""" + home = tmp_path / ".hermes" + _cadence_config(home, "every_n:3") + monkeypatch.setenv("HERMES_HOME", str(home)) + + ref_runs = [] + _install_fake_llm(monkeypatch, ref_runs) + + from agent.moa_loop import MoAChatCompletions + + events = [] + facade = MoAChatCompletions("review", reference_callback=lambda ev, **kw: events.append(ev)) + base = [{"role": "user", "content": "do the thing"}] + for msgs in _iteration_messages(base, 6): + facade.create(messages=msgs, tools=[{"type": "function"}]) + + # 1 reference model × iterations {1, 4} on-cadence = 2 advisor runs. + assert len(ref_runs) == 2 + # Display blocks only surface when references actually ran. + assert events.count("moa.reference") == 2 + assert events.count("moa.aggregating") == 2 + + +def test_every_n_off_cadence_iterations_reuse_cached_guidance(monkeypatch, tmp_path): + """Off-cadence iterations must still give the aggregator the last + on-cadence advisor guidance (cache reuse), not run advisor-less.""" + home = tmp_path / ".hermes" + _cadence_config(home, "every_n:3") + monkeypatch.setenv("HERMES_HOME", str(home)) + + ref_runs = [] + _install_fake_llm(monkeypatch, ref_runs) + + from agent.moa_loop import MoAChatCompletions + + facade = MoAChatCompletions("review") + base = [{"role": "user", "content": "task"}] + prepared = [ + facade.create(messages=msgs, tools=[], _moa_prepare_only=True) + for msgs in _iteration_messages(base, 3) + ] + + # Iteration 1 ran the references; iterations 2-3 are off-cadence. + assert len(ref_runs) == 1 + # Every iteration's aggregator request carries reference guidance... + assert all(p["guidance"] for p in prepared) + # ...and the off-cadence ones reuse iteration 1's exact advice text. + assert "advice #1" in prepared[0]["guidance"] + assert prepared[1]["guidance"] == prepared[0]["guidance"] + assert prepared[2]["guidance"] == prepared[0]["guidance"] + + +def test_every_n_off_cadence_does_not_double_charge_usage(monkeypatch, tmp_path): + """Cache-reuse iterations must not re-report advisor usage/cost.""" + home = tmp_path / ".hermes" + _cadence_config(home, "every_n:2") + monkeypatch.setenv("HERMES_HOME", str(home)) + + ref_runs = [] + _install_fake_llm(monkeypatch, ref_runs) + + from agent.moa_loop import MoAChatCompletions + + facade = MoAChatCompletions("review") + base = [{"role": "user", "content": "task"}] + it = _iteration_messages(base, 2) + + facade.create(messages=next(it), tools=[]) + usage1, _cost1 = facade.consume_reference_usage() + + facade.create(messages=next(it), tools=[]) # off-cadence: reuse + usage2, cost2 = facade.consume_reference_usage() + + assert len(ref_runs) == 1 + # The reuse iteration reports zero advisor usage and no cost. + assert usage2.input_tokens == 0 and usage2.output_tokens == 0 + assert cost2 is None + assert usage1 is not usage2 + + +def test_every_n_counter_resets_on_new_user_turn(monkeypatch, tmp_path): + """A new user message starts a new turn: iteration 1 is on-cadence again, + regardless of where the previous turn's counter stood.""" + home = tmp_path / ".hermes" + _cadence_config(home, "every_n:3") + monkeypatch.setenv("HERMES_HOME", str(home)) + + ref_runs = [] + _install_fake_llm(monkeypatch, ref_runs) + + from agent.moa_loop import MoAChatCompletions + + facade = MoAChatCompletions("review") + turn1 = [{"role": "user", "content": "turn one"}] + msgs: list = turn1 + for msgs in _iteration_messages(turn1, 2): + facade.create(messages=msgs, tools=[]) + assert len(ref_runs) == 1 # iteration 2 was off-cadence + + # New user turn appended after the tool loop → counter resets, advisors + # run immediately (fresh advice for the fresh request). + turn2 = msgs + [ + {"role": "assistant", "content": "done with turn one"}, + {"role": "user", "content": "turn two"}, + ] + facade.create(messages=turn2, tools=[]) + assert len(ref_runs) == 2 + + +def test_every_n_redundant_create_does_not_consume_cadence_slot(monkeypatch, tmp_path): + """A repeat create() with IDENTICAL state (e.g. a streaming retry) must not + advance the cadence counter — only real state changes count.""" + home = tmp_path / ".hermes" + _cadence_config(home, "every_n:2") + monkeypatch.setenv("HERMES_HOME", str(home)) + + ref_runs = [] + _install_fake_llm(monkeypatch, ref_runs) + + from agent.moa_loop import MoAChatCompletions + + facade = MoAChatCompletions("review") + base = [{"role": "user", "content": "task"}] + it = _iteration_messages(base, 3) + + first = next(it) + facade.create(messages=first, tools=[]) + facade.create(messages=first, tools=[]) # retry: same state, no slot used + assert facade._fanout_iteration_count == 1 + + facade.create(messages=next(it), tools=[]) # iteration 2: off-cadence + facade.create(messages=next(it), tools=[]) # iteration 3: on-cadence (every 2nd) + assert len(ref_runs) == 2 + + +def test_per_iteration_default_unchanged_by_cadence_state(monkeypatch, tmp_path): + """Default fanout still re-runs references on every state change.""" + home = tmp_path / ".hermes" + _cadence_config(home, "per_iteration") + monkeypatch.setenv("HERMES_HOME", str(home)) + + ref_runs = [] + _install_fake_llm(monkeypatch, ref_runs) + + from agent.moa_loop import MoAChatCompletions + + facade = MoAChatCompletions("review") + base = [{"role": "user", "content": "task"}] + for msgs in _iteration_messages(base, 3): + facade.create(messages=msgs, tools=[]) + + assert len(ref_runs) == 3 diff --git a/website/docs/user-guide/features/mixture-of-agents.md b/website/docs/user-guide/features/mixture-of-agents.md index 151a066dda49..d2d720bc3910 100644 --- a/website/docs/user-guide/features/mixture-of-agents.md +++ b/website/docs/user-guide/features/mixture-of-agents.md @@ -132,6 +132,66 @@ moa: Leave it unset (or `0`/blank) to keep the prior uncapped behavior. +### Advisor cadence with `fanout` + +By default the advisors re-run on **every tool iteration** (`fanout: +per_iteration`), so their advice always tracks the latest tool results — at +the cost of multiplying advisor latency and spend by the number of tool calls +in a turn. Two alternative cadences trade freshness for speed: + +- `fanout: user_turn` — advisors run **once per user turn**; every tool + iteration after that reuses the same upfront advice (the original MoA + shape: synthesize a plan, then let the acting model work). +- `fanout: every_n:3` — the middle ground: advisors run on the **first** + iteration of each user turn and then every **3rd** tool iteration (any + `N >= 2` works). Iterations in between reuse the cached guidance from the + last advisor run, so the aggregator still gets advice on every step — it is + just refreshed every N steps instead of every step. The counter resets on + each new user message, so every turn starts with fresh advice. The mapping + form `fanout: {mode: every_n, n: 3}` is also accepted and normalized to + the string form. + +```yaml +moa: + presets: + fast: + reference_models: + - provider: openrouter + model: anthropic/claude-opus-4.8 + aggregator: + provider: openrouter + model: openai/gpt-5.5 + fanout: every_n:3 # advisors refresh every 3rd tool iteration +``` + +Unknown or malformed values fall back to `per_iteration`. + +### Privacy filter for advisor outputs + +Advisor outputs can echo sensitive data from the conversation — emails, +formatted phone numbers, API keys, JWTs — into the reference blocks shown in +the UI, saved MoA traces, and the aggregator prompt. `moa.privacy_filter` +(off by default) redacts those surfaces: + +```yaml +moa: + privacy_filter: display # or: full +``` + +- `display` — redacts **user-visible surfaces only**: the labelled reference + blocks rendered in the UI and the records written by `save_traces`. The + aggregator still receives the raw advisor text, so answer quality is + unaffected. +- `full` — additionally redacts the advisor text injected into the + aggregator prompt (and the one-shot `/moa` synthesis input). + +Credential shapes (API-key prefixes, JWTs, private keys, DB connection +strings) are masked by Hermes' central secret redactor; the MoA filter adds +email and clearly formatted phone-number redaction on top. Patterns are +deliberately conservative for code-review-style advice: bare digit runs, line +numbers, timestamps, git SHAs, and IP addresses are never touched — only +delimited phone formats like `(555) 123-4567` or `555-123-4567` match. + ### Per-slot reasoning effort Reference and aggregator slots may also set `reasoning_effort`. Use this when