hermes-agent/tests/run_agent/test_moa_loop_mode.py
Teknium 7b65073dc9 fix(moa): tolerate SDK-shaped tool_call entries in _render_tool_calls
_render_tool_calls only handled dict-shaped entries; a SimpleNamespace-
shaped tool_call (SDK-style stream-stitched responses) rendered as
'[called tool: tool]', silently losing the function name and arguments
from the advisory view. Handle both shapes (including a namespace-shaped
nested function inside a dict entry).

One-hunk hardening salvaged from closed #59712.

Co-authored-by: SquabbyZ <601709253@qq.com>
2026-07-23 18:40:09 -07:00

2666 lines
96 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from run_agent import AIAgent
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 test_moa_virtual_provider_aggregator_is_actor(monkeypatch, tmp_path):
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openai-codex
model: gpt-5.5
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
calls = []
def fake_call_llm(**kwargs):
calls.append(kwargs)
if kwargs["task"] == "moa_reference":
return _response("reference advice")
return _response("aggregator acted")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
agent = AIAgent(
api_key="moa-virtual-provider",
base_url="http://127.0.0.1/v1",
model="review",
provider="moa",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
enabled_toolsets=["file"],
max_iterations=1,
)
monkeypatch.setattr(
agent,
"_create_request_openai_client",
lambda *_args, **_kwargs: (_ for _ in ()).throw(
AssertionError("MoA calls must use MoAClient, not a request OpenAI client")
),
)
result = agent.run_conversation("solve this")
assert result["final_response"] == "aggregator acted"
assert agent.base_url == "moa://local"
assert [(c["task"], c["provider"], c["model"]) for c in calls] == [
("moa_reference", "openai-codex", "gpt-5.5"),
("moa_aggregator", "openrouter", "anthropic/claude-opus-4.8"),
]
assert calls[1]["tools"] is not None
def test_moa_runtime_provider_uses_virtual_endpoint():
from hermes_cli.runtime_provider import resolve_runtime_provider
runtime = resolve_runtime_provider(requested="moa", target_model="review")
assert runtime["provider"] == "moa"
assert runtime["base_url"] == "moa://local"
assert runtime["api_key"] == "moa-virtual-provider"
def test_moa_primary_restore_rebuilds_virtual_facade(monkeypatch, tmp_path):
"""MoA sessions must restore from fallback without constructing OpenAI().
Regression for a long-lived MoA session that failed over to a real provider:
the next turn restored provider/model to MoA but tried to rebuild the shared
client from MoA's empty client_kwargs, raising "api_key client option must be
set" and then "Failed to recreate closed OpenAI client".
"""
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openai-codex
model: gpt-5.5
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
agent = AIAgent(
api_key="moa-virtual-provider",
base_url="moa://local",
model="review",
provider="moa",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
enabled_toolsets=["file"],
max_iterations=1,
)
primary_client = agent.client
def fail_openai_rebuild(*_args, **_kwargs):
raise AssertionError("MoA restore must not build a real OpenAI client")
monkeypatch.setattr(agent, "_create_openai_client", fail_openai_rebuild)
setattr(agent, "_fallback_activated", True)
setattr(agent, "provider", "zai")
setattr(agent, "model", "glm-5.2")
agent.base_url = "https://api.z.ai/api/coding/paas/v4"
agent.api_key = "fallback-key"
setattr(agent, "_client_kwargs", {"api_key": "fallback-key", "base_url": agent.base_url})
agent.client = SimpleNamespace(close=lambda: None, _client=SimpleNamespace(is_closed=True))
assert agent._restore_primary_runtime() is True
assert getattr(agent, "provider") == "moa"
assert getattr(agent, "model") == "review"
assert agent.client is not primary_client
assert hasattr(agent.client.chat, "completions")
assert getattr(agent, "_fallback_activated") is False
def test_moa_restored_facade_still_emits_reference_events(monkeypatch, tmp_path):
"""A restored MoA facade must keep the reference_callback relay wired.
Regression for the naive-rebuild flaw in the original #53802 approach:
``MoAClient(preset)`` without ``reference_callback`` restores a *working*
facade that silently stops emitting ``moa.reference``/``moa.aggregating``
display events for the rest of the session. The shared ``build_moa_facade``
factory rewires the relay to ``agent.tool_progress_callback`` on restore.
"""
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openai-codex
model: gpt-5.5
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
agent = AIAgent(
api_key="moa-virtual-provider",
base_url="moa://local",
model="review",
provider="moa",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
enabled_toolsets=["file"],
max_iterations=1,
)
# Simulate a fallback to a real provider, then restore.
setattr(agent, "_fallback_activated", True)
setattr(agent, "provider", "zai")
setattr(agent, "model", "glm-5.2")
agent.base_url = "https://api.z.ai/api/coding/paas/v4"
agent.api_key = "fallback-key"
setattr(agent, "_client_kwargs", {"api_key": "fallback-key", "base_url": agent.base_url})
agent.client = SimpleNamespace(close=lambda: None, _client=SimpleNamespace(is_closed=True))
assert agent._restore_primary_runtime() is True
# The relay reads tool_progress_callback at emit time — attach a recorder
# and fire the facade's internal _emit exactly as the fan-out does.
events = []
def record_progress(event, *args, **kwargs):
events.append((event, args, kwargs))
agent.tool_progress_callback = record_progress
completions = agent.client.chat.completions
assert completions.reference_callback is not None, (
"restored MoA facade lost its reference_callback relay"
)
completions._emit(
"moa.reference", index=0, count=1, label="openai-codex/gpt-5.5", text="advice"
)
completions._emit("moa.aggregating", aggregator="openrouter", ref_count=1)
assert [e[0] for e in events] == ["moa.reference", "moa.aggregating"]
ref_event = events[0]
assert ref_event[1][0] == "openai-codex/gpt-5.5"
assert ref_event[1][1] == "advice"
assert ref_event[2] == {"moa_index": 0, "moa_count": 1}
def test_moa_does_not_cap_output_tokens(monkeypatch, tmp_path):
"""MoA must not inject an output cap on reference or aggregator calls.
The preset's old hardcoded max_tokens=4096 truncated long aggregator
syntheses. MoA now passes max_tokens=None (no caller cap), so call_llm
omits the parameter and each model uses its real maximum. Regression for
the "no limit on MoA models" fix.
"""
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
max_tokens: 4096
reference_models:
- provider: openai-codex
model: gpt-5.5
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
calls = []
def fake_call_llm(**kwargs):
calls.append(kwargs)
if kwargs["task"] == "moa_reference":
return _response("reference advice")
return _response("aggregator acted")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
agent = AIAgent(
api_key="moa-virtual-provider",
base_url="moa://local",
model="review",
provider="moa",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
enabled_toolsets=["file"],
max_iterations=1,
)
agent.run_conversation("solve this")
# Even with a preset max_tokens: 4096 present in config, neither the
# reference nor the aggregator call carries a cap — MoA passes None and
# call_llm omits the parameter so the model uses its full output budget.
ref_call = next(c for c in calls if c["task"] == "moa_reference")
agg_call = next(c for c in calls if c["task"] == "moa_aggregator")
assert ref_call.get("max_tokens") is None
assert agg_call.get("max_tokens") is None
def test_moa_slots_routed_through_resolve_runtime_provider(monkeypatch):
"""Reference + aggregator slots must be called via their provider's real
runtime (resolve_runtime_provider), not a bare provider/model call.
This is the "call any model the way it's called elsewhere" contract: each
slot's resolved base_url/api_key is passed through to call_llm so the
provider's actual API surface (anthropic_messages, max_completion_tokens,
custom endpoints) applies — same as if the model were the acting model.
"""
from agent import moa_loop
resolved = []
def fake_resolve(*, requested, target_model=None):
resolved.append((requested, target_model))
return {
"provider": requested,
"api_mode": "chat_completions",
"base_url": f"https://{requested}.example/v1",
"api_key": f"key-for-{requested}",
}
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider", fake_resolve
)
rt = moa_loop._slot_runtime({"provider": "minimax", "model": "MiniMax-M2"})
assert ("minimax", "MiniMax-M2") in resolved
assert rt["provider"] == "minimax"
assert rt["model"] == "MiniMax-M2"
assert rt["base_url"] == "https://minimax.example/v1"
assert rt["api_key"] == "key-for-minimax"
def test_moa_codex_slot_preserves_provider_identity(monkeypatch):
"""Codex slots must not become custom chat-completions endpoints.
_slot_runtime forwards the resolved base_url/api_key/api_mode; the single
chokepoint that must NOT collapse openai-codex to provider=custom is
_resolve_task_provider_model (via _preserve_provider_with_base_url). If it
collapsed, the Codex auxiliary branch — Cloudflare headers + Responses
adapter for chatgpt.com/backend-api/codex — would be bypassed.
"""
from agent import moa_loop
from agent.auxiliary_client import _resolve_task_provider_model
def fake_resolve(*, requested, target_model=None):
return {
"provider": requested,
"api_mode": "codex_responses",
"base_url": "https://chatgpt.com/backend-api/codex",
"api_key": "codex-oauth-token",
}
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider", fake_resolve
)
rt = moa_loop._slot_runtime({"provider": "openai-codex", "model": "gpt-5.5"})
# _slot_runtime forwards the resolved endpoint unconditionally now.
assert rt["provider"] == "openai-codex"
assert rt["model"] == "gpt-5.5"
assert rt["base_url"] == "https://chatgpt.com/backend-api/codex"
# The chokepoint preserves openai-codex identity despite the explicit
# base_url (api_mode is forwarded to call_llm directly, not the resolver).
resolver_kwargs = {k: v for k, v in rt.items() if k != "api_mode"}
resolved_provider, _model, base_url, _api_key, _mode = _resolve_task_provider_model(
task="moa_reference",
**resolver_kwargs,
)
assert resolved_provider == "openai-codex"
assert base_url == "https://chatgpt.com/backend-api/codex"
@pytest.mark.parametrize("provider", ["minimax-oauth", "qwen-oauth"])
def test_moa_provider_backed_slot_survives_aux_resolution(monkeypatch, provider):
"""MoA can pass resolved endpoints for provider-backed slots without
call_llm flattening them to generic custom endpoints.
``_slot_runtime`` resolves a provider-backed slot to ``provider`` plus a
concrete ``base_url``/``api_key``/``api_mode``; ``_run_reference`` then
forwards that dict to ``call_llm``. ``call_llm`` resolves the routing tuple
via ``_resolve_task_provider_model`` (which takes everything except
``api_mode``, handled separately). The provider identity must survive that
resolution rather than being flattened to ``custom``.
NOTE: providers in the ``_slot_runtime`` name-preservation set (anthropic,
bedrock, nous, openai-codex, xai-oauth) are intentionally NOT forwarded —
they're covered by their own dedicated tests. This case covers the
forward-the-resolved-endpoint path for providers that are NOT in the set.
"""
from agent import moa_loop
from agent.auxiliary_client import _resolve_task_provider_model
def fake_resolve(*, requested, target_model=None):
return {
"provider": requested,
"api_mode": "anthropic_messages",
"base_url": f"https://{requested}.example/v1",
"api_key": f"token-for-{requested}",
}
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider", fake_resolve
)
rt = moa_loop._slot_runtime({"provider": provider, "model": "test-model"})
# api_mode is forwarded to call_llm directly, not to _resolve_task_provider_model.
resolver_kwargs = {k: v for k, v in rt.items() if k != "api_mode"}
resolved_provider, model, base_url, api_key, _mode = _resolve_task_provider_model(
task="moa_reference",
**resolver_kwargs,
)
assert resolved_provider == provider
assert model == "test-model"
assert base_url == f"https://{provider}.example/v1"
assert api_key == f"token-for-{provider}"
def test_moa_copilot_reference_forwards_user_initiator_header(monkeypatch):
"""Copilot MoA advisors must carry the same user-turn attribution as main calls.
Copilot Pro/Pro+ gates some premium chat models on the ``x-initiator``
request header. MoA references are direct fan-out for the user's current
turn, so Copilot advisors need ``x-initiator: user`` rather than inheriting
the Copilot language-server default attribution.
"""
from agent import moa_loop
calls = []
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda _slot: {
"provider": "copilot",
"model": "claude-sonnet-4.6",
"api_mode": "chat_completions",
"base_url": "https://api.githubcopilot.com",
"api_key": "copilot-token",
},
)
def fake_call_llm(**kwargs):
calls.append(kwargs)
return _response("copilot advice")
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
_label, text, _acct = moa_loop._run_reference(
{"provider": "copilot", "model": "claude-sonnet-4.6"},
[{"role": "user", "content": "solve this"}],
)
assert text == "copilot advice"
assert calls[0]["task"] == "moa_reference"
assert calls[0]["extra_headers"] == {"x-initiator": "user"}
def test_moa_non_copilot_reference_does_not_forward_initiator_header(monkeypatch):
"""The Copilot attribution header must stay scoped to Copilot advisors."""
from agent import moa_loop
calls = []
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda _slot: {
"provider": "openrouter",
"model": "anthropic/claude-sonnet-4.6",
"api_mode": "chat_completions",
"base_url": "https://openrouter.ai/api/v1",
"api_key": "openrouter-token",
},
)
def fake_call_llm(**kwargs):
calls.append(kwargs)
return _response("openrouter advice")
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
_label, text, _acct = moa_loop._run_reference(
{"provider": "openrouter", "model": "anthropic/claude-sonnet-4.6"},
[{"role": "user", "content": "solve this"}],
)
assert text == "openrouter advice"
assert calls[0]["task"] == "moa_reference"
assert calls[0]["extra_headers"] is None
@pytest.mark.parametrize(
"provider_spelling",
["copilot", "github-copilot", "github", "github-models", "Copilot", "copilot-acp"],
)
def test_moa_copilot_alias_spellings_forward_initiator_header(
monkeypatch, provider_spelling
):
"""Every Copilot alias spelling must trigger the x-initiator header.
Slot configs spell the provider inconsistently (github, github-copilot,
github-models, copilot-acp, mixed case); the header gate goes through the
auxiliary client's canonical alias normalization so all of them get the
user-turn attribution, not just the literal string "copilot".
"""
from agent import moa_loop
calls = []
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda _slot: {
"provider": provider_spelling,
"model": "claude-sonnet-4.6",
"api_mode": "chat_completions",
"base_url": "https://api.githubcopilot.com",
"api_key": "copilot-token",
},
)
def fake_call_llm(**kwargs):
calls.append(kwargs)
return _response("copilot advice")
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
_label, text, _acct = moa_loop._run_reference(
{"provider": provider_spelling, "model": "claude-sonnet-4.6"},
[{"role": "user", "content": "solve this"}],
)
assert text == "copilot advice"
assert calls[0]["extra_headers"] == {"x-initiator": "user"}
def test_call_llm_extra_headers_reach_transport_create(monkeypatch):
"""extra_headers must reach the SDK client's create() kwargs.
Transport-boundary regression for #60293: mocking call_llm proves nothing
about delivery — this asserts the header survives call_llm's request
building and lands in the kwargs handed to chat.completions.create().
"""
from types import SimpleNamespace
from agent import auxiliary_client as ac
captured = {}
class _Completions:
def create(self, **kwargs):
captured.update(kwargs)
return _response("ok")
fake_client = SimpleNamespace(
chat=SimpleNamespace(completions=_Completions()),
base_url="https://api.githubcopilot.com",
)
monkeypatch.setattr(
ac,
"_resolve_task_provider_model",
lambda *a, **k: (
"copilot",
"claude-sonnet-4.6",
"https://api.githubcopilot.com",
"copilot-token",
"chat_completions",
),
)
monkeypatch.setattr(ac, "_get_cached_client", lambda *a, **k: (fake_client, "claude-sonnet-4.6"))
monkeypatch.setattr(ac, "_validate_llm_response", lambda resp, task, **_kw: resp)
ac.call_llm(
provider="copilot",
model="claude-sonnet-4.6",
messages=[{"role": "user", "content": "hi"}],
extra_headers={"x-initiator": "user"},
)
assert captured.get("extra_headers") == {"x-initiator": "user"}
# And it must not leak into unrelated request fields.
assert "x-initiator" not in captured.get("extra_body", {}) if captured.get("extra_body") else True
def test_retry_same_provider_sync_preserves_extra_headers(monkeypatch):
"""The same-provider retry rebuild must carry extra_headers through.
Regression for #60293's follow-up: a credential-refresh/pool-rotation
retry rebuilds the request kwargs from scratch — without forwarding
extra_headers, the retried Copilot advisor call silently loses its
``x-initiator: user`` attribution and can be rejected.
"""
from types import SimpleNamespace
from agent import auxiliary_client as ac
captured = {}
class _Completions:
def create(self, **kwargs):
captured.update(kwargs)
return _response("retried ok")
fake_client = SimpleNamespace(
chat=SimpleNamespace(completions=_Completions()),
base_url="https://api.githubcopilot.com",
)
monkeypatch.setattr(ac, "_get_cached_client", lambda *a, **k: (fake_client, "claude-sonnet-4.6"))
monkeypatch.setattr(ac, "_validate_llm_response", lambda resp, task, **_kw: resp)
ac._retry_same_provider_sync(
task=None,
resolved_provider="copilot",
resolved_model="claude-sonnet-4.6",
resolved_base_url="https://api.githubcopilot.com",
resolved_api_key="copilot-token",
resolved_api_mode="chat_completions",
main_runtime=None,
final_model="claude-sonnet-4.6",
messages=[{"role": "user", "content": "hi"}],
temperature=None,
max_tokens=None,
tools=None,
effective_timeout=30.0,
effective_extra_body={},
reasoning_config=None,
extra_headers={"x-initiator": "user"},
)
assert captured.get("extra_headers") == {"x-initiator": "user"}
def test_moa_gemini_aggregator_sanitize_uses_real_model(monkeypatch, tmp_path):
"""MoA turns must sanitize tool_calls against the AGGREGATOR model, not the preset.
Regression for #66212 / #65092: under MoA, ``agent.model`` holds the
virtual preset name (e.g. "review"), so passing it to
_sanitize_tool_calls_for_strict_api makes
_model_consumes_thought_signature() return False and strips
``extra_content`` (Gemini thought_signature) from replayed tool_calls —
the Gemini aggregator then 400s with "Function call is missing a
thought_signature in functionCall parts."
"""
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openai-codex
model: gpt-5.5
aggregator:
provider: gemini
model: gemini-3-pro-preview
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
sanitize_models = []
tool_call = SimpleNamespace(
id="call_1",
type="function",
function=SimpleNamespace(name="read_file", arguments='{"path": "x"}'),
)
responses = iter(
[
_response(None, tool_calls=[tool_call]),
_response("aggregator done"),
]
)
def fake_call_llm(**kwargs):
if kwargs["task"] == "moa_reference":
return _response("reference advice")
return next(responses)
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
agent = AIAgent(
api_key="moa-virtual-provider",
base_url="moa://local",
model="review",
provider="moa",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
enabled_toolsets=["file"],
max_iterations=3,
)
real_sanitize = type(agent)._sanitize_tool_calls_for_strict_api
def spy_sanitize(api_msg, model=None):
sanitize_models.append(model)
return real_sanitize(api_msg, model=model)
monkeypatch.setattr(
type(agent), "_sanitize_tool_calls_for_strict_api", staticmethod(spy_sanitize)
)
monkeypatch.setattr(
agent, "execute_tool", lambda *_a, **_k: "file contents", raising=False
)
result = agent.run_conversation("read the file")
assert result["final_response"] == "aggregator done"
# Once the history contains an assistant tool_call turn, the sanitize
# pass must be asked about the REAL aggregator model — never the virtual
# preset name (which would strip Gemini's thought_signature). The very
# first API call may still see the preset (the facade hasn't resolved a
# slot yet), but no tool_calls exist in history at that point.
assert any(m == "gemini-3-pro-preview" for m in sanitize_models), sanitize_models
first_resolved = sanitize_models.index("gemini-3-pro-preview")
assert all(
m == "gemini-3-pro-preview" for m in sanitize_models[first_resolved:]
), sanitize_models
def test_moa_slot_runtime_falls_back_on_resolution_error(monkeypatch):
"""A slot whose provider can't be resolved still attempts the call with the
bare provider/model rather than aborting the whole MoA turn."""
from agent import moa_loop
def boom(*, requested, target_model=None):
raise RuntimeError("unknown provider")
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider", boom
)
rt = moa_loop._slot_runtime({"provider": "mystery", "model": "x"})
assert rt == {"provider": "mystery", "model": "x"}
assert "base_url" not in rt
assert "api_key" not in rt
def test_reference_messages_drops_system_but_renders_tools_as_text():
"""System prompt is dropped, but tool calls + results are RENDERED as text.
A reference must see what the agent did (tool calls) and what came back
(tool results) to give an informed judgement — so neither is stripped. They
are flattened to text so the view carries zero tool-role messages / no
tool_calls arrays (strict providers reject those), while the reference
still has the full picture. The view ends on a user turn.
"""
from agent.moa_loop import _reference_messages
messages = [
{"role": "system", "content": "huge hermes system prompt"},
{"role": "user", "content": "do the thing"},
{
"role": "assistant",
"content": "",
"tool_calls": [{"id": "c1", "function": {"name": "f", "arguments": "{}"}}],
},
{"role": "tool", "tool_call_id": "c1", "content": "tool result"},
{"role": "assistant", "content": "here is my answer"},
]
view = _reference_messages(messages)
# Wire-format safety: only user/assistant text, no tool roles / tool_calls.
assert all(m["role"] in ("user", "assistant") for m in view)
assert all("tool_calls" not in m for m in view)
# System prompt is gone.
assert all("huge hermes system prompt" not in m["content"] for m in view)
# The agent's action and the tool result are PRESERVED as text.
joined = "\n".join(m["content"] for m in view)
assert "[called tool: f(" in joined
assert "[tool result: tool result]" in joined
assert "here is my answer" in joined
# Ends on a user turn (advisory request appended after the final assistant).
assert view[-1]["role"] == "user"
def test_reference_messages_ends_with_user_not_assistant_prefill():
"""Advisory reference views must never end on an assistant turn.
Mid-tool-loop the conversation ends on an assistant/tool exchange. Anthropic
(and OpenRouter→Anthropic) treat a trailing assistant turn as an assistant
prefill to continue, and no-prefill models (e.g. Claude Opus 4.8) reject it
with ``400 ... must end with a user message``. We append a synthetic user
turn asking for judgement rather than DELETING the agent's latest context —
the reference must still see the current state to advise on it.
"""
from agent.moa_loop import _reference_messages
messages = [
{"role": "user", "content": "q1"},
{"role": "assistant", "content": "a1"},
{"role": "user", "content": "q2 current"},
{
"role": "assistant",
"content": "let me reason then call a tool",
"tool_calls": [{"id": "c1", "function": {"name": "f", "arguments": "{}"}}],
},
{"role": "tool", "tool_call_id": "c1", "content": "the tool output"},
]
view = _reference_messages(messages)
assert view, "advisory view should not be empty"
assert view[-1]["role"] == "user"
joined = "\n".join(m["content"] for m in view)
# The agent's latest action and its result are preserved, not dropped.
assert "let me reason then call a tool" in joined
assert "[called tool: f(" in joined
assert "[tool result: the tool output]" in joined
# Earlier context preserved too.
assert "q1" in joined and "a1" in joined and "q2 current" in joined
def test_reference_messages_truncates_large_tool_results():
"""Large tool results are previewed head+tail, not replayed verbatim."""
from agent.moa_loop import _REFERENCE_TOOL_RESULT_BUDGET, _reference_messages
huge = "A" * (_REFERENCE_TOOL_RESULT_BUDGET * 3)
messages = [
{"role": "user", "content": "q"},
{
"role": "assistant",
"content": "",
"tool_calls": [{"id": "c1", "function": {"name": "f", "arguments": "{}"}}],
},
{"role": "tool", "tool_call_id": "c1", "content": huge},
]
view = _reference_messages(messages)
joined = "\n".join(m["content"] for m in view)
assert "chars omitted" in joined
# The folded result is far smaller than the raw payload.
assert len(joined) < len(huge)
def test_reference_messages_fresh_user_turn_ends_on_that_user():
"""A fresh user prompt with no agent action yet ends on that user turn."""
from agent.moa_loop import _reference_messages
messages = [
{"role": "system", "content": "sys"},
{"role": "user", "content": "q1"},
{"role": "assistant", "content": "a1"},
{"role": "user", "content": "q2 current"},
]
view = _reference_messages(messages)
assert view[-1] == {"role": "user", "content": "q2 current"}
def test_reference_messages_drops_empty_user_turns():
"""Empty user turns must not leak into the advisory view.
A user message whose content is "" or a non-string/multimodal payload
(flattened to "" by the text-extraction step) carries nothing advisory.
Strict providers (Kimi/Moonshot and others that enforce non-empty user
content) reject such a message with
400 "message ... with role 'user' must not be empty", while lenient
providers (DeepSeek) accept it — so a fan-out over the identical rendered
view fails on one reference and passes on another. The renderer must emit
NO empty user turn, mirroring how empty assistant turns are dropped.
"""
from agent.moa_loop import _reference_messages
messages = [
{"role": "system", "content": "sys"},
{"role": "user", "content": "real question"},
{"role": "assistant", "content": "", "tool_calls": [
{"function": {"name": "read_file", "arguments": '{"path":"c.yaml"}'}}
]},
{"role": "tool", "content": "some result"},
{"role": "user", "content": ""}, # empty string user turn
{"role": "user", "content": [{"type": "text", "text": "multimodal"}]}, # non-string -> ""
]
view = _reference_messages(messages)
# No user turn in the view may be empty/whitespace-only.
empty_users = [
m for m in view
if m.get("role") == "user" and not str(m.get("content", "")).strip()
]
assert empty_users == [], f"empty user turn leaked into advisory view: {empty_users}"
# The real user prompt survives and the view still ends on a user turn.
assert view[0] == {"role": "user", "content": "real question"}
assert view[-1]["role"] == "user"
def test_run_reference_prepends_advisory_system_prompt(monkeypatch):
"""Each reference call gets the advisory-role system prompt first.
Without it the reference assumes it is the acting agent and refuses ("I
can't access repositories/URLs from here") or tries to call tools it
doesn't have. The system prompt reframes it as an analyst advising the
aggregator, and the advisory transcript still ends on a user turn.
"""
from agent.moa_loop import _REFERENCE_SYSTEM_PROMPT, _run_reference
captured = {}
def fake_call_llm(**kwargs):
captured.update(kwargs)
return _response("advice")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
label, text, _acct = _run_reference(
{"provider": "openai-codex", "model": "gpt-5.5"},
[{"role": "user", "content": "review this PR"}],
)
assert text == "advice"
msgs = captured["messages"]
assert msgs[0] == {"role": "system", "content": _REFERENCE_SYSTEM_PROMPT}
assert msgs[-1]["role"] == "user"
def test_moa_facade_references_get_trimmed_messages(monkeypatch, tmp_path):
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openai-codex
model: gpt-5.5
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
calls = []
def fake_call_llm(**kwargs):
calls.append(kwargs)
return _response("ok")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
from agent.moa_loop import MoAChatCompletions
facade = MoAChatCompletions("review")
facade.create(
messages=[
{"role": "system", "content": "system prompt"},
{"role": "user", "content": "question"},
{
"role": "assistant",
"content": "checking",
"tool_calls": [{"id": "x", "function": {"name": "lookup", "arguments": "{}"}}],
},
{"role": "tool", "tool_call_id": "x", "content": "tool output"},
],
tools=[{"type": "function"}],
)
ref_call = next(c for c in calls if c["task"] == "moa_reference")
ref_msgs = ref_call["messages"]
# Advisory-role system prompt first; the agent's own system prompt is gone.
assert ref_msgs[0]["role"] == "system"
assert "reference advisor" in ref_msgs[0]["content"].lower()
assert "system prompt" not in ref_msgs[0]["content"]
# No tool-role messages and no tool_calls arrays leak to the reference.
assert all(m["role"] in ("system", "user", "assistant") for m in ref_msgs)
assert all("tool_calls" not in m for m in ref_msgs)
# The agent's action + tool result ARE preserved, rendered as text.
joined = "\n".join(m["content"] for m in ref_msgs[1:])
assert "[called tool: lookup(" in joined
assert "[tool result: tool output]" in joined
# Ends on a user turn (advisory request after the final assistant block).
assert ref_msgs[-1]["role"] == "user"
assert ref_call.get("tools") in (None, [])
# Aggregator still receives the original messages + tool schema.
agg_call = next(c for c in calls if c["task"] == "moa_aggregator")
assert agg_call["tools"] is not None
def test_moa_disabled_preset_skips_references(monkeypatch, tmp_path):
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
enabled: false
reference_models:
- provider: openai-codex
model: gpt-5.5
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
calls = []
def fake_call_llm(**kwargs):
calls.append(kwargs)
return _response("aggregator only")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
from agent.moa_loop import MoAChatCompletions
facade = MoAChatCompletions("review")
facade.create(messages=[{"role": "user", "content": "question"}], tools=[{"type": "function"}])
tasks = [c["task"] for c in calls]
# No reference fan-out — only the aggregator runs.
assert tasks == ["moa_aggregator"]
# Aggregator gets the unmodified user message (no MoA guidance appended).
agg_call = calls[0]
assert agg_call["messages"][-1]["content"] == "question"
def test_moa_disabled_reference_is_not_called(monkeypatch, tmp_path):
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openai-codex
model: gpt-5.5
enabled: false
- provider: openrouter
model: deepseek/deepseek-v4-pro
enabled: true
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
calls = []
def fake_call_llm(**kwargs):
calls.append(kwargs)
if kwargs["task"] == "moa_reference":
return _response(f"reference from {kwargs['provider']}:{kwargs['model']}")
return _response("aggregator acted")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
from agent.moa_loop import MoAChatCompletions
facade = MoAChatCompletions("review")
facade.create(messages=[{"role": "user", "content": "question"}], tools=[{"type": "function"}])
reference_calls = [c for c in calls if c["task"] == "moa_reference"]
assert [(c["provider"], c["model"]) for c in reference_calls] == [
("openrouter", "deepseek/deepseek-v4-pro")
]
assert calls[-1]["task"] == "moa_aggregator"
def test_references_run_in_parallel(monkeypatch):
"""References fan out concurrently (delegate-batch semantics), not serially.
Each reference sleeps; wall-time must approximate the slowest single call,
not the sum. Order is preserved and a failing reference is isolated.
"""
import time
from agent import moa_loop
# Force _extract_text down its fallback path (no transport normalize).
monkeypatch.setattr(moa_loop, "get_transport", lambda *_a, **_k: None)
barrier_hits = []
def slow_call_llm(**kwargs):
barrier_hits.append(time.monotonic())
model = kwargs["model"]
if model == "boom":
raise RuntimeError("kaboom")
time.sleep(0.5)
return _response(f"resp-{kwargs['provider']}")
monkeypatch.setattr(moa_loop, "call_llm", slow_call_llm)
refs = [
{"provider": "p1", "model": "ok"},
{"provider": "moa", "model": "preset"}, # recursion guard, not dispatched
{"provider": "p2", "model": "boom"}, # failure isolated
{"provider": "p3", "model": "ok"},
]
start = time.monotonic()
out = moa_loop._run_references_parallel(
refs, [{"role": "user", "content": "hi"}], temperature=0.6, max_tokens=64
)
elapsed = time.monotonic() - start
# Two 0.5s sleeps run concurrently → well under the 1.0s serial floor.
# Threshold sits at 0.95s (not tight against 0.5s) to tolerate CI
# thread-pool startup jitter while still failing hard if the two calls
# ran serially (which would be ≥1.0s).
assert elapsed < 0.95, f"references did not run in parallel (took {elapsed:.2f}s)"
# Output order matches input order (stable Reference N labelling).
assert [label for label, _, _ in out] == ["p1:ok", "moa:preset", "p2:boom", "p3:ok"]
assert "recursively reference MoA" in out[1][1]
assert out[2][1].startswith("[failed:")
assert out[0][1] == "resp-p1"
def test_references_parallel_without_agent_is_unaffected(monkeypatch):
"""No agent passed (the pre-fix call shape) must behave exactly as
before: block until every reference completes, no interrupt check."""
import time
from agent import moa_loop
monkeypatch.setattr(moa_loop, "get_transport", lambda *_a, **_k: None)
# Poll interval shorter than the reference's own sleep so the assertion
# below would catch a regression that waits a whole poll cycle extra.
monkeypatch.setattr(moa_loop, "_REFERENCE_POLL_INTERVAL_S", 0.05)
def slow_call_llm(**kwargs):
time.sleep(0.2)
return _response(f"resp-{kwargs['provider']}")
monkeypatch.setattr(moa_loop, "call_llm", slow_call_llm)
refs = [{"provider": "p1", "model": "ok"}]
out = moa_loop._run_references_parallel(
refs, [{"role": "user", "content": "hi"}],
)
assert out[0][1] == "resp-p1"
def test_references_parallel_interrupt_aborts_wait(monkeypatch):
"""A user interrupt mid-fanout must stop the wait instead of blocking
until every reference (including a wedged one) finishes or times out on
its own — mirroring the interrupt check agent.tool_executor already
applies to its own concurrent tool batch."""
import threading
import time
from agent import moa_loop
monkeypatch.setattr(moa_loop, "get_transport", lambda *_a, **_k: None)
monkeypatch.setattr(moa_loop, "_REFERENCE_POLL_INTERVAL_S", 0.05)
fake_agent = SimpleNamespace(_interrupt_requested=False)
release_wedged = threading.Event()
def fake_call_llm(**kwargs):
if kwargs["provider"] == "fast":
# Simulate the interrupt arriving right after the fast reference
# finishes, while the wedged one is still in flight.
fake_agent._interrupt_requested = True
return _response("fast output")
# "wedged" — never returns within the test unless released, standing
# in for a reference whose own (possibly very long) timeout hasn't
# elapsed yet.
release_wedged.wait(timeout=5)
return _response("should not be observed")
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
refs = [
{"provider": "fast", "model": "m1"},
{"provider": "wedged", "model": "m2"},
]
try:
start = time.monotonic()
out = moa_loop._run_references_parallel(
refs, [{"role": "user", "content": "hi"}], agent=fake_agent,
)
elapsed = time.monotonic() - start
# Must return promptly once interrupted, not block for the wedged
# reference's full (5s test-simulated) duration.
assert elapsed < 2.0, f"interrupt did not abort the wait (took {elapsed:.2f}s)"
assert out[0][1] == "fast output"
assert "interrupted" in out[1][1]
finally:
release_wedged.set() # don't leak a blocked thread past the test
def _ref_config(home):
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openai-codex
model: gpt-5.5
- provider: openrouter
model: anthropic/claude-opus-4.8
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
def test_moa_facade_emits_reference_then_aggregating(monkeypatch, tmp_path):
"""The facade reports each reference's output, then an aggregating signal,
so frontends can render reference blocks before the aggregator acts."""
home = tmp_path / ".hermes"
_ref_config(home)
monkeypatch.setenv("HERMES_HOME", str(home))
def fake_call_llm(**kwargs):
if kwargs["task"] == "moa_reference":
return _response(f"advice from {kwargs['model']}")
return _response("aggregator acted")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
from agent.moa_loop import MoAChatCompletions
events = []
facade = MoAChatCompletions("review", reference_callback=lambda ev, **kw: events.append((ev, kw)))
facade.create(messages=[{"role": "user", "content": "q"}], tools=[{"type": "function"}])
ref_events = [e for e in events if e[0] == "moa.reference"]
agg_events = [e for e in events if e[0] == "moa.aggregating"]
# One block per reference model, labelled by source, with index/count.
assert len(ref_events) == 2
assert ref_events[0][1]["label"] == "openai-codex:gpt-5.5"
assert ref_events[0][1]["index"] == 1 and ref_events[0][1]["count"] == 2
assert "advice from" in ref_events[0][1]["text"]
# Exactly one aggregating signal, after the references, naming the aggregator.
assert len(agg_events) == 1
assert agg_events[0][1]["aggregator"] == "openrouter:anthropic/claude-opus-4.8"
assert agg_events[0][1]["ref_count"] == 2
def test_moa_facade_reruns_references_on_new_tool_result(monkeypatch, tmp_path):
"""References re-run when a new tool result advances the task state.
The agent loop calls create() once per tool-loop iteration. References must
judge the LATEST state, so a new tool result is a cache MISS and re-runs the
references — but a redundant create() call with the SAME state is a cache
HIT (no re-run, no re-emit), so we don't fire on a pure no-op re-call.
"""
home = tmp_path / ".hermes"
_ref_config(home)
monkeypatch.setenv("HERMES_HOME", str(home))
ref_runs = []
def fake_call_llm(**kwargs):
if kwargs["task"] == "moa_reference":
ref_runs.append(kwargs["model"])
return _response("advice")
return _response("acted")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
from agent.moa_loop import MoAChatCompletions
events = []
facade = MoAChatCompletions("review", reference_callback=lambda ev, **kw: events.append(ev))
base_msgs = [{"role": "user", "content": "do the thing"}]
# Iteration 1: fresh user turn — references run (2 models).
facade.create(messages=base_msgs, tools=[{"type": "function"}])
after_tool = base_msgs + [
{"role": "assistant", "content": "", "tool_calls": [{"id": "c1", "function": {"name": "f", "arguments": "{}"}}]},
{"role": "tool", "tool_call_id": "c1", "content": "result"},
]
# Iteration 2: a NEW tool result advanced the state → references re-run.
facade.create(messages=after_tool, tools=[{"type": "function"}])
# Iteration 3: identical state (no new tool/user input) → cache hit, no re-run.
facade.create(messages=after_tool, tools=[{"type": "function"}])
# 2 models × 2 distinct states (fresh turn + new tool result) = 4 runs.
# The redundant 3rd call adds none.
assert len(ref_runs) == 4
assert events.count("moa.reference") == 4
assert events.count("moa.aggregating") == 2
def test_moa_facade_reruns_references_on_new_turn(monkeypatch, tmp_path):
"""A genuinely new user message invalidates the cache and re-runs refs."""
home = tmp_path / ".hermes"
_ref_config(home)
monkeypatch.setenv("HERMES_HOME", str(home))
ref_runs = []
def fake_call_llm(**kwargs):
if kwargs["task"] == "moa_reference":
ref_runs.append(kwargs["model"])
return _response("advice")
return _response("acted")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
from agent.moa_loop import MoAChatCompletions
facade = MoAChatCompletions("review")
facade.create(messages=[{"role": "user", "content": "turn one"}], tools=[])
facade.create(messages=[{"role": "user", "content": "turn two"}], tools=[])
# 2 references × 2 distinct turns = 4 reference runs.
assert len(ref_runs) == 4
def test_slot_runtime_anthropic_oauth_routes_through_provider_branch(monkeypatch):
"""Native anthropic slots must keep their provider identity, not collapse to custom.
anthropic OAuth setup-tokens (sk-ant-oat*) require Bearer auth + the
``anthropic-beta: oauth-*`` header, which only the anthropic provider branch
of call_llm adds. _slot_runtime forwards the resolved base_url/api_key for
every provider now; the single chokepoint that must NOT collapse anthropic
to provider=custom (which would send the token as x-api-key → bare 429) is
_resolve_task_provider_model via _preserve_provider_with_base_url.
"""
from agent import moa_loop
from agent.auxiliary_client import _resolve_task_provider_model
def fake_resolve(*, requested, target_model=None):
return {
"provider": requested,
"base_url": "https://resolved.example/v1",
"api_key": "resolved-key",
}
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider", fake_resolve
)
# _slot_runtime forwards the resolved endpoint for anthropic like any slot.
anthropic_rt = moa_loop._slot_runtime(
{"provider": "anthropic", "model": "claude-opus-4-8"}
)
assert anthropic_rt["provider"] == "anthropic"
assert anthropic_rt["base_url"] == "https://resolved.example/v1"
# The chokepoint preserves anthropic identity despite the explicit base_url,
# so call_llm routes through the anthropic provider branch (not custom).
resolved_provider, _model, base_url, _api_key, _mode = _resolve_task_provider_model(
task="moa_reference",
provider="anthropic",
model="claude-opus-4-8",
base_url="https://resolved.example/v1",
api_key="resolved-key",
)
assert resolved_provider == "anthropic"
# A generic provider (openrouter) is likewise forwarded and preserved.
other_rt = moa_loop._slot_runtime(
{"provider": "openrouter", "model": "some-model"}
)
assert other_rt["provider"] == "openrouter"
assert other_rt["model"] == "some-model"
assert other_rt["base_url"] == "https://resolved.example/v1"
assert other_rt["api_key"] == "resolved-key"
def _response_with_usage(content="advice", *, prompt=100, completion=50, cached=0):
"""A fake response carrying OpenAI-style usage so normalize_usage works."""
details = SimpleNamespace(cached_tokens=cached, cache_write_tokens=0)
usage = SimpleNamespace(
prompt_tokens=prompt,
completion_tokens=completion,
prompt_tokens_details=details,
output_tokens_details=None,
)
message = SimpleNamespace(content=content, tool_calls=[])
choice = SimpleNamespace(message=message, finish_reason="stop")
return SimpleNamespace(choices=[choice], usage=usage, model="fake-model")
def test_run_reference_captures_usage_and_cost(monkeypatch):
"""A reference call returns per-advisor CanonicalUsage + priced cost.
Before this, _run_reference discarded response.usage entirely, so the
advisor fan-out was invisible to cost tracking.
"""
from agent.moa_loop import _RefAccounting, _run_reference
from agent.usage_pricing import CanonicalUsage
monkeypatch.setattr(
"agent.moa_loop.call_llm",
lambda **kw: _response_with_usage(prompt=1000, completion=200, cached=400),
)
# Keep runtime resolution + pricing deterministic.
monkeypatch.setattr(
"agent.moa_loop._slot_runtime",
lambda slot: {"provider": "openrouter", "model": slot.get("model")},
)
monkeypatch.setattr(
"agent.usage_pricing.estimate_usage_cost",
lambda *a, **k: SimpleNamespace(amount_usd=0.0123, status="estimated", source="table"),
)
label, text, acct = _run_reference(
{"provider": "openrouter", "model": "vendor/adv-model"},
[{"role": "user", "content": "state?"}],
)
assert text == "advice"
assert isinstance(acct, _RefAccounting)
assert isinstance(acct.usage, CanonicalUsage)
# prompt_tokens=1000 with 400 cached → 600 fresh input + 400 cache_read.
assert acct.usage.input_tokens == 600
assert acct.usage.cache_read_tokens == 400
assert acct.usage.output_tokens == 200
assert acct.cost_usd == 0.0123
def test_references_parallel_sum_and_consume(monkeypatch, tmp_path):
"""create() sums advisor usage + cost once per turn; consume clears it.
Repeat tool-iterations within a turn reuse the cache and contribute ZERO
additional advisor spend (otherwise advisor cost multiplies by iteration
count).
"""
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openrouter
model: adv-a
- provider: openrouter
model: adv-b
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
def fake_call_llm(**kwargs):
if kwargs["task"] == "moa_reference":
return _response_with_usage(prompt=1000, completion=100, cached=0)
return _response("aggregator acted")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
monkeypatch.setattr(
"agent.moa_loop._slot_runtime",
lambda slot: {"provider": "openrouter", "model": slot.get("model")},
)
monkeypatch.setattr(
"agent.usage_pricing.estimate_usage_cost",
lambda *a, **k: SimpleNamespace(amount_usd=0.01, status="estimated", source="table"),
)
from agent.moa_loop import MoAChatCompletions
facade = MoAChatCompletions("review")
facade.create(messages=[{"role": "user", "content": "turn one"}], tools=[])
usage, cost = facade.consume_reference_usage()
# Two advisors × (1000 input, 100 output) = 2000 input, 200 output.
assert usage.input_tokens == 2000
assert usage.output_tokens == 200
# Two advisors × $0.01 each = $0.02.
assert cost == pytest.approx(0.02)
# consume clears — a second consume with no new create() is zeroed.
usage2, cost2 = facade.consume_reference_usage()
assert usage2.input_tokens == 0
assert cost2 is None
# A repeat create() with the SAME advisory view is a cache HIT: advisors
# do not re-run, so pending advisor spend is zero (no double-charge).
facade.create(messages=[{"role": "user", "content": "turn one"}], tools=[])
usage3, cost3 = facade.consume_reference_usage()
assert usage3.input_tokens == 0
assert cost3 is None
def test_canonical_usage_add():
"""CanonicalUsage sums per bucket (used to fold advisor tokens in)."""
from agent.usage_pricing import CanonicalUsage
a = CanonicalUsage(input_tokens=100, output_tokens=20, cache_read_tokens=5)
b = CanonicalUsage(input_tokens=50, output_tokens=10, cache_write_tokens=3)
total = a + b
assert total.input_tokens == 150
assert total.output_tokens == 30
assert total.cache_read_tokens == 5
assert total.cache_write_tokens == 3
assert total.request_count == 2
def test_moa_full_trace_written_when_enabled(monkeypatch, tmp_path):
"""With moa.save_traces on, a full MoA turn is written to JSONL.
Asserts the record captures each reference's FULL input messages + output
and the aggregator's FULL input (incl. injected reference guidance) +
output — the true full turn, auditable offline.
"""
import json
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
save_traces: true
default_preset: review
presets:
review:
reference_models:
- provider: openrouter
model: adv-a
- provider: openrouter
model: adv-b
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
def fake_call_llm(**kwargs):
if kwargs["task"] == "moa_reference":
# Echo the model so we can prove per-reference output is captured.
model = kwargs.get("model", "?")
return _response_with_usage(content=f"advice from {model}", prompt=500, completion=80)
return _response("AGGREGATOR FINAL ANSWER")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
monkeypatch.setattr(
"agent.moa_loop._slot_runtime",
lambda slot: {"provider": "openrouter", "model": slot.get("model")},
)
monkeypatch.setattr(
"agent.usage_pricing.estimate_usage_cost",
lambda *a, **k: SimpleNamespace(amount_usd=0.001, status="estimated", source="table"),
)
from agent.moa_loop import MoAChatCompletions
facade = MoAChatCompletions("review")
# Non-streaming create() → aggregator output captured inline.
facade.create(messages=[{"role": "user", "content": "please review the plan"}], tools=[])
facade.consume_and_save_trace(session_id="sess-xyz")
trace_file = home / "moa-traces" / "sess-xyz.jsonl"
assert trace_file.exists(), "trace file not written"
lines = trace_file.read_text(encoding="utf-8").strip().splitlines()
assert len(lines) == 1
rec = json.loads(lines[0])
# Turn framing.
assert rec["session_id"] == "sess-xyz"
assert rec["preset"] == "review"
# Both references captured, each with FULL input messages + output.
assert len(rec["references"]) == 2
for ref in rec["references"]:
assert ref["model"] in ("adv-a", "adv-b")
assert ref["provider"] == "openrouter"
# Full input messages present (system advisory prompt + advisory view).
assert isinstance(ref["input_messages"], list) and len(ref["input_messages"]) >= 2
assert ref["input_messages"][0]["role"] == "system"
# Full output present and model-specific.
assert ref["output"] == f"advice from {ref['model']}"
assert ref["usage"]["input_tokens"] == 500
assert ref["cost_usd"] == 0.001
# Aggregator: full input (with injected reference guidance) + inline output.
agg = rec["aggregator"]
assert agg["model"] == "anthropic/claude-opus-4.8"
assert agg["streamed"] is False
assert agg["output"] == "AGGREGATOR FINAL ANSWER"
agg_text = json.dumps(agg["input_messages"])
assert "Mixture of Agents reference context" in agg_text
assert "advice from adv-a" in agg_text and "advice from adv-b" in agg_text
def test_moa_trace_not_written_when_disabled(monkeypatch, tmp_path):
"""Default (save_traces off) writes nothing."""
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openrouter
model: adv-a
aggregator:
provider: openrouter
model: anthropic/claude-opus-4.8
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
def fake_call_llm(**kwargs):
if kwargs["task"] == "moa_reference":
return _response_with_usage(content="advice")
return _response("acted")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
monkeypatch.setattr(
"agent.moa_loop._slot_runtime",
lambda slot: {"provider": "openrouter", "model": slot.get("model")},
)
from agent.moa_loop import MoAChatCompletions
facade = MoAChatCompletions("review")
facade.create(messages=[{"role": "user", "content": "hi"}], tools=[])
facade.consume_and_save_trace(session_id="sess-off")
assert not (home / "moa-traces").exists()
def test_reference_guidance_appended_at_end_in_tool_loop():
"""In an agentic loop the reference block must land at the END of the prompt.
The most recent user turn is the original task near the top of the context;
merging the per-turn (volatile) reference block into it would diverge the
prompt prefix early and defeat the server's KV-cache reuse, forcing a full
re-prefill of the whole conversation on every tool-loop step.
"""
from agent.moa_loop import _attach_reference_guidance
messages = [
{"role": "system", "content": "system prompt"},
{"role": "user", "content": "ORIGINAL TASK"},
{"role": "assistant", "content": "", "tool_calls": [{"id": "1"}]},
{"role": "tool", "content": "tool result", "tool_call_id": "1"},
]
_attach_reference_guidance(messages, "REFERENCE BLOCK")
# The original (top-of-context) user turn is untouched, so the prefix stays
# cache-reusable across steps.
assert messages[1]["content"] == "ORIGINAL TASK"
# The reference block is appended as a new trailing turn, not merged upstream.
assert messages[-1]["role"] == "user"
assert messages[-1]["content"] == "REFERENCE BLOCK"
assert len(messages) == 5
def test_reference_guidance_merges_into_trailing_user_in_plain_chat():
"""Plain chat ends on the user turn, so the block merges there (still at end)."""
from agent.moa_loop import _attach_reference_guidance
messages = [
{"role": "system", "content": "system prompt"},
{"role": "user", "content": "hello"},
]
_attach_reference_guidance(messages, "REFERENCE BLOCK")
# No extra message; the block joins the trailing user turn (which is the end).
assert len(messages) == 2
assert messages[-1]["role"] == "user"
assert messages[-1]["content"] == "hello\n\nREFERENCE BLOCK"
def test_reference_messages_flattens_cache_decorated_content():
"""Cache-decorated turns (content-part lists) must not blind the references.
conversation_loop runs apply_anthropic_cache_control BEFORE the MoA facade
when the preset's aggregator is a cache-honoring Claude route (post-#57675).
That converts string content into [{"type": "text", "text": ...,
"cache_control": ...}] lists. The advisory view previously read only string
content, so the user's ENTIRE prompt flattened to "" — Claude references
then 400'd ("messages: at least one message is required") while tolerant
models answered "no user request is present" (live incident, Jul 14 2026,
preset "closed", session 20260714_001520_28157b).
"""
from agent.moa_loop import _reference_messages
from agent.prompt_caching import apply_anthropic_cache_control
plain = [
{"role": "system", "content": "hermes system prompt"},
{"role": "user", "content": "Can we get codex usage resets into hermes?"},
]
decorated = apply_anthropic_cache_control(plain, native_anthropic=False)
# Premise: decoration really converts the user turn to a content-part list.
assert isinstance(decorated[1]["content"], list)
view = _reference_messages(decorated)
assert view == [
{"role": "user", "content": "Can we get codex usage resets into hermes?"}
]
# Invariant: decorated and undecorated transcripts produce the SAME
# advisory view — so decoration can never change what references see,
# and the advisory prefix stays byte-stable for advisor prompt caching.
assert view == _reference_messages(plain)
def test_reference_messages_flattens_multimodal_user_turn():
"""Multimodal user turns (text + image parts) keep their text in the view.
Image parts carry no advisory text and are skipped; the text part must
survive. Previously the whole turn flattened to "".
"""
from agent.moa_loop import _reference_messages
messages = [
{"role": "user", "content": [
{"type": "text", "text": "what is in this screenshot?"},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
]},
]
view = _reference_messages(messages)
assert view == [{"role": "user", "content": "what is in this screenshot?"}]
# No base64 payload leaks into the advisory view.
assert all("base64" not in m["content"] for m in view)
def test_reference_messages_image_only_user_turn_gets_placeholder():
"""An image-only user turn must not become an empty user message.
Anthropic rejects empty text blocks (the original 400 class) and silently
skipping the turn would misalign user/assistant alternation in the view —
so a placeholder stands in for the non-text content.
"""
from agent.moa_loop import _reference_messages
messages = [
{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
]},
{"role": "assistant", "content": "I see a diagram."},
{"role": "user", "content": "now explain it"},
]
view = _reference_messages(messages)
assert view[0]["role"] == "user"
assert view[0]["content"].strip(), "image-only turn must not be empty"
assert "non-text" in view[0]["content"]
assert view[-1] == {"role": "user", "content": "now explain it"}
def test_reference_messages_flattens_structured_assistant_and_tool_content():
"""Assistant and tool turns with content-part lists are flattened too.
Multimodal tool results (e.g. computer_use screenshots) and adapter-shaped
assistant turns arrive as lists; their text must reach the references and
their image parts must not leak.
"""
from agent.moa_loop import _reference_messages
messages = [
{"role": "user", "content": "check the screen"},
{
"role": "assistant",
"content": [{"type": "text", "text": "taking a screenshot"}],
"tool_calls": [{"id": "c1", "function": {"name": "capture", "arguments": "{}"}}],
},
{"role": "tool", "tool_call_id": "c1", "content": [
{"type": "text", "text": "screenshot captured: login page visible"},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,BBBB"}},
]},
]
view = _reference_messages(messages)
joined = "\n".join(m["content"] for m in view)
assert "taking a screenshot" in joined
assert "[called tool: capture(" in joined
assert "[tool result: screenshot captured: login page visible]" in joined
assert "BBBB" not in joined
assert view[-1]["role"] == "user"
def test_reference_guidance_appends_text_part_to_decorated_trailing_user():
"""A cache-decorated trailing user turn still receives the guidance block.
Decoration converts the trailing user turn to a content-part list; the
guidance must be appended as a NEW text part AFTER the cache_control-marked
part (cached prefix stays byte-stable, no consecutive-user-turn 400s), not
silently dropped and not added as a second user message.
"""
from agent.moa_loop import _attach_reference_guidance
marked_part = {
"type": "text",
"text": "hello",
"cache_control": {"type": "ephemeral"},
}
messages = [
{"role": "system", "content": "system prompt"},
{"role": "user", "content": [dict(marked_part)]},
]
_attach_reference_guidance(messages, "REFERENCE BLOCK")
# No extra message (would break user/user alternation).
assert len(messages) == 2
content = messages[-1]["content"]
assert isinstance(content, list) and len(content) == 2
# The cache-marked part is byte-identical (prefix stability).
assert content[0] == marked_part
# The guidance rides as a trailing text part outside the cached span.
assert content[1] == {"type": "text", "text": "\n\nREFERENCE BLOCK"}
def test_reference_messages_drops_whitespace_only_string_user_turn():
"""A whitespace-only STRING user turn is dropped, not placeholdered.
The non-text placeholder exists for structured content (image-only turns)
where a real turn happened that the reference should know about. A bare
whitespace string carries nothing — emitting it would 400 strict
providers (Kimi/Moonshot 'role user must not be empty'), and
placeholdering it would fabricate an attachment that never existed.
"""
from agent.moa_loop import _reference_messages
messages = [
{"role": "user", "content": " "},
{"role": "assistant", "content": "a"},
{"role": "user", "content": "real"},
]
view = _reference_messages(messages)
assert view[0] == {"role": "assistant", "content": "a"}
assert view[-1] == {"role": "user", "content": "real"}
assert all(str(m["content"]).strip() for m in view)
def test_moa_pre_api_compression_includes_reference_guidance(monkeypatch, tmp_path):
"""The aggregator must not receive guidance that pushes it past compression.
The normal pre-API check sees only the persisted conversation. MoA adds
reference guidance later, inside ``MoAChatCompletions.create()``, so this
regression drives a raw request just below the threshold and makes the
injected guidance cross it. Compression must occur before the aggregator
request and leave the rebuilt request below the threshold.
"""
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openrouter
model: advisor
aggregator:
provider: openrouter
model: aggregator
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
events = []
compression_inputs = []
aggregator_request_tokens = []
def fake_estimate(messages, *args, **kwargs):
rendered = str(messages)
raw_tokens = 80 if "PRE_COMPACTION_HISTORY" in rendered else 20
guidance_tokens = 40 if "Mixture of Agents reference context" in rendered else 0
return raw_tokens + guidance_tokens
def fake_call_llm(**kwargs):
if kwargs["task"] == "moa_reference":
events.append("reference")
return _response("advisor guidance")
events.append("aggregator")
aggregator_request_tokens.append(fake_estimate(kwargs["messages"]))
return _response("aggregator acted")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
monkeypatch.setattr("agent.turn_context.estimate_request_tokens_rough", fake_estimate)
monkeypatch.setattr("agent.conversation_loop.estimate_request_tokens_rough", fake_estimate)
agent = AIAgent(
api_key="moa-virtual-provider",
base_url="moa://local",
model="review",
provider="moa",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
enabled_toolsets=["file"],
max_iterations=3,
)
compressor = getattr(agent, "context_compressor")
compressor.threshold_tokens = 100
def fake_compress(messages, *_args, **_kwargs):
events.append("compress")
compression_inputs.append(messages)
return ([{"role": "user", "content": "SUMMARY"}], "system")
monkeypatch.setattr(agent, "_compress_context", fake_compress)
result = agent.run_conversation(
"PRE_COMPACTION_HISTORY",
conversation_history=[{"role": "assistant", "content": "prior response"}],
)
assert result["final_response"] == "aggregator acted"
assert events.index("compress") < events.index("aggregator")
assert events.count("reference") == 1
assert all("Mixture of Agents reference context" not in str(item) for item in compression_inputs)
assert aggregator_request_tokens == [60]
def test_prepared_aggregator_preserves_reasoning_config(monkeypatch):
"""Prepared MoA requests retain the acting aggregator reasoning policy."""
from agent import moa_loop
captured = {}
expected_reasoning = {"enabled": True, "effort": "high"}
def fake_call_llm(**kwargs):
captured.update(kwargs)
return _response("aggregator acted")
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
monkeypatch.setattr(moa_loop, "_aggregator_reasoning_config", lambda _slot: expected_reasoning)
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda slot: {"provider": slot["provider"], "model": slot["model"]},
)
facade = moa_loop.MoAChatCompletions("review")
facade._call_prepared_aggregator(
{
"messages": [{"role": "user", "content": "question"}],
"aggregator": {"provider": "openrouter", "model": "aggregator"},
"aggregator_temperature": None,
},
{},
)
assert captured["reasoning_config"] == expected_reasoning
def test_reference_filtering_preserves_accounting_triples():
from agent.moa_loop import (
_RefAccounting,
_failed_reference_labels,
_successful_references,
)
from agent.usage_pricing import CanonicalUsage
good_accounting = _RefAccounting(CanonicalUsage(input_tokens=7), 0.07)
failed_accounting = _RefAccounting(CanonicalUsage(input_tokens=5), 0.05)
outputs = [
("good-model", "useful advice", good_accounting),
("bad-model", "[failed: raw provider secret]", failed_accounting),
]
successful = _successful_references(outputs)
assert successful == [outputs[0]]
assert successful[0][2] is good_accounting
assert _failed_reference_labels(outputs) == ["bad-model"]
def test_reference_filtering_excludes_recursion_guard_skips():
"""[skipped: …] recursion-guard notes are internal sentinels, not advice —
they must be filtered out of the aggregator prompt like [failed: …]."""
from agent.moa_loop import (
_RefAccounting,
_failed_reference_labels,
_is_failed_reference,
_successful_references,
)
from agent.usage_pricing import CanonicalUsage
outputs = [
("good-model", "useful advice", _RefAccounting(CanonicalUsage())),
(
"moa:nested",
"[skipped: MoA presets cannot recursively reference MoA]",
_RefAccounting(CanonicalUsage()),
),
]
assert _is_failed_reference(outputs[1][1])
assert _successful_references(outputs) == [outputs[0]]
assert _failed_reference_labels(outputs) == ["moa:nested"]
def test_aggregate_moa_context_sanitizes_failed_reference_and_forwards_timeout(monkeypatch):
from agent import moa_loop
from agent.usage_pricing import CanonicalUsage
outputs = [
("good-model", "useful advice", moa_loop._RefAccounting(CanonicalUsage())),
(
"bad-model",
"[failed: HTTP 401 key=super-secret]",
moa_loop._RefAccounting(CanonicalUsage()),
),
]
fanout_kwargs = {}
aggregator_calls = []
def fake_fanout(*args, **kwargs):
fanout_kwargs.update(kwargs)
return outputs
def fake_call_llm(**kwargs):
aggregator_calls.append(kwargs)
return _response("synthesized guidance")
monkeypatch.setattr(moa_loop, "_run_references_parallel", fake_fanout)
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda slot: {"provider": slot["provider"], "model": slot["model"]},
)
result = moa_loop.aggregate_moa_context(
user_prompt="review this",
api_messages=[{"role": "user", "content": "review this"}],
reference_models=[
{"provider": "openrouter", "model": "good-model"},
{"provider": "openrouter", "model": "bad-model"},
],
aggregator={"provider": "openrouter", "model": "aggregator"},
reference_timeout=17.5,
degraded_reference_policy="loud",
)
assert fanout_kwargs["reference_timeout"] == 17.5
private_prompt = aggregator_calls[0]["messages"][0]["content"]
assert "useful advice" in private_prompt
assert "super-secret" not in private_prompt
assert "Reference models unavailable: bad-model" in private_prompt
assert "super-secret" not in result
def test_moa_facade_sanitizes_failures_without_breaking_accounting(monkeypatch, tmp_path):
from agent import moa_loop
from agent.usage_pricing import CanonicalUsage
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_timeout: 19
degraded_reference_policy: silent
reference_models:
- provider: openrouter
model: good-model
- provider: openrouter
model: bad-model
aggregator:
provider: openrouter
model: aggregator
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
outputs = [
(
"good-model",
"useful advice",
moa_loop._RefAccounting(CanonicalUsage(input_tokens=7), 0.07),
),
(
"bad-model",
"[failed: HTTP 401 key=super-secret]",
moa_loop._RefAccounting(CanonicalUsage(input_tokens=5), 0.05),
),
]
fanout_kwargs = {}
aggregator_calls = []
def fake_fanout(*args, **kwargs):
fanout_kwargs.update(kwargs)
return outputs
def fake_call_llm(**kwargs):
aggregator_calls.append(kwargs)
return _response("aggregator acted")
monkeypatch.setattr(moa_loop, "_run_references_parallel", fake_fanout)
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda slot: {"provider": slot["provider"], "model": slot["model"]},
)
facade = moa_loop.MoAChatCompletions("review")
facade.create(messages=[{"role": "user", "content": "review this"}], tools=[])
assert fanout_kwargs["reference_timeout"] == 19.0
private_prompt = str(aggregator_calls[0]["messages"])
assert "useful advice" in private_prompt
assert "super-secret" not in private_prompt
assert "Reference models unavailable" not in private_prompt
usage, cost = facade.consume_reference_usage()
assert usage.input_tokens == 12
assert cost == pytest.approx(0.12)
assert facade._pending_trace["reference_outputs"] == outputs
def test_run_reference_forwards_configured_timeout(monkeypatch):
from agent import moa_loop
calls = []
def fake_call_llm(**kwargs):
calls.append(kwargs)
return _response("advice")
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda slot: {"provider": "openrouter", "model": slot["model"]},
)
monkeypatch.setattr(
"agent.usage_pricing.estimate_usage_cost",
lambda *args, **kwargs: SimpleNamespace(
amount_usd=None, status="unavailable", source=None
),
)
label, text, accounting = moa_loop._run_reference(
{"provider": "openrouter", "model": "advisor"},
[{"role": "user", "content": "review"}],
reference_timeout=23.5,
)
assert (label, text) == ("openrouter:advisor", "advice")
assert calls[0]["timeout"] == 23.5
assert isinstance(accounting, moa_loop._RefAccounting)
def test_aggregate_skips_aggregator_when_all_references_failed(monkeypatch):
"""When every reference returns [failed: …], the aggregator is skipped entirely."""
from agent.moa_loop import aggregate_moa_context
call_count = {"n": 0}
def fake_call_llm(**kwargs):
call_count["n"] += 1
if kwargs["task"] == "moa_reference":
raise RuntimeError("provider down key=super-secret")
raise AssertionError("aggregator should not be called when all references fail")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
monkeypatch.setattr(
"agent.moa_loop._slot_runtime",
lambda slot: {"provider": slot["provider"], "model": slot["model"]},
)
result = aggregate_moa_context(
user_prompt="do something",
api_messages=[{"role": "user", "content": "do something"}],
reference_models=[
{"provider": "openai", "model": "gpt-4"},
{"provider": "anthropic", "model": "claude-opus"},
],
aggregator={"provider": "openrouter", "model": "anthropic/claude-opus-4.8"},
)
# The aggregator LLM call was never made.
assert call_count["n"] == 2 # only the two reference calls
# The result carries a sanitized unavailability notice (never raw
# provider error text) so the main agent can still act.
assert "all reference models failed" in result
assert "Reference models unavailable" in result
assert "super-secret" not in result
def test_aggregate_skips_aggregator_when_all_references_skipped(monkeypatch):
"""References that are skipped (MoA recursion guard) also trigger the early return."""
from agent.moa_loop import aggregate_moa_context
def fake_call_llm(**kwargs):
raise AssertionError("aggregator should not be called when all references are skipped")
monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
# Both reference models are "moa" — they hit the recursion guard and are
# returned as "[skipped: …]" without calling call_llm at all.
result = aggregate_moa_context(
user_prompt="do something",
api_messages=[{"role": "user", "content": "do something"}],
reference_models=[
{"provider": "moa", "model": "preset-a"},
{"provider": "moa", "model": "preset-b"},
],
aggregator={"provider": "openrouter", "model": "anthropic/claude-opus-4.8"},
)
assert "all reference models failed" in result
assert "Reference models unavailable" in result
def _facade_all_failed_fixture(monkeypatch, tmp_path, policy):
"""Common scaffolding: a 'review' preset whose references ALL fail."""
from agent import moa_loop
from agent.usage_pricing import CanonicalUsage
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
f"""
moa:
default_preset: review
presets:
review:
degraded_reference_policy: {policy}
reference_models:
- provider: openrouter
model: bad-model-a
- provider: openrouter
model: bad-model-b
aggregator:
provider: openrouter
model: aggregator
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
outputs = [
(
"bad-model-a",
"[failed: HTTP 401 key=super-secret]",
moa_loop._RefAccounting(CanonicalUsage(input_tokens=5), 0.05),
),
(
"bad-model-b",
"[failed: timeout after 900s]",
moa_loop._RefAccounting(CanonicalUsage(input_tokens=3), 0.03),
),
]
aggregator_calls = []
def fake_call_llm(**kwargs):
aggregator_calls.append(kwargs)
return _response("aggregator acted alone")
monkeypatch.setattr(moa_loop, "_run_references_parallel", lambda *a, **k: outputs)
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda slot: {"provider": slot["provider"], "model": slot["model"]},
)
return moa_loop, outputs, aggregator_calls
def test_moa_facade_acts_aggregator_alone_when_all_references_fail_loud(
monkeypatch, tmp_path
):
"""Facade path (MoAChatCompletions.create): when every reference fails,
the aggregator acts alone — no 'use the reference responses below'
guidance wrapping a wall of failure sentinels. Under the loud policy the
sanitized unavailability notice is still disclosed."""
moa_loop, outputs, aggregator_calls = _facade_all_failed_fixture(
monkeypatch, tmp_path, "loud"
)
facade = moa_loop.MoAChatCompletions("review")
response = facade.create(
messages=[{"role": "user", "content": "review this"}], tools=[]
)
# The aggregator still acted (it IS the acting model)…
assert len(aggregator_calls) == 1
prompt = str(aggregator_calls[0]["messages"])
# …but got no failure sentinels or raw provider error text…
assert "[failed:" not in prompt
assert "super-secret" not in prompt
assert "Use the reference responses below" not in prompt
# …only the sanitized loud-policy notice.
assert "Reference models unavailable" in prompt
assert "bad-model-a" in prompt
# Accounting for the failed fan-out is still folded into the turn.
usage, cost = facade.consume_reference_usage()
assert usage.input_tokens == 8
assert cost == pytest.approx(0.08)
assert response.choices[0].message.content == "aggregator acted alone"
def test_moa_facade_acts_aggregator_alone_when_all_references_fail_silent(
monkeypatch, tmp_path
):
"""Silent policy: all-failed turns attach no reference guidance at all."""
moa_loop, _outputs, aggregator_calls = _facade_all_failed_fixture(
monkeypatch, tmp_path, "silent"
)
facade = moa_loop.MoAChatCompletions("review")
facade.create(messages=[{"role": "user", "content": "review this"}], tools=[])
assert len(aggregator_calls) == 1
prompt = str(aggregator_calls[0]["messages"])
assert "[failed:" not in prompt
assert "Reference models unavailable" not in prompt
assert "Mixture of Agents reference context" not in prompt
def test_interrupted_but_completed_reference_keeps_real_accounting(monkeypatch):
"""A reference that finishes between the interrupt check and the reap
must keep its REAL output and accounting — the call billed."""
from concurrent.futures import wait as real_wait
from agent import moa_loop
monkeypatch.setattr(moa_loop, "get_transport", lambda *_a, **_k: None)
monkeypatch.setattr(moa_loop, "_REFERENCE_POLL_INTERVAL_S", 0.05)
fake_agent = SimpleNamespace(_interrupt_requested=True)
def fake_call_llm(**kwargs):
return _response_with_usage("slowish output", prompt=11, completion=4)
# Force the exact race: the wait loop reports the future as still
# pending (so the interrupt path is taken) even though the underlying
# call has already completed — the reap must then hit the done() branch
# and keep the real result instead of writing a placeholder.
def fake_wait(pending, timeout=None):
real_wait(pending) # let the call actually finish (it billed)
return set(), set(pending) # report it as still pending
monkeypatch.setattr(moa_loop, "_futures_wait", fake_wait)
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda slot: {"provider": slot["provider"], "model": slot["model"]},
)
out = moa_loop._run_references_parallel(
[{"provider": "slowish", "model": "m1"}],
[{"role": "user", "content": "hi"}],
agent=fake_agent,
)
# The completed call's real output + usage must survive the reap.
assert out[0][1] == "slowish output"
acct = out[0][2]
assert isinstance(acct, moa_loop._RefAccounting)
assert acct.usage.input_tokens == 11
def test_late_completing_interrupted_reference_feeds_accounting_sink(monkeypatch):
"""A reference still in flight at interrupt time gets a placeholder in
the results, but its eventual REAL accounting must reach the sink."""
import threading
import time
from agent import moa_loop
monkeypatch.setattr(moa_loop, "get_transport", lambda *_a, **_k: None)
monkeypatch.setattr(moa_loop, "_REFERENCE_POLL_INTERVAL_S", 0.05)
fake_agent = SimpleNamespace(_interrupt_requested=False)
release = threading.Event()
sink_calls = []
sink_seen = threading.Event()
def sink(label, accounting):
sink_calls.append((label, accounting))
sink_seen.set()
def fake_call_llm(**kwargs):
if kwargs["provider"] == "fast":
fake_agent._interrupt_requested = True
return _response("fast output")
# wedged: blocks past the interrupt, completes later.
release.wait(timeout=5)
return _response_with_usage("late output", prompt=21, completion=2)
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda slot: {"provider": slot["provider"], "model": slot["model"]},
)
out = moa_loop._run_references_parallel(
[
{"provider": "fast", "model": "m1"},
{"provider": "wedged", "model": "m2"},
],
[{"role": "user", "content": "hi"}],
agent=fake_agent,
late_accounting_sink=sink,
)
# The wedged slot returned a placeholder with zeroed accounting…
assert out[1][1] == moa_loop._INTERRUPTED_REFERENCE_NOTE
assert out[1][2].usage.input_tokens == 0
# …then completes late; its real billed usage must reach the sink.
release.set()
assert sink_seen.wait(timeout=5), "late accounting sink never called"
label, acct = sink_calls[0]
assert "wedged" in label
assert acct.usage.input_tokens == 21
def test_facade_does_not_cache_interrupted_reference_results(monkeypatch, tmp_path):
"""An interrupted fan-out is a partial snapshot — caching it would replay
placeholder notes on every later iteration of the turn. The facade must
leave the cache empty so the next create() re-runs the references, and
a late-completing reference's real spend must land in pending usage."""
from agent import moa_loop
from agent.usage_pricing import CanonicalUsage
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"""
moa:
default_preset: review
presets:
review:
reference_models:
- provider: openrouter
model: advisor
aggregator:
provider: openrouter
model: aggregator
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(home))
interrupted_outputs = [
(
"openrouter:advisor",
moa_loop._INTERRUPTED_REFERENCE_NOTE,
moa_loop._RefAccounting(CanonicalUsage()),
)
]
def fake_fanout(*args, **kwargs):
return list(interrupted_outputs)
monkeypatch.setattr(moa_loop, "_run_references_parallel", fake_fanout)
monkeypatch.setattr(moa_loop, "call_llm", lambda **k: _response("acted"))
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda slot: {"provider": slot["provider"], "model": slot["model"]},
)
facade = moa_loop.MoAChatCompletions("review")
facade.create(messages=[{"role": "user", "content": "go"}], tools=[])
# Interrupted results must not be cached as this state's advice.
assert facade._ref_cache_key is None
assert facade._ref_cache_outputs == []
# A late completion depositing real spend is picked up by consume().
facade._record_late_reference_accounting(
"openrouter:advisor",
moa_loop._RefAccounting(CanonicalUsage(input_tokens=33), 0.42),
)
usage, cost = facade.consume_reference_usage()
assert usage.input_tokens == 33
assert cost == pytest.approx(0.42)
# And consume() drained it — no double count.
usage2, cost2 = facade.consume_reference_usage()
assert usage2.input_tokens == 0
assert cost2 is None
class _CountingCtxLen:
"""Stub for get_model_context_length that counts resolutions."""
def __init__(self, value):
self.value = value
self.calls = 0
def __call__(self, **kwargs):
self.calls += 1
if isinstance(self.value, Exception):
raise self.value
return self.value
def _trim(messages, *, window=1000, reserve=None, cache=None, counting=None,
monkeypatch=None):
from agent import model_metadata, moa_loop
stub = counting or _CountingCtxLen(window)
monkeypatch.setattr(model_metadata, "get_model_context_length", stub)
return moa_loop._trim_messages_for_reference(
messages,
{"provider": "openrouter", "model": "small-window"},
{"provider": "openrouter", "model": "small-window"},
reserve_output_tokens=reserve,
context_length_cache=cache,
)
def _advisory_view(n_pairs, chunk="x" * 400):
"""A text-only advisory view: system + n user/assistant pairs + trailing user."""
msgs = [{"role": "system", "content": "advisory system prompt"}]
for i in range(n_pairs):
msgs.append({"role": "user", "content": f"u{i} {chunk}"})
msgs.append({"role": "assistant", "content": f"a{i} {chunk}"})
msgs.append({"role": "user", "content": "judge the state above"})
return msgs
def test_reference_trim_untouched_when_within_window(monkeypatch):
msgs = _advisory_view(2)
out = _trim(list(msgs), window=10_000_000, monkeypatch=monkeypatch)
assert out == msgs
def test_reference_trim_drops_oldest_and_keeps_invariants(monkeypatch):
from agent.model_metadata import estimate_messages_tokens_rough
msgs = _advisory_view(30)
out = _trim(list(msgs), window=4000, reserve=100, monkeypatch=monkeypatch)
# Something was dropped and it fits the (window*0.9 - reserve) budget…
assert len(out) < len(msgs)
# System prompt survives at index 0.
assert out[0]["role"] == "system"
# User-first after the system prompt — never assistant-first.
assert out[1]["role"] == "user"
# Trailing synthetic user turn survives.
assert out[-1] == msgs[-1]
# Oldest frames were the ones dropped: the kept body is a contiguous
# suffix of the original body.
assert out[1:] == msgs[len(msgs) - len(out) + 1:]
def test_reference_trim_estimates_after_system_prompt(monkeypatch):
"""The advisory system prompt counts against the budget: a view that fits
without it but not with it must still be trimmed."""
from agent import model_metadata, moa_loop
msgs = _advisory_view(6)
body_tokens = model_metadata.estimate_messages_tokens_rough(msgs[1:])
total_tokens = model_metadata.estimate_messages_tokens_rough(msgs)
assert total_tokens > body_tokens
# Pick a window where (body fits) but (system+body does not):
# budget = window*0.9 - reserve(100) must sit between the two.
window = int((body_tokens + (total_tokens - body_tokens) / 2 + 100) / 0.9)
out = _trim(list(msgs), window=window, reserve=100, monkeypatch=monkeypatch)
assert len(out) < len(msgs)
assert out[0]["role"] == "system"
def test_reference_trim_reserves_output_tokens(monkeypatch):
"""With a huge output reserve the budget shrinks and forces a trim that
a reserve-less estimate would not need."""
msgs = _advisory_view(10)
from agent.model_metadata import estimate_messages_tokens_rough
total = estimate_messages_tokens_rough(msgs)
window = int((total + 500) / 0.9) # fits easily with a small reserve
out_small = _trim(list(msgs), window=window, reserve=100, monkeypatch=monkeypatch)
assert out_small == msgs
out_big = _trim(list(msgs), window=window, reserve=total, monkeypatch=monkeypatch)
assert len(out_big) < len(msgs)
def test_reference_trim_unresolvable_window_is_a_noop(monkeypatch):
msgs = _advisory_view(3)
stub = _CountingCtxLen(RuntimeError("metadata down"))
out = _trim(list(msgs), counting=stub, monkeypatch=monkeypatch)
assert out == msgs
def test_reference_trim_context_length_cache_hits_once(monkeypatch):
"""A shared per-turn cache resolves each (provider, model) window once."""
cache = {}
stub = _CountingCtxLen(10_000_000)
msgs = _advisory_view(2)
for _ in range(4):
_trim(list(msgs), cache=cache, counting=stub, monkeypatch=monkeypatch)
assert stub.calls == 1
assert cache == {("openrouter", "small-window"): 10_000_000}
def test_reference_trim_caches_resolution_failures(monkeypatch):
"""A failing metadata source is probed once, not per reference call."""
cache = {}
stub = _CountingCtxLen(RuntimeError("metadata down"))
msgs = _advisory_view(2)
for _ in range(3):
out = _trim(list(msgs), cache=cache, counting=stub, monkeypatch=monkeypatch)
assert out == msgs
assert stub.calls == 1
assert cache == {("openrouter", "small-window"): None}
def test_run_reference_trims_oversized_view_before_calling(monkeypatch):
"""End-to-end: _run_reference sends a trimmed request for a small-window
model instead of letting the provider 400."""
from agent import model_metadata, moa_loop
sent = {}
def fake_call_llm(**kwargs):
sent.update(kwargs)
return _response_with_usage("advice")
monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm)
monkeypatch.setattr(
moa_loop,
"_slot_runtime",
lambda slot: {"provider": "openrouter", "model": slot["model"]},
)
monkeypatch.setattr(
model_metadata, "get_model_context_length", lambda **k: 3000
)
view = _advisory_view(40)[1:] # advisory view has no system prompt
label, text, _acct = moa_loop._run_reference(
{"provider": "openrouter", "model": "small-window"},
view,
max_tokens=200,
)
assert text == "advice"
sent_messages = sent["messages"]
# System prompt was prepended and survived the trim.
assert sent_messages[0]["role"] == "system"
# The request actually shrank.
assert len(sent_messages) < len(view) + 1
# And ends with the synthetic trailing user turn.
assert sent_messages[-1]["role"] == "user"
def test_render_tool_calls_tolerates_namespace_shapes():
"""SDK-shaped (SimpleNamespace) tool_call entries must render their real
function name+args, not degrade to '[called tool: tool]'."""
from agent.moa_loop import _render_tool_calls
ns_call = SimpleNamespace(
function=SimpleNamespace(name="web_search", arguments='{"query": "x"}')
)
dict_call = {"function": {"name": "read_file", "arguments": '{"path": "y"}'}}
mixed = _render_tool_calls([ns_call, dict_call])
assert '[called tool: web_search({"query": "x"})]' in mixed
assert '[called tool: read_file({"path": "y"})]' in mixed
# Dict entry with a namespace-shaped nested function also renders.
hybrid = {"function": SimpleNamespace(name="terminal", arguments=None)}
assert _render_tool_calls([hybrid]) == "[called tool: terminal]"
# Degenerate shapes still fall back safely.
assert _render_tool_calls([SimpleNamespace()]) == "[called tool: tool]"