test(blackbox): cover MoA attribution end to end

tests/agent/test_moa_aggregator_cost_slot.py

@@ -134,3 +134,8 @@ def fake_call_llm(**kwargs):
     # Consume-once matches consume_reference_usage: cached tool iterations do
     # not multiply advisor billing rows.
     assert client.consume_reference_pricing_calls() == []
+
+    cached_messages = [{"role": "user", "content": "clean another db"}]
+    client.chat.completions.create(model="closed", messages=cached_messages)
+    client.chat.completions.create(model="closed", messages=cached_messages)
+    assert client.consume_reference_pricing_calls() == []

tests/plugins/blackbox/test_cost_perclass.py

@@ -135,6 +135,33 @@ def fake_estimate(model, usage, *, provider=None, base_url=None):
     ]
 
 
+def test_nested_moa_unknown_physical_route_is_honestly_partial(monkeypatch):
+    """A nonzero unknown physical call withholds the per-class split."""
+    priced = CostResult(
+        amount_usd=Decimal("0.01"), status="estimated",
+        source="official_docs_snapshot", label="~$0.01",
+        cost_input_usd=Decimal("0.01"), cost_output_usd=Decimal("0"),
+        cost_cache_read_usd=Decimal("0"), cost_cache_write_usd=Decimal("0"),
+    )
+    unknown = CostResult(
+        amount_usd=None, status="unknown", source="none", label="n/a"
+    )
+    seq = iter([priced, unknown])
+    monkeypatch.setattr(cost_mod, "estimate_usage_cost", lambda *a, **k: next(seq))
+    calls = [{"pricing_calls": [
+        {"model": "advisor-model", "provider": "advisor-provider",
+         "input_tokens": 100, "output_tokens": 10},
+        {"model": "unknown-model", "provider": "unknown-provider",
+         "input_tokens": 30, "output_tokens": 20},
+    ]}]
+
+    total, status, perclass = compute_turn_cost("default", "moa", None, calls)
+
+    assert total == 0.01
+    assert status == "partial"
+    assert all(value is None for value in perclass.values())
+
+
 def test_zero_token_failed_advisor_does_not_make_moa_turn_partial(monkeypatch):
     """A failed advisor emitted no billable usage, so its unknown route costs $0."""
     def fake_estimate(model, usage, *, provider=None, base_url=None):

tests/plugins/blackbox/test_moa_seam_integration.py

