fix(ccr): make continuation-usage publication additive for batch paths

set_pending_ccr_continuation_usage now appends to a list instead of
replacing, so multiple handle_response calls in a batch (e.g.
BatchResultProcessor.process_results) accumulate rather than overwrite.
consume sums the list and returns a single 4-tuple, preserving the
existing consumer interface.

Fixes the CHANGES_REQUESTED from @JerrettDavis: with two CCR-bearing
batch items, item 2 was overwriting item 1's continuation overhead,
so cost_tracker recorded only the last item's dropped rounds.

Added test_batch_two_ccr_items_both_reach_cost_view: two set_pending
calls + one emit, asserting both items' overhead reaches
cost_tracker.record_tokens (sum), while record_request stays on the
base outcome (cost-only fold, unchanged). 14/14 pass.

Author: EvolveAegis

headroom/proxy/outcome.py

@@ -65,24 +65,36 @@ def consume_pending_proactive_retrieval() -> tuple[int, int] | None:
 # 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
-)
+_pending_ccr_continuation_usage: ContextVar[
+    list[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)
+    """Append the dropped continuation rounds' (uncached_in, cache_read, cache_write, output) to this request.
+
+    Additive: each call appends to a list so batch processing (multiple
+    handle_response calls before a single emit) accumulates all items.
+    consume_pending_ccr_continuation_usage sums the list.
+    """
+    if value is None:
+        return
+    items = _pending_ccr_continuation_usage.get()
+    if items is None:
+        items = []
+        _pending_ccr_continuation_usage.set(items)
+    items.append(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
+    """Read, sum, and clear all pending continuation usage for this request."""
+    items = _pending_ccr_continuation_usage.get()
+    _pending_ccr_continuation_usage.set(None)
+    if not items:
+        return None
+    return tuple(sum(group) for group in zip(*items))
 
 
 def clear_pending_outcome_side_channels() -> None:

tests/test_ccr_continuation_cost_fold.py

@@ -487,3 +487,48 @@ async def main():
     assert metrics_rec["uncached_input_tokens"] == U2["input_tokens"]
     assert metrics_rec["cache_read_tokens"] == U2["cache_read_input_tokens"]
     assert metrics_rec["output_tokens"] == U2["output_tokens"]
+
+
+# --- batch path: multiple CCR items accumulate, not overwrite ------------
+
+
+def test_batch_two_ccr_items_both_reach_cost_view():
+    """Two CCR-bearing batch items: both items' continuation overhead
+    reaches cost_tracker.record_tokens, not just the last.
+
+    Simulates BatchResultProcessor: multiple handle_response calls
+    (each set_pending), then one emit_request_outcome (one consume).
+    """
+    handler = _Handler()
+    handler.cost_tracker = _CostTracker()
+    base_out, base_uin, base_cr, base_cw = 800, 110_000, 30_000, 3_000
+    item1 = (205_000, 80_000, 11_000, 1_100)
+    item2 = (100_000, 50_000, 5_000, 500)
+
+    outcome = RequestOutcome(
+        request_id="req-batch", provider="anthropic", model="m", status_code=200,
+        original_tokens=base_uin, optimized_tokens=base_uin, output_tokens=base_out,
+        tokens_saved=0, attempted_input_tokens=base_uin,
+        uncached_input_tokens=base_uin, cache_read_tokens=base_cr, cache_write_tokens=base_cw,
+    )
+
+    async def main():
+        consume_pending_ccr_continuation_usage()
+        set_pending_ccr_continuation_usage(item1)
+        set_pending_ccr_continuation_usage(item2)
+        await emit_request_outcome(handler, outcome)
+        return consume_pending_ccr_continuation_usage()
+
+    leftover = asyncio.run(main())
+
+    assert handler.cost_tracker.recorded, "cost_tracker.record_tokens was not called"
+    ct = handler.cost_tracker.recorded[0]
+    # Both items accumulated into cost_tracker, not just the last
+    assert ct["uncached_tokens"] == base_uin + item1[0] + item2[0]
+    assert ct["cache_read_tokens"] == base_cr + item1[1] + item2[1]
+    assert ct["cache_write_tokens"] == base_cw + item1[2] + item2[2]
+    assert ct["output_tokens"] == base_out + item1[3] + item2[3]
+    # record_request stays on base outcome (cost-only fold)
+    assert handler.metrics.requested[0]["uncached_input_tokens"] == base_uin
+    # Consume cleared after emit
+    assert leftover is None