Merge 5cfec1fc19 into 851743adcf

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):

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)

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:
@@ -384,18 +385,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(
@@ -403,6 +408,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:

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")
 

strix/report/usage.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import logging
+import math
+import threading
 from typing import Any
 
 from agents.usage import Usage, deserialize_usage, serialize_usage
@@ -67,7 +69,23 @@ 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] = {}
+        # Running totals over the two maps, kept in step on every write so
+        # reading ``total_cost`` (each save, each budget check) is O(1):
+        # the cost of id-matched responses, and charges with no estimate.
+        self._paired_cost = 0.0
+        self._unpaired_observed_by_id = 0.0
+        # Estimates arrive from the event loop, charges from provider
+        # callbacks; one lock keeps a map and its running total in step.
+        self._response_lock = threading.Lock()
+        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 +101,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 +118,50 @@ 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._set_response_cost(response_id, estimated=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._set_response_cost(response_id, observed=float(cost))
+            else:
+                self._observed_cost += float(cost)
+
+    def _response_contribution(self, response_id: str) -> tuple[float, float]:
+        """This response's share of (paired cost, unpaired keyed charges)."""
+        observed = self._observed_by_response.get(response_id)
+        estimated = self._estimated_by_response.get(response_id)
+        if estimated is not None:
+            return (estimated if observed is None else observed), 0.0
+        return 0.0, (observed or 0.0)
+
+    def _set_response_cost(
+        self,
+        response_id: str,
+        *,
+        estimated: float | None = None,
+        observed: float | None = None,
+    ) -> None:
+        with self._response_lock:
+            old_paired, old_unpaired = self._response_contribution(response_id)
+            if estimated is not None:
+                self._estimated_by_response[response_id] = estimated
+            if observed is not None:
+                self._observed_by_response[response_id] = observed
+            new_paired, new_unpaired = self._response_contribution(response_id)
+            self._paired_cost += new_paired - old_paired
+            self._unpaired_observed_by_id += new_unpaired - old_unpaired
 
     def record_provider(
         self,
@@ -150,7 +199,18 @@ 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)
+        # Matched by response id: the provider's charge replaces that response's
+        # estimate, and a response without a charge keeps its own estimate.
+        # 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.
+        with self._response_lock:
+            paired = self._paired_cost
+            unpaired_observed = self._observed_cost + self._unpaired_observed_by_id
+        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 +258,11 @@ 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._paired_cost = 0.0
+        self._unpaired_observed_by_id = 0.0
+        self._persisted_cost = 0.0
         self._providers = {}
 
         if not isinstance(raw_usage, dict):
@@ -216,8 +280,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):

tests/test_cost_response_pairing.py

@@ -0,0 +1,187 @@
+"""Provider charges pair with their own response's estimate by response id."""
+
+from __future__ import annotations
+
+import random
+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)
+
+
+def test_running_totals_match_a_full_recount(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        usage_module,
+        "_estimate_litellm_cost",
+        lambda used, _model: used.input_tokens / 1000,
+    )
+    rng = random.Random(1642)  # noqa: S311 - a seeded, repeatable test sequence
+    ledger = LLMUsageLedger()
+    estimated: dict[str, float] = {}
+    observed: dict[str, float] = {}
+    for _ in range(500):
+        response_id = f"r{rng.randrange(60)}"
+        if rng.random() < 0.5:
+            tokens = rng.randrange(1, 900)
+            ledger.record(agent_id="a", usage=_usage(tokens), model="m", response_id=response_id)
+            estimated[response_id] = tokens / 1000
+        else:
+            cost = rng.randrange(1, 900) / 1000
+            ledger.record_observed_cost(cost, response_id=response_id)
+            observed[response_id] = cost
+        paired = sum(observed.get(rid, est) for rid, est in estimated.items())
+        unpaired = sum(cost for rid, cost in observed.items() if rid not in estimated)
+        assert ledger.total_cost == pytest.approx(paired + unpaired, abs=1e-9)

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()
 
 

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