fix(cost_router): record unavailable race-loser usage instead of dropping it

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

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,16 +1411,32 @@ def rollup(
                     "total_tokens": 0,
                     "cost_amount": Decimal("0"),
                     "currency_code": row.get("currency_code", "USD"),
+                    "_measurement_statuses": set(),
                 },
             )
             bucket["record_count"] += 1
             bucket["prompt_tokens"] += int(row.get("prompt_tokens", 0))
             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]]:

contextual_orchestrator/cost_router.py

@@ -190,23 +190,42 @@ 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="",
             route_mode=context["route_mode"],
             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,

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,