fix(usage): cache-read percentage exceeded 100% in streaming-ui display

amplifier_module_provider_github_copilot/sdk_adapter/event_helpers.py

@@ -143,21 +143,51 @@ def extract_usage_data(sdk_event: Any) -> dict[str, int | None] | None:
 
     Fields returned:
 
-    * ``input_tokens`` (int, required) — **fresh (uncacheable) tokens only**.
-      The Copilot SDK's ``assistant.usage`` event reports ``input_tokens`` as the
-      billing total (``fresh + cache_read + cache_write``). The kernel
-      ``Usage.input_tokens`` convention — established by the Anthropic provider
-      and consumed by the Amplifier streaming UI — defines this field as the
-      fresh/uncacheable portion only. The streaming UI computes the display total
-      as ``input_tokens + cache_read + cache_write``, so subtracting both cache
-      buckets means the display exactly recovers the SDK billing total.
-      Formula: ``max(0, sdk_input_tokens - cache_read_tokens - cache_write_tokens)``.
+    * ``input_tokens`` (int, required) — equal to the SDK ``inputTokens`` field
+      minus ``cacheWriteTokens`` (treated as ``0`` when absent or ``None``).
+      The Copilot SDK reports four token fields (``inputTokens``,
+      ``outputTokens``, ``cacheReadTokens``, ``cacheWriteTokens``); the SDK
+      v0.3.0 docs (``copilot-sdk/v0.3.0/docs/features/streaming-events.md``)
+      describe each field neutrally and do not state whether the cache
+      buckets are included in ``inputTokens`` or additive to it. A captured
+      SDK v1.0.0b4 ``session.shutdown`` event (model ``claude-sonnet-4.6``,
+      2026-05-17, captured after commit ``eab7989`` bumped the SDK from
+      v0.3.0 to v1.0.0b4 — the billing schema change in v1.0.0b4 is exactly
+      what made the gross-total interpretation of ``inputTokens`` load-bearing)
+      reports ``inputTokens=34554``, ``cacheReadTokens=26075``,
+      ``cacheWriteTokens=8475`` alongside a sibling ``tokenDetails`` block
+      with ``input.tokenCount=4``. The arithmetic identity
+      ``4 + 26075 + 8475 == 34554`` is direct empirical proof that
+      ``inputTokens`` is the gross billing total
+      ``fresh + cacheReadTokens + cacheWriteTokens``. The kernel ``Usage``
+      schema (``amplifier-core`` ``docs/contracts/PROVIDER_CONTRACT.md``
+      ``llm:response``) requires ``input_tokens`` to be the gross-total form
+      that excludes ``cache_write_tokens`` (i.e. ``fresh + cache_read``),
+      treating ``cache_write_tokens`` as a separate additive bucket billed
+      on top. Subtracting only ``cacheWriteTokens`` produces that
+      kernel-mandated shape; subtracting ``cacheReadTokens`` would
+      double-remove it. The resulting value is compatible with the
+      streaming-ui hook's display total, which adds ``cache_write_tokens``
+      on top of the kernel ``input_tokens`` per the same kernel contract.
+      The ``session.shutdown`` event above gives the *direct* arithmetic
+      proof because it carries a sibling ``tokenDetails.input.tokenCount=4``
+      field. ``assistant.usage`` events (the event type this function
+      actually processes) do not carry ``tokenDetails``, so the same
+      gross-total identity is verified end-to-end against four
+      production-captured ``assistant.usage`` shapes (plus one defensive
+      all-fields-None shape) in ``tests/test_event_classification_overlap.py``
+      (``TestStreamingUIPercentageInvariantWithRealCapturedShapes``); each
+      non-degenerate shape asserts the round-trip
+      ``input_tokens + (cache_write_tokens or 0) == sdk_inputTokens`` and
+      the resulting streaming-ui denominator invariant
+      ``denom >= cache_read_tokens``.
+      Formula: ``max(0, sdk_inputTokens - (cache_write_tokens or 0))``.
     * ``output_tokens`` (int, required) — output tokens generated.
