lean_ctx/proxy/
usage.rs

1//! Real provider-reported token usage extraction.
2//!
3//! The proxy already rewrites requests; this module reads the *response* so the
4//! dashboard/terminal can show **measured** cost (the user's real provider bill)
5//! instead of an estimate. All three providers report the exact model and the
6//! billed token breakdown — including prompt-cache reads/writes — in the final
7//! event of a stream (or the body of a non-streaming response):
8//!
9//! - Anthropic: `message_start` carries model + input/cache tokens, the final
10//!   `message_delta` carries `output_tokens`. Non-streaming: one `usage` object.
11//! - OpenAI Responses: the `response.completed` event nests `response.usage`.
12//! - OpenAI Chat Completions: the final chunk carries `usage` (needs
13//!   `stream_options.include_usage`, which the proxy injects).
14//! - Gemini: every chunk carries `usageMetadata`; the last one has the totals.
15//!
16//! [`RealUsage`] normalizes every provider onto the four billable buckets that
17//! [`crate::core::gain::model_pricing::ModelCost::estimate_usd`] prices:
18//! uncached input, output (incl. reasoning/thoughts), cache-read, cache-write.
19
20use futures::{Stream, StreamExt};
21use serde_json::Value;
22
23/// LLM provider whose response shape a [`Scanner`] understands.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum Provider {
26    Anthropic,
27    /// Covers both Chat Completions and the Responses API (same `usage` dialects
28    /// are detected by field name).
29    OpenAi,
30    Gemini,
31}
32
33impl Provider {
34    /// Maps the proxy's `provider_label` (`"Anthropic"`, `"OpenAI"`/`"ChatGPT"`, else Gemini).
35    pub fn from_label(label: &str) -> Self {
36        match label {
37            "Anthropic" => Self::Anthropic,
38            "OpenAI" | "ChatGPT" => Self::OpenAi,
39            _ => Self::Gemini,
40        }
41    }
42}
43
44/// One LLM turn's real, provider-reported usage, normalized to billable buckets.
45///
46/// `output_tokens` already includes reasoning/thinking tokens (they are billed at
47/// the output rate); `reasoning_tokens` is retained only for display.
48#[derive(Debug, Clone, Default, PartialEq)]
49pub struct RealUsage {
50    pub model: String,
51    /// Input tokens billed at the input rate (cache reads/writes excluded).
52    pub input_tokens: u64,
53    /// Output tokens billed at the output rate (includes reasoning/thoughts).
54    pub output_tokens: u64,
55    /// Input tokens served from the prompt cache (billed at the cache-read rate).
56    pub cache_read_tokens: u64,
57    /// Input tokens written to the prompt cache (Anthropic + OpenRouter models
58    /// with explicit cache-write pricing; 0 elsewhere).
59    pub cache_write_tokens: u64,
60    /// Reasoning/thinking subset of `output_tokens` (display only).
61    pub reasoning_tokens: u64,
62    /// USD the provider *actually charged* for this turn, when the response
63    /// reports it (OpenRouter: `usage.cost` in credits ≡ USD, plus the BYOK
64    /// upstream share from `cost_details.upstream_inference_cost`). The
65    /// measured figure beats any price-table estimate wherever both exist.
66    /// `None` for providers that report tokens only (Anthropic/OpenAI/Gemini).
67    pub provider_cost_usd: Option<f64>,
68    /// Output-savings experiment arm for this turn (#895 Track B), or `None` when
69    /// no holdout is active. Stamped from the request, not parsed from the
70    /// response — it identifies whether this turn was output-shaped.
71    pub cohort: Option<super::holdout::Arm>,
72    /// Request-side gateway context (enterprise#11/#17/#18): identity tags,
73    /// compression savings and baseline inputs, stamped from the request before
74    /// it left for the upstream. `None` outside the forward path (e.g. tests
75    /// that only parse response bodies).
76    pub wire: Option<Box<WireContext>>,
77}
78
79/// Request-side context the forward path knows and the response scanner does
80/// not: who sent the request (gateway identity, enterprise#11), what the proxy
81/// saved on the wire, and the counterfactual-baseline inputs (enterprise#18).
82/// Travels inside [`RealUsage`] so `usage_meter::record` stays the single
83/// choke-point through which every measured turn flows.
84#[derive(Debug, Clone, Default, PartialEq)]
85pub struct WireContext {
86    /// Provider label (`Anthropic|OpenAI|ChatGPT|Gemini`) of the serving route.
87    pub provider: String,
88    /// Person tag from the gateway key (enterprise#11).
89    pub person: Option<String>,
90    /// Team tag from the gateway key.
91    pub team: Option<String>,
92    /// Project: `x-leanctx-project` header wins over the key's default project.
93    pub project: Option<String>,
94    /// Estimated tokens this request saved through wire compression
95    /// (bytes/4 heuristic, same basis as the proxy stats).
96    pub saved_tokens: u64,
97    /// Estimated request tokens BEFORE lean-ctx compression (bytes/4) — the
98    /// SEE-attribution input of the avoided-cost baseline (enterprise#18).
99    pub uncompressed_input_tokens: u64,
100    /// True when the serving upstream is a local/loopback endpoint — billed via
101    /// the transparent `local_shadow_rate`, never 0 (enterprise#15/#18).
102    pub is_local: bool,
103    /// Originally requested model when the router downgraded/aliased it
104    /// (enterprise#13); `None` for passthrough.
105    pub routed_from: Option<String>,
106    /// Provider-counted input tokens of the ORIGINAL (uncompressed) request,
107    /// from the free `count_tokens` probe (#701). `None` when metering is off
108    /// or no probe was spawned; an empty slot at read time (probe failed or
109    /// still in flight) degrades the row to the local estimate.
110    pub counterfactual: Option<super::counterfactual::CounterfactualSlot>,
111}
112
113impl RealUsage {
114    /// True once any model, token or measured-cost field has been observed —
115    /// the gate for recording. Avoids emitting empty rows for streams that
116    /// never reported usage.
117    fn is_meaningful(&self) -> bool {
118        !self.model.is_empty()
119            || self.input_tokens > 0
120            || self.output_tokens > 0
121            || self.cache_read_tokens > 0
122            || self.cache_write_tokens > 0
123            || self.provider_cost_usd.is_some()
124    }
125}
126
127/// Upper bound on a single buffered line before we give up on it. Usage events
128/// are tiny; this only guards against a pathological newline-free stream.
129const MAX_LINE_BYTES: usize = 1 << 20; // 1 MiB
130
131/// Incrementally extracts [`RealUsage`] from a response stream (or a full body).
132///
133/// `feed` is called with raw response chunks and keeps only the trailing partial
134/// line buffered (O(1) memory beyond one line); `finalize` returns the merged
135/// usage once the stream ends.
136pub struct Scanner {
137    provider: Provider,
138    /// Model parsed from the request URL (Gemini puts it there, not in the body).
139    url_model: Option<String>,
140    /// Output-savings arm (#895), stamped onto the usage at finalize.
141    cohort: Option<super::holdout::Arm>,
142    /// Request-side gateway context (enterprise#11/#18), stamped at finalize.
143    wire: Option<Box<WireContext>>,
144    /// Billed USD from a gateway response header (#1189), stamped at finalize
145    /// unless the body already reported the charge.
146    header_cost: Option<f64>,
147    buf: Vec<u8>,
148    usage: RealUsage,
149}
150
151impl Scanner {
152    pub fn new(provider: Provider, url_model: Option<String>) -> Self {
153        Self {
154            provider,
155            url_model,
156            cohort: None,
157            wire: None,
158            header_cost: None,
159            buf: Vec::new(),
160            usage: RealUsage::default(),
161        }
162    }
163
164    /// Tags the usage this scanner produces with an output-savings arm (#895).
165    #[must_use]
166    pub fn with_cohort(mut self, cohort: Option<super::holdout::Arm>) -> Self {
167        self.cohort = cohort;
168        self
169    }
170
171    /// Attaches the request-side gateway context (identity tags, wire savings,
172    /// baseline inputs — enterprise#11/#17/#18) stamped onto the usage record.
173    #[must_use]
174    pub fn with_wire_context(mut self, wire: Option<Box<WireContext>>) -> Self {
175        self.wire = wire;
176        self
177    }
178
179    /// Attaches a billed USD figure reported by the upstream via a response
180    /// header (LiteLLM `x-litellm-response-cost`, or the operator-configured
181    /// `[proxy] cost_response_header`, #1189). Applied at finalize only when
182    /// the body did not already carry a measured cost — a body figure
183    /// (OpenRouter `usage.cost`) is the bill itself and always wins.
184    #[must_use]
185    pub fn with_header_cost(mut self, cost: Option<f64>) -> Self {
186        self.header_cost = cost.filter(|c| c.is_finite() && *c >= 0.0);
187        self
188    }
189
190    /// Feeds a raw streaming chunk, scanning every complete line it completes.
191    pub fn feed(&mut self, chunk: &[u8]) {
192        self.buf.extend_from_slice(chunk);
193        while let Some(nl) = self.buf.iter().position(|&b| b == b'\n') {
194            let mut line: Vec<u8> = self.buf.drain(..=nl).collect();
195            line.pop(); // drop '\n'
196            if line.last() == Some(&b'\r') {
197                line.pop();
198            }
199            self.scan_line(&line);
200        }
201        if self.buf.len() > MAX_LINE_BYTES {
202            self.buf.clear();
203        }
204    }
205
206    /// Feeds a complete non-streaming JSON response body.
207    pub fn feed_body(&mut self, body: &[u8]) {
208        if let Ok(v) = serde_json::from_slice::<Value>(body) {
209            self.absorb(&v);
210        }
211    }
212
213    /// Consumes the scanner, flushing any trailing partial line (a final event
214    /// may arrive without a newline) and returning the merged usage if any.
215    pub fn finalize(mut self) -> Option<RealUsage> {
216        if !self.buf.is_empty() {
217            let line = std::mem::take(&mut self.buf);
218            self.scan_line(&line);
219        }
220        // Gateway header cost (#1189): measured, but the body figure is the
221        // bill itself (OpenRouter usage.cost) and keeps priority when present.
222        if self.usage.provider_cost_usd.is_none() {
223            self.usage.provider_cost_usd = self.header_cost;
224        }
225        if self.usage.is_meaningful() {
226            self.usage.cohort = self.cohort;
227            self.usage.wire = self.wire;
228            Some(self.usage)
229        } else {
230            None
231        }
232    }
233
234    fn scan_line(&mut self, line: &[u8]) {
235        let Ok(text) = std::str::from_utf8(line) else {
236            return;
237        };
238        let trimmed = text.trim();
239        if trimmed.is_empty() {
240            return;
241        }
242        // Cheap pre-filter: skip the bulk of the stream (content deltas) and only
243        // JSON-parse lines that can carry usage or the model name.
244        if !self.line_might_be_relevant(trimmed) {
245            return;
246        }
247        let json_str = if let Some(rest) = trimmed.strip_prefix("data:") {
248            // SSE (Anthropic, OpenAI, Gemini with alt=sse).
249            let r = rest.trim();
250            if r.is_empty() || r == "[DONE]" {
251                return;
252            }
253            r
254        } else if trimmed.starts_with('{') {
255            // NDJSON / array-element line (Gemini x-ndjson). Tolerate the array
256            // punctuation a streamed JSON array puts around an element.
257            trimmed
258                .trim_start_matches([',', '['])
259                .trim_end_matches([',', ']'])
260                .trim()
261        } else {
262            return;
263        };
264        if let Ok(v) = serde_json::from_str::<Value>(json_str) {
265            self.absorb(&v);
266        }
267    }
268
269    fn line_might_be_relevant(&self, s: &str) -> bool {
270        match self.provider {
271            // Anthropic `message_start`/`message_delta` and OpenAI `usage`/
272            // `response.*` events all contain the substring "usage".
273            Provider::Anthropic | Provider::OpenAi => s.contains("usage"),
274            Provider::Gemini => s.contains("usageMetadata"),
275        }
276    }
277
278    fn absorb(&mut self, v: &Value) {
279        match self.provider {
280            Provider::Anthropic => absorb_anthropic(&mut self.usage, v),
281            Provider::OpenAi => absorb_openai(&mut self.usage, v),
282            Provider::Gemini => absorb_gemini(&mut self.usage, v, self.url_model.as_deref()),
283        }
284    }
285}
286
287/// Anthropic: model + input/cache live on `message` (streaming `message_start`
288/// or a non-streaming body); `output_tokens` arrives later on the event-level
289/// `usage` of `message_delta`. Latest non-zero wins, so the cumulative final
290/// delta is authoritative.
291fn absorb_anthropic(u: &mut RealUsage, v: &Value) {
292    let msg = v.get("message").unwrap_or(v);
293    if let Some(model) = msg.get("model").and_then(Value::as_str)
294        && !model.is_empty()
295    {
296        u.model = model.to_string();
297    }
298    let Some(usage) = msg.get("usage").or_else(|| v.get("usage")) else {
299        return;
300    };
301    if let Some(n) = usage.get("input_tokens").and_then(Value::as_u64) {
302        u.input_tokens = n;
303    }
304    if let Some(n) = usage.get("cache_read_input_tokens").and_then(Value::as_u64) {
305        u.cache_read_tokens = n;
306    }
307    if let Some(n) = usage
308        .get("cache_creation_input_tokens")
309        .and_then(Value::as_u64)
310    {
311        u.cache_write_tokens = n;
312    }
313    if let Some(n) = usage.get("output_tokens").and_then(Value::as_u64)
314        && n > 0
315    {
316        u.output_tokens = n;
317    }
318}
319
320/// OpenAI Chat Completions + Responses. `response.completed` nests the payload
321/// under `response`; chat chunks and non-streaming bodies are top-level. Both
322/// `usage` dialects are accepted (Responses: `input_tokens`/`output_tokens`;
323/// Chat: `prompt_tokens`/`completion_tokens`). `cached_tokens` is the cache-read
324/// portion of the reported input; OpenAI bills no separate cache write, but
325/// OpenRouter reports one (`prompt_tokens_details.cache_write_tokens`) for
326/// models with explicit cache-write pricing.
327///
328/// OpenRouter usage accounting additionally carries the money actually charged:
329/// `usage.cost` (credits ≡ USD) and, for BYOK requests, the upstream provider's
330/// own bill under `cost_details.upstream_inference_cost`. Their sum is this
331/// turn's real price — measured, not table-derived.
332fn absorb_openai(u: &mut RealUsage, v: &Value) {
333    let root = v.get("response").unwrap_or(v);
334    if let Some(model) = root.get("model").and_then(Value::as_str)
335        && !model.is_empty()
336    {
337        u.model = model.to_string();
338    }
339    let Some(usage) = root.get("usage") else {
340        return;
341    };
342    if usage.is_null() {
343        // `response.created` / `response.in_progress` carry `usage: null`.
344        return;
345    }
346
347    // Measured cost (OpenRouter dialect). Parsed before the token guard so a
348    // cost-bearing usage object is never lost, and `0` is preserved — a
349    // `:free` model's real price IS zero, not "unknown".
350    if let Some(cost) = usage.get("cost").and_then(Value::as_f64) {
351        let upstream = usage
352            .get("cost_details")
353            .and_then(|d| d.get("upstream_inference_cost"))
354            .and_then(Value::as_f64)
355            .unwrap_or(0.0);
356        u.provider_cost_usd = Some(cost + upstream.max(0.0));
357    }
358
359    let total_input = usage
360        .get("input_tokens")
361        .or_else(|| usage.get("prompt_tokens"))
362        .and_then(Value::as_u64)
363        .unwrap_or(0);
364    let total_output = usage
365        .get("output_tokens")
366        .or_else(|| usage.get("completion_tokens"))
367        .and_then(Value::as_u64)
368        .unwrap_or(0);
369    let input_details = usage
370        .get("input_tokens_details")
371        .or_else(|| usage.get("prompt_tokens_details"));
372    let cached = input_details
373        .and_then(|d| d.get("cached_tokens"))
374        .and_then(Value::as_u64)
375        .unwrap_or(0);
376    let cache_write = input_details
377        .and_then(|d| d.get("cache_write_tokens"))
378        .and_then(Value::as_u64)
379        .unwrap_or(0);
380    let reasoning = usage
381        .get("output_tokens_details")
382        .or_else(|| usage.get("completion_tokens_details"))
383        .and_then(|d| d.get("reasoning_tokens"))
384        .and_then(Value::as_u64)
385        .unwrap_or(0);
386
387    if total_input == 0 && total_output == 0 {
388        return;
389    }
390    // OpenRouter counts cache writes inside prompt_tokens (unlike Anthropic's
391    // separate bucket) — subtract both cached reads and writes so the three
392    // buckets stay disjoint and are never double-priced.
393    u.input_tokens = total_input
394        .saturating_sub(cached)
395        .saturating_sub(cache_write);
396    u.cache_read_tokens = cached;
397    u.cache_write_tokens = cache_write;
398    u.output_tokens = total_output;
399    u.reasoning_tokens = reasoning;
400}
401
402/// Gemini: `usageMetadata` carries the counts; `modelVersion` (or the request
403/// URL) the model. `thoughtsTokenCount` is billed at the output rate and is not
404/// part of `candidatesTokenCount`, so it is added into `output_tokens`.
405fn absorb_gemini(u: &mut RealUsage, v: &Value, url_model: Option<&str>) {
406    if let Some(mv) = v.get("modelVersion").and_then(Value::as_str)
407        && !mv.is_empty()
408    {
409        u.model = mv.to_string();
410    } else if u.model.is_empty()
411        && let Some(m) = url_model
412        && !m.is_empty()
413    {
414        u.model = m.to_string();
415    }
416    let Some(um) = v.get("usageMetadata") else {
417        return;
418    };
419    let prompt = um
420        .get("promptTokenCount")
421        .and_then(Value::as_u64)
422        .unwrap_or(0);
423    let candidates = um
424        .get("candidatesTokenCount")
425        .and_then(Value::as_u64)
426        .unwrap_or(0);
427    let cached = um
428        .get("cachedContentTokenCount")
429        .and_then(Value::as_u64)
430        .unwrap_or(0);
431    let thoughts = um
432        .get("thoughtsTokenCount")
433        .and_then(Value::as_u64)
434        .unwrap_or(0);
435    if prompt == 0 && candidates == 0 && thoughts == 0 {
436        return;
437    }
438    u.input_tokens = prompt.saturating_sub(cached);
439    u.cache_read_tokens = cached;
440    u.cache_write_tokens = 0;
441    u.output_tokens = candidates + thoughts;
442    u.reasoning_tokens = thoughts;
443}
444
445/// Extracts the model from a Gemini request path
446/// (`/v1beta/models/{model}:generateContent`). Returns `None` for other paths.
447pub fn gemini_model_from_path(path: &str) -> Option<String> {
448    let after = path.rsplit_once("/models/").map(|(_, m)| m)?;
449    let model = after.split(':').next().unwrap_or(after).trim();
450    if model.is_empty() {
451        None
452    } else {
453        Some(model.to_string())
454    }
455}
456
457/// Wraps a response byte stream so every chunk is forwarded **byte-for-byte**
458/// while a [`Scanner`] observes it; on stream end the merged usage is recorded.
459/// Memory overhead is one buffered SSE line, never the whole response.
460pub fn tee_stream<S, B, E>(
461    inner: S,
462    scanner: Scanner,
463) -> impl Stream<Item = Result<B, E>> + Send + 'static
464where
465    S: Stream<Item = Result<B, E>> + Send + Unpin + 'static,
466    B: AsRef<[u8]> + Send + 'static,
467    E: Send + 'static,
468{
469    futures::stream::unfold(
470        (inner, Some(scanner)),
471        |(mut inner, mut scanner)| async move {
472            match inner.next().await {
473                Some(Ok(chunk)) => {
474                    if let Some(s) = scanner.as_mut() {
475                        s.feed(chunk.as_ref());
476                    }
477                    Some((Ok(chunk), (inner, scanner)))
478                }
479                Some(err) => Some((err, (inner, scanner))),
480                None => {
481                    if let Some(s) = scanner.take()
482                        && let Some(usage) = s.finalize()
483                    {
484                        super::usage_meter::record(&usage);
485                    }
486                    None
487                }
488            }
489        },
490    )
491}
492
493#[cfg(test)]
494mod tests {
495    use super::*;
496
497    fn feed_lines(
498        provider: Provider,
499        url_model: Option<&str>,
500        lines: &[&str],
501    ) -> Option<RealUsage> {
502        let mut s = Scanner::new(provider, url_model.map(str::to_string));
503        for line in lines {
504            s.feed(line.as_bytes());
505            s.feed(b"\n");
506        }
507        s.finalize()
508    }
509
510    #[test]
511    fn anthropic_merges_message_start_and_delta() {
512        let u = feed_lines(
513            Provider::Anthropic,
514            None,
515            &[
516                r#"data: {"type":"message_start","message":{"model":"claude-opus-4-5-20251101","usage":{"input_tokens":100,"cache_read_input_tokens":2000,"cache_creation_input_tokens":50,"output_tokens":1}}}"#,
517                r#"data: {"type":"content_block_delta","index":0,"delta":{"text":"hello"}}"#,
518                r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":73}}"#,
519                "data: {\"type\":\"message_stop\"}",
520            ],
521        )
522        .expect("usage");
523        assert_eq!(u.model, "claude-opus-4-5-20251101");
524        assert_eq!(u.input_tokens, 100);
525        assert_eq!(u.cache_read_tokens, 2000);
526        assert_eq!(u.cache_write_tokens, 50);
527        assert_eq!(u.output_tokens, 73);
528    }
529
530    #[test]
531    fn anthropic_non_streaming_body() {
532        let mut s = Scanner::new(Provider::Anthropic, None);
533        s.feed_body(
534            br#"{"model":"claude-sonnet-4-5","usage":{"input_tokens":24,"output_tokens":18,"cache_creation_input_tokens":0,"cache_read_input_tokens":0}}"#,
535        );
536        let u = s.finalize().expect("usage");
537        assert_eq!(u.model, "claude-sonnet-4-5");
538        assert_eq!(u.input_tokens, 24);
539        assert_eq!(u.output_tokens, 18);
540    }
541
542    #[test]
543    fn scanner_stamps_wire_context_onto_usage() {
544        // enterprise#11/#17: the request-side context must survive scanning and
545        // arrive on the finalized record that usage_meter/store consume.
546        let wire = Box::new(WireContext {
547            provider: "Anthropic".into(),
548            person: Some("yves".into()),
549            team: None,
550            project: Some("billing".into()),
551            saved_tokens: 42,
552            uncompressed_input_tokens: 500,
553            is_local: false,
554            routed_from: None,
555            counterfactual: None,
556        });
557        let mut s = Scanner::new(Provider::Anthropic, None).with_wire_context(Some(wire.clone()));
558        s.feed_body(
559            br#"{"model":"claude-sonnet-4-5","usage":{"input_tokens":24,"output_tokens":18}}"#,
560        );
561        let u = s.finalize().expect("usage");
562        assert_eq!(u.wire, Some(wire));
563    }
564
565    #[test]
566    fn openai_responses_completed_event() {
567        let u = feed_lines(
568            Provider::OpenAi,
569            None,
570            &[
571                r#"data: {"type":"response.created","response":{"model":"gpt-5.4","usage":null}}"#,
572                r#"data: {"type":"response.completed","response":{"model":"gpt-5.4","usage":{"input_tokens":1289,"input_tokens_details":{"cached_tokens":289},"output_tokens":685,"output_tokens_details":{"reasoning_tokens":640},"total_tokens":1974}}}"#,
573            ],
574        )
575        .expect("usage");
576        assert_eq!(u.model, "gpt-5.4");
577        assert_eq!(u.input_tokens, 1000); // 1289 - 289 cached
578        assert_eq!(u.cache_read_tokens, 289);
579        assert_eq!(u.output_tokens, 685);
580        assert_eq!(u.reasoning_tokens, 640);
581    }
582
583    #[test]
584    fn openai_chat_final_usage_chunk() {
585        let u = feed_lines(
586            Provider::OpenAi,
587            None,
588            &[
589                r#"data: {"choices":[{"delta":{"content":"hi"}}],"model":"gpt-5.4-mini"}"#,
590                r#"data: {"choices":[],"model":"gpt-5.4-mini","usage":{"prompt_tokens":500,"prompt_tokens_details":{"cached_tokens":100},"completion_tokens":40,"total_tokens":540}}"#,
591                "data: [DONE]",
592            ],
593        )
594        .expect("usage");
595        assert_eq!(u.model, "gpt-5.4-mini");
596        assert_eq!(u.input_tokens, 400);
597        assert_eq!(u.cache_read_tokens, 100);
598        assert_eq!(u.output_tokens, 40);
599        assert_eq!(u.provider_cost_usd, None, "OpenAI reports no usage.cost");
600    }
601
602    #[test]
603    fn openrouter_cost_and_cache_writes_are_measured() {
604        // #1179: OpenRouter usage accounting — the final streamed chunk carries
605        // the billed USD (`cost`) and the cache-write bucket inside
606        // prompt_tokens_details. All three token buckets must stay disjoint.
607        let u = feed_lines(
608            Provider::OpenAi,
609            None,
610            &[
611                r#"data: {"choices":[{"delta":{"content":"hi"}}],"model":"deepseek/deepseek-v4-flash-20260423"}"#,
612                r#"data: {"choices":[],"model":"deepseek/deepseek-v4-flash-20260423","usage":{"prompt_tokens":700,"prompt_tokens_details":{"cached_tokens":150,"cache_write_tokens":50},"completion_tokens":40,"cost":0.0123,"cost_details":{"upstream_inference_cost":null},"total_tokens":740}}"#,
613                "data: [DONE]",
614            ],
615        )
616        .expect("usage");
617        assert_eq!(u.input_tokens, 500, "700 - 150 cached - 50 cache-write");
618        assert_eq!(u.cache_read_tokens, 150);
619        assert_eq!(u.cache_write_tokens, 50);
620        assert_eq!(u.output_tokens, 40);
621        let cost = u.provider_cost_usd.expect("measured cost");
622        assert!((cost - 0.0123).abs() < 1e-12);
623    }
624
625    #[test]
626    fn openrouter_byok_adds_upstream_inference_cost() {
627        let mut s = Scanner::new(Provider::OpenAi, None);
628        s.feed_body(
629            br#"{"model":"anthropic/claude-sonnet-5","usage":{"prompt_tokens":100,"completion_tokens":10,"cost":0.05,"cost_details":{"upstream_inference_cost":0.95}}}"#,
630        );
631        let u = s.finalize().expect("usage");
632        let cost = u.provider_cost_usd.expect("measured cost");
633        assert!(
634            (cost - 1.0).abs() < 1e-12,
635            "OpenRouter fee + BYOK upstream bill"
636        );
637    }
638
639    #[test]
640    fn gateway_header_cost_fills_when_body_has_none() {
641        // #1189: a LiteLLM-style gateway reports the bill in a header; tokens
642        // come from a normal OpenAI-shaped body without `usage.cost`.
643        let mut s = Scanner::new(Provider::OpenAi, None).with_header_cost(Some(0.0042));
644        s.feed_body(
645            br#"{"model":"azure-gpt-4o","usage":{"prompt_tokens":100,"completion_tokens":10}}"#,
646        );
647        let u = s.finalize().expect("usage");
648        assert_eq!(u.provider_cost_usd, Some(0.0042), "header is measured");
649    }
650
651    #[test]
652    fn body_reported_cost_beats_the_header_figure() {
653        // OpenRouter behind another gateway: `usage.cost` is the bill itself.
654        let mut s = Scanner::new(Provider::OpenAi, None).with_header_cost(Some(9.99));
655        s.feed_body(
656            br#"{"model":"deepseek/deepseek-v4-flash","usage":{"prompt_tokens":100,"completion_tokens":10,"cost":0.05}}"#,
657        );
658        let u = s.finalize().expect("usage");
659        assert_eq!(u.provider_cost_usd, Some(0.05), "body wins over header");
660    }
661
662    #[test]
663    fn openrouter_free_model_reports_zero_cost_as_measured() {
664        let mut s = Scanner::new(Provider::OpenAi, None);
665        s.feed_body(
666            br#"{"model":"poolside/laguna-xs-2.1:free","usage":{"prompt_tokens":80,"completion_tokens":20,"cost":0}}"#,
667        );
668        let u = s.finalize().expect("usage");
669        assert_eq!(u.provider_cost_usd, Some(0.0), "free is a price, not a gap");
670    }
671
672    #[test]
673    fn gemini_usage_metadata_with_url_model() {
674        let u = feed_lines(
675            Provider::Gemini,
676            Some("gemini-2.5-pro"),
677            &[
678                r#"data: {"candidates":[{"content":{"parts":[{"text":"hi"}]}}],"usageMetadata":{"promptTokenCount":25,"candidatesTokenCount":7,"thoughtsTokenCount":39,"totalTokenCount":71}}"#,
679            ],
680        )
681        .expect("usage");
682        assert_eq!(u.model, "gemini-2.5-pro");
683        assert_eq!(u.input_tokens, 25);
684        assert_eq!(u.output_tokens, 46); // candidates 7 + thoughts 39
685        assert_eq!(u.reasoning_tokens, 39);
686    }
687
688    #[test]
689    fn gemini_prefers_model_version_over_url() {
690        let u = feed_lines(
691            Provider::Gemini,
692            Some("gemini-2.5-pro"),
693            &[
694                r#"data: {"modelVersion":"gemini-2.5-pro-002","usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":5,"cachedContentTokenCount":4,"totalTokenCount":15}}"#,
695            ],
696        )
697        .expect("usage");
698        assert_eq!(u.model, "gemini-2.5-pro-002");
699        assert_eq!(u.input_tokens, 6); // 10 - 4 cached
700        assert_eq!(u.cache_read_tokens, 4);
701        assert_eq!(u.output_tokens, 5);
702    }
703
704    #[test]
705    fn split_chunks_reassemble_across_feed_calls() {
706        let mut s = Scanner::new(Provider::Anthropic, None);
707        let line = r#"data: {"type":"message_delta","delta":{},"usage":{"output_tokens":42}}"#;
708        let bytes = format!("{line}\n");
709        let (a, b) = bytes.as_bytes().split_at(20);
710        s.feed(a);
711        s.feed(b);
712        let u = s.finalize().expect("usage");
713        assert_eq!(u.output_tokens, 42);
714    }
715
716    #[test]
717    fn no_usage_yields_none() {
718        let out = feed_lines(
719            Provider::OpenAi,
720            None,
721            &[r#"data: {"choices":[{"delta":{"content":"hi"}}],"model":"gpt-5.4"}"#],
722        );
723        assert!(out.is_none(), "content-only stream reports no usage");
724    }
725
726    #[test]
727    fn final_event_without_trailing_newline() {
728        let mut s = Scanner::new(Provider::Anthropic, None);
729        s.feed(br#"data: {"type":"message_delta","delta":{},"usage":{"output_tokens":7}}"#);
730        let u = s.finalize().expect("flushes trailing partial line");
731        assert_eq!(u.output_tokens, 7);
732    }
733
734    #[test]
735    fn gemini_model_from_path_extracts() {
736        assert_eq!(
737            gemini_model_from_path("/v1beta/models/gemini-2.5-pro:streamGenerateContent")
738                .as_deref(),
739            Some("gemini-2.5-pro")
740        );
741        assert_eq!(
742            gemini_model_from_path("/v1beta/models/gemini-2.5-flash:generateContent").as_deref(),
743            Some("gemini-2.5-flash")
744        );
745        assert_eq!(gemini_model_from_path("/v1/chat/completions"), None);
746    }
747
748    #[tokio::test]
749    async fn tee_stream_passes_bytes_through_and_records() {
750        let chunks: Vec<Result<Vec<u8>, std::convert::Infallible>> = vec![
751            Ok(b"data: {\"type\":\"message_start\",\"message\":{\"model\":\"claude-haiku-4.5\",\"usage\":{\"input_tokens\":5,\"output_tokens\":1}}}\n".to_vec()),
752            Ok(b"data: {\"type\":\"message_delta\",\"delta\":{},\"usage\":{\"output_tokens\":9}}\n".to_vec()),
753        ];
754        let inner = futures::stream::iter(chunks);
755        let scanner = Scanner::new(Provider::Anthropic, None);
756        let teed = tee_stream(inner, scanner);
757        let collected: Vec<_> = teed.collect().await;
758        // Byte-for-byte passthrough preserved.
759        assert_eq!(collected.len(), 2);
760        assert!(collected.iter().all(std::result::Result::is_ok));
761    }
762}
lean_ctx/proxy/usage.rs

lean_ctx/proxy/
usage.rs
