diff --git a/agent/agent_init.py b/agent/agent_init.py index f9d2d1e4759..807c44b78e8 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -722,7 +722,47 @@ def init_agent( elif agent.provider == "moa": from agent.moa_loop import MoAClient agent.api_mode = "chat_completions" - agent.client = MoAClient(agent.model or "default") + + # Route reference-model outputs to the agent's tool_progress_callback so + # every surface that already consumes it (CLI spinner/scrollback, TUI, + # desktop, gateway) can show each reference's answer as a labelled block + # before the aggregator acts. The facade emits "moa.reference" and + # "moa.aggregating" events; we forward them through the same callback + # the tool lifecycle uses. Best-effort and cache-safe — these are + # display-only events, they never touch the message history. + def _moa_reference_relay(event: str, **kwargs: Any) -> None: + cb = getattr(agent, "tool_progress_callback", None) + if cb is None: + return + try: + if event == "moa.reference": + label = str(kwargs.get("label") or "") + text = str(kwargs.get("text") or "") + idx = kwargs.get("index") + count = kwargs.get("count") + cb( + "moa.reference", + label, + text, + None, + moa_index=idx, + moa_count=count, + ) + elif event == "moa.aggregating": + cb( + "moa.aggregating", + str(kwargs.get("aggregator") or ""), + None, + None, + moa_ref_count=kwargs.get("ref_count"), + ) + except Exception: + pass + + agent.client = MoAClient( + agent.model or "default", + reference_callback=_moa_reference_relay, + ) agent._client_kwargs = {} agent.api_key = api_key or "moa-virtual-provider" agent.base_url = base_url or "moa://local" diff --git a/agent/moa_loop.py b/agent/moa_loop.py index 34a4c551dca..bab36fb4010 100644 --- a/agent/moa_loop.py +++ b/agent/moa_loop.py @@ -8,6 +8,7 @@ iteration. from __future__ import annotations +import hashlib import logging from concurrent.futures import ThreadPoolExecutor from typing import Any @@ -280,8 +281,37 @@ def aggregate_moa_context( class MoAChatCompletions: """OpenAI-chat-compatible facade where the aggregator is the acting model.""" - def __init__(self, preset_name: str): + def __init__(self, preset_name: str, reference_callback: Any = None): self.preset_name = preset_name or "default" + # Optional display hook. Called as reference outputs become available so + # frontends can show each reference model's answer as a labelled block + # before the aggregator acts. Signature: + # reference_callback(event, **kwargs) + # where event is one of: + # "moa.reference" kwargs: index, count, label, text + # "moa.aggregating" kwargs: aggregator (label), ref_count + # Never raises into the model call — display is best-effort. + self.reference_callback = reference_callback + # Turn-scoped reference cache. The agent loop calls create() once per + # tool-loop iteration, but references are advisory for the whole turn: + # the advisory message view (_reference_messages) is identical across + # iterations (it strips tool/tool_call turns) until a new user message + # arrives. Re-running references every iteration would multiply their + # API cost by the tool-loop depth AND re-emit the same blocks to the + # display on every iteration. So cache outputs keyed by the advisory + # view's signature and reuse them — running and showing references once + # per user turn. + self._ref_cache_key: tuple | None = None + self._ref_cache_outputs: list[tuple[str, str]] = [] + + def _emit(self, event: str, **kwargs: Any) -> None: + cb = self.reference_callback + if cb is None: + return + try: + cb(event, **kwargs) + except Exception as exc: # pragma: no cover - display must never break the turn + logger.debug("MoA reference_callback failed for %s: %s", event, exc) def create(self, **api_kwargs: Any) -> Any: from hermes_cli.config import load_config @@ -306,12 +336,52 @@ class MoAChatCompletions: reference_outputs: list[tuple[str, str]] = [] ref_messages = _reference_messages(messages) - reference_outputs = _run_references_parallel( - reference_models, - ref_messages, - temperature=temperature, - max_tokens=None, - ) + + # 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 with the same advisory view; + # reuse the cached outputs and skip both the re-run and the re-emit. + _sig = hashlib.sha256( + "\u0000".join( + f"{m.get('role')}:{m.get('content')}" for m in ref_messages + ).encode("utf-8", "replace") + ).hexdigest() + _cache_key = (self.preset_name, _sig, tuple(_slot_label(s) for s in reference_models)) + _refs_from_cache = _cache_key == self._ref_cache_key and bool(self._ref_cache_outputs) + + if _refs_from_cache: + reference_outputs = list(self._ref_cache_outputs) + else: + reference_outputs = _run_references_parallel( + reference_models, + ref_messages, + temperature=temperature, + max_tokens=None, + ) + self._ref_cache_key = _cache_key + self._ref_cache_outputs = list(reference_outputs) + + # Surface each reference model's answer to the display BEFORE the + # aggregator acts — once per turn (only on the iteration that + # 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. + _ref_count = len(reference_outputs) + for _idx, (_label, _text) in enumerate(reference_outputs, start=1): + self._emit( + "moa.reference", + index=_idx, + count=_ref_count, + label=_label, + text=_text, + ) + if _ref_count: + self._emit( + "moa.aggregating", + aggregator=_slot_label(aggregator), + ref_count=_ref_count, + ) agg_messages = [dict(m) for m in messages] if reference_outputs: @@ -359,6 +429,6 @@ class MoAChatCompletions: class MoAClient: - def __init__(self, preset_name: str): + def __init__(self, preset_name: str, reference_callback: Any = None): self.chat = type("_MoAChat", (), {})() - self.chat.completions = MoAChatCompletions(preset_name) + self.chat.completions = MoAChatCompletions(preset_name, reference_callback=reference_callback) diff --git a/cli.py b/cli.py index e62b6f2ef2b..101ac997d80 100644 --- a/cli.py +++ b/cli.py @@ -10474,6 +10474,36 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): stacked line to scrollback on tool.completed so users can see the full history of tool calls (not just the current one in the spinner). """ + # MoA reference-model outputs: render each reference's answer as a + # labelled thinking-style block BEFORE the aggregator acts, so the user + # sees the mixture-of-agents process instead of a silent pause. These + # are display-only events emitted by the MoA facade (agent_init relay); + # they never enter message history. + if event_type == "moa.reference": + label = function_name or "reference" + text = preview or "" + idx = kwargs.get("moa_index") + count = kwargs.get("moa_count") + header = f"Reference {idx}/{count} — {label}" if idx and count else f"Reference — {label}" + try: + self._flush_reasoning_preview(force=True) + except Exception: + pass + _cprint(f" {_DIM}┊ ◇ {header}{_RST}") + try: + self._emit_reasoning_preview(text) + except Exception: + # Fallback: print the raw text dimmed if the preview helper fails. + if text.strip(): + _cprint(f" {_DIM}{text.strip()}{_RST}") + self._invalidate() + return + if event_type == "moa.aggregating": + agg = function_name or "" + self._spinner_text = f"◆ aggregating ({agg})" if agg else "◆ aggregating" + self._invalidate() + return + # Feed the pet: tools mean "running" (not reasoning); a failed tool # latches the turn so it ends on a sulk. if event_type == "tool.started": diff --git a/tests/cli/test_tool_progress_scrollback.py b/tests/cli/test_tool_progress_scrollback.py index 906cfaf7822..2f5f6a8952d 100644 --- a/tests/cli/test_tool_progress_scrollback.py +++ b/tests/cli/test_tool_progress_scrollback.py @@ -251,3 +251,45 @@ class TestToolProgressScrollback: # First entry consumed, second remains assert len(cli._pending_tool_info.get("terminal", [])) == 1 assert cli._pending_tool_info["terminal"][0] == {"command": "pwd"} + + +class TestMoAReferenceBlocks: + """moa.reference renders a labelled thinking-style block; moa.aggregating + updates the spinner. Both are display-only and must commit regardless of + tool_progress_mode (MoA is non-streaming).""" + + def test_reference_event_prints_labelled_block(self): + cli = _make_cli(tool_progress="all") + with patch.object(_cli_mod, "_cprint") as mock_print: + cli._on_tool_progress( + "moa.reference", + "openrouter:openai/gpt-5.5", + "Paris is the capital.", + None, + moa_index=1, + moa_count=2, + ) + printed = " ".join(str(c.args[0]) for c in mock_print.call_args_list) + # Header names the source model + index/count; body carries the text. + assert "openrouter:openai/gpt-5.5" in printed + assert "Reference 1/2" in printed + assert "Paris is the capital." in printed + + def test_reference_event_prints_even_when_progress_off(self): + """Reference blocks are the MoA process view, not tool progress — they + must show even with tool_progress: off.""" + cli = _make_cli(tool_progress="off") + with patch.object(_cli_mod, "_cprint") as mock_print: + cli._on_tool_progress( + "moa.reference", "openrouter:anthropic/claude-opus-4.8", "Four.", None, + moa_index=2, moa_count=2, + ) + assert mock_print.called + + def test_aggregating_event_updates_spinner_only(self): + cli = _make_cli(tool_progress="all") + with patch.object(_cli_mod, "_cprint") as mock_print: + cli._on_tool_progress("moa.aggregating", "openrouter:anthropic/claude-opus-4.8", None, None) + assert "aggregating" in cli._spinner_text + # aggregating is a spinner-only transition; no committed scrollback line. + mock_print.assert_not_called() diff --git a/tests/run_agent/test_moa_loop_mode.py b/tests/run_agent/test_moa_loop_mode.py index ba8f86f03a4..af02cc1ff59 100644 --- a/tests/run_agent/test_moa_loop_mode.py +++ b/tests/run_agent/test_moa_loop_mode.py @@ -333,3 +333,130 @@ def test_references_run_in_parallel(monkeypatch): assert out[2][1].startswith("[failed:") assert out[0][1] == "resp-p1" + +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_caches_references_within_a_turn(monkeypatch, tmp_path): + """References run + emit ONCE per user turn, not per tool-loop iteration. + + The agent loop calls create() once per iteration; the advisory message + view is identical across iterations (tool/tool_call turns are stripped), + so re-running references would multiply their cost and re-spam the display. + """ + 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: model emits a tool call. + facade.create(messages=base_msgs, tools=[{"type": "function"}]) + # Iteration 2: same turn — a tool result was appended, but the advisory + # view (which strips tool turns) is unchanged, so references must be reused. + facade.create( + messages=base_msgs + + [ + {"role": "assistant", "content": "", "tool_calls": [{"id": "c1", "function": {"name": "f", "arguments": "{}"}}]}, + {"role": "tool", "tool_call_id": "c1", "content": "result"}, + ], + tools=[{"type": "function"}], + ) + + # 2 reference models, run once total (not once per iteration). + assert len(ref_runs) == 2 + # Reference blocks emitted once (2 reference events + 1 aggregating). + assert events.count("moa.reference") == 2 + assert events.count("moa.aggregating") == 1 + + +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 +