diff --git a/CHANGELOG.md b/CHANGELOG.md index 16436c84d..1d5a60c3f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,22 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) ### Fixed +- `CostRoutingCoordinator._record_race_endpoint_usage()` no longer silently + drops a completed, billable race-loser call's spend when its usage payload + can't be parsed. It now writes a `measurement_status="unavailable"` ledger + row (0 tokens) instead of returning without any row at all, mirroring + `record_stream_usage()`'s existing "call happened, can't measure it" + fallback. (Devin review on #955) Both of `complete()`'s cost-aggregation + paths (the `provider_request`/race-proxy path and the ordinary + `orchestrator.run()` sync path) previously only checked for + `measurement_status="estimated"` when rolling records up into one + response `cost` block, so a measured winner plus this new "unavailable" + loser row still reported the whole completion as confidently `"measured"` + and silently summed the loser's unknown cost as `0`. Both paths now use + the same unavailable-outranks-estimated-outranks-measured precedence as + `record_stream_usage()`, and `cost_amount` becomes `None` rather than a + partial sum whenever any contributing record is unavailable. Mixed-currency + responses also suppress their otherwise partial `currency_components`. - `check-fast-mlsirm --help` now shows help and exits instead of running the diagnostic: the subcommand took no arguments and ignored everything after its own name, so `--help` was silently swallowed and the real diagnostic diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index 96be7772a..1622fcda1 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -334,6 +334,7 @@ def from_record( "contextual_orchestrator.usage_record_id": record.usage_record_id, "contextual_orchestrator.request_channel": record.request_channel, "contextual_orchestrator.usage.export_state": export_state, + "contextual_orchestrator.usage.measurement_status": record.measurement_status, } if record.workflow_run_id: attributes["contextual_orchestrator.workflow_run_id"] = record.workflow_run_id @@ -344,13 +345,16 @@ def from_record( if error_type: attributes["error.type"] = error_type - metrics = { - "gen_ai.usage.input_tokens": float(record.prompt_tokens), - "gen_ai.usage.output_tokens": float(record.completion_tokens), - "gen_ai.usage.total_tokens": float(record.total_tokens), - "gen_ai.usage.cost": float(record.cost_amount), - "contextual_orchestrator.usage.records": 1.0, - } + metrics = {"contextual_orchestrator.usage.records": 1.0} + if record.measurement_status != "unavailable": + metrics.update( + { + "gen_ai.usage.input_tokens": float(record.prompt_tokens), + "gen_ai.usage.output_tokens": float(record.completion_tokens), + "gen_ai.usage.total_tokens": float(record.total_tokens), + "gen_ai.usage.cost": float(record.cost_amount), + } + ) if error_type: metrics["contextual_orchestrator.usage.export_failures"] = 1.0 return cls( @@ -1407,6 +1411,7 @@ def rollup( "total_tokens": 0, "cost_amount": Decimal("0"), "currency_code": row.get("currency_code", "USD"), + "_measurement_statuses": set(), }, ) bucket["record_count"] += 1 @@ -1414,9 +1419,24 @@ def rollup( bucket["completion_tokens"] += int(row.get("completion_tokens", 0)) bucket["total_tokens"] += int(row.get("total_tokens", 0)) bucket["cost_amount"] += Decimal(str(row.get("cost_amount", 0))) + bucket["_measurement_statuses"].add( + row.get("measurement_status", "unavailable") + ) for bucket in buckets.values(): - bucket["cost_amount"] = float( - bucket["cost_amount"].quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP) + statuses = bucket.pop("_measurement_statuses") + bucket["measurement_status"] = ( + "unavailable" if "unavailable" in statuses + else "estimated" if "estimated" in statuses + else "measured" + ) + bucket["cost_amount"] = ( + None + if bucket["measurement_status"] == "unavailable" + else float( + bucket["cost_amount"].quantize( + Decimal("0.000001"), rounding=ROUND_HALF_UP + ) + ) ) return buckets @@ -1429,7 +1449,12 @@ def report( """Return a report envelope: per-value rollup plus a grand total.""" buckets = self.rollup(dimension, start, end) items = sorted( - buckets.values(), key=lambda item: item["cost_amount"], reverse=True + buckets.values(), + key=lambda item: ( + item["cost_amount"] is not None, + item["cost_amount"] or 0.0, + ), + reverse=True, ) grand_total = self.total(start, end) return { @@ -1443,12 +1468,23 @@ def total(self, start: Optional[int] = None, end: Optional[int] = None) -> Dict[ """Return grand totals (cost + tokens + record count) over the window.""" rows = self.store.query(start, end) cost = sum((Decimal(str(row.get("cost_amount", 0))) for row in rows), Decimal("0")) + statuses = {row.get("measurement_status", "unavailable") for row in rows} + measurement_status = ( + "unavailable" if "unavailable" in statuses + else "estimated" if "estimated" in statuses + else "measured" + ) return { "record_count": len(rows), "prompt_tokens": sum(int(row.get("prompt_tokens", 0)) for row in rows), "completion_tokens": sum(int(row.get("completion_tokens", 0)) for row in rows), "total_tokens": sum(int(row.get("total_tokens", 0)) for row in rows), - "cost_amount": float(cost.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)), + "cost_amount": ( + None + if measurement_status == "unavailable" + else float(cost.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)) + ), + "measurement_status": measurement_status, } def records(self, start: Optional[int] = None, end: Optional[int] = None) -> List[Dict[str, Any]]: diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index 438a26fde..c29fe4a84 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -190,15 +190,34 @@ def _record_race_endpoint_usage(self, endpoint_id: str, value: Any) -> None: usage = value[2] elif isinstance(value, dict): usage = value.get("usage") - counts = self._provider_usage(usage) - if counts is None: - return agent = next( (item for item in self.orchestrator.candidates if item.id == endpoint_id), None, ) if agent is None: # pragma: no cover - endpoint came from the current pool return + counts = self._provider_usage(usage) + provider_model = self._agent_provider_model(agent, context["model_name"]) + if counts is None: + # The provider call genuinely completed and is billable, but its + # usage payload could not be parsed. Record an honest + # "unavailable" row rather than silently dropping this spend — + # mirrors record_stream_usage's measurement_status="unavailable" + # fallback for the same "call happened, can't measure it" case. + provider, model = provider_model + record = self.ledger.record_usage( + provider=provider, + model=model, + prompt_tokens=0, + completion_tokens=0, + request_channel="sync", + route_mode=context["route_mode"], + workflow_run_id=context["workflow_run_id"], + attribution=context["attribution"], + measurement_status="unavailable", + ) + context["records"].append(record) + return record = self._record_completion( messages=[], answer="", @@ -206,7 +225,7 @@ def _record_race_endpoint_usage(self, endpoint_id: str, value: Any) -> None: request_channel="sync", attribution=context["attribution"], model_name=context["model_name"], - provider_model=self._agent_provider_model(agent, context["model_name"]), + provider_model=provider_model, workflow_run_id=context["workflow_run_id"], prompt_tokens=counts[0], completion_tokens=counts[1], @@ -374,20 +393,27 @@ def complete( provider_response["usage_record_ids"] = [ record.usage_record_id for record in records ] + # "unavailable" (a billable race-loser call whose usage payload + # could not be parsed, see _record_race_endpoint_usage) outranks + # "estimated": a completion combining a measured winner with an + # unavailable loser must not present a confident-looking summed + # total, the same honesty precedence record_stream_usage uses. + statuses = {record.measurement_status for record in records} + aggregate_measurement_status = ( + "unavailable" if "unavailable" in statuses + else "estimated" if "estimated" in statuses + else "measured" + ) provider_response["cost"] = { "cost_amount": ( round(sum(record.cost_amount for record in records), 6) - if len(currencies) == 1 + if len(currencies) == 1 and aggregate_measurement_status != "unavailable" else None ), "currency_code": next(iter(currencies)) if len(currencies) == 1 else "MIXED", - "measurement_status": ( - "estimated" - if any(record.measurement_status == "estimated" for record in records) - else "measured" - ), + "measurement_status": aggregate_measurement_status, } - if len(currencies) > 1: + if len(currencies) > 1 and aggregate_measurement_status != "unavailable": provider_response["cost"]["currency_components"] = [ { "currency_code": currency, @@ -506,20 +532,25 @@ def complete( "total_tokens": sum(item.total_tokens for item in client_usage_records), } currencies = {item.currency_code for item in records} + # Same honesty precedence as the provider_request path above and + # record_stream_usage: a billable race-loser call recorded + # "unavailable" must not be summed into a confident-looking total. + statuses = {item.measurement_status for item in records} + aggregate_measurement_status = ( + "unavailable" if "unavailable" in statuses + else "estimated" if "estimated" in statuses + else "measured" + ) result["cost"] = { "cost_amount": ( round(sum(item.cost_amount for item in records), 6) - if len(currencies) == 1 + if len(currencies) == 1 and aggregate_measurement_status != "unavailable" else None ), "currency_code": next(iter(currencies)) if len(currencies) == 1 else "MIXED", - "measurement_status": ( - "estimated" - if any(item.measurement_status == "estimated" for item in records) - else "measured" - ), + "measurement_status": aggregate_measurement_status, } - if len(currencies) > 1: + if len(currencies) > 1 and aggregate_measurement_status != "unavailable": result["cost"]["currency_components"] = [ { "currency_code": currency, diff --git a/tests/test_cost_ledger.py b/tests/test_cost_ledger.py index 709a30014..ef454ad53 100644 --- a/tests/test_cost_ledger.py +++ b/tests/test_cost_ledger.py @@ -250,6 +250,23 @@ def test_usage_telemetry_event_is_prompt_and_answer_safe() -> None: assert all("answer" not in key for key in event.attributes) +def test_unavailable_usage_telemetry_does_not_export_false_zero_metrics() -> None: + sink = InMemoryUsageTelemetrySink() + ledger = _priced_ledger(telemetry_sink=sink) + + ledger.record_usage( + provider="openai", + model="gpt-x", + prompt_tokens=0, + completion_tokens=0, + measurement_status="unavailable", + ) + + event = sink.events()[-1] + assert event.attributes["contextual_orchestrator.usage.measurement_status"] == "unavailable" + assert event.metrics == {"contextual_orchestrator.usage.records": 1.0} + + def test_non_blocking_store_records_p2028_like_failure_as_telemetry_only() -> None: sink = InMemoryUsageTelemetrySink() ledger = _priced_ledger( @@ -313,6 +330,26 @@ def test_multi_dimensional_rollup_correctness() -> None: assert ledger.total()["cost_amount"] == 17.0 +def test_unavailable_usage_nulls_rollup_and_total_cost() -> None: + ledger = _priced_ledger() + ledger.record_usage( + provider="openai", model="gpt-x", prompt_tokens=1000, + completion_tokens=0, attribution={"team": "alpha"}, + ) + ledger.record_usage( + provider="openai", model="gpt-x", prompt_tokens=0, + completion_tokens=0, attribution={"team": "alpha"}, + measurement_status="unavailable", + ) + + bucket = ledger.rollup("team")["alpha"] + assert bucket["measurement_status"] == "unavailable" + assert bucket["cost_amount"] is None + assert ledger.total()["measurement_status"] == "unavailable" + assert ledger.total()["cost_amount"] is None + assert ledger.report("team")["grand_total"]["cost_amount"] is None + + def test_rollup_by_every_declared_dimension_is_supported() -> None: ledger = _priced_ledger() ledger.record_usage(provider="openai", model="gpt-x", prompt_tokens=100, completion_tokens=100, diff --git a/tests/test_cost_router.py b/tests/test_cost_router.py index 1b88f26c6..63eafa062 100644 --- a/tests/test_cost_router.py +++ b/tests/test_cost_router.py @@ -194,6 +194,37 @@ def test_completed_race_loser_usage_is_recorded_as_measured_provider_spend() -> assert record["workflow_run_id"] == "run_race" +def test_race_loser_with_unparseable_usage_is_recorded_as_unavailable() -> None: + """A billable race-loser call with malformed usage still gets a ledger row.""" + coordinator = _coordinator() + context = { + "route_mode": "route", + "attribution": {"team": "alpha"}, + "model_name": "contextual-orchestrator", + "workflow_run_id": "run_race_unmeasurable", + "workflow_ready": True, + "records": [], + "pending_usage": [], + } + token = coordinator._race_usage_context.set(context) + try: + coordinator._record_race_endpoint_usage( + "mock_worker", + ("duplicate", "mock_worker", None), + ) + finally: + coordinator._race_usage_context.reset(token) + records = coordinator.ledger.records() + assert len(records) == 1 + record = records[0] + assert record["measurement_status"] == "unavailable" + assert record["prompt_tokens"] == 0 + assert record["completion_tokens"] == 0 + assert record["provider_name"] == "mock" + assert record["model_name"] == "mock-a" + assert record["workflow_run_id"] == "run_race_unmeasurable" + + def test_race_loser_derives_provider_from_base_url_when_name_is_absent() -> None: """Race-loser spend uses the same provider identity as winner accounting.""" agent = ModelAgent( @@ -275,6 +306,45 @@ def run(*_args, **_kwargs): assert result["cost"]["cost_amount"] == 0.022 +def test_sync_cost_reports_unavailable_when_a_race_loser_cannot_be_measured() -> None: + """Devin review (#955): the plain orchestrator.run() sync path's own cost + aggregation needs the same unavailable-outranks-estimated precedence as + the provider_request path -- a measured winner plus an unavailable race + loser must not report a confident "measured" total or sum a real cost + over an unknown one. + """ + coordinator = _coordinator() + + def run(*_args, **_kwargs): + coordinator.orchestrator._race_usage_sink( + "mock_worker", + ("duplicate", "mock_worker", None), # unparseable usage + ) + return { + "workflow_run_id": "run_race_unavailable", + "mode": "route", + "answer": "winner", + "trace": [ + { + "agent_id": "mock_worker", + "output": "winner", + "usage": {"prompt_tokens": 7, "completion_tokens": 3}, + } + ], + } + + coordinator.orchestrator.run = run # type: ignore[method-assign] + result = coordinator.complete([{"role": "user", "content": "race"}], mode="route") + + assert result["cost"]["measurement_status"] == "unavailable" + assert result["cost"]["cost_amount"] is None + assert "currency_components" not in result["cost"] + assert {record["measurement_status"] for record in coordinator.ledger.records()} == { + "measured", + "unavailable", + } + + def test_ready_race_usage_without_workflow_id_is_not_discarded() -> None: coordinator = _coordinator() context = { @@ -407,6 +477,105 @@ def proxy_completion(*_args, **_kwargs): assert len(result["usage_record_ids"]) == 2 +def test_structured_cost_reports_unavailable_when_a_race_loser_cannot_be_measured() -> None: + """Devin review (#955): a measured winner plus an unavailable race loser + must not roll up into a confident "measured" total -- the aggregate cost + status and amount need the same honesty precedence record_stream_usage + already applies, not just the raw per-record ledger label. + """ + coordinator = _coordinator() + + def proxy_completion(*_args, **_kwargs): + coordinator.orchestrator._race_usage_sink( + "mock_worker", + ("duplicate", "mock_worker", None), # unparseable usage + ) + return { + "model": "mock-a", + "usage": {"prompt_tokens": 7, "completion_tokens": 3}, + "orchestration": {"workflow_run_id": "run_unavailable_loser"}, + } + + coordinator.orchestrator.proxy_completion = proxy_completion # type: ignore[method-assign] + coordinator.orchestrator.get_workflow_run = lambda _run_id: { # type: ignore[method-assign] + "workflow_run_id": "run_unavailable_loser", + "mode": "route", + "answer": "winner", + "trace": None, + } + result = coordinator.complete( + [{"role": "user", "content": "race unavailable"}], + provider_request={ + "model": "mock-a", + "messages": [{"role": "user", "content": "race unavailable"}], + "response_format": {"type": "json_object"}, + }, + ) + + assert result["cost"]["measurement_status"] == "unavailable" + assert result["cost"]["cost_amount"] is None + assert "currency_components" not in result["cost"] + records = coordinator.ledger.records() + assert {record["measurement_status"] for record in records} == {"measured", "unavailable"} + + +@pytest.mark.parametrize("structured", [False, True]) +def test_unavailable_mixed_currency_cost_suppresses_partial_components( + structured: bool, +) -> None: + """Unknown loser spend must not expose complete-looking currency subtotals.""" + agents = [ + ModelAgent("winner_agent", "winner-model", provider_name="winner_provider"), + ModelAgent("loser_agent", "loser-model", provider_name="loser_provider"), + ] + orchestrator = TaskOrchestrator(agents) + config = InMemoryConfigStore() + price_book = PriceBook(config) + price_book.set_price(PriceEntry("winner_provider", "winner-model", 1, 1, "USD")) + price_book.set_price(PriceEntry("loser_provider", "loser-model", 1, 1, "EUR")) + coordinator = CostRoutingCoordinator(orchestrator, config, price_book=price_book) + + def workflow(): + return { + "workflow_run_id": "run_mixed_unavailable", + "mode": "route", + "answer": "winner", + "trace": [{ + "agent_id": "winner_agent", + "output": "winner", + "usage": {"prompt_tokens": 7, "completion_tokens": 3}, + }], + } + + def emit_loser(): + orchestrator._race_usage_sink( + "loser_agent", ("duplicate", "loser_agent", None) + ) + + if structured: + def proxy_completion(*_args, **_kwargs): + emit_loser() + return {"orchestration": {"workflow_run_id": "run_mixed_unavailable"}} + orchestrator.proxy_completion = proxy_completion # type: ignore[method-assign] + orchestrator.get_workflow_run = lambda _run_id: workflow() # type: ignore[method-assign] + result = coordinator.complete( + [{"role": "user", "content": "mixed"}], + provider_request={"model": "winner-model", "messages": []}, + ) + else: + def run(*_args, **_kwargs): + emit_loser() + return workflow() + orchestrator.run = run # type: ignore[method-assign] + result = coordinator.complete([{"role": "user", "content": "mixed"}]) + + assert result["cost"] == { + "cost_amount": None, + "currency_code": "MIXED", + "measurement_status": "unavailable", + } + + def test_structured_provider_workflow_estimates_each_unreported_call() -> None: """Mixed usage bills each measured call plus one fallback prompt estimate.""" coordinator = _coordinator()