usage.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/openai_compat/usage.py
 
"""
Usage-block normalisation across the OpenAI-compatible dialect variants.
 
Purpose
-------
:func:`normalise_usage` folds the ``usage`` block every lane returns into
one :class:`UsageStats`. The lanes all claim the OpenAI chat-completions
dialect, but they differ in which optional sub-blocks they populate:
Groq and Cerebras report ``completion_tokens_details.reasoning_tokens``;
OpenRouter adds a top-level ``cost`` (USD, ignored here — pricing is
``oc_pricing``'s job) and may report ``prompt_tokens_details``; Bitdeer
and NVIDIA NIM report neither and leave the optional blocks absent.
 
Responsibilities
----------------
- :func:`normalise_usage` — total, defensive dict → :class:`UsageStats`.
- :func:`_as_dict` — coerce a pydantic sub-block (or anything else) to a
  plain dict so ``isinstance(x, dict)`` checks in callers hold.
- :func:`_as_int` — coerce a reported count to a non-negative int.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Total: any input — ``None``, a non-dict, a partial block, a block with
  string counts — yields a :class:`UsageStats` rather than raising. A
  usage block is telemetry; a malformed one must never abort a turn.
- Absent fields normalise to 0, which callers read as "no data" (the
  contract's own zero-hit-zero-miss convention).
- ``total_tokens`` is derived as prompt+completion ONLY when the lane did
  not report it; a reported total is never overwritten.
- ``cache_hit_tokens`` comes from ``prompt_tokens_details.cached_tokens``
  (the OpenAI-standard prefix-cache field). The miss side is the reported
  prompt total minus the hit, floored at 0 — never negative on a lane
  that reports a cached count larger than its own prompt count.
- Money is NOT parsed here. ``usage.cost`` (OpenRouter/Vercel) is
  deliberately dropped: plan 202's fail-closed pricing path owns cost,
  and a provider-asserted figure entering by this door would bypass it.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- option_c_agent_contracts (UsageStats)
"""
from __future__ import annotations
 
from typing import Any
 
from oc_diagnostics import _dbg
 
from option_c_agent_contracts import UsageStats
 
_dbg("OC_CHAT.CLIENT", "module", "loaded", 4)
 
 
def _as_dict(value: Any) -> dict:
    """Return *value* as a plain dict, or ``{}`` when it is not dict-like."""
    if isinstance(value, dict):
        return value
    if hasattr(value, "model_dump"):
        dumped = value.model_dump()
        return dumped if isinstance(dumped, dict) else {}
    return {}
 
 
def _as_int(value: Any) -> int:
    """Return *value* as a non-negative int, or 0 when it is not a count."""
    func = "_as_int"
    if value is None or isinstance(value, bool):
        return 0
    try:
        return max(0, int(value))
    except (TypeError, ValueError) as exc:
        # VALIDATION_ERROR — a lane reported a non-numeric count; telemetry
        # degrades to zero rather than aborting the turn.
        _dbg("OC_CHAT.CLIENT", func,
             f"VALIDATION_ERROR: {value!r}: {type(exc).__name__}: {exc}", 3)
        return 0
 
 
def normalise_usage(usage: Any) -> UsageStats:
    """Fold a lane's ``usage`` block into :class:`UsageStats`.
 
    Handles the dialect variants named in the module docstring; unknown
    extra keys are ignored, so a lane that grows a field does not break
    this parser.
    """
    func = "normalise_usage"
    block = _as_dict(usage)
    if not block:
        return UsageStats()
 
    prompt = _as_int(block.get("prompt_tokens"))
    completion = _as_int(block.get("completion_tokens"))
    reported_total = _as_int(block.get("total_tokens"))
    total = reported_total or (prompt + completion)
 
    completion_details = _as_dict(block.get("completion_tokens_details"))
    prompt_details = _as_dict(block.get("prompt_tokens_details"))
 
    reasoning = _as_int(completion_details.get("reasoning_tokens"))
    cached = _as_int(prompt_details.get("cached_tokens"))
    hit = min(cached, prompt) if prompt else cached
    miss = max(0, prompt - hit)
 
    _dbg("OC_CHAT.CLIENT", func,
         f"prompt={prompt} completion={completion} total={total} "
         f"reasoning={reasoning} cache_hit={hit}", 4)
    return UsageStats(
        prompt_tokens=prompt,
        completion_tokens=completion,
        total_tokens=total,
        # Zero hit AND zero miss is the contract's "no cache data" signal, so a
        # lane with no cached count must not report the whole prompt as a miss.
        cache_hit_tokens=hit if cached else 0,
        cache_miss_tokens=miss if cached else 0,
        reasoning_tokens=reasoning,
    )
 
 
__all__ = ["normalise_usage"]