From 8c71c4a38715cd9b06d4df655ea588d5ce5f9e29 Mon Sep 17 00:00:00 2001 From: satoridev01 Date: Mon, 2 Nov 2026 07:51:40 -0300 Subject: [PATCH] fix(cost): pair provider charges with their own response's estimate Once any response reported a provider charge, total_cost summed only the reported charges and dropped every estimate, so responses that came back without a charge (an unpriced provider, a stream cut before its usage chunk) were no longer counted. The scan cost was understated and --max-budget stopped later than configured. Key estimates and charges by response id. A charge replaces its own response's estimate, a response without a charge keeps its estimate, and whatever cannot be matched by id keeps the previous behavior. The SDK leaves Chat Completions' response_id empty or a placeholder, so the id is read from the output items' provider_data, or from the request log when neither has it. Manually configured prices still take precedence. --- strix/config/models.py | 3 +- strix/core/hooks.py | 3 +- strix/report/dedupe.py | 32 +++--- strix/report/state.py | 70 +++++++++++- strix/report/usage.py | 59 ++++++++-- tests/test_cost_response_pairing.py | 161 ++++++++++++++++++++++++++++ tests/test_cost_tracking.py | 4 +- tests/test_pricing.py | 74 +++++++++++++ 8 files changed, 375 insertions(+), 31 deletions(-) create mode 100644 tests/test_cost_response_pairing.py diff --git a/strix/config/models.py b/strix/config/models.py index 102d132cf..85b575b17 100644 --- a/strix/config/models.py +++ b/strix/config/models.py @@ -705,9 +705,10 @@ def configure_sdk_model_defaults(settings: Settings) -> None: """Apply Strix config to SDK-native defaults.""" llm = settings.llm set_tracing_disabled(True) - from strix.report.state import record_llm_request + from strix.report.state import correlate_llm_response, record_llm_request request_log.install() + request_log.register_sink(correlate_llm_response) if settings.telemetry.run_record_timings: request_log.register_sink(record_llm_request) if codex.subscription_model(llm.model): diff --git a/strix/core/hooks.py b/strix/core/hooks.py index 458246689..72229e3c8 100644 --- a/strix/core/hooks.py +++ b/strix/core/hooks.py @@ -11,7 +11,7 @@ from typing import TYPE_CHECKING, Any from agents.lifecycle import RunHooks from strix.core.agents import BudgetPolicy, coordinator_from_context -from strix.report.state import get_global_report_state, record_tool_call +from strix.report.state import get_global_report_state, record_tool_call, sdk_response_id if TYPE_CHECKING: @@ -418,6 +418,7 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]): agent_name=agent_name, model=self._model, usage=response.usage, + response_id=sdk_response_id(response), ) except Exception: logger.exception("failed to record SDK usage for agent %s", agent_id) diff --git a/strix/report/dedupe.py b/strix/report/dedupe.py index b1d3e5370..789656681 100644 --- a/strix/report/dedupe.py +++ b/strix/report/dedupe.py @@ -16,7 +16,8 @@ from strix.config.models import ( configure_sdk_model_defaults, ) from strix.core.inputs import make_model_settings -from strix.report.state import get_global_report_state +from strix.llm import request_log +from strix.report.state import get_global_report_state, sdk_response_id if TYPE_CHECKING: @@ -367,18 +368,22 @@ async def check_duplicate( configure_sdk_model_defaults(settings) resolved_model = model_name.strip() model = resolve_dedupe_model(dedupe, resolved_model) - response = await model.get_response( - system_instructions=DEDUPE_SYSTEM_PROMPT, - input=user_msg, - model_settings=_dedupe_model_settings(dedupe, resolved_model, settings.llm.timeout), - tools=[], - output_schema=None, - handoffs=[], - tracing=ModelTracing.DISABLED, - previous_response_id=None, - conversation_id=None, - prompt=None, - ) + token = request_log.bind_call_context("dedupe", "dedupe") + try: + response = await model.get_response( + system_instructions=DEDUPE_SYSTEM_PROMPT, + input=user_msg, + model_settings=_dedupe_model_settings(dedupe, resolved_model, settings.llm.timeout), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ) + finally: + request_log.reset_call_context(token) report_state = get_global_report_state() if report_state is not None: report_state.record_sdk_usage( @@ -386,6 +391,7 @@ async def check_duplicate( agent_name="dedupe", model=resolved_model, usage=response.usage, + response_id=sdk_response_id(response), ) content = _extract_text(response) if not content: diff --git a/strix/report/state.py b/strix/report/state.py index 104e4af4c..8934077f7 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -3,7 +3,9 @@ import logging import re import subprocess import threading +from collections import deque from collections.abc import Callable +from contextlib import suppress from datetime import UTC, datetime from importlib.metadata import PackageNotFoundError, version from pathlib import Path @@ -236,6 +238,11 @@ class ReportState: } self._run_dir: Path | None = None self._saved_vuln_ids: set[str] = set() + # Chat Completions' SDK ModelResponse has response_id=None even when + # LiteLLM's provider reply has an id. The request log sees the latter; + # queue successful ids per agent so the usage hook can pair its estimate + # with the provider callback for that exact response. + self._pending_response_ids: dict[str, deque[str]] = {} self.caido_url: str | None = None self.vulnerability_found_callback: Callable[[dict[str, Any]], None] | None = None @@ -646,18 +653,37 @@ class ReportState: usage: "Usage | None", agent_name: str | None = None, model: str | None = None, + response_id: str | None = None, ) -> None: """Record SDK-native token usage for one completed model run/cycle.""" + pending = self._pending_response_ids.get(agent_id) + if pending: + if response_id: + with suppress(ValueError): + pending.remove(response_id) + elif len(pending) == 1: + response_id = pending.popleft() + else: + # Multiple provider successes can precede one SDK hook after + # concurrent calls or a conversion failure. Their order is + # ambiguous; leave estimates separate rather than pair the + # wrong charge and understate the scan's budget. + pending.clear() if self._llm_usage.record( agent_id=agent_id, agent_name=agent_name, model=model, usage=usage, + response_id=response_id, ): self.save_run_data() - def record_observed_llm_cost(self, cost: float) -> None: - self._llm_usage.record_observed_cost(cost) + def note_llm_response(self, agent_id: str | None, response_id: str | None) -> None: + if agent_id and response_id: + self._pending_response_ids.setdefault(agent_id, deque()).append(response_id) + + def record_observed_llm_cost(self, cost: float, *, response_id: str | None = None) -> None: + self._llm_usage.record_observed_cost(cost, response_id=response_id) def record_llm_provider( self, @@ -983,6 +1009,31 @@ def _response_id(completion_response: Any) -> str | None: return response_id if isinstance(response_id, str) and response_id else None +def sdk_response_id(response: Any) -> str | None: + """Recover Chat Completions' provider id from SDK output item metadata. + + The Agents SDK leaves ``ModelResponse.response_id`` empty for non-streamed + Chat Completions and sets the constant ``FAKE_RESPONSES_ID`` for streamed + ones, but places the provider's chunk id in each output item's + ``provider_data``. The placeholder is not an id: every response would share + it, so it is skipped. Refuse to guess if the output names different ids. + """ + # Deferred: strix.report must not import the agents SDK at import time. + from agents.models.fake_id import FAKE_RESPONSES_ID + + response_id = getattr(response, "response_id", None) + if isinstance(response_id, str) and response_id and response_id != FAKE_RESPONSES_ID: + return response_id + ids: set[str] = set() + for item in getattr(response, "output", ()) or (): + provider_data = getattr(item, "provider_data", None) + if isinstance(provider_data, dict): + value = provider_data.get("response_id") + if isinstance(value, str) and value: + ids.add(value) + return next(iter(ids)) if len(ids) == 1 else None + + class StreamedOpenRouterCosts: """Correlates OpenRouter's per-stream cost from the parser to the cost callback. @@ -1021,6 +1072,15 @@ class StreamedOpenRouterCosts: streamed_openrouter_costs = StreamedOpenRouterCosts() +def correlate_llm_response(event: "LlmRequestEvent") -> None: + """Match provider response ids to SDK usage even without timing diagnostics.""" + if event.outcome != "success": + return + report_state = get_global_report_state() + if report_state is not None and hasattr(report_state, "_pending_response_ids"): + report_state.note_llm_response(event.agent_id, event.response_id) + + def record_llm_request(event: "LlmRequestEvent") -> None: """Request-log sink: each provider attempt's timing, kept in run.json's ``llm_requests``.""" report_state = get_global_report_state() @@ -1134,7 +1194,11 @@ def litellm_cost_callback( if report_state is None: return try: - report_state.record_observed_llm_cost(cost) + response_id = _response_id(completion_response) + if response_id: + report_state.record_observed_llm_cost(cost, response_id=response_id) + else: + report_state.record_observed_llm_cost(cost) except Exception: logger.exception("Failed to record observed LiteLLM cost") diff --git a/strix/report/usage.py b/strix/report/usage.py index 3f3f10d01..cccaba8da 100644 --- a/strix/report/usage.py +++ b/strix/report/usage.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +import math from typing import Any from agents.usage import Usage, deserialize_usage, serialize_usage @@ -67,7 +68,15 @@ class LLMUsageLedger: self._agent_estimated_cost: dict[str, float] = {} self._observed_cost = 0.0 self._estimated_cost = 0.0 - self._has_observed_cost = False + # A provider callback and the SDK usage hook can describe the same + # response. Match them by response id so an observed charge replaces + # that response's estimate, while an unpriced response keeps its own + # estimate. This matters when only some OpenRouter calls report cost. + # What cannot be matched by id (no id on one side) keeps the upstream + # contract, see ``total_cost``. + self._estimated_by_response: dict[str, float] = {} + self._observed_by_response: dict[str, float] = {} + self._persisted_cost = 0.0 # Keyed by upstream provider name, e.g. "Z.AI" or "DeepInfra". self._providers: dict[str, ProviderUsage] = {} # Each agent's last prompt size, which its next call should find cached. @@ -83,6 +92,7 @@ class LLMUsageLedger: usage: Usage | None, agent_name: str | None = None, model: str | None = None, + response_id: str | None = None, ) -> bool: if usage is None or not _usage_has_activity(usage): return False @@ -99,20 +109,25 @@ class LLMUsageLedger: if not self.zero_cost: estimated = _estimate_litellm_cost(usage, model) - if estimated: - self._estimated_cost += estimated + if estimated and math.isfinite(estimated): + if response_id: + self._estimated_by_response[response_id] = estimated + else: + self._estimated_cost += estimated self._agent_estimated_cost[normalized_agent_id] = ( self._agent_estimated_cost.get(normalized_agent_id, 0.0) + estimated ) return True - def record_observed_cost(self, cost: float) -> None: + def record_observed_cost(self, cost: float, *, response_id: str | None = None) -> None: if self.zero_cost: return - if isinstance(cost, int | float) and cost > 0: - self._observed_cost += float(cost) - self._has_observed_cost = True + if isinstance(cost, int | float) and math.isfinite(cost) and cost > 0: + if response_id: + self._observed_by_response[response_id] = float(cost) + else: + self._observed_cost += float(cost) def record_provider( self, @@ -150,7 +165,28 @@ class LLMUsageLedger: return _round_cost( sum(self._configured_agent_cost(aid, model, prices) for aid in self._agent_usage) ) - return _round_cost(self._observed_cost if self._has_observed_cost else self._estimated_cost) + # Copies: tool threads read the total while the event loop records new + # responses, and iterating a dict that grows raises RuntimeError. + estimated_by_response = dict(self._estimated_by_response) + observed_by_response = dict(self._observed_by_response) + # Matched by response id: the provider's charge replaces that response's + # estimate, and a response without a charge keeps its own estimate. + paired = sum( + observed_by_response.get(response_id, estimate) + for response_id, estimate in estimated_by_response.items() + ) + # Whatever id matching cannot settle keeps the upstream contract: a + # provider charge with no estimate to match replaces the unkeyed + # estimates instead of adding to them, because without a shared id they + # may describe the same response. Adding both would overstate the spend + # and stop a scan early at --max-budget. + unpaired_observed = self._observed_cost + sum( + cost + for response_id, cost in observed_by_response.items() + if response_id not in estimated_by_response + ) + unpaired = unpaired_observed if unpaired_observed > 0 else self._estimated_cost + return _round_cost(self._persisted_cost + paired + unpaired) def _configured_agent_cost(self, agent_id: str, model: str | None, prices: Prices) -> float: if self._agent_metadata.get(agent_id, {}).get("model") == model: @@ -198,7 +234,9 @@ class LLMUsageLedger: self._agent_estimated_cost.clear() self._observed_cost = 0.0 self._estimated_cost = 0.0 - self._has_observed_cost = False + self._estimated_by_response.clear() + self._observed_by_response.clear() + self._persisted_cost = 0.0 self._providers = {} if not isinstance(raw_usage, dict): @@ -216,8 +254,7 @@ class LLMUsageLedger: self._total_usage = Usage() persisted_cost = _float_or_zero(raw_usage.get("cost")) - self._observed_cost = persisted_cost - self._estimated_cost = persisted_cost + self._persisted_cost = persisted_cost for raw_agent in raw_usage.get("agents") or []: if not isinstance(raw_agent, dict): diff --git a/tests/test_cost_response_pairing.py b/tests/test_cost_response_pairing.py new file mode 100644 index 000000000..dc2283abc --- /dev/null +++ b/tests/test_cost_response_pairing.py @@ -0,0 +1,161 @@ +"""Provider charges pair with their own response's estimate by response id.""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import TYPE_CHECKING + +import pytest +from agents.models.fake_id import FAKE_RESPONSES_ID +from agents.usage import Usage + +from strix.core.hooks import ReportUsageHooks +from strix.report import usage as usage_module +from strix.report.state import ( + ReportState, + litellm_cost_callback, + sdk_response_id, + set_global_report_state, +) +from strix.report.usage import LLMUsageLedger + + +if TYPE_CHECKING: + from pathlib import Path + + +def _usage(tokens: int) -> Usage: + return Usage(requests=1, input_tokens=tokens, output_tokens=0, total_tokens=tokens) + + +def test_sdk_chat_completion_output_preserves_provider_response_id() -> None: + response = SimpleNamespace( + response_id=None, + output=[SimpleNamespace(provider_data={"response_id": "gen-123"})], + ) + assert sdk_response_id(response) == "gen-123" + response.output.append(SimpleNamespace(provider_data={"response_id": "gen-other"})) + assert sdk_response_id(response) is None + + +def test_sdk_response_id_skips_the_placeholder_streamed_chat_completions_carry() -> None: + # A streamed Chat Completions response comes back from the SDK with the + # constant FAKE_RESPONSES_ID, which says nothing about the response. + streamed = SimpleNamespace( + response_id=FAKE_RESPONSES_ID, + output=[SimpleNamespace(provider_data={"response_id": "gen-123"})], + ) + assert sdk_response_id(streamed) == "gen-123" + + assert sdk_response_id(SimpleNamespace(response_id=FAKE_RESPONSES_ID, output=[])) is None + assert sdk_response_id(SimpleNamespace(response_id="resp_1", output=[])) == "resp_1" + + +async def test_streamed_chat_completion_charge_replaces_its_sdk_estimate( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + monkeypatch.setattr( + usage_module, + "_estimate_litellm_cost", + lambda used, _model: used.input_tokens / 1000, + ) + state = ReportState(run_name="streamed") + hooks = ReportUsageHooks(model="openrouter/z-ai/glm-5.3-flash") + context = SimpleNamespace(context={"agent_id": "root"}) + agent = SimpleNamespace(name="root") + + def sdk_response(tokens: int, provider_id: str) -> SimpleNamespace: + return SimpleNamespace( + response_id=FAKE_RESPONSES_ID, + usage=_usage(tokens), + output=[SimpleNamespace(provider_data={"response_id": provider_id})], + ) + + try: + set_global_report_state(state) + for tokens, provider_id, charge in [(400, "gen-1", 0.25), (700, "gen-2", 0.5)]: + # The provider callback runs before the SDK hook, as it does in a + # real stream; the order must not matter. + litellm_cost_callback( + {"response_cost": charge}, + SimpleNamespace(id=provider_id, usage=SimpleNamespace(cost=charge)), + ) + await hooks.on_llm_end(context, agent, sdk_response(tokens, provider_id)) # type: ignore[arg-type] + # The SDK hook runs first for this one. + await hooks.on_llm_end(context, agent, sdk_response(600, "gen-3")) # type: ignore[arg-type] + litellm_cost_callback( + {"response_cost": 0.125}, + SimpleNamespace(id="gen-3", usage=SimpleNamespace(cost=0.125)), + ) + finally: + set_global_report_state(None) + + # Each charge replaced its own estimate: 0.25 + 0.5 + 0.125, no estimate on top. + assert state.get_total_llm_cost() == pytest.approx(0.875) + + +def test_mixed_provider_prices_keep_estimate_for_unpriced_response( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + usage_module, + "_estimate_litellm_cost", + lambda used, _model: used.input_tokens / 1000, + ) + ledger = LLMUsageLedger() + ledger.record(agent_id="a", usage=_usage(400), model="model", response_id="charged") + ledger.record(agent_id="a", usage=_usage(700), model="model", response_id="unpriced") + ledger.record_observed_cost(0.2, response_id="charged") + assert ledger.total_cost == pytest.approx(0.9) + + # A callback can arrive before the SDK usage hook and must still replace + # only its own estimate. A provider-only call is included as well. + ledger.record_observed_cost(0.3, response_id="early") + ledger.record(agent_id="a", usage=_usage(600), model="model", response_id="early") + ledger.record_observed_cost(0.1, response_id="provider-only") + assert ledger.total_cost == pytest.approx(1.3) + + restored = LLMUsageLedger() + restored.hydrate(ledger.to_record()) + assert restored.total_cost == pytest.approx(1.3) + restored.record(agent_id="a", usage=_usage(200), model="model", response_id="new") + assert restored.total_cost == pytest.approx(1.5) + + +def test_sdk_usage_correlates_request_ids_per_agent_and_keeps_upstream_contract_when_ambiguous( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + monkeypatch.setattr( + usage_module, + "_estimate_litellm_cost", + lambda used, _model: used.input_tokens / 1000, + ) + state = ReportState(run_name="pricing") + + # A failed provider attempt is never queued. The retry's success is paired + # with its own SDK usage, even as another agent completes in between. + state.note_llm_response("root", "retry-ok") + state.note_llm_response("child", "child-ok") + state.record_observed_llm_cost(0.12, response_id="child-ok") + state.record_sdk_usage(agent_id="child", usage=_usage(500), model="model") + state.record_observed_llm_cost(0.25, response_id="retry-ok") + state.record_sdk_usage(agent_id="root", usage=_usage(400), model="model") + assert state.get_total_llm_cost() == pytest.approx(0.37) + + # A later root call cannot consume the previous one's id. + state.note_llm_response("root", "next") + state.record_sdk_usage(agent_id="root", usage=_usage(300), model="model") + assert state.get_total_llm_cost() == pytest.approx(0.67) + + # Two successes before a single SDK hook are ambiguous (for example, + # concurrent calls or a response conversion error), so the estimate stays + # unkeyed. Both charges count and replace it, as upstream does: adding the + # estimate on top would count the same calls twice and stop a scan early. + state.note_llm_response("root", "ambiguous-a") + state.note_llm_response("root", "ambiguous-b") + state.record_observed_llm_cost(0.1, response_id="ambiguous-a") + state.record_observed_llm_cost(0.2, response_id="ambiguous-b") + state.record_sdk_usage(agent_id="root", usage=_usage(500), model="model") + assert state.get_total_llm_cost() == pytest.approx(0.97) diff --git a/tests/test_cost_tracking.py b/tests/test_cost_tracking.py index 4f12bc0ea..7bc742b8c 100644 --- a/tests/test_cost_tracking.py +++ b/tests/test_cost_tracking.py @@ -208,7 +208,7 @@ def test_cost_callback_recovers_streamed_openrouter_cost_by_response_id() -> Non ): litellm_cost_callback({"response_cost": None, "model": "moonshotai/kimi-k3"}, response) - report_state.record_observed_llm_cost.assert_called_once_with(0.42) + report_state.record_observed_llm_cost.assert_called_once_with(0.42, response_id="gen-abc") # The entry is consumed so a later response cannot double-count it. assert streamed_openrouter_costs.take(response) is None @@ -228,7 +228,7 @@ def test_streamed_openrouter_cost_prefers_provider_report_over_estimate() -> Non ): litellm_cost_callback({"response_cost": None, "model": "moonshotai/kimi-k3"}, response) - report_state.record_observed_llm_cost.assert_called_once_with(0.9) + report_state.record_observed_llm_cost.assert_called_once_with(0.9, response_id="gen-xyz") estimate.assert_not_called() diff --git a/tests/test_pricing.py b/tests/test_pricing.py index ee1130e66..842fd36d9 100644 --- a/tests/test_pricing.py +++ b/tests/test_pricing.py @@ -217,6 +217,80 @@ def test_hydrated_estimate_continues_accumulating_new_estimates() -> None: assert ledger.total_cost == 0.59 +def _single_call(tokens: int) -> Usage: + return Usage(requests=1, input_tokens=tokens, output_tokens=0, total_tokens=tokens) + + +def _estimates_per_thousand_tokens(): + """Deterministic estimates: 400 tokens cost 0.4, 700 tokens cost 0.7.""" + return patch( + "strix.report.usage._estimate_litellm_cost", + side_effect=lambda usage, _model: usage.input_tokens / 1000, + ) + + +def test_ledger_pairs_observed_and_estimate_by_response_id() -> None: + ledger = LLMUsageLedger() + with _estimates_per_thousand_tokens(): + ledger.record(agent_id="a", usage=_single_call(400), model="m", response_id="charged") + ledger.record(agent_id="a", usage=_single_call(700), model="m", response_id="unpriced") + ledger.record_observed_cost(0.2, response_id="charged") + + # The charge replaces its own response's estimate, and the response with no + # charge keeps its estimate. + assert ledger.total_cost == pytest.approx(0.9) + + # Arrival order does not matter: a charge can land before the usage hook. + ledger.record_observed_cost(0.3, response_id="early") + with _estimates_per_thousand_tokens(): + ledger.record(agent_id="a", usage=_single_call(600), model="m", response_id="early") + assert ledger.total_cost == pytest.approx(1.2) + + +def test_ledger_keyed_charge_replaces_unkeyed_estimate_when_ids_cannot_pair() -> None: + # The provider callback named the response, the SDK hook had no id for it. + # They may be the same response, so the upstream contract applies: the + # charge replaces the estimate instead of adding to it. + ledger = LLMUsageLedger() + with _estimates_per_thousand_tokens(): + ledger.record(agent_id="a", usage=_single_call(500), model="m") + ledger.record_observed_cost(0.17, response_id="gen-1") + + assert ledger.total_cost == pytest.approx(0.17) + + +def test_ledger_keeps_unkeyed_estimate_until_a_charge_is_left_unmatched() -> None: + ledger = LLMUsageLedger() + with _estimates_per_thousand_tokens(): + ledger.record(agent_id="a", usage=_single_call(400), model="m", response_id="paired") + ledger.record(agent_id="a", usage=_single_call(500), model="m") + ledger.record_observed_cost(0.2, response_id="paired") + + # Every charge is matched by id, so nothing replaces the id-less estimate. + assert ledger.total_cost == pytest.approx(0.7) + + ledger.record_observed_cost(0.3) + + assert ledger.total_cost == pytest.approx(0.5) + + +def test_hydrated_cost_is_kept_once_when_a_charge_replaces_new_estimates() -> None: + ledger = LLMUsageLedger() + ledger.hydrate({"cost": 0.42}) + with _estimates_per_thousand_tokens(): + ledger.record(agent_id="a", usage=_single_call(1000), model="m") + ledger.record_observed_cost(0.17) + + # 0.42 persisted + 0.17 charged, not + the 1.0 estimate. + assert ledger.total_cost == pytest.approx(0.59) + + with _estimates_per_thousand_tokens(): + ledger.record(agent_id="a", usage=_single_call(300), model="m", response_id="after-resume") + ledger.record_observed_cost(0.05, response_id="after-resume") + + assert ledger.total_cost == pytest.approx(0.64) + + def test_zero_cost_disables_both_observed_and_estimated_costs() -> None: usage = Usage() usage.requests = 1