Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 49 additions & 0 deletions agent/conversation_loop.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,29 @@
INTERRUPT_WAITING_FOR_MODEL_PREFIX = "Operation interrupted: waiting for model response ("


def _build_moa_pricing_calls(
advisor_calls: list[dict[str, Any]],
aggregator_usage: Any,
*,
aggregator_model: str,
aggregator_provider: str | None,
aggregator_base_url: str | None,
) -> list[dict[str, Any]]:
"""Return physical advisor + aggregator calls for Blackbox pricing."""
calls = [dict(call) for call in advisor_calls if isinstance(call, dict)]
calls.append({
"model": aggregator_model,
"provider": aggregator_provider,
"base_url": aggregator_base_url,
"input_tokens": aggregator_usage.input_tokens,
"output_tokens": aggregator_usage.output_tokens,
"cache_read_tokens": aggregator_usage.cache_read_tokens,
"cache_write_tokens": aggregator_usage.cache_write_tokens,
"reasoning_tokens": aggregator_usage.reasoning_tokens,
})
return calls


def _is_auth_resolution_error(api_error: Exception) -> bool:
"""True for the Anthropic SDK's keyless-client ``TypeError``.

Expand Down Expand Up @@ -2276,6 +2299,7 @@ def _perform_api_call(next_api_kwargs):
# usage, so without this the entire advisor spend — usually
# the bulk of a MoA turn — is invisible in token counts.
_moa_ref_cost = None
_moa_ref_pricing_calls: list[dict[str, Any]] = []
_moa_client = getattr(agent, "client", None)
if _moa_client is not None and hasattr(_moa_client, "consume_reference_usage"):
try:
Expand All @@ -2284,6 +2308,18 @@ def _perform_api_call(next_api_kwargs):
canonical_usage = canonical_usage + _ref_usage
except Exception as _moa_acct_exc: # pragma: no cover - defensive
logger.debug("MoA reference usage accounting failed: %s", _moa_acct_exc)
if _moa_client is not None and hasattr(
_moa_client, "consume_reference_pricing_calls"
):
try:
_moa_ref_pricing_calls = (
_moa_client.consume_reference_pricing_calls() or []
)
except Exception as _moa_pricing_exc: # pragma: no cover - defensive
logger.debug(
"MoA reference pricing-call accounting failed: %s",
_moa_pricing_exc,
)
# Flush the full-turn MoA trace (references + aggregator I/O)
# to disk when moa.save_traces is on. No-op otherwise and
# for non-MoA clients. Uses the live session_id so traces
Expand Down Expand Up @@ -2366,6 +2402,7 @@ def _perform_api_call(next_api_kwargs):
# run_conversation), never an agent attribute, so concurrent
# subagents running their own run_conversation frame cannot
# stomp each other's accumulator.
_turn_call = None
try:
_turn_calls.append({
"input_tokens": canonical_usage.input_tokens,
Expand All @@ -2379,6 +2416,7 @@ def _perform_api_call(next_api_kwargs):
"latency_s": api_duration,
"composition": _call_composition,
})
_turn_call = _turn_calls[-1]
except Exception:
pass # telemetry must never break the conversation loop

Expand Down Expand Up @@ -2420,6 +2458,17 @@ def _perform_api_call(next_api_kwargs):
_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
try:
if _turn_call is not None:
_turn_call["pricing_calls"] = _build_moa_pricing_calls(
_moa_ref_pricing_calls,
aggregator_usage,
aggregator_model=_agg_cost_model,
aggregator_provider=_agg_cost_provider,
aggregator_base_url=_agg_cost_base_url,
)
except Exception:
pass # telemetry must never break the conversation loop
cost_result = estimate_usage_cost(
_agg_cost_model,
aggregator_usage,
Expand Down
30 changes: 30 additions & 0 deletions agent/moa_loop.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ class _RefAccounting:
"output",
"model",
"provider",
"base_url",
"temperature",
)

Expand All @@ -68,6 +69,7 @@ def __init__(
output: str | None = None,
model: str | None = None,
provider: str | None = None,
base_url: str | None = None,
temperature: Any = None,
):
self.usage = usage
Expand All @@ -78,6 +80,7 @@ def __init__(
self.output = output
self.model = model
self.provider = provider
self.base_url = base_url
self.temperature = temperature

# Per-tool-result character budget for the advisory reference view. Tool
Expand Down Expand Up @@ -318,6 +321,7 @@ def _run_reference(
output=_output_text,
model=slot.get("model"),
provider=runtime.get("provider") or slot.get("provider"),
base_url=runtime.get("base_url"),
temperature=temperature,
)
return label, _output_text, acct
Expand All @@ -329,6 +333,7 @@ def _run_reference(
output=f"[failed: {exc}]",
model=slot.get("model"),
provider=runtime.get("provider") or slot.get("provider"),
base_url=runtime.get("base_url"),
temperature=temperature,
)

Expand Down Expand Up @@ -715,6 +720,7 @@ def __init__(self, preset_name: str, reference_callback: Any = None):

self._pending_reference_usage: Any = CanonicalUsage()
self._pending_reference_cost: Any = None
self._pending_reference_pricing_calls: list[dict[str, Any]] = []
# 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.
Expand All @@ -741,6 +747,12 @@ def consume_reference_usage(self) -> tuple[Any, Any]:
self._pending_reference_cost = None
return usage, cost

def consume_reference_pricing_calls(self) -> list[dict[str, Any]]:
"""Pop physical advisor calls for mixed-model Blackbox pricing."""
calls = list(self._pending_reference_pricing_calls)
self._pending_reference_pricing_calls = []
return calls

def consume_and_save_trace(
self, session_id: Any = None, aggregator_output_fallback: Any = None
) -> None:
Expand Down Expand Up @@ -892,6 +904,7 @@ def create(self, **api_kwargs: Any) -> Any:
# advisor spend by the tool-iteration count, so pending is zero.
self._pending_reference_usage = CanonicalUsage()
self._pending_reference_cost = None
self._pending_reference_pricing_calls = []
# Likewise no trace on a cache HIT — the full turn was already
# traced on the MISS that ran the references. A repeat iteration is
# not a new MoA turn.
Expand All @@ -915,14 +928,27 @@ def create(self, **api_kwargs: Any) -> Any:
# NOT be repriced at the aggregator's rate.
_ref_usage = CanonicalUsage()
_ref_cost: Any = None
_ref_pricing_calls: list[dict[str, Any]] = []
for _lbl, _txt, _acct in reference_outputs:
if isinstance(_acct, _RefAccounting):
if isinstance(_acct.usage, CanonicalUsage):
_ref_usage = _ref_usage + _acct.usage
if _acct.model:
_ref_pricing_calls.append({
"model": _acct.model,
"provider": _acct.provider,
"base_url": _acct.base_url,
"input_tokens": _acct.usage.input_tokens,
"output_tokens": _acct.usage.output_tokens,
"cache_read_tokens": _acct.usage.cache_read_tokens,
"cache_write_tokens": _acct.usage.cache_write_tokens,
"reasoning_tokens": _acct.usage.reasoning_tokens,
})
if _acct.cost_usd is not None:
_ref_cost = (_ref_cost or 0) + _acct.cost_usd
self._pending_reference_usage = _ref_usage
self._pending_reference_cost = _ref_cost
self._pending_reference_pricing_calls = _ref_pricing_calls
# Stash the full reference fan-out for trace persistence. The
# aggregator input/label are filled in below once agg_messages is
# built; the aggregator OUTPUT is stitched in by the caller
Expand Down Expand Up @@ -1051,6 +1077,10 @@ def consume_reference_usage(self) -> Any:
"""
return self.chat.completions.consume_reference_usage()