@@ -0,0 +1,223 @@
+"""MoA execution → turn finalizer → Blackbox SQLite seam coverage."""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from agent.conversation_loop import _build_moa_pricing_calls
+from agent.turn_finalizer import finalize_turn
+from agent.usage_pricing import CanonicalUsage
+from plugins.blackbox.cost import compute_turn_cost
+
+
+class _Agent:
+    def __init__(self):
+        self.max_iterations = 90
+        self.iteration_budget = SimpleNamespace(remaining=10, used=1, max_total=90)
+        self.quiet_mode = True
+        self.model = "default"
+        self.provider = "moa"
+        self.base_url = ""
+        self.session_id = "moa-seam"
+        self.context_compressor = SimpleNamespace(last_prompt_tokens=0, context_length=1_000_000)
+        self.session_input_tokens = 0
+        self.session_output_tokens = 0
+        self.session_cache_read_tokens = 0
+        self.session_cache_write_tokens = 0
+        self.session_reasoning_tokens = 0
+        self.session_prompt_tokens = 0
+        self.session_completion_tokens = 0
+        self.session_total_tokens = 0
+        self.session_estimated_cost_usd = 0
+        self.session_cost_status = "unknown"
+        self.session_cost_source = "test"
+        self._tool_guardrail_halt_decision = None
+        self._interrupt_message = None
+        self._response_was_previewed = False
+        self._skill_nudge_interval = 0
+        self._iters_since_skill = 0
+        self.valid_tool_names = []
+
+    def _handle_max_iterations(self, *_args):
+        raise AssertionError("not expected")
+
+    def _emit_status(self, *_args, **_kwargs):
+        pass
+
+    def _safe_print(self, *_args, **_kwargs):
+        pass
+
+    def _save_trajectory(self, *_args, **_kwargs):
+        pass
+
+    def _cleanup_task_resources(self, *_args, **_kwargs):
+        pass
+
+    def _drop_trailing_empty_response_scaffolding(self, _messages):
+        pass
+
+    def _persist_session(self, *_args, **_kwargs):
+        pass
+
+    def _file_mutation_verifier_enabled(self):
+        return False
+
+    def _turn_completion_explainer_enabled(self):
+        return False
+
+    def _drain_pending_steer(self):
+        return None
+
+    def clear_interrupt(self):
+        pass
+
+    def _sync_external_memory_for_turn(self, **_kwargs):
+        pass
+
+
+def _response(content: str, prompt_tokens: int, completion_tokens: int):
+    message = SimpleNamespace(content=content, tool_calls=[])
+    choice = SimpleNamespace(message=message, finish_reason="stop")
+    usage = SimpleNamespace(
+        prompt_tokens=prompt_tokens,
+        completion_tokens=completion_tokens,
+        total_tokens=prompt_tokens + completion_tokens,
+    )
+    return SimpleNamespace(choices=[choice], usage=usage, model="fake")
+
+
+def test_moa_execution_persists_priced_explicit_model_through_finalizer(
+    tmp_path, monkeypatch
+):
+    home = tmp_path / "hermes"
+    home.mkdir()
+    (home / "config.yaml").write_text(
+        """
+moa:
+  default_preset: default
+  presets:
+    default:
+      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))
+
+    import hermes_constants
+    import plugins.blackbox as bb
+    import plugins.blackbox.store as store
+
+    repo_root = Path(__file__).resolve().parents[3]
+    assert Path(bb.__file__).resolve().is_relative_to(repo_root)
+
+    importlib.reload(hermes_constants)
+    importlib.reload(store)
+    importlib.reload(bb)
+    bb._sessions.clear()
+    monkeypatch.setattr(
+        bb,
+        "_config",
+        lambda: {
+            "enabled": True,
+            "cost_alert_threshold_usd": 999.0,
+            "store_text": False,
+            "record_subagents": True,
+        },
+    )
+    monkeypatch.setattr(bb, "store", store, raising=False)
+    monkeypatch.setitem(sys.modules, "plugins.blackbox.store", store)
+    monkeypatch.setattr(bb, "_turn_id", lambda: "turn_moa_seam")
+
+    def fake_call_llm(**kwargs):
+        if kwargs.get("task") == "moa_reference":
+            return _response("advice", 1_000, 100)
+        return _response("acted", 400, 40)
+
+    monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
+    from agent.moa_loop import MoAClient
+
+    client = MoAClient("default")
+    response = client.chat.completions.create(
+        model="default",
+        messages=[{"role": "user", "content": "clean the db"}],
+    )
+    reference_usage, _ = client.consume_reference_usage()
+    reference_calls = client.consume_reference_pricing_calls()
+    aggregator_usage = CanonicalUsage(
+        input_tokens=response.usage.prompt_tokens,
+        output_tokens=response.usage.completion_tokens,
+    )
+    combined_usage = aggregator_usage + reference_usage
+    slot = client.last_aggregator_slot
+    assert slot is not None
+    pricing_calls = _build_moa_pricing_calls(
+        reference_calls,
+        aggregator_usage,
+        aggregator_model=slot["model"],
+        aggregator_provider=slot["provider"],
+        aggregator_base_url=slot.get("base_url"),
+    )
+    turn_call = {
+        "input_tokens": combined_usage.input_tokens,
+        "output_tokens": combined_usage.output_tokens,
+        "cache_read_tokens": combined_usage.cache_read_tokens,
+        "cache_write_tokens": combined_usage.cache_write_tokens,
+        "reasoning_tokens": combined_usage.reasoning_tokens,
+        "prompt_tokens": combined_usage.prompt_tokens,
+        "completion_tokens": combined_usage.output_tokens,
+        "total_tokens": combined_usage.total_tokens,
+        "latency_s": 0.1,
+        "composition": None,
+        "pricing_calls": pricing_calls,
+    }
+    expected_cost, expected_status, _ = compute_turn_cost(
+        "moa/default", "moa", None, [turn_call]
+    )
+    assert expected_cost is not None
+
+    def invoke_hook(name, **kwargs):
+        if name == "on_session_end":
+            bb._on_session_end(**kwargs)
+        return []
+
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", invoke_hook)
+    bb._on_session_start(session_id="moa-seam")
+    finalize_turn(
+        _Agent(),
+        final_response=response.choices[0].message.content,
+        api_call_count=1,
+        interrupted=False,
+        failed=False,
+        messages=[{"role": "user", "content": "clean the db"}],
+        conversation_history=[],
+        effective_task_id="task",
+        turn_id="turn",
+        user_message="clean the db",
+        original_user_message="clean the db",
+        _should_review_memory=False,
+        _turn_exit_reason="text_response(stop)",
+        _turn_calls=[turn_call],
+    )
+
+    row = store.get_turn("turn_moa_seam")
+    assert row is not None
+    assert row["provider"] == "moa"
+    assert row["model"] == "moa/default"
+    assert row["api_calls"] == 1
+    assert row["input_tokens"] == combined_usage.input_tokens
+    assert row["output_tokens"] == combined_usage.output_tokens
+    assert row["cost_status"] == expected_status == "estimated"
+    assert row["cost_usd"] == pytest.approx(expected_cost)