fix(ccr): fold dropped continuation rounds' usage into cost_with_headroom

handle_response returns only the final continuation round, so the dropped
rounds' usage was captured into /stats ccr_overhead but excluded from
cost_with_headroom. The cost.py comment claiming it was 'already inside
cost_with_headroom' did not hold: handle_response returns only current_response,
the handler adopts it as resp_json, and the outcome funnel reads usage from
that single final-round response. The intermediate rounds go through a raw
api_call_fn and skip the funnel.

Fold the dropped rounds' usage into the billed totals, cache-split per
provider, reusing the proactive-expansion ContextVar pattern already in this
PR. Cache-split matters: Anthropic input_tokens is the uncached portion, but
OpenAI prompt_tokens is gross (includes cached_tokens), so collapsing to one
input number would price cached continuation tokens at full list instead of
the discounted rate.

- response_handler.handle_response: sum each dropped round's usage into locals
  as (uncached_input, cache_read, cache_write, output) via the new
  _extract_round_cost_usage (provider-aware; mirrors the per-handler parsing),
  and publish via set_pending_ccr_continuation_usage before returning. Locals,
  not the shared instance counters, so concurrent requests on one handler do
  not mix. The instance accumulators stay for the /stats attribution view.
  Signature and all call sites unchanged.
- outcome.emit_request_outcome: consume it and add each bucket to the field
  cost_with_headroom prices (uncached_input_tokens at list, cache_read_tokens
  discounted, cache_write_tokens premium; output_tokens to the budget/metrics
  total, since cost_with_headroom itself is input-only). Client-facing
  response.usage is untouched. The >=500 short-circuit still books nothing.
- cost.py: fix the comment.

OpenAI cache_write inference (the _infer_openai_cache_write_tokens heuristic)
is not replicated in _extract_round_cost_usage; only explicitly reported
cache_creation is credited. The batch path (multiple handle_response calls,
one emit) is a known follow-up.

Verified: 148 passed across CCR / proactive-expansion / stats / outcome, 0
regressions.

Author: EvolveAegis

headroom/ccr/response_handler.py

