fix(usage): carry cumulative cache tokens in result.usage

eval/harbor/clawcodex_agent.py

@@ -686,8 +686,13 @@ def populate_context_post_run(self, context: AgentContext) -> None:
 
         # Leaderboard token/cost columns. Prefer the session's authoritative
         # BILLING totals (input + cache, summed per turn — matches
-        # claude-code); fall back to the stream-json usage (incomplete: no
-        # cumulative cache) only when no session cost block was synced.
+        # claude-code); fall back to the stream-json usage only when no
+        # session cost block was synced. That fallback now carries cumulative
+        # cache counters, so the sum below is no longer short by the cached
+        # portion — but it is still not COMPLETE: result.usage only ever sees
+        # the main loop's assistant messages, so subagent and compaction
+        # tokens never reach it. The cost block includes them, which is one
+        # reason it stays preferred.
         if totals is not None:
             context.n_input_tokens = totals["prompt"]
             context.n_cache_tokens = totals["cached"]
@@ -748,8 +753,10 @@ def _final_metrics(
         totals: dict[str, Any] | None,
     ) -> FinalMetrics | None:
         """``totals`` = authoritative session billing totals (preferred).
-        Falls back to the stream-json usage (incomplete — no cumulative
-        cache) only when no session cost block was synced."""
+        Falls back to the stream-json usage when no session cost block was
+        synced; that fallback now carries cumulative cache counters, so it is
+        no longer short by the cached portion — though it still omits
+        subagent and compaction tokens, which only the cost block sees."""
         usage = (result_event or {}).get("usage")
         if not isinstance(usage, dict):
             usage = {}
@@ -814,15 +821,21 @@ def _billing_totals_from_cost_block(
     ) -> dict[str, Any] | None:
         """Authoritative BILLING token totals from a session ``cost`` block.
 
