From 8da853b9068329528218a1fe627e47068a0ed8f4 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 06:25:07 +0000 Subject: [PATCH 01/11] fix(cost): stop fabricating $0.00 prices for unpriced models MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PriceBook.compute_cost() silently returned a $0.00 price for any provider/model without a configured price row, and that fabricated zero was indistinguishable from a real free price once recorded — the ledger asserted "measured, cost nothing" when the truth was "price unknown". This contradicted the repo's own honest-metrics convention and the two sibling honest-degradation patterns already in this codebase (model_discovery's unset Bytez per-token pricing, and orchestrator's separate unpriced_models reporting key). compute_cost() now returns a (cost_amount, currency_code, price_known) 3-tuple. UsageRecord gains a price_known: bool field, persisted through a new usage_price_knowledge satellite table joined the same way usage_measurements already is (including the conservative "unknown" default for rows migrated from the pre-price_known flattened schema). CostLedger.rollup()/.report()/.total() add an additive cost_amount_by_status/record_count_by_status breakdown (measured/estimated/unavailable) alongside the existing flat cost_amount total, so measured, estimated, and unavailable-priced spend are no longer opaquely blended into one authoritative-looking number. cost_amount's existing meaning and value are unchanged. --- CHANGELOG.md | 12 ++ contextual_orchestrator/batch_routing.py | 2 +- contextual_orchestrator/cost_ledger.py | 139 ++++++++++++++++++--- contextual_orchestrator/cost_router.py | 8 +- contextual_orchestrator/model_discovery.py | 2 +- tests/test_batch_routing_boundaries.py | 4 +- tests/test_cost_ledger.py | 93 ++++++++++++++ tests/test_cost_ledger_boundaries.py | 3 +- 8 files changed, 241 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b38e0770..c73e37332 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,18 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) ### Fixed +- The cost ledger no longer fabricates a $0.00 price for an unpriced + provider/model. `PriceBook.compute_cost()` now returns a + `(cost_amount, currency_code, price_known)` 3-tuple; `UsageRecord` gains a + `price_known: bool` field (persisted through a new `usage_price_knowledge` + satellite table, joined the same way `usage_measurements` already is) so a + measured-but-unpriced request is distinguishable from a genuinely free one. + `CostLedger.rollup()`/`.report()`/`.total()` add an additive + `cost_amount_by_status`/`record_count_by_status` breakdown + (measured/estimated/unavailable) alongside the existing flat `cost_amount` + total, so measured, estimated, and unavailable-priced spend are no longer + opaquely blended into one authoritative-looking number. + - OpenRouter discovery no longer marks the entire credential account evidence-only. Authenticated catalog rows may serve ordinary requests, while ZDR-only requests still require explicit route-level ZDR evidence. diff --git a/contextual_orchestrator/batch_routing.py b/contextual_orchestrator/batch_routing.py index 82ce36600..bf4c1075f 100644 --- a/contextual_orchestrator/batch_routing.py +++ b/contextual_orchestrator/batch_routing.py @@ -141,7 +141,7 @@ def cheapest_upstream( for candidate in candidates: provider = candidate.get("provider", "") model = candidate.get("model", "") - cost, _currency = price_book.compute_cost( + cost, _currency, _price_known = price_book.compute_cost( provider, model, assumed_prompt_tokens, assumed_completion_tokens ) if best_cost is None or cost < best_cost: diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index 96be7772a..f4daeca3e 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -23,7 +23,8 @@ DB object names are two-or-more-word snake_case per the repository convention: ``llm_usage_records``, ``cost_attribution_dimensions``, ``llm_price_entries``, -``cost_attribution_values``, and ``usage_record_attributions``. +``cost_attribution_values``, ``usage_record_attributions``, +``usage_measurements``, and ``usage_price_knowledge``. """ from __future__ import annotations @@ -218,15 +219,18 @@ def compute_cost( model: str, prompt_tokens: int, completion_tokens: int, - ) -> tuple[float, str]: - """Return ``(cost_amount, currency_code)`` for a request. + ) -> tuple[float, str, bool]: + """Return ``(cost_amount, currency_code, price_known)`` for a request. An unpriced provider/model yields ``0.0`` in the default currency so - recording never fails on a missing price row. + recording never fails on a missing price row — but ``price_known`` is + ``False`` for that case so callers can tell "priced at zero" apart + from "we do not know the price" instead of silently fabricating a + free price for an unpriced model. """ entry = self.get_price(provider, model) if entry is None: - return 0.0, self.default_currency + return 0.0, self.default_currency, False prompt_cost = (Decimal(prompt_tokens) / Decimal(1000)) * Decimal( str(entry.prompt_price_per_1k) ) @@ -236,13 +240,18 @@ def compute_cost( total = (prompt_cost + completion_cost).quantize( Decimal("0.000001"), rounding=ROUND_HALF_UP ) - return float(total), entry.currency_code + return float(total), entry.currency_code, True # --------------------------------------------------------------------------- # Usage records + stores # --------------------------------------------------------------------------- +# The only recognized provenance labels for a recorded usage row's token +# counts. Shared between ``record_usage``'s validation and the rollup/total +# per-status breakdown so the two never drift apart. +MEASUREMENT_STATUSES: tuple[str, ...] = ("measured", "estimated", "unavailable") + @dataclass class UsageRecord: @@ -261,6 +270,12 @@ class UsageRecord: cost_amount: float currency_code: str measurement_status: str = "measured" + # Whether ``cost_amount`` reflects a real configured price rather than the + # fallback zero used when a provider/model has no price row. Defaults to + # ``True`` so callers that set ``cost_amount`` explicitly (bypassing + # ``PriceBook.compute_cost``) are not force-changed by this field's + # addition. + price_known: bool = True attribution: AttributionDimensions = field(default_factory=AttributionDimensions) def as_dict(self) -> Dict[str, Any]: @@ -281,6 +296,7 @@ def as_dict(self) -> Dict[str, Any]: "cost_amount": self.cost_amount, "currency_code": self.currency_code, "measurement_status": self.measurement_status, + "price_known": self.price_known, } row.update( { @@ -721,6 +737,14 @@ def __len__(self) -> int: ON DELETE CASCADE ); +CREATE TABLE IF NOT EXISTS usage_price_knowledge ( + usage_record_id TEXT PRIMARY KEY, + price_known INTEGER NOT NULL, + CONSTRAINT usage_price_knowledge_record_foreign_key + FOREIGN KEY (usage_record_id) REFERENCES llm_usage_records(usage_record_id) + ON DELETE CASCADE +); + CREATE TABLE IF NOT EXISTS cost_attribution_values ( dimension_name TEXT NOT NULL, dimension_value TEXT NOT NULL, @@ -779,6 +803,7 @@ def __len__(self) -> int: "cost_amount", "currency_code", "measurement_status", + "price_known", ) _RELATIONAL_ATTRIBUTION_COLUMNS = { "account": "account_name", @@ -870,6 +895,16 @@ def __len__(self) -> int: "VALUES (%s, %s) ON CONFLICT (usage_record_id) DO NOTHING" ), } +_USAGE_PRICE_KNOWLEDGE_INSERT_SQL = { + "qmark": ( + "INSERT INTO usage_price_knowledge (usage_record_id, price_known) " + "VALUES (?, ?) ON CONFLICT (usage_record_id) DO NOTHING" + ), + "pyformat": ( + "INSERT INTO usage_price_knowledge (usage_record_id, price_known) " + "VALUES (%s, %s) ON CONFLICT (usage_record_id) DO NOTHING" + ), +} _USAGE_SELECT_SQL = ( "SELECT u.usage_record_id, u.created_at, u.workflow_run_id, u.request_channel, " "u.route_mode, u.provider_name, u.model_name, " @@ -880,15 +915,18 @@ def __len__(self) -> int: "COALESCE(MAX(CASE WHEN a.dimension_name = 'group' THEN a.dimension_value END), 'unattributed') AS group_name, " "COALESCE(MAX(CASE WHEN a.dimension_name = 'company' THEN a.dimension_value END), 'unattributed') AS company_name, " "u.prompt_tokens, u.completion_tokens, u.total_tokens, u.cost_amount, u.currency_code, " - "COALESCE(m.measurement_status, 'unavailable') AS measurement_status " + "COALESCE(m.measurement_status, 'unavailable') AS measurement_status, " + "COALESCE(k.price_known, 0) AS price_known " "FROM llm_usage_records AS u " "LEFT JOIN usage_record_attributions AS a ON a.usage_record_id = u.usage_record_id " - "LEFT JOIN usage_measurements AS m ON m.usage_record_id = u.usage_record_id" + "LEFT JOIN usage_measurements AS m ON m.usage_record_id = u.usage_record_id " + "LEFT JOIN usage_price_knowledge AS k ON k.usage_record_id = u.usage_record_id" ) _USAGE_GROUP_ORDER_SQL = ( " GROUP BY u.usage_record_id, u.created_at, u.workflow_run_id, u.request_channel, " "u.route_mode, u.provider_name, u.model_name, u.prompt_tokens, " - "u.completion_tokens, u.total_tokens, u.cost_amount, u.currency_code, m.measurement_status " + "u.completion_tokens, u.total_tokens, u.cost_amount, u.currency_code, m.measurement_status, " + "k.price_known " "ORDER BY u.created_at, u.usage_record_id" ) _USAGE_QUERY_SQL = { @@ -1012,6 +1050,13 @@ def _copy_flattened_usage_rows(self, cur: Any, source_table: str) -> None: _USAGE_MEASUREMENT_INSERT_SQL[self._paramstyle], (row["usage_record_id"], "unavailable"), ) + # A flattened legacy row predates this signal entirely, so + # whether its stored cost reflects a real price is genuinely + # unknown — mark it unknown (0) rather than assume it was known. + cur.execute( + _USAGE_PRICE_KNOWLEDGE_INSERT_SQL[self._paramstyle], + (row["usage_record_id"], 0), + ) self._insert_normalized_attribution(cur, row) def _create_schema(self) -> None: @@ -1134,6 +1179,10 @@ def _append_locked(self, record: UsageRecord) -> bool: _USAGE_MEASUREMENT_INSERT_SQL[self._paramstyle], (row["usage_record_id"], row.get("measurement_status", "unavailable")), ) + cur.execute( + _USAGE_PRICE_KNOWLEDGE_INSERT_SQL[self._paramstyle], + (row["usage_record_id"], 1 if row.get("price_known", True) else 0), + ) self._insert_normalized_attribution(cur, row) if outer_transaction: cur.execute("RELEASE SAVEPOINT usage_record_append") @@ -1201,6 +1250,18 @@ def _within_window(created_at: int, start: Optional[int], end: Optional[int]) -> return True +def _measurement_status_of(row: Dict[str, Any]) -> str: + """Return a row's recognized measurement status, defaulting to unavailable. + + Covers a row whose ``measurement_status`` is missing entirely (a caller + or store bypassing the normal write path) the same way a genuinely + unrecognized value is covered: fail closed to the most conservative + label rather than silently attributing it to ``measured``. + """ + status = row.get("measurement_status") + return status if status in MEASUREMENT_STATUSES else "unavailable" + + # --------------------------------------------------------------------------- # The ledger # --------------------------------------------------------------------------- @@ -1286,10 +1347,10 @@ def record_usage( if provider: dims.upstream_api = provider - cost_amount, currency = self.price_book.compute_cost( + cost_amount, currency, price_known = self.price_book.compute_cost( provider, model, prompt_tokens, completion_tokens ) - if measurement_status not in {"measured", "estimated", "unavailable"}: + if measurement_status not in MEASUREMENT_STATUSES: raise ValueError("measurement_status must be measured, estimated, or unavailable") record = UsageRecord( usage_record_id=usage_record_id or f"usage_{uuid.uuid4().hex}", @@ -1305,6 +1366,7 @@ def record_usage( cost_amount=cost_amount, currency_code=currency, measurement_status=measurement_status, + price_known=price_known, attribution=dims, ) accepted = False @@ -1386,6 +1448,14 @@ def rollup( ``dimension`` is one of the attribution dimension names (``account``, ``service``, ``upstream_api``/``provider``, ``model_name``, ``team``, ``group``, ``company``). Returns a mapping of dimension value to totals. + + Each bucket's ``cost_amount`` sums every row regardless of provenance + (unchanged, backward-compatible meaning). Alongside it, + ``cost_amount_by_status``/``record_count_by_status`` break the same + bucket down by row ``measurement_status`` (``measured``, ``estimated``, + ``unavailable``) so a caller can tell how much of a total is real + provider-measured spend versus an estimate or an unpriced/unmeasured + row, without those being silently blended into one number. """ column = _DIMENSION_TO_COLUMN.get(dimension) if column is None: @@ -1407,17 +1477,29 @@ def rollup( "total_tokens": 0, "cost_amount": Decimal("0"), "currency_code": row.get("currency_code", "USD"), + "cost_amount_by_status": { + status: Decimal("0") for status in MEASUREMENT_STATUSES + }, + "record_count_by_status": {status: 0 for status in MEASUREMENT_STATUSES}, }, ) + status = _measurement_status_of(row) + row_cost = Decimal(str(row.get("cost_amount", 0))) 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["cost_amount"] += row_cost + bucket["cost_amount_by_status"][status] += row_cost + bucket["record_count_by_status"][status] += 1 for bucket in buckets.values(): bucket["cost_amount"] = float( bucket["cost_amount"].quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP) ) + bucket["cost_amount_by_status"] = { + status: float(amount.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)) + for status, amount in bucket["cost_amount_by_status"].items() + } return buckets def report( @@ -1426,7 +1508,13 @@ def report( start: Optional[int] = None, end: Optional[int] = None, ) -> Dict[str, Any]: - """Return a report envelope: per-value rollup plus a grand total.""" + """Return a report envelope: per-value rollup plus a grand total. + + Both ``items`` (from :meth:`rollup`) and ``grand_total`` (from + :meth:`total`) carry the ``cost_amount_by_status``/ + ``record_count_by_status`` measured/estimated/unavailable breakdown + alongside their existing flat ``cost_amount`` totals. + """ buckets = self.rollup(dimension, start, end) items = sorted( buckets.values(), key=lambda item: item["cost_amount"], reverse=True @@ -1440,15 +1528,34 @@ def report( } def total(self, start: Optional[int] = None, end: Optional[int] = None) -> Dict[str, Any]: - """Return grand totals (cost + tokens + record count) over the window.""" + """Return grand totals (cost + tokens + record count) over the window. + + ``cost_amount`` remains the sum over every row regardless of + provenance (unchanged, backward-compatible meaning). It is joined by + ``cost_amount_by_status``/``record_count_by_status``, the same total + broken down by row ``measurement_status`` (``measured``, ``estimated``, + ``unavailable``) so measured, estimated, and unavailable-priced spend + are never opaquely blended into one authoritative-looking number. + """ rows = self.store.query(start, end) - cost = sum((Decimal(str(row.get("cost_amount", 0))) for row in rows), Decimal("0")) + cost_amount_by_status = {status: Decimal("0") for status in MEASUREMENT_STATUSES} + record_count_by_status = {status: 0 for status in MEASUREMENT_STATUSES} + for row in rows: + status = _measurement_status_of(row) + cost_amount_by_status[status] += Decimal(str(row.get("cost_amount", 0))) + record_count_by_status[status] += 1 + total_cost = sum(cost_amount_by_status.values(), Decimal("0")) 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": float(total_cost.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)), + "cost_amount_by_status": { + status: float(amount.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)) + for status, amount in cost_amount_by_status.items() + }, + "record_count_by_status": record_count_by_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..e14b2cdf7 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -1139,7 +1139,13 @@ def cost_report( start: Optional[int] = None, end: Optional[int] = None, ) -> Dict[str, Any]: - """Return a cost rollup report grouped by ``dimension`` over a window.""" + """Return a cost rollup report grouped by ``dimension`` over a window. + + The envelope's ``items`` and ``grand_total`` each carry a + ``cost_amount_by_status``/``record_count_by_status`` breakdown + (measured/estimated/unavailable) alongside their flat ``cost_amount`` + total — see :meth:`CostLedger.rollup`. + """ return self.ledger.report(dimension, start, end) diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py index a0c9b3131..ce92ea99e 100644 --- a/contextual_orchestrator/model_discovery.py +++ b/contextual_orchestrator/model_discovery.py @@ -1622,7 +1622,7 @@ def _discovery_price_key( ): return unknown try: - cost, currency = price_book.compute_cost( + cost, currency, _price_known = price_book.compute_cost( model.provider_name, model.model_id, 1000, diff --git a/tests/test_batch_routing_boundaries.py b/tests/test_batch_routing_boundaries.py index bb04dbbdf..1f07e1884 100644 --- a/tests/test_batch_routing_boundaries.py +++ b/tests/test_batch_routing_boundaries.py @@ -28,9 +28,9 @@ def __init__(self, cost: float) -> None: def compute_cost( self, provider: str, model: str, prompt_tokens: int, completion_tokens: int - ) -> tuple[float, str]: + ) -> tuple[float, str, bool]: self.queries.append((provider, model)) - return self._cost, "USD" + return self._cost, "USD", True def test_cheapest_upstream_returns_none_for_no_candidates() -> None: diff --git a/tests/test_cost_ledger.py b/tests/test_cost_ledger.py index 709a30014..f23df905e 100644 --- a/tests/test_cost_ledger.py +++ b/tests/test_cost_ledger.py @@ -139,9 +139,35 @@ def test_unpriced_model_costs_zero_and_still_records() -> None: provider="mystery", model="unpriced", prompt_tokens=100, completion_tokens=50 ) assert record.cost_amount == 0.0 + # The zero must be distinguishable from a real free price: an unpriced + # provider/model is unknown, never fabricated as $0.00. + assert record.price_known is False assert len(ledger.records()) == 1 +def test_priced_model_marks_price_known_true() -> None: + ledger = _priced_ledger() + record = ledger.record_usage( + provider="openai", model="gpt-x", prompt_tokens=100, completion_tokens=50 + ) + assert record.price_known is True + + +def test_compute_cost_returns_price_known_flag() -> None: + """PriceBook.compute_cost's 3-tuple distinguishes known from unknown price.""" + config = InMemoryConfigStore() + price_book = PriceBook(config) + price_book.set_price( + PriceEntry("openai", "gpt-x", prompt_price_per_1k=2.0, completion_price_per_1k=4.0) + ) + + priced = price_book.compute_cost("openai", "gpt-x", 1000, 500) + assert priced == (4.0, "USD", True) + + unpriced = price_book.compute_cost("mystery", "unpriced", 100, 50) + assert unpriced == (0.0, "USD", False) + + def test_provider_wildcard_price_entry() -> None: config = InMemoryConfigStore() price_book = PriceBook(config) @@ -355,6 +381,48 @@ def test_report_envelope_sorts_by_cost_desc_and_includes_grand_total() -> None: assert report["grand_total"]["cost_amount"] == 11.0 +def test_rollup_report_total_break_down_cost_by_measurement_status() -> None: + """cost_amount stays a flat sum; the new by-status fields attribute it.""" + ledger = _priced_ledger() + ledger.record_usage( + provider="openai", model="gpt-x", prompt_tokens=1000, completion_tokens=0, + measurement_status="measured", + ) # 2.0, priced + ledger.record_usage( + provider="openai", model="gpt-x", prompt_tokens=500, completion_tokens=0, + measurement_status="estimated", + ) # 1.0, priced + ledger.record_usage( + provider="mystery", model="unpriced", prompt_tokens=1000, completion_tokens=1000, + measurement_status="unavailable", + ) # 0.0, unpriced + + total = ledger.total() + assert total["cost_amount"] == 3.0 # unchanged, backward-compatible flat total + assert total["cost_amount_by_status"] == { + "measured": 2.0, "estimated": 1.0, "unavailable": 0.0, + } + assert total["record_count_by_status"] == { + "measured": 1, "estimated": 1, "unavailable": 1, + } + # The breakdown must sum back to the flat total exactly. + assert sum(total["cost_amount_by_status"].values()) == total["cost_amount"] + + by_provider = ledger.rollup("provider") + assert by_provider["openai"]["cost_amount"] == 3.0 + assert by_provider["openai"]["cost_amount_by_status"] == { + "measured": 2.0, "estimated": 1.0, "unavailable": 0.0, + } + assert by_provider["openai"]["record_count_by_status"] == { + "measured": 1, "estimated": 1, "unavailable": 0, + } + assert by_provider["mystery"]["cost_amount_by_status"]["unavailable"] == 0.0 + assert by_provider["mystery"]["record_count_by_status"]["unavailable"] == 1 + + report = ledger.report("provider") + assert report["grand_total"] == total + + def test_sql_ledger_store_on_sqlite_creates_objects_and_rolls_up() -> None: conn = sqlite3.connect(":memory:") store = SqlLedgerStore(conn, paramstyle="qmark") @@ -377,6 +445,23 @@ def test_sql_ledger_store_on_sqlite_creates_objects_and_rolls_up() -> None: assert by_company["acme"]["cost_amount"] == 15.0 # 6 + 9 +def test_sql_ledger_persists_price_known_flag() -> None: + """price_known round-trips through the SQL store's satellite table.""" + conn = sqlite3.connect(":memory:") + store = SqlLedgerStore(conn, paramstyle="qmark") + ledger = _priced_ledger(store=store) + ledger.record_usage(provider="openai", model="gpt-x", prompt_tokens=1000, completion_tokens=0) + ledger.record_usage(provider="mystery", model="unpriced", prompt_tokens=1000, completion_tokens=0) + + assert conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='usage_price_knowledge'" + ).fetchone() == ("usage_price_knowledge",) + + rows = {row["provider_name"]: row for row in store.query()} + assert bool(rows["openai"]["price_known"]) is True + assert bool(rows["mystery"]["price_known"]) is False + + def test_table_columns_honors_its_argument_on_qmark() -> None: """The qmark branch must inspect the requested table, not a hardcoded one.""" conn = sqlite3.connect(":memory:") @@ -544,6 +629,10 @@ def test_sql_ledger_migrates_flattened_usage_rows() -> None: assert row["service_name"] == "unattributed" assert row["team_name"] == "unattributed" assert row["company_name"] == "acme" + # A flattened legacy row predates the price_known signal: whether its + # stored cost reflects a real price is unknown, so migration marks it + # unknown rather than assuming it was known. + assert bool(row["price_known"]) is False def test_sql_ledger_maps_null_legacy_attribution_to_unattributed() -> None: @@ -824,6 +913,8 @@ def test_orphaned_legacy_generation_is_adopted_and_dropped() -> None: rows = store.query(None, None) assert len(rows) == 1 assert rows[0]["usage_record_id"] == "usage_guard_t1" + # Adopted from a pre-price_known legacy table: unknown, not assumed known. + assert bool(rows[0]["price_known"]) is False legacy_tables = connection.execute( "SELECT COUNT(*) FROM sqlite_master WHERE name='llm_usage_records_legacy'" ).fetchone()[0] @@ -910,6 +1001,8 @@ def test_ledger_table_names_follow_two_word_snake_case() -> None: "llm_price_entries", "cost_attribution_values", "usage_record_attributions", + "usage_measurements", + "usage_price_knowledge", ): assert is_two_word_snake_case(name) diff --git a/tests/test_cost_ledger_boundaries.py b/tests/test_cost_ledger_boundaries.py index 77fcba4d4..a4f0efbad 100644 --- a/tests/test_cost_ledger_boundaries.py +++ b/tests/test_cost_ledger_boundaries.py @@ -268,9 +268,10 @@ def test_corrupt_price_component_text_is_rejected_not_parsed() -> None: ) book = PriceBook(config) assert book.get_price("openai", "corrupt-model") is None - cost, currency = book.compute_cost("openai", "corrupt-model", 1000, 1000) + cost, currency, price_known = book.compute_cost("openai", "corrupt-model", 1000, 1000) assert cost == 0.0 assert currency == "USD" + assert price_known is False def test_working_background_store_marks_records_stored() -> None: From ddbfe6d10149ee157c2b356613892ae58a33bbee Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 01:06:31 +0900 Subject: [PATCH 02/11] fix(cost): keep unknown prices out of measured spend --- contextual_orchestrator/batch_routing.py | 8 ++++--- contextual_orchestrator/cost_ledger.py | 30 +++++++++++++++++++++--- contextual_orchestrator/cost_router.py | 3 ++- tests/test_batch_routing_boundaries.py | 14 +++++++++++ tests/test_cost_ledger.py | 4 ++++ 5 files changed, 52 insertions(+), 7 deletions(-) diff --git a/contextual_orchestrator/batch_routing.py b/contextual_orchestrator/batch_routing.py index bf4c1075f..cb9aae429 100644 --- a/contextual_orchestrator/batch_routing.py +++ b/contextual_orchestrator/batch_routing.py @@ -131,8 +131,8 @@ def cheapest_upstream( Cost-optimising upstream selection for load balancing: given candidate provider/model pairs, price each against the configurable price table for a representative request shape and return the cheapest. Unpriced candidates - cost ``0`` and are treated as free (explicit, so a missing price is visible - rather than silently expensive). Ties keep input order. + are excluded because an unknown price is not free. Ties keep input order; + ``None`` is returned when no candidate has a known price. """ if not candidates: return None @@ -141,9 +141,11 @@ def cheapest_upstream( for candidate in candidates: provider = candidate.get("provider", "") model = candidate.get("model", "") - cost, _currency, _price_known = price_book.compute_cost( + cost, _currency, price_known = price_book.compute_cost( provider, model, assumed_prompt_tokens, assumed_completion_tokens ) + if not price_known: + continue if best_cost is None or cost < best_cost: best_cost = cost best = candidate diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index f4daeca3e..b3c075bd4 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -1455,7 +1455,8 @@ def rollup( bucket down by row ``measurement_status`` (``measured``, ``estimated``, ``unavailable``) so a caller can tell how much of a total is real provider-measured spend versus an estimate or an unpriced/unmeasured - row, without those being silently blended into one number. + row. The corresponding ``*_by_price_status`` fields distinguish known + prices from unknown-price fallback zeros. """ column = _DIMENSION_TO_COLUMN.get(dimension) if column is None: @@ -1481,9 +1482,14 @@ def rollup( status: Decimal("0") for status in MEASUREMENT_STATUSES }, "record_count_by_status": {status: 0 for status in MEASUREMENT_STATUSES}, + "cost_amount_by_price_status": { + "known": Decimal("0"), "unknown": Decimal("0") + }, + "record_count_by_price_status": {"known": 0, "unknown": 0}, }, ) status = _measurement_status_of(row) + price_status = "known" if row.get("price_known") else "unknown" row_cost = Decimal(str(row.get("cost_amount", 0))) bucket["record_count"] += 1 bucket["prompt_tokens"] += int(row.get("prompt_tokens", 0)) @@ -1492,6 +1498,8 @@ def rollup( bucket["cost_amount"] += row_cost bucket["cost_amount_by_status"][status] += row_cost bucket["record_count_by_status"][status] += 1 + bucket["cost_amount_by_price_status"][price_status] += row_cost + bucket["record_count_by_price_status"][price_status] += 1 for bucket in buckets.values(): bucket["cost_amount"] = float( bucket["cost_amount"].quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP) @@ -1500,6 +1508,10 @@ def rollup( status: float(amount.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)) for status, amount in bucket["cost_amount_by_status"].items() } + bucket["cost_amount_by_price_status"] = { + status: float(amount.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)) + for status, amount in bucket["cost_amount_by_price_status"].items() + } return buckets def report( @@ -1535,15 +1547,22 @@ def total(self, start: Optional[int] = None, end: Optional[int] = None) -> Dict[ ``cost_amount_by_status``/``record_count_by_status``, the same total broken down by row ``measurement_status`` (``measured``, ``estimated``, ``unavailable``) so measured, estimated, and unavailable-priced spend - are never opaquely blended into one authoritative-looking number. + are never opaquely blended into one authoritative-looking number. The + corresponding ``*_by_price_status`` fields expose unknown-price rows. """ rows = self.store.query(start, end) cost_amount_by_status = {status: Decimal("0") for status in MEASUREMENT_STATUSES} record_count_by_status = {status: 0 for status in MEASUREMENT_STATUSES} + cost_amount_by_price_status = {"known": Decimal("0"), "unknown": Decimal("0")} + record_count_by_price_status = {"known": 0, "unknown": 0} for row in rows: status = _measurement_status_of(row) - cost_amount_by_status[status] += Decimal(str(row.get("cost_amount", 0))) + price_status = "known" if row.get("price_known") else "unknown" + row_cost = Decimal(str(row.get("cost_amount", 0))) + cost_amount_by_status[status] += row_cost record_count_by_status[status] += 1 + cost_amount_by_price_status[price_status] += row_cost + record_count_by_price_status[price_status] += 1 total_cost = sum(cost_amount_by_status.values(), Decimal("0")) return { "record_count": len(rows), @@ -1556,6 +1575,11 @@ def total(self, start: Optional[int] = None, end: Optional[int] = None) -> Dict[ for status, amount in cost_amount_by_status.items() }, "record_count_by_status": record_count_by_status, + "cost_amount_by_price_status": { + status: float(amount.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)) + for status, amount in cost_amount_by_price_status.items() + }, + "record_count_by_price_status": record_count_by_price_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 e14b2cdf7..257690a92 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -1144,7 +1144,8 @@ def cost_report( The envelope's ``items`` and ``grand_total`` each carry a ``cost_amount_by_status``/``record_count_by_status`` breakdown (measured/estimated/unavailable) alongside their flat ``cost_amount`` - total — see :meth:`CostLedger.rollup`. + total, plus known/unknown ``*_by_price_status`` fields — see + :meth:`CostLedger.rollup`. """ return self.ledger.report(dimension, start, end) diff --git a/tests/test_batch_routing_boundaries.py b/tests/test_batch_routing_boundaries.py index 1f07e1884..511750457 100644 --- a/tests/test_batch_routing_boundaries.py +++ b/tests/test_batch_routing_boundaries.py @@ -33,6 +33,13 @@ def compute_cost( return self._cost, "USD", True +class _MixedPriceBook: + def compute_cost( + self, provider: str, model: str, prompt_tokens: int, completion_tokens: int + ) -> tuple[float, str, bool]: + return (2.0, "USD", True) if provider == "known" else (0.0, "USD", False) + + def test_cheapest_upstream_returns_none_for_no_candidates() -> None: assert cheapest_upstream([], _StaticPriceBook(1.0)) is None @@ -44,6 +51,13 @@ def test_cheapest_upstream_tie_keeps_input_order() -> None: assert best is first # strict less-than keeps the earlier candidate on ties +def test_cheapest_upstream_excludes_unknown_prices() -> None: + known = {"provider": "known", "model": "priced"} + unknown = {"provider": "unknown", "model": "unpriced"} + assert cheapest_upstream([unknown, known], _MixedPriceBook()) is known + assert cheapest_upstream([unknown], _MixedPriceBook()) is None + + def test_local_backend_rejects_invalid_concurrency() -> None: with pytest.raises(ValueError, match="max_concurrency"): LocalBatchBackend(lambda messages, mode: {}, max_concurrency=0) diff --git a/tests/test_cost_ledger.py b/tests/test_cost_ledger.py index f23df905e..af9ae9b44 100644 --- a/tests/test_cost_ledger.py +++ b/tests/test_cost_ledger.py @@ -143,6 +143,10 @@ def test_unpriced_model_costs_zero_and_still_records() -> None: # provider/model is unknown, never fabricated as $0.00. assert record.price_known is False assert len(ledger.records()) == 1 + assert ledger.total()["record_count_by_price_status"] == {"known": 0, "unknown": 1} + assert ledger.rollup("provider")["mystery"]["record_count_by_price_status"] == { + "known": 0, "unknown": 1, + } def test_priced_model_marks_price_known_true() -> None: From 7d070f8beb836208d9560ea5d8d90a6a41bd2f50 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 17:35:16 +0000 Subject: [PATCH 03/11] fix(cost_ledger): keep a rejected duplicate append fully idempotent Devin review on #956: SqlLedgerStore._append_locked()'s satellite writes (usage_measurements, usage_price_knowledge, attribution) ran unconditionally even when the parent llm_usage_records insert was rejected as a duplicate usage_record_id. A retried append for a parent row that predates one of these satellite tables (e.g. an upgrade-migrated row with no usage_price_knowledge child, intentionally read as price-unknown) silently backfilled that child using the retry's current price/measurement state instead of what actually priced the original spend -- relabeling historical unknown-price provenance from an unrelated later call. The three satellite inserts now only run when the parent insert is actually accepted, so append() is a true no-op on a rejected duplicate. New regression test appends the same usage_record_id twice with different measurement_status/price_known values and asserts the second (rejected) append leaves the original row's provenance untouched. --- CHANGELOG.md | 20 ++++++++++++++++++ contextual_orchestrator/cost_ledger.py | 29 ++++++++++++++++++-------- tests/test_cost_ledger.py | 29 ++++++++++++++++++++++++++ 3 files changed, 69 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c73e37332..6a96d76b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,26 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) (measured/estimated/unavailable) alongside the existing flat `cost_amount` total, so measured, estimated, and unavailable-priced spend are no longer opaquely blended into one authoritative-looking number. +- `CostLedger.rollup()`/`.report()` add the same treatment one level further: + an additive `cost_amount_by_price_status`/`record_count_by_price_status` + breakdown (`known`/`unknown`) alongside the measured/estimated/unavailable + one, so an unpriced request's spend is visible even after rolling many + records up into one bucket. `cheapest_upstream()` no longer treats an + unpriced candidate as free (cost `0`) when selecting the lowest-cost + upstream — an unknown price is excluded from the comparison entirely + rather than winning it by default; `None` is returned when no candidate + has a known price. +- (Devin review on #956) `SqlLedgerStore._append_locked()`'s satellite + writes (`usage_measurements`, `usage_price_knowledge`, attribution) now + only run when the parent `llm_usage_records` insert is actually accepted. + A retried `usage_record_id` whose parent insert is correctly rejected as + a duplicate previously still ran these satellite inserts unconditionally + — for a parent row that predates one of these tables (e.g. an + upgrade-migrated row with no `usage_price_knowledge` child, intentionally + read as price-unknown) that silently backfilled its provenance from the + retry's current price/measurement state instead of what actually priced + the original spend. `append()` is now a true no-op on a rejected + duplicate. - OpenRouter discovery no longer marks the entire credential account evidence-only. Authenticated catalog rows may serve ordinary requests, while diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index b3c075bd4..4d93c6460 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -1175,15 +1175,26 @@ def _append_locked(self, record: UsageRecord) -> bool: tuple(row.get(column) for column in _CORE_USAGE_COLUMNS), ) accepted = getattr(cur, "rowcount", 1) != 0 - cur.execute( - _USAGE_MEASUREMENT_INSERT_SQL[self._paramstyle], - (row["usage_record_id"], row.get("measurement_status", "unavailable")), - ) - cur.execute( - _USAGE_PRICE_KNOWLEDGE_INSERT_SQL[self._paramstyle], - (row["usage_record_id"], 1 if row.get("price_known", True) else 0), - ) - self._insert_normalized_attribution(cur, row) + # A rejected duplicate parent insert (a retried usage_record_id) + # must stay fully idempotent: skipping the satellite writes below + # keeps a duplicate append a true no-op. Without this guard, a + # retry whose parent row already exists but predates one of + # these satellite tables (e.g. a pre-upgrade row with no + # usage_price_knowledge child, intentionally read as + # price-unknown) would insert that child using the retry's + # *current* price/measurement state -- silently relabeling a + # historical unknown-price row's provenance from an unrelated + # later call, not from what actually priced that original spend. + if accepted: + cur.execute( + _USAGE_MEASUREMENT_INSERT_SQL[self._paramstyle], + (row["usage_record_id"], row.get("measurement_status", "unavailable")), + ) + cur.execute( + _USAGE_PRICE_KNOWLEDGE_INSERT_SQL[self._paramstyle], + (row["usage_record_id"], 1 if row.get("price_known", True) else 0), + ) + self._insert_normalized_attribution(cur, row) if outer_transaction: cur.execute("RELEASE SAVEPOINT usage_record_append") else: diff --git a/tests/test_cost_ledger.py b/tests/test_cost_ledger.py index af9ae9b44..97fe6cc5c 100644 --- a/tests/test_cost_ledger.py +++ b/tests/test_cost_ledger.py @@ -962,6 +962,35 @@ def test_append_preserves_caller_owned_sqlite_transaction() -> None: assert store.query(None, None) == [] +def test_duplicate_append_does_not_rewrite_price_provenance() -> None: + """Devin review (#956): a rejected duplicate parent insert must stay a + true no-op. Previously the satellite usage_measurements/ + usage_price_knowledge/attribution writes ran unconditionally even when + the parent row already existed, so a retry with today's price state + could backfill a historical price-unknown row's provenance from an + unrelated later call instead of what actually priced that original + spend (the exact gap an upgrade-migrated pre-price-knowledge row is + supposed to preserve). + """ + connection = sqlite3.connect(":memory:", isolation_level=None) + store = SqlLedgerStore(connection, paramstyle="qmark") + + accepted_first = store.append( + _guard_usage_record(measurement_status="unavailable", price_known=False) + ) + assert accepted_first is True + + accepted_retry = store.append( + _guard_usage_record(measurement_status="measured", price_known=True) + ) + assert accepted_retry is False + + rows = store.query(None, None) + assert len(rows) == 1 + assert rows[0]["measurement_status"] == "unavailable" + assert not rows[0]["price_known"] + + def test_sql_ledger_rejects_unknown_parameter_style() -> None: try: SqlLedgerStore(sqlite3.connect(":memory:"), paramstyle="named") From 2b0b2f306b3d8c152d357f90b4c0abccdace55db Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 03:55:09 +0900 Subject: [PATCH 04/11] fix: expose unknown price across usage surfaces --- contextual_orchestrator/cost_ledger.py | 4 +- contextual_orchestrator/cost_router.py | 46 ++++++++++++---------- tests/test_cost_ledger.py | 14 +++++++ tests/test_cost_router.py | 4 +- tests/test_cost_router_boundaries.py | 54 +++++++++++++++++++++++++- 5 files changed, 98 insertions(+), 24 deletions(-) diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index 4d93c6460..621b0bc57 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -350,6 +350,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.price_known": record.price_known, } if record.workflow_run_id: attributes["contextual_orchestrator.workflow_run_id"] = record.workflow_run_id @@ -364,9 +365,10 @@ def from_record( "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, } + if record.price_known: + metrics["gen_ai.usage.cost"] = float(record.cost_amount) if error_type: metrics["contextual_orchestrator.usage.export_failures"] = 1.0 return cls( diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index 257690a92..6cee609e9 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -371,16 +371,18 @@ def complete( ) ) currencies = {record.currency_code for record in records} + price_known = all(record.price_known for record in records) provider_response["usage_record_ids"] = [ record.usage_record_id for record in records ] provider_response["cost"] = { "cost_amount": ( round(sum(record.cost_amount for record in records), 6) - if len(currencies) == 1 + if price_known and len(currencies) == 1 else None ), "currency_code": next(iter(currencies)) if len(currencies) == 1 else "MIXED", + "price_known": price_known, "measurement_status": ( "estimated" if any(record.measurement_status == "estimated" for record in records) @@ -391,14 +393,12 @@ def complete( provider_response["cost"]["currency_components"] = [ { "currency_code": currency, - "cost_amount": round( - sum( - record.cost_amount - for record in records - if record.currency_code == currency - ), - 6, + "cost_amount": ( + round(sum(record.cost_amount for record in records if record.currency_code == currency), 6) + if all(record.price_known for record in records if record.currency_code == currency) + else None ), + "price_known": all(record.price_known for record in records if record.currency_code == currency), } for currency in sorted(currencies) ] @@ -506,13 +506,15 @@ def complete( "total_tokens": sum(item.total_tokens for item in client_usage_records), } currencies = {item.currency_code for item in records} + price_known = all(item.price_known for item in records) result["cost"] = { "cost_amount": ( round(sum(item.cost_amount for item in records), 6) - if len(currencies) == 1 + if price_known and len(currencies) == 1 else None ), "currency_code": next(iter(currencies)) if len(currencies) == 1 else "MIXED", + "price_known": price_known, "measurement_status": ( "estimated" if any(item.measurement_status == "estimated" for item in records) @@ -523,14 +525,12 @@ def complete( result["cost"]["currency_components"] = [ { "currency_code": currency, - "cost_amount": round( - sum( - item.cost_amount - for item in records - if item.currency_code == currency - ), - 6, + "cost_amount": ( + round(sum(item.cost_amount for item in records if item.currency_code == currency), 6) + if all(item.price_known for item in records if item.currency_code == currency) + else None ), + "price_known": all(item.price_known for item in records if item.currency_code == currency), } for currency in sorted(currencies) ] @@ -629,6 +629,7 @@ def record_stream_usage( else "measured" ) currencies = {record.currency_code for record in records} + price_known = all(record.price_known for record in records) return { "usage_record_ids": [record.usage_record_id for record in records], "usage": ( @@ -643,11 +644,12 @@ def record_stream_usage( "cost": { "cost_amount": ( round(sum(record.cost_amount for record in records), 6) - if measurement_status == "measured" and len(currencies) == 1 + if measurement_status == "measured" and price_known and len(currencies) == 1 else None ), "currency_code": next(iter(currencies)) if len(currencies) == 1 else "MIXED", "measurement_status": measurement_status, + "price_known": price_known, }, } @@ -730,8 +732,9 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict "custom_id": item.custom_id, "answer": item.answer, "usage_record_id": record.usage_record_id, - "cost_amount": record.cost_amount, + "cost_amount": record.cost_amount if record.price_known else None, "currency_code": record.currency_code, + "price_known": record.price_known, "prompt_tokens": record.prompt_tokens, "completion_tokens": record.completion_tokens, "measurement_status": record.measurement_status, @@ -1041,6 +1044,7 @@ def embeddings_batch_document(self, batch_id: str) -> Dict[str, Any]: embeddings: List[Dict[str, Any]] = [] token_counts: List[int] = [] total_cost_amount = 0.0 + price_known = True currency_code = "USD" for source_index in range(input_count): parts = sorted(parts_by_source.get(source_index, []), key=lambda item: item["part_index"]) @@ -1065,6 +1069,7 @@ def embeddings_batch_document(self, batch_id: str) -> Dict[str, Any]: attribution=attribution, ) total_cost_amount += float(record.cost_amount) + price_known = price_known and record.price_known currency_code = record.currency_code token_counts.append(record.prompt_tokens) embeddings.append( @@ -1090,9 +1095,10 @@ def embeddings_batch_document(self, batch_id: str) -> Dict[str, Any]: "strategy": "token_budgeted_embedding_parts_weighted_average", **part_limits, }, - "cost_amount": round(total_cost_amount, 6), + "cost_amount": round(total_cost_amount, 6) if price_known else None, "currency_code": currency_code, - "cost_micro_usd": int(round(total_cost_amount * 1_000_000)), + "price_known": price_known, + "cost_micro_usd": int(round(total_cost_amount * 1_000_000)) if price_known else None, } self._embedding_documents[batch_id] = document return document diff --git a/tests/test_cost_ledger.py b/tests/test_cost_ledger.py index 97fe6cc5c..e26452881 100644 --- a/tests/test_cost_ledger.py +++ b/tests/test_cost_ledger.py @@ -314,6 +314,20 @@ def test_non_blocking_store_records_p2028_like_failure_as_telemetry_only() -> No assert "secret answer" not in repr(sink.events()) +def test_unpriced_usage_telemetry_omits_cost_metric() -> None: + sink = InMemoryUsageTelemetrySink() + ledger = CostLedger(PriceBook(InMemoryConfigStore()), telemetry_sink=sink) + ledger.record_usage( + provider="unknown", + model="unpriced", + prompt_tokens=1, + completion_tokens=1, + ) + event = sink.events()[0] + assert event.attributes["contextual_orchestrator.usage.price_known"] is False + assert "gen_ai.usage.cost" not in event.metrics + + def test_multi_dimensional_rollup_correctness() -> None: ledger = _priced_ledger() ledger.record_usage(provider="openai", model="gpt-x", prompt_tokens=1000, completion_tokens=1000, diff --git a/tests/test_cost_router.py b/tests/test_cost_router.py index 1b88f26c6..fd8b84558 100644 --- a/tests/test_cost_router.py +++ b/tests/test_cost_router.py @@ -527,8 +527,8 @@ def test_structured_mixed_currency_costs_are_never_implicitly_converted() -> Non assert result["cost"]["cost_amount"] is None assert result["cost"]["currency_code"] == "MIXED" assert result["cost"]["currency_components"] == [ - {"currency_code": "EUR", "cost_amount": 2.0}, - {"currency_code": "USD", "cost_amount": 1.0}, + {"currency_code": "EUR", "cost_amount": 2.0, "price_known": True}, + {"currency_code": "USD", "cost_amount": 1.0, "price_known": True}, ] assert "approved exchange-rate source" in result["cost"]["customer_action"] diff --git a/tests/test_cost_router_boundaries.py b/tests/test_cost_router_boundaries.py index d09d75ab5..36d270d0a 100644 --- a/tests/test_cost_router_boundaries.py +++ b/tests/test_cost_router_boundaries.py @@ -15,6 +15,7 @@ ) from contextual_orchestrator.batch_routing import ( BatchJob, + BatchRequest, BatchResultItem, EmbeddingBatchResultItem, ) @@ -96,6 +97,49 @@ def test_stream_usage_aggregates_trace_steps_without_text_estimates() -> None: assert len(coordinator.ledger.records()) == 2 +def test_unpriced_stream_usage_omits_cost_total() -> None: + result = _coordinator().record_stream_usage( + result={ + "workflow_run_id": "unpriced-stream", + "mode": "route", + "trace": [{"agent_id": "mock_worker", "usage": {"prompt_tokens": 1, "completion_tokens": 1}}], + }, + attribution=None, + model_name="mock-a", + ) + assert result["cost"]["price_known"] is False + assert result["cost"]["cost_amount"] is None + + +def test_unpriced_sync_cost_omits_total() -> None: + result = _coordinator().complete([{"role": "user", "content": "hello"}]) + assert result["cost"]["price_known"] is False + assert result["cost"]["cost_amount"] is None + + +def test_unpriced_provider_request_cost_omits_total() -> None: + result = _coordinator().complete( + [{"role": "user", "content": "return json"}], + provider_request={ + "model": "mock-a", + "messages": [{"role": "user", "content": "return json"}], + "response_format": {"type": "json_object"}, + }, + ) + assert result["cost"]["price_known"] is False + assert result["cost"]["cost_amount"] is None + + +def test_unpriced_batch_item_omits_cost() -> None: + coordinator = _coordinator() + job = coordinator.submit_batch( + [BatchRequest(messages=[{"role": "user", "content": "batch"}], model="mock-a")] + ) + item = coordinator.retrieve_batch(job.job_id)["results"][0] + assert item["price_known"] is False + assert item["cost_amount"] is None + + def test_complete_rejects_non_boolean_cache_bypass() -> None: coordinator = _coordinator() with pytest.raises(TypeError, match="cache_bypass"): @@ -327,6 +371,13 @@ def test_embeddings_document_is_idempotent_after_completion() -> None: assert backend.polled.count(job.job_id) == 1 +def test_unpriced_embeddings_document_omits_cost() -> None: + document = _coordinator().complete_embeddings_batch(["unpriced embedding"]) + assert document["price_known"] is False + assert document["cost_amount"] is None + assert document["cost_micro_usd"] is None + + def test_embeddings_document_requires_known_batch() -> None: coordinator = _coordinator() with pytest.raises(KeyError, match="embeddings batch job"): @@ -442,7 +493,8 @@ def test_complete_embeddings_batch_round_trips_locally() -> None: ) assert document["status"] == "completed" assert document["embeddings"][0]["index"] == 0 - assert document["cost_micro_usd"] >= 0 + assert document["cost_micro_usd"] is None + assert document["price_known"] is False From c37932f70c237638a58740a026195c8522a6e473 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 19:53:23 +0000 Subject: [PATCH 05/11] fix(cost_ledger): zero usage is always a known zero cost PriceBook.compute_cost() treated a zero-token record the same as a missing price row: both returned price_known=False. That's wrong for zero usage specifically -- zero tokens cost zero regardless of whether the per-token price is known, since zero times any finite price is still zero, so there is no unknown quantity left to guess. This broke the required "Full unit and contract suite" check on 2b0b2f30 ("expose unknown price across usage surfaces"), which started surfacing price_known on complete()'s cost dict: a cache hit records its answer via a synthetic ("cache", "response") provider/model that never has a real price row, so every cache hit was newly (and wrongly) reported as an unpriced request with cost_amount=None instead of the honest $0.00 it actually is. compute_cost() now special-cases prompt_tokens == completion_tokens == 0 before ever consulting the price book, returning price_known=True with cost_amount=0.0 (using the price entry's own currency when one exists, the ledger's default currency otherwise). Updated the pre-existing cache-hit test to expect the now-exposed price_known: True field. Full suite: 2868 passed, 1 skipped, 1 known pre-existing fast_mlsirm gap (Python 3.11 sandbox vs the package's python_full_version >= 3.12 requirement). --- CHANGELOG.md | 12 ++++++++++++ contextual_orchestrator/cost_ledger.py | 8 ++++++++ tests/test_distributed_cache_truth_and_isolation.py | 1 + 3 files changed, 21 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a96d76b1..5b73c6ff5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,18 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) retry's current price/measurement state instead of what actually priced the original spend. `append()` is now a true no-op on a rejected duplicate. +- `price_known` now propagates from the ledger into every downstream usage + surface: `CostRoutingCoordinator.complete()`'s sync/provider-request cost + dicts and their per-currency components, `record_stream_usage()`, + `retrieve_batch()`'s per-item results, and `embeddings_batch_document()`. + An unpriced request's `cost_amount` is `null` rather than a silent `0` + wherever it surfaces, not just in the ledger's own rollups. + `PriceBook.compute_cost()` treats zero token usage as the one exception: + zero tokens cost zero regardless of whether the provider/model's price is + known (zero times any finite price is still zero), so a cache hit — whose + synthetic `("cache", "response")` provider/model never has a price row — + is always `price_known=True`, `cost_amount=0.0` rather than being wrongly + reported as an unpriced request. - OpenRouter discovery no longer marks the entire credential account evidence-only. Authenticated catalog rows may serve ordinary requests, while diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index 621b0bc57..90b835ec1 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -227,8 +227,16 @@ def compute_cost( ``False`` for that case so callers can tell "priced at zero" apart from "we do not know the price" instead of silently fabricating a free price for an unpriced model. + + Zero usage is the one exception: zero tokens cost zero regardless of + whether this provider/model's per-token price is known, since zero + times any finite price is still zero. A cache hit (and any other + genuinely zero-token record, priced or not) is therefore always + ``price_known=True`` -- there is no unknown quantity left to guess. """ entry = self.get_price(provider, model) + if prompt_tokens == 0 and completion_tokens == 0: + return 0.0, entry.currency_code if entry is not None else self.default_currency, True if entry is None: return 0.0, self.default_currency, False prompt_cost = (Decimal(prompt_tokens) / Decimal(1000)) * Decimal( diff --git a/tests/test_distributed_cache_truth_and_isolation.py b/tests/test_distributed_cache_truth_and_isolation.py index 13dbcf1ff..1d45ebe8b 100644 --- a/tests/test_distributed_cache_truth_and_isolation.py +++ b/tests/test_distributed_cache_truth_and_isolation.py @@ -111,6 +111,7 @@ def test_cache_hit_records_zero_provider_usage_instead_of_rebilling_inference() assert second["cost"] == { "cost_amount": 0.0, "currency_code": "USD", + "price_known": True, "measurement_status": "measured", } From 231434116ba32b9e57c73f1161c7c02778dfcc02 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 05:09:02 +0900 Subject: [PATCH 06/11] fix(cost): preserve authoritative zero batch usage --- contextual_orchestrator/batch_routing.py | 14 ++++++++++-- contextual_orchestrator/cost_router.py | 15 +++++++++++-- tests/test_cost_router_boundaries.py | 28 ++++++++++++++++++++++++ 3 files changed, 53 insertions(+), 4 deletions(-) diff --git a/contextual_orchestrator/batch_routing.py b/contextual_orchestrator/batch_routing.py index cb9aae429..b1984e153 100644 --- a/contextual_orchestrator/batch_routing.py +++ b/contextual_orchestrator/batch_routing.py @@ -205,6 +205,7 @@ class BatchResultItem: attribution: Dict[str, Any] = field(default_factory=dict) model: str = "contextual-orchestrator" mode: str = "auto" + usage_valid: Optional[bool] = None class BatchBackend(Protocol): @@ -405,17 +406,26 @@ async def _download() -> Dict[str, Any]: body = (entry.get("response") or {}).get("body", {}) answer = _extract_answer(body) usage = body.get("usage", {}) or {} + raw_prompt_tokens = usage.get("prompt_tokens") + raw_completion_tokens = usage.get("completion_tokens") + usage_valid = ( + type(raw_prompt_tokens) is int + and raw_prompt_tokens >= 0 + and type(raw_completion_tokens) is int + and raw_completion_tokens >= 0 + ) raw_request = tracked.get(custom_id) request = BatchRequest(**raw_request) if raw_request else None items.append( BatchResultItem( custom_id=custom_id, answer=answer, - prompt_tokens=int(usage.get("prompt_tokens", 0)), - completion_tokens=int(usage.get("completion_tokens", 0)), + prompt_tokens=raw_prompt_tokens if usage_valid else 0, + completion_tokens=raw_completion_tokens if usage_valid else 0, attribution=dict(request.attribution) if request else {}, model=request.model if request else "contextual-orchestrator", mode=request.mode if request else "auto", + usage_valid=usage_valid, ) ) return items diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index 6cee609e9..393b0ae89 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -715,6 +715,17 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict recorded: List[Dict[str, Any]] = [] for item in items: provider_model = self._resolve_batch_provider_model(item) + usage_valid = ( + item.prompt_tokens >= 0 + and item.completion_tokens >= 0 + and ( + item.usage_valid is True + or ( + item.usage_valid is None + and (item.prompt_tokens > 0 or item.completion_tokens > 0) + ) + ) + ) record = self._record_completion( messages=[{"role": "user", "content": ""}], answer=item.answer, @@ -724,8 +735,8 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict model_name=item.model, provider_model=provider_model, workflow_run_id=job.job_id, - prompt_tokens=item.prompt_tokens or None, - completion_tokens=item.completion_tokens or None, + prompt_tokens=item.prompt_tokens if usage_valid else None, + completion_tokens=item.completion_tokens if usage_valid else None, ) recorded.append( { diff --git a/tests/test_cost_router_boundaries.py b/tests/test_cost_router_boundaries.py index 36d270d0a..e31dad274 100644 --- a/tests/test_cost_router_boundaries.py +++ b/tests/test_cost_router_boundaries.py @@ -140,6 +140,34 @@ def test_unpriced_batch_item_omits_cost() -> None: assert item["cost_amount"] is None +def test_provider_confirmed_zero_usage_stays_measured_and_price_known() -> None: + class ZeroUsageBackend: + name = "zero-usage" + + def submit(self, requests, metadata=None): # type: ignore[no-untyped-def] + del requests, metadata + return BatchJob("zero-usage-job", self.name, request_count=1) + + def retrieve(self, job): # type: ignore[no-untyped-def] + del job + return [BatchResultItem( + "zero-result", "", prompt_tokens=0, completion_tokens=0, + model="unpriced-model", usage_valid=True, + )] + + coordinator = _coordinator(batch_backend=ZeroUsageBackend()) + job = coordinator.submit_batch([BatchRequest( + messages=[{"role": "user", "content": "ignored"}], model="unpriced-model" + )]) + + item = coordinator.retrieve_batch(job.job_id)["results"][0] + + assert item["prompt_tokens"] == item["completion_tokens"] == 0 + assert item["measurement_status"] == "measured" + assert item["price_known"] is True + assert item["cost_amount"] == 0.0 + + def test_complete_rejects_non_boolean_cache_bypass() -> None: coordinator = _coordinator() with pytest.raises(TypeError, match="cache_bypass"): From ca88b4fb779bd951f02447324490972efb416e3c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 20:33:46 +0000 Subject: [PATCH 07/11] fix(cost_router): estimate invalid batch usage from the real prompt CodeRabbit review on #956: retrieve_batch's invalid-usage estimation fallback passed a hardcoded [{"role": "user", "content": ""}] placeholder into _record_completion instead of the request actually submitted. When a provider marks a batch item's usage invalid, _record_completion estimates prompt tokens from that messages argument -- an empty placeholder always estimates near-zero prompt tokens regardless of the real prompt's size, silently understating batch cost for any large-input request whose usage came back invalid. submit_batch() now keeps the submitted BatchRequest list keyed by job_id (self._batch_requests, mirroring the existing self._embedding_requests pattern for embeddings batches). retrieve_batch() looks the original request up by custom_id and uses its real messages for estimation whenever usage_valid is False, falling back to the empty placeholder only when no matching original request is on record (e.g. a backend that never went through submit_batch). New regression test (test_invalid_batch_usage_estimates_prompt_tokens_from_original_request) submits a ~500-word prompt through a backend that reports invalid usage, and asserts the recorded prompt_tokens reflects the real prompt rather than the near-zero an empty placeholder would give. Full suite: 2870 passed, 1 skipped, 1 known pre-existing fast_mlsirm gap (Python 3.11 sandbox vs the package's python_full_version >= 3.12 requirement). --- CHANGELOG.md | 14 +++++++++ contextual_orchestrator/cost_router.py | 23 ++++++++++++++- tests/test_cost_router_boundaries.py | 40 ++++++++++++++++++++++++++ 3 files changed, 76 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b73c6ff5..37e30dc18 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -55,6 +55,20 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) synthetic `("cache", "response")` provider/model never has a price row — is always `price_known=True`, `cost_amount=0.0` rather than being wrongly reported as an unpriced request. +- `retrieve_batch()`'s `item.prompt_tokens or None` treated a legitimately + reported zero token count the same as a missing one (Python's falsy-zero), + so a genuinely confirmed zero-usage batch item was silently downgraded to + an unmeasured estimate instead of staying a real, priced `measured` `0`. + `PgLlmBatchBackend.retrieve()` and `BatchResultItem` now carry an explicit + `usage_valid` tri-state (confirmed-typed non-negative counts vs. missing/ + malformed usage), and `retrieve_batch()` reads that instead of relying on + truthiness. That same estimation fallback also passed a hardcoded + empty-content placeholder instead of the request actually submitted, so a + large prompt whose provider marked usage invalid was undercounted to + near-zero prompt tokens — understating batch cost. (CodeRabbit review) + `submit_batch()` now keeps the submitted `BatchRequest` list keyed by job + id so `retrieve_batch()` can look the original request up by `custom_id` + and estimate prompt tokens from what was actually sent. - OpenRouter discovery no longer marks the entire credential account evidence-only. Authenticated catalog rows may serve ordinary requests, while diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index 393b0ae89..a272c5c80 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -117,6 +117,12 @@ def __init__( ) # job_id -> submitted BatchJob (so poll/retrieve can be driven by id) self._batch_jobs = registry.mapping("batch_jobs", decode=lambda raw: BatchJob(**raw)) + # job_id -> submitted BatchRequest list, so retrieve_batch can estimate + # prompt tokens from the real submitted prompt (not an empty + # placeholder) whenever a provider marks a batch item's usage invalid. + self._batch_requests = registry.mapping( + "batch_requests", decode=lambda raw: BatchRequest(**raw) + ) # embeddings batch state: job handle + submitted requests + cached doc, # keyed by batch id so poll/retrieve is idempotent (usage recorded once). self._embedding_jobs = registry.mapping("embedding_jobs", decode=lambda raw: BatchJob(**raw)) @@ -674,6 +680,7 @@ def submit_batch( job = self.batch_backend.submit(prepared_requests, metadata=metadata) job.owner_id = owner_id self._batch_jobs[job.job_id] = job + self._batch_requests[job.job_id] = prepared_requests return job def _resolve_batch_request(self, request: BatchRequest) -> BatchRequest: @@ -712,6 +719,9 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict """Retrieve results for a batch owned by ``owner_id`` and record usage.""" job = self._require_job(job_id, owner_id=owner_id) items: List[BatchResultItem] = self.batch_backend.retrieve(job) + request_by_custom_id = { + request.custom_id: request for request in self._batch_requests.get(job_id, []) + } recorded: List[Dict[str, Any]] = [] for item in items: provider_model = self._resolve_batch_provider_model(item) @@ -726,8 +736,19 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict ) ) ) + original_request = request_by_custom_id.get(item.custom_id) + # An invalid/missing provider usage falls back to estimating + # prompt tokens from the request messages (see _record_completion). + # A hardcoded empty placeholder would silently undercount prompt + # tokens for a real, possibly large submitted prompt -- use the + # actual submitted request when it is available. + estimation_messages = ( + original_request.messages + if not usage_valid and original_request is not None + else [{"role": "user", "content": ""}] + ) record = self._record_completion( - messages=[{"role": "user", "content": ""}], + messages=estimation_messages, answer=item.answer, route_mode=item.mode, request_channel="batch", diff --git a/tests/test_cost_router_boundaries.py b/tests/test_cost_router_boundaries.py index e31dad274..898cfdc84 100644 --- a/tests/test_cost_router_boundaries.py +++ b/tests/test_cost_router_boundaries.py @@ -168,6 +168,46 @@ def retrieve(self, job): # type: ignore[no-untyped-def] assert item["cost_amount"] == 0.0 +def test_invalid_batch_usage_estimates_prompt_tokens_from_original_request() -> None: + """usage_valid=False must estimate from the real submitted prompt. + + CodeRabbit review (PR #956): retrieve_batch's invalid-usage fallback + passed a hardcoded empty-content placeholder into the estimation path + instead of the request actually submitted, so a large prompt whose + provider marked usage invalid was undercounted to near-zero prompt + tokens -- understating batch cost. The original request is now looked + up by custom_id and used for estimation instead. + """ + class InvalidUsageBackend: + name = "invalid-usage" + + def submit(self, requests, metadata=None): # type: ignore[no-untyped-def] + del metadata + self._custom_id = requests[0].custom_id + return BatchJob("invalid-usage-job", self.name, request_count=1) + + def retrieve(self, job): # type: ignore[no-untyped-def] + del job + return [BatchResultItem( + self._custom_id, "short answer", prompt_tokens=-1, completion_tokens=2, + model="mock-a", usage_valid=False, + )] + + coordinator = _coordinator(batch_backend=InvalidUsageBackend()) + large_prompt = "word " * 500 + job = coordinator.submit_batch([ + BatchRequest(messages=[{"role": "user", "content": large_prompt}], model="mock-a") + ]) + + item = coordinator.retrieve_batch(job.job_id)["results"][0] + + assert item["measurement_status"] == "estimated" + # A hardcoded empty placeholder would estimate near-zero prompt tokens + # regardless of the real prompt's size; the actual submitted prompt must + # drive the estimate instead. + assert item["prompt_tokens"] > 100 + + def test_complete_rejects_non_boolean_cache_bypass() -> None: coordinator = _coordinator() with pytest.raises(TypeError, match="cache_bypass"): From 2bfcaf6d3ee1bb19ed8b461139fcc3c9b717ee52 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 05:42:56 +0900 Subject: [PATCH 08/11] fix(cost_router): keep batch prompt fallback metadata safe --- contextual_orchestrator/batch_routing.py | 3 ++ contextual_orchestrator/cost_router.py | 36 +++++++++--------------- tests/test_cost_router_boundaries.py | 36 ++++++++++++++++++++---- 3 files changed, 46 insertions(+), 29 deletions(-) diff --git a/contextual_orchestrator/batch_routing.py b/contextual_orchestrator/batch_routing.py index b1984e153..2292c2f73 100644 --- a/contextual_orchestrator/batch_routing.py +++ b/contextual_orchestrator/batch_routing.py @@ -192,6 +192,9 @@ class BatchJob: # HTTP callers bind this opaque digest to the authenticated principal; # library-only jobs may remain unowned for standalone use. owner_id: Optional[str] = None + # Prompt-token fallback estimates are safe metadata, stored atomically with + # the job handle rather than retaining submitted prompt text. + prompt_token_estimates: Dict[str, int] = field(default_factory=dict) @dataclass diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index a272c5c80..faf812062 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -117,12 +117,6 @@ def __init__( ) # job_id -> submitted BatchJob (so poll/retrieve can be driven by id) self._batch_jobs = registry.mapping("batch_jobs", decode=lambda raw: BatchJob(**raw)) - # job_id -> submitted BatchRequest list, so retrieve_batch can estimate - # prompt tokens from the real submitted prompt (not an empty - # placeholder) whenever a provider marks a batch item's usage invalid. - self._batch_requests = registry.mapping( - "batch_requests", decode=lambda raw: BatchRequest(**raw) - ) # embeddings batch state: job handle + submitted requests + cached doc, # keyed by batch id so poll/retrieve is idempotent (usage recorded once). self._embedding_jobs = registry.mapping("embedding_jobs", decode=lambda raw: BatchJob(**raw)) @@ -677,10 +671,16 @@ def submit_batch( raise BatchModelSelectionError( "no eligible model-group member is available for this batch request" ) from exc + prompt_token_estimates = { + request.custom_id: self.token_counter.count_messages( + request.messages, request.model + ) + for request in prepared_requests + } job = self.batch_backend.submit(prepared_requests, metadata=metadata) job.owner_id = owner_id + job.prompt_token_estimates = prompt_token_estimates self._batch_jobs[job.job_id] = job - self._batch_requests[job.job_id] = prepared_requests return job def _resolve_batch_request(self, request: BatchRequest) -> BatchRequest: @@ -719,9 +719,6 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict """Retrieve results for a batch owned by ``owner_id`` and record usage.""" job = self._require_job(job_id, owner_id=owner_id) items: List[BatchResultItem] = self.batch_backend.retrieve(job) - request_by_custom_id = { - request.custom_id: request for request in self._batch_requests.get(job_id, []) - } recorded: List[Dict[str, Any]] = [] for item in items: provider_model = self._resolve_batch_provider_model(item) @@ -736,19 +733,8 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict ) ) ) - original_request = request_by_custom_id.get(item.custom_id) - # An invalid/missing provider usage falls back to estimating - # prompt tokens from the request messages (see _record_completion). - # A hardcoded empty placeholder would silently undercount prompt - # tokens for a real, possibly large submitted prompt -- use the - # actual submitted request when it is available. - estimation_messages = ( - original_request.messages - if not usage_valid and original_request is not None - else [{"role": "user", "content": ""}] - ) record = self._record_completion( - messages=estimation_messages, + messages=[{"role": "user", "content": ""}], answer=item.answer, route_mode=item.mode, request_channel="batch", @@ -756,7 +742,11 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict model_name=item.model, provider_model=provider_model, workflow_run_id=job.job_id, - prompt_tokens=item.prompt_tokens if usage_valid else None, + prompt_tokens=( + item.prompt_tokens + if usage_valid + else job.prompt_token_estimates.get(item.custom_id) + ), completion_tokens=item.completion_tokens if usage_valid else None, ) recorded.append( diff --git a/tests/test_cost_router_boundaries.py b/tests/test_cost_router_boundaries.py index 898cfdc84..7a48533a2 100644 --- a/tests/test_cost_router_boundaries.py +++ b/tests/test_cost_router_boundaries.py @@ -19,6 +19,7 @@ BatchResultItem, EmbeddingBatchResultItem, ) +from contextual_orchestrator.batch_job_registry import JobRegistryFactory from contextual_orchestrator.cost_router import ( CostRoutingCoordinator as Coordinator, ) @@ -171,12 +172,9 @@ def retrieve(self, job): # type: ignore[no-untyped-def] def test_invalid_batch_usage_estimates_prompt_tokens_from_original_request() -> None: """usage_valid=False must estimate from the real submitted prompt. - CodeRabbit review (PR #956): retrieve_batch's invalid-usage fallback - passed a hardcoded empty-content placeholder into the estimation path - instead of the request actually submitted, so a large prompt whose - provider marked usage invalid was undercounted to near-zero prompt - tokens -- understating batch cost. The original request is now looked - up by custom_id and used for estimation instead. + The fallback count is computed before submission and stored as safe + metadata on the durable job record. Raw prompts are never copied into the + shared registry, and the accepted job has only one publication write. """ class InvalidUsageBackend: name = "invalid-usage" @@ -199,6 +197,10 @@ def retrieve(self, job): # type: ignore[no-untyped-def] BatchRequest(messages=[{"role": "user", "content": large_prompt}], model="mock-a") ]) + stored_job = coordinator._batch_jobs[job.job_id] # noqa: SLF001 + assert stored_job.prompt_token_estimates == job.prompt_token_estimates + assert large_prompt not in repr(stored_job) + item = coordinator.retrieve_batch(job.job_id)["results"][0] assert item["measurement_status"] == "estimated" @@ -208,6 +210,28 @@ def retrieve(self, job): # type: ignore[no-untyped-def] assert item["prompt_tokens"] > 100 +def test_batch_prompt_fallback_has_no_separate_registry_publication() -> None: + """Accepted jobs publish their safe fallback metadata in one job record.""" + class RecordingRegistry(JobRegistryFactory): + def __init__(self) -> None: + super().__init__() + self.names: list[str] = [] + + def mapping(self, name, *, decode=None): # type: ignore[no-untyped-def] + self.names.append(name) + return super().mapping(name, decode=decode) + + registry = RecordingRegistry() + coordinator = _coordinator(job_registry=registry) + + coordinator.submit_batch([ + BatchRequest(messages=[{"role": "user", "content": "private prompt"}]) + ]) + + assert "batch_jobs" in registry.names + assert "batch_requests" not in registry.names + + def test_complete_rejects_non_boolean_cache_bypass() -> None: coordinator = _coordinator() with pytest.raises(TypeError, match="cache_bypass"): From f0945bfa7b37239d0da9d48f7d78dd56ffaa2c6d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 20:44:51 +0000 Subject: [PATCH 09/11] docs(changelog): document the batch prompt fallback metadata fix Updates the CHANGELOG entry to describe the actual shipped design from 2bfcaf6d (fix(cost_router): keep batch prompt fallback metadata safe) rather than the superseded intermediate approach: the prompt- token estimate lives on the durable BatchJob record itself (one publication write, no raw prompt content ever persisted), not a separate registry holding the original BatchRequest list. --- CHANGELOG.md | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 37e30dc18..6d92a3f46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,9 +66,19 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) empty-content placeholder instead of the request actually submitted, so a large prompt whose provider marked usage invalid was undercounted to near-zero prompt tokens — understating batch cost. (CodeRabbit review) - `submit_batch()` now keeps the submitted `BatchRequest` list keyed by job - id so `retrieve_batch()` can look the original request up by `custom_id` - and estimate prompt tokens from what was actually sent. + `submit_batch()` now computes a real prompt-token estimate per + `custom_id` before submission and stores it as a `prompt_token_estimates` + field on the durable `BatchJob` record itself, rather than the raw + submitted messages (Devin review: a batch registry can be Valkey-backed + with a multi-day retention shared across processes, and a submitted + prompt may be ZDR-flagged or otherwise sensitive — a token count carries + no reconstructable prompt content). Publishing it on the existing + `BatchJob` write also means an accepted job still has exactly one + publication write, so a metadata-only estimate can never orphan an + already-accepted (and possibly billed) backend job behind a raised + exception with no job id ever returned to the caller. `retrieve_batch()` + reads that stored estimate instead of falling back to an empty + placeholder. - OpenRouter discovery no longer marks the entire credential account evidence-only. Authenticated catalog rows may serve ordinary requests, while From 1b986f54f42218b37429c1996fe1832b275e84dd Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 05:53:08 +0900 Subject: [PATCH 10/11] fix(cost_router): preserve legacy batch prompt estimates --- contextual_orchestrator/cost_router.py | 33 ++++++++++++++++++- tests/test_cost_router_boundaries.py | 44 ++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index faf812062..70116fa0a 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -115,6 +115,7 @@ def __init__( token_counter=self.token_counter, job_registry=registry ) ) + self._job_registry = registry # job_id -> submitted BatchJob (so poll/retrieve can be driven by id) self._batch_jobs = registry.mapping("batch_jobs", decode=lambda raw: BatchJob(**raw)) # embeddings batch state: job handle + submitted requests + cached doc, @@ -719,6 +720,8 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict """Retrieve results for a batch owned by ``owner_id`` and record usage.""" job = self._require_job(job_id, owner_id=owner_id) items: List[BatchResultItem] = self.batch_backend.retrieve(job) + request_by_custom_id = self._legacy_batch_requests(job) + prompt_token_estimates = dict(job.prompt_token_estimates) recorded: List[Dict[str, Any]] = [] for item in items: provider_model = self._resolve_batch_provider_model(item) @@ -733,6 +736,12 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict ) ) ) + if not usage_valid and item.custom_id not in prompt_token_estimates: + original_request = request_by_custom_id.get(item.custom_id) + if original_request is not None: + prompt_token_estimates[item.custom_id] = self.token_counter.count_messages( + original_request.messages, item.model + ) record = self._record_completion( messages=[{"role": "user", "content": ""}], answer=item.answer, @@ -745,7 +754,7 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict prompt_tokens=( item.prompt_tokens if usage_valid - else job.prompt_token_estimates.get(item.custom_id) + else prompt_token_estimates.get(item.custom_id) ), completion_tokens=item.completion_tokens if usage_valid else None, ) @@ -762,6 +771,9 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict "measurement_status": record.measurement_status, } ) + if prompt_token_estimates != job.prompt_token_estimates: + job.prompt_token_estimates = prompt_token_estimates + self._batch_jobs[job.job_id] = job return { "job_id": job_id, "backend": job.backend, @@ -769,6 +781,25 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict "results": recorded, } + def _legacy_batch_requests(self, job: BatchJob) -> Dict[str, BatchRequest]: + """Read pre-upgrade batch requests only when a stored estimate is absent.""" + if job.prompt_token_estimates: + return {} + legacy_requests = self._job_registry.mapping( + "batch_requests", decode=lambda raw: BatchRequest(**raw) + ) + try: + requests = legacy_requests[job.job_id] + except KeyError: + return {} + if not isinstance(requests, list): + return {} + return { + request.custom_id: request + for request in requests + if isinstance(request, BatchRequest) + } + def _resolve_batch_provider_model(self, item: BatchResultItem) -> tuple[str, str]: provider = str(item.attribution.get("provider") or item.attribution.get("upstream_api") or "") if not provider: diff --git a/tests/test_cost_router_boundaries.py b/tests/test_cost_router_boundaries.py index 7a48533a2..dc063f218 100644 --- a/tests/test_cost_router_boundaries.py +++ b/tests/test_cost_router_boundaries.py @@ -210,6 +210,50 @@ def retrieve(self, job): # type: ignore[no-untyped-def] assert item["prompt_tokens"] > 100 +def test_invalid_batch_usage_uses_legacy_request_registry_when_job_predates_prompt_estimates() -> None: + """Pre-upgrade jobs keep old prompt fallback compatibility without new writes.""" + class LegacyRegistry(JobRegistryFactory): + def __init__(self) -> None: + super().__init__() + self._mappings: dict[str, Any] = {} + + def mapping(self, name, *, decode=None): # type: ignore[no-untyped-def] + if name not in self._mappings: + self._mappings[name] = {} + return self._mappings[name] + + class InvalidUsageBackend: + name = "invalid-usage" + + def retrieve(self, job): # type: ignore[no-untyped-def] + del job + return [BatchResultItem( + "legacy-request", "short answer", prompt_tokens=-1, completion_tokens=2, + model="mock-a", usage_valid=False, + )] + + registry = LegacyRegistry() + coordinator = _coordinator(batch_backend=InvalidUsageBackend(), job_registry=registry) + coordinator._batch_jobs["legacy-job"] = BatchJob( # noqa: SLF001 + "legacy-job", "invalid-usage", request_count=1 + ) + registry.mapping("batch_requests", decode=lambda raw: BatchRequest(**raw))["legacy-job"] = [ + BatchRequest( + messages=[{"role": "user", "content": "word " * 500}], + model="mock-a", + custom_id="legacy-request", + ) + ] + + item = coordinator.retrieve_batch("legacy-job")["results"][0] + + assert item["measurement_status"] == "estimated" + assert item["prompt_tokens"] > 100 + assert coordinator._batch_jobs["legacy-job"].prompt_token_estimates == { # noqa: SLF001 + "legacy-request": item["prompt_tokens"] + } + + def test_batch_prompt_fallback_has_no_separate_registry_publication() -> None: """Accepted jobs publish their safe fallback metadata in one job record.""" class RecordingRegistry(JobRegistryFactory): From 3e45e34498223461e22ca16518b302d07f2c8d61 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 21:03:22 +0000 Subject: [PATCH 11/11] fix(cost_router): keep legacy batch estimates alive across partial retrievals Devin review on #956: _legacy_batch_requests() gated its lookup on `if job.prompt_token_estimates: return {}` -- meaning any non-empty dict short-circuited the legacy read, not just a fully-populated one. A legacy (pre-fix) job's estimates can be filled in incrementally across multiple retrieve_batch() calls (e.g. polling for still- processing items); once the first partial retrieval stored even one custom_id's estimate, every subsequent retrieval for that same job stopped consulting the legacy registry entirely, silently leaving any other still-unestimated custom_id's cost understated with no way to recover it later. retrieve_batch() now decides whether a legacy lookup is needed from the current retrieval's own items (does any item still lack a stored estimate), not from whether the job's dict happens to be non-empty. _legacy_batch_requests() itself is now a plain, ungated read -- for a normal post-fix job (fully estimated at submission time), the caller never invokes it at all, so there is no added cost to the common case. Extracted the usage-validity check the caller and the pre-scan both need into _batch_item_usage_valid() rather than duplicating it. New regression test drives a legacy job through two separate retrieve_batch() calls, each surfacing one of two legacy custom_ids, and asserts both get correctly estimated from the real submitted prompt -- reproducing the exact gap Devin flagged, which the prior gate would have left the second call's estimate at None. Targeted: tests/test_cost_router_boundaries.py + test_cost_router.py + test_batch_routing_boundaries.py + test_cost_ledger.py + test_cost_ledger_boundaries.py, 133 passed. Ruff passed; interrogate 100%. Full suite validation running in the background. --- CHANGELOG.md | 20 +++++++- contextual_orchestrator/cost_router.py | 50 +++++++++++++------ tests/test_cost_router_boundaries.py | 69 ++++++++++++++++++++++++++ 3 files changed, 123 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d92a3f46..bfd020b9b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -78,7 +78,25 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) already-accepted (and possibly billed) backend job behind a raised exception with no job id ever returned to the caller. `retrieve_batch()` reads that stored estimate instead of falling back to an empty - placeholder. + placeholder. (Devin review) A job accepted before this fix has no + `prompt_token_estimates` at all, and its original request never lived + anywhere durable that a post-fix retrieval could still read — so + `retrieve_batch()` now also reads a legacy, pre-fix `batch_requests` + registry entry (still populated only for jobs submitted before this + change; nothing writes new entries there anymore) whenever a custom_id + has no stored estimate, computes the real prompt-token count from it the + same way a fresh submission would have, and persists that estimate back + onto the job so a re-retrieval never repeats the lookup. That legacy + lookup initially gated on whether `job.prompt_token_estimates` was + non-empty at all — wrong once any one custom_id had already picked up an + estimate (including from an earlier partial retrieval of the same job), + since every other still-unestimated custom_id would then silently stop + being looked up for the rest of that job's lifetime. It now gates on + whether the current retrieval actually has an item that still needs the + legacy lookup, so a legacy job's estimates can be filled in correctly + across as many partial retrievals as it takes; a job that has never + needed the legacy path (everything submitted after this fix) still never + touches that registry. - OpenRouter discovery no longer marks the entire credential account evidence-only. Authenticated catalog rows may serve ordinary requests, while diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index 70116fa0a..5521e8457 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -720,22 +720,19 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict """Retrieve results for a batch owned by ``owner_id`` and record usage.""" job = self._require_job(job_id, owner_id=owner_id) items: List[BatchResultItem] = self.batch_backend.retrieve(job) - request_by_custom_id = self._legacy_batch_requests(job) prompt_token_estimates = dict(job.prompt_token_estimates) + needs_legacy_lookup = any( + not self._batch_item_usage_valid(item) + and item.custom_id not in prompt_token_estimates + for item in items + ) + request_by_custom_id = ( + self._legacy_batch_requests(job) if needs_legacy_lookup else {} + ) recorded: List[Dict[str, Any]] = [] for item in items: provider_model = self._resolve_batch_provider_model(item) - usage_valid = ( - item.prompt_tokens >= 0 - and item.completion_tokens >= 0 - and ( - item.usage_valid is True - or ( - item.usage_valid is None - and (item.prompt_tokens > 0 or item.completion_tokens > 0) - ) - ) - ) + usage_valid = self._batch_item_usage_valid(item) if not usage_valid and item.custom_id not in prompt_token_estimates: original_request = request_by_custom_id.get(item.custom_id) if original_request is not None: @@ -781,10 +778,33 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict "results": recorded, } + @staticmethod + def _batch_item_usage_valid(item: BatchResultItem) -> bool: + """True when a batch result item's provider-reported usage is trustworthy.""" + return ( + item.prompt_tokens >= 0 + and item.completion_tokens >= 0 + and ( + item.usage_valid is True + or ( + item.usage_valid is None + and (item.prompt_tokens > 0 or item.completion_tokens > 0) + ) + ) + ) + def _legacy_batch_requests(self, job: BatchJob) -> Dict[str, BatchRequest]: - """Read pre-upgrade batch requests only when a stored estimate is absent.""" - if job.prompt_token_estimates: - return {} + """Read pre-upgrade batch requests for a job with an unestimated custom_id. + + Callers gate this on whether the current retrieval actually needs a + legacy lookup (some item still lacks a stored/computed estimate), not + on whether ``job.prompt_token_estimates`` is merely non-empty -- a + job's estimates can be filled in incrementally across multiple + ``retrieve_batch`` calls, and gating on non-emptiness alone would stop + looking up any custom_id not yet covered by an earlier partial + retrieval. A job never seen by the legacy registry (every job + submitted after this fix) costs one cheap KeyError-guarded miss. + """ legacy_requests = self._job_registry.mapping( "batch_requests", decode=lambda raw: BatchRequest(**raw) ) diff --git a/tests/test_cost_router_boundaries.py b/tests/test_cost_router_boundaries.py index dc063f218..aa778f865 100644 --- a/tests/test_cost_router_boundaries.py +++ b/tests/test_cost_router_boundaries.py @@ -254,6 +254,75 @@ def retrieve(self, job): # type: ignore[no-untyped-def] } +def test_legacy_batch_requests_stay_available_across_partial_retrievals() -> None: + """A partial retrieval must not stop a later one from reaching legacy fallback. + + (Devin review on #956) Once one legacy custom_id picks up a stored + estimate, ``job.prompt_token_estimates`` becomes non-empty -- gating the + legacy registry lookup on mere non-emptiness would silently stop looking + up every other still-unestimated custom_id from that point on, even + though the legacy request registry still holds their original prompts. + """ + class LegacyRegistry(JobRegistryFactory): + def __init__(self) -> None: + super().__init__() + self._mappings: dict[str, Any] = {} + + def mapping(self, name, *, decode=None): # type: ignore[no-untyped-def] + if name not in self._mappings: + self._mappings[name] = {} + return self._mappings[name] + + class TwoPassInvalidUsageBackend: + name = "invalid-usage" + + def __init__(self) -> None: + self.calls = 0 + + def retrieve(self, job): # type: ignore[no-untyped-def] + del job + self.calls += 1 + custom_id = "legacy-request-one" if self.calls == 1 else "legacy-request-two" + return [BatchResultItem( + custom_id, "short answer", prompt_tokens=-1, completion_tokens=2, + model="mock-a", usage_valid=False, + )] + + registry = LegacyRegistry() + backend = TwoPassInvalidUsageBackend() + coordinator = _coordinator(batch_backend=backend, job_registry=registry) + coordinator._batch_jobs["legacy-job"] = BatchJob( # noqa: SLF001 + "legacy-job", "invalid-usage", request_count=2 + ) + registry.mapping("batch_requests", decode=lambda raw: BatchRequest(**raw))["legacy-job"] = [ + BatchRequest( + messages=[{"role": "user", "content": "word " * 500}], + model="mock-a", + custom_id="legacy-request-one", + ), + BatchRequest( + messages=[{"role": "user", "content": "word " * 700}], + model="mock-a", + custom_id="legacy-request-two", + ), + ] + + first = coordinator.retrieve_batch("legacy-job")["results"][0] + assert first["prompt_tokens"] > 100 # first partial retrieval: legacy lookup works + + second = coordinator.retrieve_batch("legacy-job")["results"][0] + # Pre-fix: the first retrieval already left prompt_token_estimates + # non-empty, so this second retrieval's legacy lookup was skipped + # entirely and this custom_id's real prompt was never found. + assert second["custom_id"] == "legacy-request-two" + assert second["prompt_tokens"] > 100 + + assert coordinator._batch_jobs["legacy-job"].prompt_token_estimates == { # noqa: SLF001 + "legacy-request-one": first["prompt_tokens"], + "legacy-request-two": second["prompt_tokens"], + } + + def test_batch_prompt_fallback_has_no_separate_registry_publication() -> None: """Accepted jobs publish their safe fallback metadata in one job record.""" class RecordingRegistry(JobRegistryFactory):