-    * ``total_tokens`` (int, required) — ``input_tokens + output_tokens``
-      (fresh + output). Computed here because the SDK does not populate
-      ``totalTokens`` in ``assistant.usage`` events (``Data.total_tokens`` is
-      ``float | None = None`` in the schema; it is ``None`` in usage payloads).
-      The kernel ``Usage.total_tokens`` is non-optional so we compute it.
+    * ``total_tokens`` (int, required) — ``input_tokens + output_tokens``.
+      Computed here because the SDK does not populate ``totalTokens`` in
+      ``assistant.usage`` events (``Data.total_tokens`` is ``float | None = None``
+      in the schema; it is ``None`` in usage payloads). The kernel
+      ``Usage.total_tokens`` is non-optional so we compute it.
     * ``cache_read_tokens`` (int | None) — tokens served from the upstream LLM's
       prompt cache. ``None`` when the field is absent from the event, which is
       semantically distinct from ``0`` (SDK reported a confirmed zero, meaning no
@@ -200,21 +230,19 @@ def extract_usage_data(sdk_event: Any) -> dict[str, int | None] | None:
                 cache_write: int | None = (
                     int(raw_cache_write) if raw_cache_write is not None else None
                 )
-                # Contract: streaming-contract:usage:MUST:3 — kernel Usage.input_tokens
-                # must be the fresh (uncacheable) portion only.
-                # SDK input_tokens = fresh + cache_read + cache_write (billing total).
-                # Subtract both buckets so the computation is exact in all cases.
-                # The streaming UI adds cache_read + cache_write back for display, so
-                # display total = fresh + cache_read + cache_write = sdk_input_tokens.
-                fresh_tok = max(0, in_tok - (cache_read or 0) - (cache_write or 0))
+                # Contract: streaming-contract:usage:MUST:3 — kernel
+                # Usage.input_tokens equals the SDK billing total minus
+                # cache_write_tokens; cache_read remains inside input_tokens
+                # so the gross-input shape matches the kernel contract.
+                # cache_write is treated as 0 when absent or None.
+                adjusted_input = max(0, in_tok - (cache_write or 0))
                 return {
-                    "input_tokens": fresh_tok,
+                    "input_tokens": adjusted_input,
                     "output_tokens": out_tok,
                     # SDK assistant.usage does not send total_tokens — compute it.
                     # Kernel Usage.total_tokens: int is required (not Optional).
-                    # total_tokens = fresh + output (consistent with input_tokens convention).
-                    # Contract: streaming-contract:usage:MUST:1, MUST:3
-                    "total_tokens": fresh_tok + out_tok,
+                    # Contract: streaming-contract:usage:MUST:3
+                    "total_tokens": adjusted_input + out_tok,
                     # Contract: streaming-contract:usage:MUST:2
                     "cache_read_tokens": cache_read,
                     "cache_write_tokens": cache_write,
@@ -236,16 +264,16 @@ def extract_usage_data(sdk_event: Any) -> dict[str, int | None] | None:
             raw_cache_write = getattr(data, "cache_write_tokens", None)
             cache_read = int(raw_cache_read) if raw_cache_read is not None else None
             cache_write = int(raw_cache_write) if raw_cache_write is not None else None
-            # Contract: streaming-contract:usage:MUST:3 — kernel Usage.input_tokens
-            # must be the fresh (uncacheable) portion only.
-            # SDK input_tokens = fresh + cache_read + cache_write (billing total).
-            fresh_tok = max(0, in_tok - (cache_read or 0) - (cache_write or 0))
+            # Contract: streaming-contract:usage:MUST:3 — kernel
+            # Usage.input_tokens equals the SDK billing total minus
+            # cache_write_tokens; cache_read remains inside input_tokens
+            # so the gross-input shape matches the kernel contract.
+            adjusted_input = max(0, in_tok - (cache_write or 0))
             return {
-                "input_tokens": fresh_tok,
+                "input_tokens": adjusted_input,
                 "output_tokens": out_tok,
-                # total_tokens = fresh + output (consistent with input_tokens convention).
-                # Contract: streaming-contract:usage:MUST:1, MUST:3
-                "total_tokens": fresh_tok + out_tok,
+                # Contract: streaming-contract:usage:MUST:3
+                "total_tokens": adjusted_input + out_tok,
                 # Contract: streaming-contract:usage:MUST:2
                 "cache_read_tokens": cache_read,
                 "cache_write_tokens": cache_write,

amplifier_module_provider_github_copilot/streaming.py

@@ -807,9 +807,12 @@ def translate_event(sdk_event: dict[str, Any], config: EventConfig) -> DomainEve
 
     domain_type, block_type = config.bridge_mappings[event_type]
 
-    # For USAGE_UPDATE events, use extract_usage_data to apply the
-    # fresh-only input_tokens computation (MUST:3) rather than raw extraction.
-    # Raw _extract_event_data would pass the SDK's billing total (fresh + cache_read).
+    # For USAGE_UPDATE events, route through extract_usage_data so that
+    # Usage.input_tokens follows the kernel-mandated gross shape
+    # (fresh + cache_read), produced by subtracting cache_write_tokens from
+    # the SDK's billing-total inputTokens. Raw _extract_event_data would
+    # pass the SDK's billing total unchanged (i.e. still containing
+    # cache_write), violating streaming-contract:usage:MUST:3.
     # Contract: streaming-contract:usage:MUST:3
     if domain_type == DomainEventType.USAGE_UPDATE:
         from .sdk_adapter.event_helpers import extract_usage_data

contracts/streaming-contract.md

@@ -1,13 +1,14 @@
 # Contract: Streaming
 
 ## Version
-- **Current:** 1.4 (EventRouter-Extracted)
+- **Current:** 1.5 (Cache-Token Semantics Aligned with Cross-Provider Convention)
 - **Module Reference:** 
   - `amplifier_module_provider_github_copilot/streaming.py` (accumulator, event translation)
   - `amplifier_module_provider_github_copilot/event_router.py` (SDK event routing)
 - **Kernel Types:** `amplifier_core.message_models` (Pydantic, NOT content_models dataclass)
 - **Status:** Specification
 - **History:**
+  - **1.5** — `usage:MUST:3` rewritten: `Usage.input_tokens` excludes only `cache_write_tokens`; cache_read remains inside input_tokens, aligning with `amplifier-core` `PROVIDER_CONTRACT.md` gross-input shape (PR #69, 2026-05-06)
   - **1.4** — Extracted EventRouter from provider.py (separation of concerns)
   - **1.3** — Added completion guard, usage capture, session lifecycle anchors (bug fix documentation)
   - **1.2** — Fixed anchor prefix from `streaming:` to `streaming-contract:`, added missing anchors
@@ -169,15 +170,9 @@ The SDK sends `assistant.usage` events AFTER `session.idle` (turn completion). T
 
 **Rationale:** The kernel `Usage` model includes optional `cache_read_tokens` and `cache_write_tokens` fields (Pydantic `int | None`, default `None`). The SDK populates `cacheReadTokens` in the `assistant.usage` event when the upstream LLM provider's (e.g., Anthropic/Claude) prompt cache is hit. If these fields are dropped, Amplifier session analytics cannot distinguish cached from uncached token cost — masking significant cost differences (cache-read tokens bill at 0.10× the base rate). The SDK schema (`session_events.Data`) defines both `cache_read_tokens` and `cache_write_tokens` as `float | None`; `cache_write_tokens` is not currently populated by the SDK even when a cache write occurs, but the provider MUST extract it when present so the implementation is correct as soon as the SDK populates it. `None` is semantically distinct from `0`: `None` means the field was not reported; `0` means the SDK reported a confirmed zero.
 
-**streaming-contract:usage:MUST:3**: The provider MUST set `Usage.input_tokens` to the **fresh (uncacheable) token count only** — not the SDK's billing total. When cache fields are present, `input_tokens = sdk_input_tokens - cache_read_tokens - cache_write_tokens`. `total_tokens` MUST equal `input_tokens + output_tokens`.
+**streaming-contract:usage:MUST:3**: When the SDK `assistant.usage` event is processed, the provider MUST set `Usage.input_tokens = max(0, sdk_inputTokens - (cache_write_tokens or 0))`. Cache token fields are forwarded per `streaming-contract:usage:MUST:2`. `Usage.total_tokens` MUST equal `Usage.input_tokens + Usage.output_tokens`.
 
-**Rationale:** The Copilot SDK's `assistant.usage` event reports `input_tokens` as the billing total (`fresh + cache_read + cache_write`). The kernel `Usage.input_tokens` convention — established by the Anthropic provider and consumed by the Amplifier streaming UI — defines `input_tokens` as the fresh/uncacheable portion only. The streaming UI computes the display total as `input_tokens + cache_read + cache_write`, so if the provider passes the SDK's billing total, the UI double-counts the cache buckets. Subtracting both `cache_read` and `cache_write` means the streaming UI display exactly recovers the SDK billing total: `fresh + cache_read + cache_write = sdk_input_tokens`.
-
-**Evidence:** Streaming UI source (`amplifier-module-hooks-streaming-ui/__init__.py`):
-```python
-# When caching is active, input_tokens is just the uncacheable portion
-total_input = input_tokens + cache_read + cache_create
-```
+**Rationale:** The Copilot SDK reports four token fields (`inputTokens`, `outputTokens`, `cacheReadTokens`, `cacheWriteTokens`) without documenting their inclusion semantics; the SDK schema and docs (`copilot-sdk/v0.3.0/docs/features/streaming-events.md`) describe each field neutrally. The kernel `Usage` schema (`amplifier-core` `docs/contracts/PROVIDER_CONTRACT.md` `llm:response`, normative since `1548042c` / PR #69, 2026-05-06) requires `input_tokens` to be the gross total (fresh + `cache_read` combined) with `cache_write_tokens` as a separate additive bucket billed on top. This provider subtracts only `cacheWriteTokens` from `inputTokens` to produce that kernel-mandated shape; subtracting `cacheReadTokens` would double-remove it. The resulting value is compatible with the streaming-ui hook's display total, which adds `cache_write_tokens` on top of gross per kernel contract (`microsoft/amplifier-module-hooks-streaming-ui@79dde49`, post-PR #10 `dda82a9`).
 
 **Implementation:**
 ```python
@@ -422,7 +417,7 @@ class StreamingChatResponse(ChatResponse):
 | `streaming-contract:completion:MUST:2` | Events after ERROR are ignored (except usage) |
 | `streaming-contract:usage:MUST:1` | Usage events captured even after completion |
 | `streaming-contract:usage:MUST:2` | Cache token fields forwarded to kernel Usage when present |
-| `streaming-contract:usage:MUST:3` | `Usage.input_tokens` = fresh only (`sdk_input - cache_read - cache_write`); `total_tokens = input + output` |
+| `streaming-contract:usage:MUST:3` | `Usage.input_tokens = max(0, sdk_inputTokens - (cache_write_tokens or 0))`; cache fields forwarded per `usage:MUST:2`; `total_tokens = input_tokens + output_tokens` |
 | `streaming-contract:SessionLifecycle:MUST:1` | Provider validates idle_events config at load time |
 | `streaming-contract:ProgressiveStreaming:SHOULD:1` | Emit llm:content_block events |
 | `streaming-contract:ProgressiveStreaming:SHOULD:2` | Async fire-and-forget emission |

tests/test_contract_events.py

@@ -166,11 +166,28 @@ def test_nested_dict_data_promoted_to_top_level(self) -> None:
         )
 
     def test_usage_token_fields_at_top_level(self) -> None:
-        """USAGE_UPDATE DomainEvent.data must have token fields at the top level.
+        """USAGE_UPDATE DomainEvent.data must have token fields at the top level
+        AND ``input_tokens`` MUST already be in the kernel-mandated gross shape
+        (cache_write subtracted) when it leaves ``translate_event``.
+
+        Contracts:
+          - event-vocabulary:Bridge:MUST:3 (top-level promotion)
+          - streaming-contract:usage:MUST:3 (cache_write subtraction)
 
-        Contract: event-vocabulary:Bridge:MUST:3
         Rationale: StreamingAccumulator stores event.data directly as self.usage;
-        token fields buried under a nested 'data' key produce silent zero usage counts.
+        token fields buried under a nested 'data' key produce silent zero usage
+        counts. Additionally, ``translate_event`` MUST route assistant.usage
+        through ``extract_usage_data`` so the kernel-shape transform (subtract
+        cache_write, leave cache_read inside input_tokens) is applied before
+        the DomainEvent reaches the accumulator. The mixed-cache fixture
+        below would silently regress if a future refactor bypassed
+        ``extract_usage_data`` on the USAGE_UPDATE branch.
+
+        Sibling caller: ``event_router.py`` (immediate usage capture for the
+        idle-race window) also routes through ``extract_usage_data`` and is
+        therefore covered transitively. The helper's own dict-vs-object paths,
+        cache-only / write-only / mixed shapes, and the ``max(0, ...)`` clamp
+        are exercised directly in ``test_event_classification_overlap.py``.
         """
         from amplifier_module_provider_github_copilot.streaming import (
             DomainEvent,
@@ -180,23 +197,40 @@ def test_usage_token_fields_at_top_level(self) -> None:
         )
 
         config = load_event_config()
+        # Mixed-cache shape (cache_read > 0 AND cache_write > 0) exercises both
+        # contracts in a single payload: 17 - 7 = 10 (subtract only cache_write;
+        # cache_read=3 stays inside input_tokens).
         sdk_event = {
             "type": "assistant.usage",
-            "data": {"input_tokens": 10, "output_tokens": 20, "total_tokens": 30},
+            "data": {
+                "input_tokens": 17,
+                "output_tokens": 20,
+                "total_tokens": 37,
+                "cache_read_tokens": 3,
+                "cache_write_tokens": 7,
+            },
         }
         result = translate_event(sdk_event, config)
 
         assert isinstance(result, DomainEvent)
         assert result.type == DomainEventType.USAGE_UPDATE
-        # MUST:3 — no residual 'data' key; all token fields at top level
+        # Bridge:MUST:3 — no residual 'data' key; all token fields at top level
         assert "data" not in result.data, (
             "translate_event MUST NOT leave a nested 'data' key in DomainEvent.data"
         )
+        # usage:MUST:3 — input_tokens already cache_write-subtracted at this layer
         assert result.data["input_tokens"] == 10, (
-            "translate_event MUST promote nested usage token fields to the top level"
+            "translate_event MUST route USAGE_UPDATE through extract_usage_data; "
+            f"expected input_tokens=10 (=17-7), got {result.data['input_tokens']}"
         )
         assert result.data["output_tokens"] == 20
-        assert result.data["total_tokens"] == 30
+        # total_tokens recomputed against the post-subtraction input
+        assert result.data["total_tokens"] == 30, (
+            "total_tokens MUST equal input_tokens + output_tokens after subtraction"
+        )
+        # Bridge:MUST:3 — cache fields forwarded
+        assert result.data["cache_read_tokens"] == 3
+        assert result.data["cache_write_tokens"] == 7
 
     def test_sdk_object_in_data_not_preserved(self) -> None:
         """When sdk_event contains a raw SDK data object, it must not appear in DomainEvent.data