def consume_reference_pricing_calls(self) -> list[dict[str, Any]]:
"""Pop physical advisor calls for Blackbox mixed-model pricing."""
return self.chat.completions.consume_reference_pricing_calls()

@property
def last_aggregator_slot(self) -> Any:
"""Resolved aggregator slot ({provider, model, ...}) from the most
Expand Down
6 changes: 6 additions & 0 deletions plugins/blackbox/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -274,6 +274,12 @@ def _build_record(
# perturb live inference/prompt-caching; also feeds the corrected provider to
# cost pricing below so the row prices on the right route.
provider, model = _recover_provider_from_model(model, provider)
# MoA's model is a virtual PRESET name, not an unnamed physical model. Keep
# the raw provider column and make that identity explicit in Spend-by-model.
if str(provider or "").strip().lower() == "moa" and model:
model_text = str(model).strip()
if model_text and not model_text.lower().startswith("moa/"):
model = f"moa/{model_text}"

now = time.time()
state = _session_state(session_id)
Expand Down
33 changes: 28 additions & 5 deletions plugins/blackbox/cost.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,24 @@ def compute_turn_cost(
if not calls:
return 0.0, "included", dict(_PERCLASS_NONE)

# A logical MoA iteration contains several physical provider calls: one per
# advisor plus the acting aggregator. The outer turn still carries the
# virtual preset identity (provider="moa", model="default"), so price nested
# calls at their own routes when the conversation loop supplies them.
pricing_calls: list[dict] = []
for call in calls:
nested = call.get("pricing_calls") if isinstance(call, dict) else None
# An explicitly empty list intentionally falls through to the outer
# call: nonzero usage with missing physical attribution must remain
# unknown, never be silently reclassified as included/$0.
if isinstance(nested, list) and nested:
pricing_calls.extend(item for item in nested if isinstance(item, dict))
elif isinstance(call, dict):
pricing_calls.append(call)

if not pricing_calls:
return 0.0, "included", dict(_PERCLASS_NONE)
Comment thread
Kyzcreig marked this conversation as resolved.

# M3 (SPEC §5D / INV-7): a turn whose every billed token class is zero is
# costless, not unpriced. Return a concrete $0 split so the dashboard treats
# it as priced; ANY nonzero class falls through to normal pricing.
Expand All @@ -58,14 +76,19 @@ def _billed_tokens(call: dict) -> int:
+ _int_value(call.get("cache_write_tokens"))
)

if all(_billed_tokens(call) == 0 for call in calls):
if all(_billed_tokens(call) == 0 for call in pricing_calls):
return 0.0, "priced_zero", {
"uncached": 0.0,
"cache_read": 0.0,
"cache_write": 0.0,
"output": 0.0,
}

# A zero-token physical call is concretely $0 even if its route has no
# pricing entry (for example a failed MoA advisor). It must not make an
# otherwise fully-priced mixed-model turn look partial.
pricing_calls = [call for call in pricing_calls if _billed_tokens(call) > 0]

known_total = Decimal("0")
known_count = 0
unknown_count = 0
Expand All @@ -76,7 +99,7 @@ def _billed_tokens(call: dict) -> int:
acc_cache_read = Decimal("0")
acc_cache_write = Decimal("0")

for call in calls:
for call in pricing_calls:
try:
usage = CanonicalUsage(
input_tokens=_int_value(call.get("input_tokens")),
Expand All @@ -86,10 +109,10 @@ def _billed_tokens(call: dict) -> int:
reasoning_tokens=_int_value(call.get("reasoning_tokens")),
)
result = estimate_usage_cost(
model,
call.get("model") or model,
usage,
provider=provider,
base_url=base_url,
provider=call.get("provider") or provider,
base_url=call.get("base_url") or base_url,
)
except Exception:
unknown_count += 1
Expand Down
36 changes: 36 additions & 0 deletions tests/agent/test_moa_aggregator_cost_slot.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,3 +98,39 @@ def fake_call_llm(**kwargs):
assert slot is not None
assert slot["model"] == "anthropic/claude-opus-4.8"
assert slot["provider"] == "openrouter"


def test_reference_pricing_calls_preserve_each_advisor_route(moa_config, monkeypatch):
"""Blackbox can retrieve physical advisor usage without repricing it as MoA."""
from agent.moa_loop import MoAClient

def fake_call_llm(**kwargs):
response = _response(
"acted" if kwargs.get("task") != "moa_reference" else "advice"
)
response.usage = SimpleNamespace(
prompt_tokens=1000,
completion_tokens=100,
total_tokens=1100,
)
return response

monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm)
client = MoAClient("closed")
client.chat.completions.create(
model="closed",
messages=[{"role": "user", "content": "clean the db"}],
)

calls = client.consume_reference_pricing_calls()
assert [c["model"] for c in calls] == [
"anthropic/claude-opus-4.8",
"openai/gpt-5.5",
]
assert [c["provider"] for c in calls] == ["openrouter", "openrouter"]
assert [c["input_tokens"] for c in calls] == [1000, 1000]
assert [c["output_tokens"] for c in calls] == [100, 100]

# Consume-once matches consume_reference_usage: cached tool iterations do
# not multiply advisor billing rows.
assert client.consume_reference_pricing_calls() == []
47 changes: 47 additions & 0 deletions tests/agent/test_moa_blackbox_attribution.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
"""Physical-call attribution passed from the MoA loop to Blackbox."""

from agent.usage_pricing import CanonicalUsage


def test_build_moa_pricing_calls_appends_real_aggregator_after_advisors():
from agent.conversation_loop import _build_moa_pricing_calls

advisor_calls = [{
"model": "gpt-5.5",
"provider": "openai-codex",
"base_url": "https://codex.invalid",
"input_tokens": 1000,
"output_tokens": 100,
"cache_read_tokens": 0,
"cache_write_tokens": 0,
"reasoning_tokens": 0,
}]
aggregator_usage = CanonicalUsage(
input_tokens=200,
output_tokens=50,
cache_read_tokens=300,
cache_write_tokens=40,
reasoning_tokens=10,
)

calls = _build_moa_pricing_calls(
advisor_calls,
aggregator_usage,
aggregator_model="anthropic/claude-opus-4.8",
aggregator_provider="openrouter",
aggregator_base_url="https://openrouter.ai/api/v1",
)

assert calls[0] == advisor_calls[0]
assert calls[1] == {
"model": "anthropic/claude-opus-4.8",
"provider": "openrouter",
"base_url": "https://openrouter.ai/api/v1",
"input_tokens": 200,
"output_tokens": 50,
"cache_read_tokens": 300,
"cache_write_tokens": 40,
"reasoning_tokens": 10,
}
# The helper must not mutate the facade-owned pending list.
assert len(advisor_calls) == 1
20 changes: 20 additions & 0 deletions tests/plugins/blackbox/test_composite_model_recovery.py
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,26 @@ def test_build_record_writes_the_split_pair():
assert rec.provider == "claude-api-proxy-f6"


def test_build_record_names_virtual_moa_preset_explicitly():
"""Spend-by-model must not show the ambiguous bare model name ``default``."""
rec = bb._build_record(
session_id="moa-session",
interrupted=False,
model="default",
platform="cli",
provider="moa",
user_message="ping",
final_response="pong",
turn_usage={"api_calls": 1, "input_tokens": 0, "output_tokens": 0,
"calls": []},
cfg={"store_text": False},
kwargs={},
)
assert rec is not None
assert rec.model == "moa/default"
assert rec.provider == "moa"


if __name__ == "__main__": # pragma: no cover
import pytest

Expand Down
Loading
Loading