diff --git a/amplifier_module_provider_github_copilot/sdk_adapter/event_helpers.py b/amplifier_module_provider_github_copilot/sdk_adapter/event_helpers.py index d343eeb..46de474 100644 --- a/amplifier_module_provider_github_copilot/sdk_adapter/event_helpers.py +++ b/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, diff --git a/amplifier_module_provider_github_copilot/streaming.py b/amplifier_module_provider_github_copilot/streaming.py index 1074ee4..a18c60e 100644 --- a/amplifier_module_provider_github_copilot/streaming.py +++ b/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 diff --git a/contracts/streaming-contract.md b/contracts/streaming-contract.md index 19f7616..9e486d2 100644 --- a/contracts/streaming-contract.md +++ b/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 | diff --git a/tests/test_contract_events.py b/tests/test_contract_events.py index 77d56cf..347a630 100644 --- a/tests/test_contract_events.py +++ b/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 diff --git a/tests/test_event_classification_overlap.py b/tests/test_event_classification_overlap.py index ed1de99..c06d8b7 100644 --- a/tests/test_event_classification_overlap.py +++ b/tests/test_event_classification_overlap.py @@ -13,7 +13,7 @@ from amplifier_module_provider_github_copilot._compat import ConfigurationError from amplifier_module_provider_github_copilot.streaming import ( DomainEventType, - _validate_no_classification_overlap, + _validate_no_classification_overlap, # pyright: ignore[reportPrivateUsage] load_event_config, ) @@ -545,27 +545,53 @@ class MockEvent: f"got {result.get('cache_write_tokens')}" ) - def test_extract_usage_data_cache_tokens_none_when_absent(self) -> None: + @pytest.mark.parametrize( + "shape,reported_zero,expected", + [ + pytest.param("dict", False, None, id="dict_absent_yields_none"), + pytest.param("dict", True, 0, id="dict_explicit_zero_preserved"), + pytest.param("object", False, None, id="object_absent_yields_none"), + pytest.param("object", True, 0, id="object_explicit_zero_preserved"), + ], + ) + def test_extract_usage_data_cache_tokens_none_vs_zero( + self, shape: str, reported_zero: bool, expected: int | None + ) -> None: """Contract: streaming-contract:usage:MUST:2 - When cache_read_tokens and cache_write_tokens are absent from the SDK event, - extract_usage_data MUST return None for each — not zero. None is semantically - distinct from 0: None means the SDK did not report the field; 0 means the SDK - reported a confirmed zero. Conflating them hides whether caching occurred. + ``None`` (SDK did not report the field) MUST remain distinct from ``0`` + (SDK reported a confirmed zero). Conflating them hides whether caching + occurred. Both event shapes must preserve the distinction. """ from amplifier_module_provider_github_copilot.sdk_adapter.event_helpers import ( extract_usage_data, ) - event = {"data": {"input_tokens": 100, "output_tokens": 50}} + if shape == "dict": + data: dict[str, object] = {"input_tokens": 100, "output_tokens": 50} + if reported_zero: + data["cache_read_tokens"] = 0 + data["cache_write_tokens"] = 0 + event: object = {"data": data} + else: + + class _Data: + input_tokens = 100 + output_tokens = 50 + + if reported_zero: + _Data.cache_read_tokens = 0.0 # type: ignore[attr-defined] + _Data.cache_write_tokens = 0.0 # type: ignore[attr-defined] + + class _Event: + data = _Data() + + event = _Event() + result = extract_usage_data(event) assert result is not None # narrowed for pyright - assert result["cache_read_tokens"] is None, ( - "cache_read_tokens absent from event must be None, not 0" - ) - assert result["cache_write_tokens"] is None, ( - "cache_write_tokens absent from event must be None, not 0" - ) + assert result["cache_read_tokens"] == expected + assert result["cache_write_tokens"] == expected def test_stream_accumulator_build_response_passes_cache_tokens_to_usage(self) -> None: """Contract: streaming-contract:usage:MUST:2 @@ -603,67 +629,184 @@ def test_stream_accumulator_build_response_passes_cache_tokens_to_usage(self) -> f"got {response.usage.cache_write_tokens}" ) - def test_extract_usage_data_input_tokens_is_fresh_only_when_cache_hit_dict(self) -> None: + @pytest.mark.parametrize( + "sdk_input,output,cache_read,cache_write,expected_input", + [ + pytest.param(70436, 23, 63128, None, 70436, id="read_only_cache_write_absent"), + # prod_cw_zero: cache_read > 0, cache_write explicitly 0. Distinct + # from read_only_cache_write_absent (cw=None / field absent): here + # cw=0 is explicit, exercising the None-vs-0 preservation path on + # the cache_write field while leaving the subtraction a no-op. + pytest.param(84119, 119, 60034, 0, 84119, id="prod_cw_zero"), + # write_only_no_read: cache_read=0, cache_write > 0. Synthetic + # shape that isolates the cache_write subtraction branch from + # cache_read by zeroing the latter. Drives a non-trivial + # subtraction even when the read bucket is inactive, so dropping + # ``- (cache_write or 0)`` makes this row fail in isolation. + pytest.param(64295, 473, 0, 64285, 10, id="write_only_no_read"), + # mixed_real_shape: numbers scaled from a captured SDK v1.0.0b4 + # ``session.shutdown`` event (2026-05-17, model claude-sonnet-4.6, + # 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) + # whose ``modelMetrics`` block carries inputTokens=34554, + # cacheReadTokens=26075, cacheWriteTokens=8475 with a sibling + # ``tokenDetails.input.tokenCount=4``; the identity + # ``4 + 26075 + 8475 == 34554`` confirms ``inputTokens`` is the + # gross billing total (fresh + cache_read + cache_write). This row + # exercises the BOTH-cache-buckets-non-zero arithmetic path that + # the production capture proves the SDK is capable of emitting. + pytest.param(67719, 6, 60034, 7682, 60037, id="mixed_real_shape"), + ], + ) + def test_extract_usage_data_subtracts_only_cache_write_when_cache_hit_dict( + self, + sdk_input: int, + output: int, + cache_read: int, + cache_write: int | None, + expected_input: int, + ) -> None: """Contract: streaming-contract:usage:MUST:3 - When the SDK sends input_tokens=70436 (billing total = fresh + cached) and - cache_read_tokens=63128, extract_usage_data MUST set input_tokens to the fresh - portion only (70436 - 63128 = 7308). The streaming UI convention is that - input_tokens = uncacheable/fresh portion; it adds cache_read separately to - compute the display total. Passing the billing total causes the UI to double- - count cache_read and display 133K instead of 70K. + Behavioural contract: ``Usage.input_tokens == + max(0, sdk_inputTokens - (cache_write or 0))``; cache_read remains + inside ``input_tokens`` (it is NOT subtracted), and the + ``input_tokens + (cache_write or 0)`` round-trip MUST recover the SDK's + original ``inputTokens`` value. The mixed and write-only rows kill + mutations that the read-only fixture cannot reach: dropping the cache_write + subtraction (write_only row), or accidentally subtracting cache_read + (mixed row), both turn at least one assertion red. """ from amplifier_module_provider_github_copilot.sdk_adapter.event_helpers import ( extract_usage_data, ) - event = { - "data": { - "input_tokens": 70436, - "output_tokens": 23, - "cache_read_tokens": 63128, - } + data: dict[str, object] = { + "input_tokens": sdk_input, + "output_tokens": output, + "cache_read_tokens": cache_read, } - result = extract_usage_data(event) + if cache_write is not None: + data["cache_write_tokens"] = cache_write + result = extract_usage_data({"data": data}) assert result is not None # narrowed for pyright - assert result["input_tokens"] == 7308, ( - f"input_tokens must be fresh-only (70436 - 63128 = 7308), " - f"got {result.get('input_tokens')} — " - "provider is double-counting cache_read in input_tokens" - ) - assert result["total_tokens"] == 7308 + 23, ( - f"total_tokens must be fresh + output = 7331, got {result.get('total_tokens')}" - ) - assert result["cache_read_tokens"] == 63128 - - def test_extract_usage_data_input_tokens_is_fresh_only_when_cache_hit_object(self) -> None: + assert result["input_tokens"] == expected_input + assert result["output_tokens"] == output + assert result["total_tokens"] == expected_input + output + assert result["cache_read_tokens"] == cache_read + assert result["cache_write_tokens"] == cache_write + # Round-trip: streaming-UI recovers SDK billing total via + # input_tokens + (cache_write or 0). + in_t = result["input_tokens"] or 0 + cw_t = result["cache_write_tokens"] or 0 + assert in_t + cw_t == sdk_input + + @pytest.mark.parametrize( + "sdk_input,output,cache_read,cache_write,expected_input", + [ + pytest.param(70436, 23, 63128.0, None, 70436, id="read_only_cache_write_absent"), + # prod_cw_zero: see provenance note on the dict-path row above. + pytest.param(84119, 119, 60034.0, 0.0, 84119, id="prod_cw_zero"), + # write_only_no_read: see provenance note on the dict-path row above. + pytest.param(64295, 473, 0.0, 64285.0, 10, id="write_only_no_read"), + # mixed_real_shape: see provenance note on the dict-path row above. + pytest.param(67719, 6, 60034.0, 7682.0, 60037, id="mixed_real_shape"), + ], + ) + def test_extract_usage_data_subtracts_only_cache_write_when_cache_hit_object( + self, + sdk_input: int, + output: int, + cache_read: float, + cache_write: float | None, + expected_input: int, + ) -> None: """Contract: streaming-contract:usage:MUST:3 - Object-path (real SDK): same fresh-only requirement on the object event path. - SDK Data object has input_tokens=70436 (billing total) and - cache_read_tokens=63128.0 (float). Provider must compute 70436 - 63128 = 7308. + Object-path (attribute-bearing stand-in mimicking the SDK + ``session_events.Data`` shape; ``float | None`` cache fields). Same + behavioural contract as the dict path: cache_read stays inside + ``input_tokens`` and only cache_write is subtracted, clamped at zero. + The object-path branch in ``extract_usage_data`` is physically distinct + from the dict-path branch (separate ``getattr`` reads vs. ``dict.get``), + so both must be exercised independently. """ from amplifier_module_provider_github_copilot.sdk_adapter.event_helpers import ( extract_usage_data, ) class MockData: - input_tokens = 70436 - output_tokens = 23 - cache_read_tokens = 63128.0 # SDK sends float - cache_write_tokens = None + pass + + MockData.input_tokens = sdk_input # type: ignore[attr-defined] + MockData.output_tokens = output # type: ignore[attr-defined] + MockData.cache_read_tokens = cache_read # type: ignore[attr-defined] + MockData.cache_write_tokens = cache_write # type: ignore[attr-defined] class MockEvent: data = MockData() result = extract_usage_data(MockEvent()) assert result is not None # narrowed for pyright - assert result["input_tokens"] == 7308, ( - f"input_tokens must be fresh-only (70436 - 63128 = 7308), " - f"got {result.get('input_tokens')}" + assert result["input_tokens"] == expected_input + assert result["output_tokens"] == output + assert result["total_tokens"] == expected_input + output + assert result["cache_read_tokens"] == int(cache_read) + expected_cw = int(cache_write) if cache_write is not None else None + assert result["cache_write_tokens"] == expected_cw + in_t = result["input_tokens"] or 0 + cw_t = result["cache_write_tokens"] or 0 + assert in_t + cw_t == sdk_input + + @pytest.mark.parametrize( + "shape", + ["dict", "object"], + ) + def test_extract_usage_data_clamps_input_to_zero_when_cache_write_exceeds_input( + self, shape: str + ) -> None: + """Contract: streaming-contract:usage:MUST:3 + + The MUST clause requires the result to be ``clamped to zero`` when the + cache_write subtraction would underflow. Without ``max(0, ...)`` the + kernel ``Usage.input_tokens`` would be negative, breaking downstream + billing math. This is the only test that exercises the clamp branch. + """ + from amplifier_module_provider_github_copilot.sdk_adapter.event_helpers import ( + extract_usage_data, + ) + + if shape == "dict": + event: object = { + "data": { + "input_tokens": 100, + "output_tokens": 5, + "cache_read_tokens": 0, + "cache_write_tokens": 250, + } + } + else: + + class _Data: + input_tokens = 100 + output_tokens = 5 + cache_read_tokens = 0.0 + cache_write_tokens = 250.0 + + class _Event: + data = _Data() + + event = _Event() + + result = extract_usage_data(event) + assert result is not None # narrowed for pyright + assert result["input_tokens"] == 0, ( + "Cache-write underflow MUST clamp input_tokens to 0, not " + f"{result.get('input_tokens')}" ) - assert result["total_tokens"] == 7308 + 23, ( - f"total_tokens must be fresh + output = 7331, got {result.get('total_tokens')}" + assert result["total_tokens"] == 5, ( + "total_tokens = clamped_input + output = 0 + 5" ) def test_extract_usage_data_input_tokens_unchanged_when_no_cache(self) -> None: @@ -685,6 +828,216 @@ def test_extract_usage_data_input_tokens_unchanged_when_no_cache(self) -> None: assert result["total_tokens"] == 150 +class TestStreamingUIPercentageInvariantWithRealCapturedShapes: + """Anti-1633% regression: the streaming-ui hook's cache-percentage display + must never exceed 100%. + + Contract: streaming-contract:usage:MUST:3 (kernel-mandated gross input shape) + Anchor (upstream): amplifier-core PROVIDER_CONTRACT.md L176-187 -- input_tokens + "MUST" be "gross total (fresh + cache_read combined)". + Anchor (downstream): amplifier-module-hooks-streaming-ui __init__.py L74-96 -- + `_compute_total_input` returns `input_tokens + cache_create`, then displays + `cache_pct = int((cache_read / total_input) * 100)`. + + Why this class exists separately from the per-row parametrize tables above: + those rows cover the post-transform field values. This class enforces the + end-to-end ANTI-1633% invariant -- that for every shape the SDK can plausibly + emit, the post-`extract_usage_data` Usage dict, when fed into the streaming-ui + formula, produces a percentage in [0, 100]. + + Shapes are CAPTURED FROM LIVE amplifier runs against the github-copilot SDK + (4 distinct production shapes plus one defensive all-fields-None shape; + identities verified by replaying the kernel formula). Each shape pairs the + SDK-side raw fields with the kernel-mandated post-transform result. + + Cited by ``extract_usage_data`` docstring in + ``amplifier_module_provider_github_copilot/sdk_adapter/event_helpers.py`` + as the assistant.usage proof chain for streaming-contract:usage:MUST:3 -- + keep class name in sync if renamed. + """ + + # Five live-captured shapes from `amplifier run` probe matrix: + # (1) claude-sonnet text-only mid-session cache hit + # (2) claude-sonnet tool-using turn (tool_calls > 0) + # (3) claude-haiku WRITE-ONLY first turn (huge cache_write, no cache_read) + # (4) gpt-5.5 no-caching turn (cache fields zero across the board) + # (5) defensive: SDK emitted Usage with all fields None + # Each row: (label, sdk_input, sdk_cache_read, sdk_cache_write, sdk_output, + # expected_input_after_transform, expected_total_after_transform). + CAPTURED_SHAPES = [ + ("claude_sonnet_cache_hit", 63390, 60068, 3319, 6, 60071, 60077), + ("claude_sonnet_tool_use", 63393, 60068, 3322, 90, 60071, 60161), + ("claude_haiku_write_only", 63552, 0, 63542, 96, 10, 106), + ("gpt55_no_cache", 53611, 0, 0, 58, 53611, 53669), + ("all_none_defensive", None, None, None, None, None, None), + ] + + @pytest.mark.parametrize( + "label,sdk_input,sdk_cr,sdk_cw,sdk_out,exp_in,exp_total", + CAPTURED_SHAPES, + ids=[r[0] for r in CAPTURED_SHAPES], + ) + def test_real_captured_shape_satisfies_streaming_ui_invariant( + self, + label: str, + sdk_input: int | None, + sdk_cr: int | None, + sdk_cw: int | None, + sdk_out: int | None, + exp_in: int | None, + exp_total: int | None, + ) -> None: + """For every captured production shape, the streaming-ui denominator + (`input_tokens + cache_write_tokens`) must be >= cache_read_tokens. + + This is the direct anti-regression assertion for the 1633%-display bug. + """ + from amplifier_module_provider_github_copilot.sdk_adapter.event_helpers import ( + extract_usage_data, + ) + + event = { + "data": { + "input_tokens": sdk_input, + "output_tokens": sdk_out, + "cache_read_tokens": sdk_cr, + "cache_write_tokens": sdk_cw, + } + } + result = extract_usage_data(event) + + if sdk_input is None and sdk_out is None: + # Defensive: SDK emitted a Usage event with no numbers. + # The helper should either return None or a Usage with input=None. + # The streaming-ui hook coerces None to 0 (`_compute_total_input` L90-96) + # and the percentage display path is gated on `total_input > 0`, so + # no division-by-zero crash and no percentage shown -- acceptable. + if result is not None: + # If a Usage was synthesised, its input_tokens must be 0 or None, + # and cache_read must be 0 or None. + assert (result.get("input_tokens") or 0) == 0 + assert (result.get("cache_read_tokens") or 0) == 0 + return + + assert isinstance(result, dict), ( + f"shape={label}: extract_usage_data returned {type(result).__name__}, expected dict" + ) + assert result["input_tokens"] == exp_in, ( + f"shape={label}: post-transform input_tokens " + f"expected {exp_in}, got {result['input_tokens']}" + ) + assert result["total_tokens"] == exp_total, ( + f"shape={label}: total_tokens expected {exp_total}, got {result['total_tokens']}" + ) + + # The streaming-ui invariant: denominator must be >= numerator. + # Denominator = _compute_total_input = input_tokens + cache_write_tokens + # Numerator = cache_read_tokens + input_tokens = result["input_tokens"] or 0 + cache_read = result.get("cache_read_tokens") or 0 + cache_write = result.get("cache_write_tokens") or 0 + denom = input_tokens + cache_write + + assert denom >= cache_read, ( + f"shape={label}: streaming-ui denominator ({denom}) < cache_read ({cache_read}); " + "this would produce a >100% display (the 1633% regression)" + ) + if denom > 0: + pct = (cache_read / denom) * 100 + assert 0 <= pct <= 100, ( + f"shape={label}: cache percentage {pct:.2f}% outside [0, 100]" + ) + + def test_round_trip_identity_holds_for_all_captured_shapes(self) -> None: + """For every non-degenerate captured shape, the SDK-side gross input + must equal (post-transform input_tokens) + (cache_write_tokens). + + This is the identity that proves `extract_usage_data` only subtracts + cache_write -- not cache_read. + """ + from amplifier_module_provider_github_copilot.sdk_adapter.event_helpers import ( + extract_usage_data, + ) + + for label, sdk_in, sdk_cr, sdk_cw, sdk_out, _exp_in, _exp_total in ( + self.CAPTURED_SHAPES + ): + if sdk_in is None: + continue # degenerate row covered by parametrized test above + event = { + "data": { + "input_tokens": sdk_in, + "output_tokens": sdk_out, + "cache_read_tokens": sdk_cr, + "cache_write_tokens": sdk_cw, + } + } + result = extract_usage_data(event) + assert isinstance(result, dict) + it = result["input_tokens"] or 0 + cw = result.get("cache_write_tokens") or 0 + assert it + cw == sdk_in, ( + f"shape={label}: round-trip identity violated: " + f"input_tokens({it}) + cache_write({cw}) != sdk_input({sdk_in})" + ) + + def test_gpt55_no_cache_shape_displays_no_percentage(self) -> None: + """gpt-5.5 produced (cr=0, cw=0). Streaming-ui must not display a cache % + on this shape -- there is nothing to display. The numerator is 0, so any + well-formed display formula yields 0% (or suppresses entirely). + """ + from amplifier_module_provider_github_copilot.sdk_adapter.event_helpers import ( + extract_usage_data, + ) + + event = { + "data": { + "input_tokens": 53611, + "output_tokens": 58, + "cache_read_tokens": 0, + "cache_write_tokens": 0, + } + } + result = extract_usage_data(event) + assert isinstance(result, dict) + assert result["input_tokens"] == 53611, "no-cache shape: input must pass through unchanged" + assert result["cache_read_tokens"] == 0 + assert result["cache_write_tokens"] == 0 + # streaming-ui formula: pct = cr / (it + cw) = 0 / 53611 = 0.0 + denom = (result["input_tokens"] or 0) + (result.get("cache_write_tokens") or 0) + assert denom == 53611 + pct = (result.get("cache_read_tokens") or 0) / denom * 100 + assert pct == 0.0 + + def test_haiku_write_only_first_turn_shape(self) -> None: + """claude-haiku first-turn write-only: SDK input=63552 with cw=63542, cr=0. + This is the WRITE phase before any cache reads occur. Post-transform + input_tokens must collapse to the fresh portion (10). + """ + from amplifier_module_provider_github_copilot.sdk_adapter.event_helpers import ( + extract_usage_data, + ) + + event = { + "data": { + "input_tokens": 63552, + "output_tokens": 96, + "cache_read_tokens": 0, + "cache_write_tokens": 63542, + } + } + result = extract_usage_data(event) + assert isinstance(result, dict) + assert result["input_tokens"] == 10, ( + "haiku write-only first turn: post-transform input = sdk_input - cache_write" + ) + # Streaming-ui denom = 10 + 63542 = 63552; numerator (cache_read) = 0. + denom = (result["input_tokens"] or 0) + (result.get("cache_write_tokens") or 0) + assert denom == 63552 + pct = (result.get("cache_read_tokens") or 0) / denom * 100 + assert pct == 0.0 + + class TestValidateNoClassificationOverlapValidator: """Direct unit tests for every raise path in _validate_no_classification_overlap. @@ -821,7 +1174,7 @@ def test_empty_idle_events_raises_configuration_error(self, tmp_path: Path) -> N import yaml # Minimal YAML with empty idle_events — triggers the fail-fast guard - bad_yaml = { + bad_yaml: dict[str, object] = { "event_classifications": { "bridge": [], "consume": [],