feat(moa): show each reference model's output as a labelled block before the aggregator

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"

agent/moa_loop.py

@@ -8,6 +8,7 @@
 
 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 @@ def create(self, **api_kwargs: Any) -> Any:
 
         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 @@ def create(self, **api_kwargs: Any) -> Any:
 
 
 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)

cli.py

@@ -10474,6 +10474,36 @@ def _on_tool_progress(self, event_type: str, function_name: str = None, preview:
         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":

tests/cli/test_tool_progress_scrollback.py

@@ -251,3 +251,45 @@ def test_pending_info_consumed_on_completed(self):
         # 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()

tests/run_agent/test_moa_loop_mode.py

@@ -333,3 +333,130 @@ def slow_call_llm(**kwargs):
     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
+