@@ -459,6 +459,19 @@ async def handle_response(
         current_response = response
         current_messages = list(messages)  # Copy to avoid mutation
         rounds = 0
+        # Per-call sum of the dropped rounds' usage, split into the cost buckets
+        # the outcome funnel prices (uncached input / cache_read / cache_write /
+        # output). Kept in locals (NOT the shared instance counters at 542-543)
+        # so concurrent requests on one handler can't mix their overhead; the
+        # instance counters stay for the /stats ccr_overhead attribution view.
+        # Cache-split (not just gross input) matters: OpenAI prompt_tokens and
+        # Anthropic input_tokens have different cache semantics, and pricing
+        # cached tokens at the discounted/premium rate vs full-list uncached is
+        # the difference between an accurate fold and an over-count.
+        call_overhead_uncached_in = 0
+        call_overhead_cache_read = 0
+        call_overhead_cache_write = 0
+        call_overhead_output = 0
 
         while rounds < self.config.max_retrieval_rounds:
             # Check for CCR tool calls
@@ -541,6 +554,11 @@ async def handle_response(
                 self._retrieved_tokens += sum(r.tokens_retrieved for r in results)
                 self._ccr_overhead_input_tokens += _in_tok
                 self._ccr_overhead_output_tokens += _out_tok
+                _uin, _cr, _cw, _uout = self._extract_round_cost_usage(current_response, provider)
+                call_overhead_uncached_in += _uin
+                call_overhead_cache_read += _cr
+                call_overhead_cache_write += _cw
+                call_overhead_output += _uout
 
             # Durable ledger: record this round's proxy-side retrieval so
             # `headroom savings` reflects net (not just gross) for proxy
@@ -555,6 +573,28 @@ async def handle_response(
                 f"returning response with possible unhandled CCR calls"
             )
 
+        # Publish this call's dropped-round usage (cache-split) to the outcome
+        # funnel (per-asyncio-task ContextVar). The funnel folds each bucket
+        # into the matching cost field so cost_with_headroom prices continuation
+        # cache at the right rate. Guarded so a no-continuation call (rounds ==
+        # 0) publishes nothing.
+        if (
+            call_overhead_uncached_in
+            or call_overhead_cache_read
+            or call_overhead_cache_write
+            or call_overhead_output
+        ):
+            from headroom.proxy.outcome import set_pending_ccr_continuation_usage
+
+            set_pending_ccr_continuation_usage(
+                (
+                    call_overhead_uncached_in,
+                    call_overhead_cache_read,
+                    call_overhead_cache_write,
+                    call_overhead_output,
+                )
+            )
+
         return current_response
 
     def _record_retrieval_savings(self, results: list[CCRToolResult]) -> None:
@@ -616,6 +656,59 @@ def _extract_usage_tokens(resp: Any) -> tuple[int, int]:
         except (TypeError, ValueError):
             return 0, 0
 
+    @staticmethod
+    def _extract_round_cost_usage(resp: Any, provider: str) -> tuple[int, int, int, int]:
+        """Return (uncached_input, cache_read, cache_write, output) for one
+        continuation round, split into the cost buckets the outcome funnel
+        prices.
+
+        Provider-aware so cached tokens land in cache_read / cache_write
+        (priced discounted / premium by cost_with_headroom) instead of being
+        collapsed into uncached input at full list price. Mirrors the per-handler
+        usage parsing (handlers/anthropic.py, handlers/openai.py) for the dropped
+        continuation rounds, which those handlers only parse for the final round.
+
+        OpenAI cache_write *inference* (_infer_openai_cache_write_tokens) is
+        deliberately NOT replicated here: it is a heuristic, and for the cost
+        fold we only credit cache_write that the provider reported explicitly
+        (cache_creation_input_tokens). Cached-token reads prefer the
+        Anthropic/Bedrock top-level keys when present (authoritative), then fall
+        back to the OpenAI prompt_tokens_details / input_tokens_details shapes.
+        """
+        if not isinstance(resp, dict):
+            return 0, 0, 0, 0
+        usage = resp.get("usage") or {}
+        if not isinstance(usage, dict):
+            return 0, 0, 0, 0
+
+        def _i(value: Any, default: int = 0) -> int:
+            try:
+                return max(int(value), 0)
+            except (TypeError, ValueError):
+                return default
+
+        if provider in ("openai", "openai_responses"):
+            gross = _i(usage.get("prompt_tokens", usage.get("input_tokens", 0)))
+            out = _i(usage.get("completion_tokens", usage.get("output_tokens", 0)))
+            cr = _i(usage.get("cache_read_input_tokens", 0))
+            cw = _i(usage.get("cache_creation_input_tokens", 0))
+            if cr == 0:
+                details = (
+                    usage.get("prompt_tokens_details") or usage.get("input_tokens_details") or {}
+                )
+                if isinstance(details, dict):
+                    cr = _i(details.get("cached_tokens", 0))
+            uncached = max(0, gross - cr - cw)
+            return uncached, cr, cw, out
+
+        # anthropic (and default): input_tokens is already the uncached portion.
+        return (
+            _i(usage.get("input_tokens", 0)),
+            _i(usage.get("cache_read_input_tokens", 0)),
+            _i(usage.get("cache_creation_input_tokens", 0)),
+            _i(usage.get("output_tokens", 0)),
+        )
+
     def get_stats(self) -> dict[str, Any]:
         """Get handler statistics."""
         with self._retrieval_count_lock:

headroom/proxy/cost.py

@@ -670,9 +670,16 @@ def _entry_number(entry: Any, attr: str) -> int | float:
     summary["compression"]["retrievals_total"] = retrievals_total
     summary["compression"]["net_tokens_saved"] = net_tokens_saved
 
-    # Cost view = ATTRIBUTION ONLY, never debited: the CCR continuation's usage
-    # is already inside cost_with_headroom (handlers/anthropic.py reads usage
-    # AFTER the continuation replaced resp_json), so debiting here double-counts.
+    # Attribution view only. The dropped CCR continuation rounds' usage is now
+    # folded into the billed totals at the outcome funnel
+    # (set_pending_ccr_continuation_usage -> uncached_input_tokens /
+    # cache_read_tokens / cache_write_tokens, which cost_with_headroom prices, and
+    # output_tokens for the budget/metrics total). So the dropped rounds ARE
+    # inside cost_with_headroom (their input side). This block surfaces them
+    # again purely as a breakdown; do NOT add ccr_overhead into a second cost
+    # total, that double-counts. (The earlier "already inside cost_with_headroom"
+    # claim was inaccurate: handle_response returns only the final round, so
+    # without the funnel fold the dropped rounds were never costed at all.)
     summary["cost"]["retrieval_cost_usd"] = round(
         estimate_cost_usd(primary_model, tokens_retrieved + overhead_tokens), 6
     )

headroom/proxy/outcome.py

@@ -56,6 +56,35 @@ def consume_pending_proactive_retrieval() -> tuple[int, int] | None:
     return value
 
 
+# CCR reactive-continuation overhead (the dropped rounds' real billed usage),
+# bound per-request via a ContextVar for the same reason as the proactive one:
+# the outcome funnel can fold it into the billed token totals WITHOUT threading
+# a parameter through every handler. handle_response publishes the per-call sum
+# (kept in locals, so it is safe across concurrent requests on a shared handler);
+# consumed only in emit's success section, so a failed forward books nothing.
+# The payload is cache-split (uncached_input, cache_read, cache_write, output) so
+# cost_with_headroom prices continuation cache at the right rate rather than
+# folding cached tokens into uncached input at full list price.
+_pending_ccr_continuation_usage: ContextVar[tuple[int, int, int, int] | None] = ContextVar(
+    "headroom_ccr_continuation_pending", default=None
+)
+
+
+def set_pending_ccr_continuation_usage(
+    value: tuple[int, int, int, int] | None,
+) -> None:
+    """Bind the dropped continuation rounds' (uncached_in, cache_read, cache_write, output) to this request."""
+    _pending_ccr_continuation_usage.set(value)
+
+
+def consume_pending_ccr_continuation_usage() -> tuple[int, int, int, int] | None:
+    """Read and clear the pending continuation usage for this request."""
+    value = _pending_ccr_continuation_usage.get()
+    if value is not None:
+        _pending_ccr_continuation_usage.set(None)
+    return value
+
+
 @dataclass(frozen=True)
 class RequestOutcome:
     """Immutable, value-equal snapshot of a completed request.
@@ -362,6 +391,8 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
     and is awaitable-compatible. We could lift this to a typing.Protocol
     if/when another contract surface emerges, but YAGNI.
     """
+    import dataclasses
+
     from headroom.copilot_auth import consume_request_routed_to_copilot
     from headroom.proxy.cost import _summarize_transforms
     from headroom.proxy.models import RequestLog
@@ -385,6 +416,12 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
     # >=500 short-circuit must never book a retrieval that was never billed.
     _pending_proactive = consume_pending_proactive_retrieval()
 
+    # Consume (read + clear) the dropped CCR continuation rounds' usage for this
+    # request. Folded into the billed totals only in the success section below —
+    # a >=500 short-circuit must never book continuation tokens that, by the
+    # account-after-success rule in handle_response, were never billed.
+    _pending_continuation = consume_pending_ccr_continuation_usage()
+
     # Upstream failure (>= 500, e.g. a 529 Overloaded surfaced after retry
     # exhaustion) must not feed the savings/cost/log success stats; that would
     # let a failed request inflate the save-rate. Record it as failed and stop,
@@ -403,6 +440,25 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
         if _proactive_rec is not None:
             _proactive_rec(*_pending_proactive)
 
+    # Fold the dropped continuation rounds' real billed usage into the cost-
+    # relevant token totals. handle_response returns only the final round, so
+    # without this the intermediate rounds' usage never reaches cost_with_headroom
+    # (it sat in the /stats ccr_overhead side-channel but was deliberately
+    # excluded from the total). Each bucket lands in the field cost_with_headroom
+    # prices: uncached input at list, cache_read at the discounted rate,
+    # cache_write at the premium rate, and output into the budget/metrics total
+    # (cost_with_headroom itself is input-only). Client-facing response.usage is
+    # unchanged; only the internal billed totals move.
+    if _pending_continuation is not None:
+        _ct_uin, _ct_cr, _ct_cw, _ct_out = _pending_continuation
+        outcome = dataclasses.replace(
+            outcome,
+            output_tokens=outcome.output_tokens + _ct_out,
+            uncached_input_tokens=outcome.uncached_input_tokens + _ct_uin,
+            cache_read_tokens=outcome.cache_read_tokens + _ct_cr,
+            cache_write_tokens=outcome.cache_write_tokens + _ct_cw,
+        )
+
     # Output-shaping savings ledger (counterfactual estimator). The shaper
     # tags each request's (arm, stratum) onto ``transforms_applied``; feed the
     # observed output tokens to the recorder so it can produce an honest