fix(moa): price aggregator turn at its real model so session cost isn't advisor-only (#56394)

On the MoA path agent.model/provider are the virtual preset name (e.g.
"closed") and "moa", which have no pricing entry. estimate_usage_cost()
returned None for the aggregator turn, so the `if amount_usd is not None`
guard skipped it and the session's estimated_cost_usd reflected only the
advisor fan-out — a ~50% undercount when the aggregator does the full acting
loop (verified: $0.91 advisor-only vs $1.96 true, aggregator = 54%).

MoAChatCompletions.create() now stashes the resolved aggregator slot as
last_aggregator_slot (exposed via MoAClient); conversation_loop reads it to
price the aggregator turn at its real model/provider. cost_source flips from
'none' to 'provider_models_api'.

agent/conversation_loop.py

@@ -2034,11 +2034,27 @@ def run_conversation(
                         api_duration, _cache_pct,
                     )
 
+                    # On the MoA path, agent.model/provider are the virtual
+                    # preset name ("closed") and "moa", which have no pricing
+                    # entry — estimating against them returns None and silently
+                    # drops the aggregator's own spend, leaving the session cost
+                    # as advisor-fan-out only (a ~50% undercount when the
+                    # aggregator does the full acting loop). Price the aggregator
+                    # turn at its REAL model/provider, read from the MoA client's
+                    # resolved aggregator slot.
+                    _agg_cost_model = agent.model
+                    _agg_cost_provider = agent.provider
+                    _agg_cost_base_url = agent.base_url
+                    _agg_slot = getattr(_moa_client, "last_aggregator_slot", None) if _moa_client is not None else None
+                    if _agg_slot and _agg_slot.get("model"):
+                        _agg_cost_model = _agg_slot["model"]
+                        _agg_cost_provider = _agg_slot.get("provider") or agent.provider
+                        _agg_cost_base_url = _agg_slot.get("base_url") or agent.base_url
                     cost_result = estimate_usage_cost(
-                        agent.model,
+                        _agg_cost_model,
                         aggregator_usage,
-                        provider=agent.provider,
-                        base_url=agent.base_url,
+                        provider=_agg_cost_provider,
+                        base_url=_agg_cost_base_url,
                         api_key=getattr(agent, "api_key", ""),
                     )
                     if cost_result.amount_usd is not None:

agent/moa_loop.py

@@ -618,6 +618,10 @@ class MoAChatCompletions:
 
         self._pending_reference_usage: Any = CanonicalUsage()
         self._pending_reference_cost: Any = None
+        # Resolved aggregator slot ({provider, model, ...}) from the most recent
+        # create(); read by session cost accounting to price the aggregator's
+        # acting turn at its real model instead of the virtual preset name.
+        self.last_aggregator_slot: Any = None
         # Full-turn trace parts stashed on a cache-MISS create(), awaiting the
         # caller to stitch in the live session_id + resolved aggregator output
         # and flush to the trace file (only when moa.save_traces is on).
@@ -704,6 +708,13 @@ class MoAChatCompletions:
         messages = list(api_kwargs.get("messages") or [])
         reference_models = preset.get("reference_models") or []
         aggregator = preset.get("aggregator") or {}
+        # Expose the resolved aggregator slot so session cost accounting can
+        # price the aggregator's acting turn at its REAL model/provider. The
+        # agent's model/provider on the MoA path are the virtual preset name
+        # ("closed") and "moa", which have no pricing entry — without this the
+        # aggregator's spend (often the bulk of the turn) is silently dropped
+        # and the session cost reflects advisor fan-out only.
+        self.last_aggregator_slot = dict(aggregator) if aggregator else None
         # MoA does not cap reference or aggregator output: each model uses its
         # own maximum. Passing max_tokens=None makes call_llm omit the parameter
         # (it never caps by default), so a long aggregator synthesis is never
@@ -901,6 +912,14 @@ class MoAClient:
         """
         return self.chat.completions.consume_reference_usage()
 
+    @property
+    def last_aggregator_slot(self) -> Any:
+        """Resolved aggregator slot ({provider, model, ...}) from the most
+        recent create(), or None. Read by session cost accounting to price the
+        aggregator's acting turn at its real model instead of the virtual
+        preset name."""
+        return getattr(self.chat.completions, "last_aggregator_slot", None)
+
     def consume_and_save_trace(
         self, session_id: Any = None, aggregator_output_fallback: Any = None
     ) -> None:

tests/agent/test_moa_aggregator_cost_slot.py

@@ -0,0 +1,100 @@
+"""Tests for MoA aggregator-slot exposure used by session cost accounting.
+
+Regression guard for the ~50% MoA cost undercount: on the MoA path the
+agent's model/provider are the virtual preset name (e.g. "closed") and "moa",
+which have no pricing entry. Session cost accounting must price the
+aggregator's acting turn at its REAL model/provider, read from the MoA
+client's ``last_aggregator_slot``. Before the fix that slot did not exist and
+the aggregator's spend (often >50% of the turn) was silently dropped, leaving
+the session cost as advisor-fan-out only.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+
+def _response(content="ok"):
+    message = SimpleNamespace(content=content, tool_calls=[])
+    choice = SimpleNamespace(message=message, finish_reason="stop")
+    return SimpleNamespace(choices=[choice], usage=None, model="fake")
+
+
+@pytest.fixture
+def moa_config(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    (home / "config.yaml").write_text(
+        """
+moa:
+  default_preset: closed
+  presets:
+    closed:
+      enabled: true
+      reference_models:
+        - provider: openrouter
+          model: anthropic/claude-opus-4.8
+        - provider: openrouter
+          model: openai/gpt-5.5
+      aggregator:
+        provider: openrouter
+        model: anthropic/claude-opus-4.8
+""".strip(),
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    return home
+
+
+def test_create_populates_last_aggregator_slot(moa_config, monkeypatch):
+    """After a create() turn, last_aggregator_slot carries the REAL aggregator
+    model/provider — not the virtual preset name."""
+    from agent.moa_loop import MoAChatCompletions
+
+    def fake_call_llm(**kwargs):
+        return _response("acted" if kwargs.get("task") != "moa_reference" else "advice")
+
+    monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
+
+    facade = MoAChatCompletions("closed")
+    # Slot is unset before any turn runs.
+    assert facade.last_aggregator_slot is None
+
+    facade.create(
+        model="closed",
+        messages=[{"role": "user", "content": "clean the db"}],
+    )
+
+    slot = facade.last_aggregator_slot
+    assert slot is not None
+    # The virtual preset name / "moa" must NOT leak into the priced slot.
+    assert slot["model"] == "anthropic/claude-opus-4.8"
+    assert slot["provider"] == "openrouter"
+    assert slot["model"] != "closed"
+
+
+def test_client_exposes_last_aggregator_slot(moa_config, monkeypatch):
+    """MoAClient delegates last_aggregator_slot to its completions facade so
+    session accounting can read it without touching internals."""
+    from agent.moa_loop import MoAClient
+
+    def fake_call_llm(**kwargs):
+        return _response("acted" if kwargs.get("task") != "moa_reference" else "advice")
+
+    monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
+
+    client = MoAClient("closed")
+    assert client.last_aggregator_slot is None
+
+    client.chat.completions.create(
+        model="closed",
+        messages=[{"role": "user", "content": "clean the db"}],
+    )
+
+    slot = client.last_aggregator_slot
+    assert slot is not None
+    assert slot["model"] == "anthropic/claude-opus-4.8"
+    assert slot["provider"] == "openrouter"