-        clawcodex's stream-json ``result.usage`` is built for live-CONTEXT
-        measurement, not accounting: ``input_tokens`` is the running sum of
-        NON-cached input and it drops cumulative cache tokens (only a
-        ``last_*`` snapshot survives). For a heavily prompt-cached opus run
-        that under-reports total prompt tokens by orders of magnitude
-        (observed: 6 vs the real ~412 K). The session cost block's
-        ``model_usage`` accumulates the real per-turn billing counters
-        (input + cache_read + cache_creation, summed across turns), which is
-        exactly the convention the built-in claude-code agent reports.
+        Historically clawcodex's stream-json ``result.usage`` summed only
+        input+output, and ``input_tokens`` is the running sum of NON-cached
+        input, so cumulative cache tokens were dropped entirely (only a
+        ``last_*`` snapshot survived). For a heavily prompt-cached opus run
+        that under-reported total prompt tokens by orders of magnitude —
+        observed here as 6 against the real ~412 K, which is why this cost
+        block path exists. ``result.usage`` now carries the cumulative cache
+        counters, so the fallback is sound; this remains PREFERRED because
+        the cost block also supplies the real dollar cost, and because a run
+        against an older clawcodex build still needs it.
+
+        The session cost block's ``model_usage`` accumulates the real
+        per-turn billing counters (input + cache_read + cache_creation,
+        summed across turns), which is exactly the convention the built-in
+        claude-code agent reports.
         Returns ``{prompt, completion, cached, cost}`` or ``None``.
         """
         model_usage = cost.get("model_usage")

src/entrypoints/headless.py

@@ -64,6 +64,49 @@
 from src.utils.abort_controller import AbortController, AbortError
 
 
+def _accumulate_usage(total: dict[str, int], usage: dict[str, int]) -> None:
+    """Fold one prompt's usage into the running totals, in place.
+
+    Two different kinds of key live in the same dict and must not be treated
+    alike. The cumulative counters (``input_tokens``, ``output_tokens`` and
+    the two cache counters) ADD across prompts — together they are the
+    billing total. The ``last_*`` keys are a snapshot of the most recent
+    response, used as the live-context measure, so they REPLACE: adding a
+    snapshot to itself across prompts yields a number that measures nothing,
+    and it ships in the stream-json ``result`` payload.
+    """
+    for key, value in usage.items():
+        # Provider usage dicts carry non-numeric entries — MinimaxProvider
+        # puts a ``service_tier`` string in its. The aggregator upstream
+        # whitelists keys today, so none arrive; skipping rather than
+        # raising means the natural next edit there ("stop dropping keys")
+        # cannot take down a headless run with ``int("standard")``.
+        try:
+            numeric = int(value)
+        except (TypeError, ValueError):
+            continue
+        if key.startswith("last_"):
+            total[key] = numeric
+        else:
+            total[key] = int(total.get(key, 0) or 0) + numeric
+
+
+#: The cumulative counters that together make up what was actually billed.
+#: ``input_tokens`` is only the NON-cached part of each prompt, so any total
+#: built from input+output alone silently omits every cached token.
+_BILLED_TOKEN_KEYS = (
+    "input_tokens",
+    "output_tokens",
+    "cache_read_input_tokens",
+    "cache_creation_input_tokens",
+)
+
+
+def _billed_token_total(usage_total: dict[str, int]) -> int:
+    """Every billed token across the run — what ``/goal`` spends against."""
+    return int(sum(int(usage_total.get(key, 0) or 0) for key in _BILLED_TOKEN_KEYS))
+
+
 def _json_result_subtype(
     exit_code: int, early_stop_subtype: str | None
 ) -> tuple[str, bool]:
@@ -824,8 +867,7 @@ def _persist(msg: Any) -> None:
 
                 num_turns_total += result.num_turns
                 if result.usage:
-                    for key, value in result.usage.items():
-                        usage_total[key] = usage_total.get(key, 0) + int(value)
+                    _accumulate_usage(usage_total, result.usage)
 
                 if writer is not None:
                     writer.write(AssistantEvent(text=result.response_text))
@@ -841,10 +883,7 @@ def _persist(msg: Any) -> None:
                     evidence = collect_turn_evidence(
                         list(session.conversation.messages)
                     ) or (result.response_text or "")
-                    tokens_now = int(
-                        usage_total.get("input_tokens", 0)
-                        + usage_total.get("output_tokens", 0)
-                    )
+                    tokens_now = _billed_token_total(usage_total)
                     decision = goal_mgr.evaluate_after_turn(
                         evidence, tokens_now=tokens_now,
                     )

src/query/agent_loop_compat.py

@@ -664,7 +664,12 @@ async def run_query_as_agent_loop(
 
     holder = TerminalHolder()
     response_text_parts: list[str] = []
-    usage: dict[str, int] = {"input_tokens": 0, "output_tokens": 0}
+    usage: dict[str, int] = {
+        "input_tokens": 0,
+        "output_tokens": 0,
+        "cache_read_input_tokens": 0,
+        "cache_creation_input_tokens": 0,
+    }
     num_turns = 0
     last_assistant_text = ""
     last_api_error_text = ""
@@ -717,16 +722,36 @@ async def run_query_as_agent_loop(
             num_turns += 1
             # Sum usage across turns.
             mu = getattr(msg, "usage", None) or {}
-            usage["input_tokens"] += mu.get("input_tokens", 0)
-            usage["output_tokens"] += mu.get("output_tokens", 0)
+            usage["input_tokens"] += mu.get("input_tokens", 0) or 0
+            usage["output_tokens"] += mu.get("output_tokens", 0) or 0
+            # The cache counters are summed too, because ``input_tokens`` is
+            # only the NON-cached part of each prompt — that is the shared
+            # convention across providers (Anthropic reports it natively;
+            # OpenAI-compatible providers since the cache-read split). Summing
+            # input+output alone therefore drops every cached token from the
+            # total, and ``agent_server`` feeds this very dict to
+            # ``compute_cost``, which then bills the cached portion at nothing:
+            # a real 2613-token turn with a 2560-token hit lost 71.7% of its
+            # cost. ``eval/harbor/clawcodex_agent.py`` measured the same defect
+            # from the other end — 6 tokens reported against a real ~412K on a
+            # prompt-cached opus run — and works around it by reading the
+            # session cost block instead.
+            #
+            # Purely additive: ``input_tokens`` keeps its meaning, so existing
+            # readers are untouched and only gain the ability to see the rest.
+            usage["cache_read_input_tokens"] += mu.get("cache_read_input_tokens", 0) or 0
+            usage["cache_creation_input_tokens"] += (
+                mu.get("cache_creation_input_tokens", 0) or 0
+            )
             # C3a: also keep the LAST response's FULL usage (all four
             # keys, last-wins — TS getCurrentUsage, utils/tokens.ts:
-            # 152-171). The cumulative sum above double-counts context
-            # across a multi-tool-call run and drops the cache keys, so
-            # it must NOT be used as a live-context measure
-            # (tokens.ts:407-420 warning). Last-wins also covers the
-            # parallel-tool-call case where N assistant records share
-            # one usage object.
+            # 152-171). The cumulative sum above is a BILLING total and
+            # still must NOT be used as a live-context measure: it
+            # double-counts context across a multi-tool-call run, because
+            # each turn re-sends the whole conversation (tokens.ts:407-420
+            # warning). Last-wins is the context measure, and it also covers
+            # the parallel-tool-call case where N assistant records share one
+            # usage object.
             if mu:
                 usage["last_input_tokens"] = int(mu.get("input_tokens", 0) or 0)
                 usage["last_output_tokens"] = int(mu.get("output_tokens", 0) or 0)

src/server/agent_server.py

@@ -4782,6 +4782,12 @@ def on_message(message: Any) -> None:
             # subagents spawn from the Agent tool's captured FULL registry.
             from src.coordinator.mode import coordinator_main_loop_registry
 
+            # Cost is read as a DELTA of the tracker's running total rather
+            # than recomputed from ``result.usage`` below — see the note at
+            # the ``_cost`` assignment for why the aggregate cannot be priced.
+            from src.bootstrap.state import get_total_cost_usd
+
+            _cost_before = get_total_cost_usd()
             result = asyncio.run(run_query_as_agent_loop(
                 initial_messages=list(self.session.conversation.messages),
                 provider=turn_provider,
@@ -4848,14 +4854,23 @@ def on_message(message: Any) -> None:
                 msgs.extend(_btw_snapshot)
 
         _usage = result.usage if result.num_turns > 0 else None
+        # The turn's cost is the tracker's delta, NOT ``compute_cost`` over
+        # ``result.usage``. That dict is the sum across every loop turn, and
+        # ``get_pricing`` selects a tier from a PER-REQUEST threshold
+        # (gpt-5.6-luna at 272K, MiniMax-M3 at 512K). Cache reads sum to
+        # roughly turns x conversation-size, so a long loop of small requests
+        # crosses a boundary no single request came near: a 25-turn loop over
+        # a 15K conversation prices at 1.76x the true cost.
+        #
+        # ``cost_tracker.record_api_usage`` already prices each response as it
+        # arrives, with the tier chosen from that one request, so the running
+        # total is correct by construction. Reading its delta also drops a
+        # duplicate cost implementation rather than patching one.
         _cost = 0.0
-        if _usage:
-            try:
-                from src.services.pricing import compute_cost
-
-                _cost = compute_cost(getattr(self.provider, "model", None) or self.config.model or "", _usage)
-            except Exception:  # noqa: BLE001 — cost is best-effort, never break the turn
-                _cost = 0.0
+        try:
+            _cost = max(0.0, get_total_cost_usd() - _cost_before)
+        except Exception:  # noqa: BLE001 — cost is best-effort, never break the turn
+            _cost = 0.0
         # One more completed user turn. Internal (notification) and btw
         # (ephemeral, rolled-back) turns don't move the odometer — same rule
         # as the deleted REPL, which only counted real prompt→response rounds.

tests/server/test_agent_server_workflows.py

@@ -921,3 +921,93 @@ def _stop(*a, **k):
         {"subtype": "error_during_execution", "response_text": "[Stopped: x]"},
     )
     assert not sess2._inbox.empty(), "control failed — nothing ever enqueues"
+
+
+class _TieredCacheProvider:
+    """Many small requests whose cache reads only cross a tier once summed.
+
+    Each request is ~15.4K prompt tokens — far below gpt-5.6-luna's 272K
+    long-context threshold — but the loop's cumulative total reaches 385K.
+    """
+
+    PER_REQUEST_USAGE = {
+        "input_tokens": 400,
+        "output_tokens": 300,
+        "cache_read_input_tokens": 15_000,
+        "cache_creation_input_tokens": 0,
+    }
+    TURNS = 25
+
+    def __init__(self, api_key=None, base_url=None, model=None):
+        self.model = "openai/gpt-5.6-luna"
+        self._calls = 0
+
+    def chat(self, messages, tools=None, **kw):
+        from src.providers.base import ChatResponse
+
+        self._calls += 1
+        last = self._calls >= self.TURNS
+        return ChatResponse(
+            content="done" if last else "thinking",
+            model=self.model,
+            usage=dict(self.PER_REQUEST_USAGE),
+            finish_reason="end_turn" if last else "tool_use",
+            tool_uses=None
+            if last
+            else [
+                {
+                    "id": f"tool_{self._calls}",
+                    "name": "Bash",
+                    "input": {"command": "true", "description": "noop"},
+                }
+            ],
+        )
+
+    def chat_stream_response(self, *a, **kw):  # force fallback to chat()
+        raise NotImplementedError
+
+
+async def test_turn_cost_is_not_priced_from_the_cumulative_usage(tmp_path):
+    """A long loop must not be billed at the long-context rate.
+
+    ``get_pricing`` picks a tier from a PER-REQUEST threshold, but
+    ``result.usage`` is the sum across every loop turn. Once the cumulative
+    dict carried cache reads — which sum to roughly turns x conversation
+    size — pricing it crossed a boundary no single request came near, and
+    this turn billed at 1.76x its true cost.
+
+    The cost is now a delta of ``cost_tracker``'s running total, which prices
+    each response as it arrives with the tier chosen from that one request.
+
+    BEHAVIOURAL: asserted on the emitted ``total_cost_usd``, because the
+    failure mode is a plausible-looking number rather than an error.
+    """
+    from src.services.pricing import compute_cost
+
+    per_request = _TieredCacheProvider.PER_REQUEST_USAGE
+    turns = _TieredCacheProvider.TURNS
+    model = "openai/gpt-5.6-luna"
+
+    truth = sum(compute_cost(model, per_request) for _ in range(turns))
+    aggregate = {k: v * turns for k, v in per_request.items()}
+    if_priced_as_aggregate = compute_cost(model, aggregate)
+    assert if_priced_as_aggregate > truth * 1.5, "fixture must cross the tier"
+
+    async with _spawned(tmp_path, _TieredCacheProvider) as (handle, gen):
+        await handle.send_to_agent(
+            {"type": "user", "message": {"role": "user", "content": "go"}}
+        )
+        result = None
+        for _ in range(200):
+            msg = await asyncio.wait_for(gen.__anext__(), timeout=20)
+            if msg.get("type") == "result":
+                result = msg
+                break
+
+    assert result is not None, "no result frame emitted"
+    reported = float(result.get("total_cost_usd") or 0.0)
+    assert reported > 0, "the turn reported no cost at all"
+    assert reported < if_priced_as_aggregate * 0.8, (
+        f"turn billed at the long-context rate: {reported} vs aggregate "
+        f"{if_priced_as_aggregate} (true {truth})"
+    )