diff --git a/routstr/upstream/terminal_outcome_tracking.py b/routstr/upstream/terminal_outcome_tracking.py index 1c124ac0..f790ca43 100644 --- a/routstr/upstream/terminal_outcome_tracking.py +++ b/routstr/upstream/terminal_outcome_tracking.py @@ -187,9 +187,17 @@ def observe_terminal_sse_bytes( pending = b"" for event in events: + lines = event.split(b"\n") + # A Routstr upstream appends its own cost summary, with cached tokens + # folded into input_tokens; it is not provider usage. + if any( + line.startswith(b"event:") and line[len(b"event:") :].strip() == b"cost" + for line in lines + ): + continue data_lines = [ line[len(b"data:") :].lstrip(b" ") - for line in event.split(b"\n") + for line in lines if line.startswith(b"data:") ] if not data_lines: diff --git a/tests/unit/test_streaming_billing_finalization.py b/tests/unit/test_streaming_billing_finalization.py index a4082220..c7b4126d 100644 --- a/tests/unit/test_streaming_billing_finalization.py +++ b/tests/unit/test_streaming_billing_finalization.py @@ -1329,6 +1329,29 @@ def test_stream_cut_inside_a_character_does_not_raise() -> None: assert state.settlement_context() is None +def test_routstr_upstream_cost_event_is_not_provider_usage() -> None: + state = TerminalOutcomeState( + TerminalOutcomeContext(outcome_id="routstr-upstream", model_identifier="m") + ) + stream = ( + b'event: message_start\ndata: {"type":"message_start","message":{"usage":' + b'{"input_tokens":100,"cache_read_input_tokens":1000,"output_tokens":1}}}\n\n' + b'event: message_delta\ndata: {"type":"message_delta",' + b'"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":50}}\n\n' + b'event: message_stop\ndata: {"type":"message_stop"}\n\n' + # A Routstr upstream ends with its own cost summary, cache folded into input. + b'event: cost\ndata: {"model":"m","usage":{"input_tokens":1100,' + b'"cache_read_input_tokens":1000,"output_tokens":50}}\n\n' + ) + + assert observe_terminal_sse_bytes(state, b"", stream, final=True) == b"" + assert state.usage == { + "input_tokens": 100, + "cache_read_input_tokens": 1000, + "output_tokens": 50, + } + + @pytest.mark.asyncio async def test_cross_key_reservation_snapshot_is_rejected_without_mutation() -> None: engine = await _engine()