Blame for core/model_usage.py

# -*- coding: utf-8 -*-
"""Provider-neutral model usage telemetry for SmartPaper benchmarks.

A3-R5 intentionally keeps this module independent from product entitlement,
payment, quota, and production routing.  It normalizes provider usage metadata
and emits append-only benchmark records without retaining prompts or secrets.
"""
from __future__ import annotations

from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import threading
import uuid
from typing import Any, Mapping


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _safe_int(value: Any) -> int:
    try:
        return max(0, int(value or 0))
    except (TypeError, ValueError):
        return 0


def _safe_text(value: Any, limit: int = 240) -> str:
    text = str(value or "").replace("\r", " ").replace("\n", " ").strip()
    return text[:limit]


def sanitize_metadata(value: Any, *, depth: int = 0) -> Any:
    """Return a bounded, secret-safe JSON-compatible metadata value."""
    if depth > 4:
        return "<max-depth>"
    if value is None or isinstance(value, (bool, int, float)):
        return value
    if isinstance(value, str):
        lowered = value.lower()
        if any(marker in lowered for marker in ("bearer ", "api_key", "api-key", "x-goog-api-key")):
            return "<redacted>"
        return value[:500]
    if isinstance(value, Mapping):
        clean = {}
        for key, item in list(value.items())[:40]:
            name = str(key)
            lowered = name.lower()
            if any(marker in lowered for marker in ("key", "token", "authorization", "secret", "credential", "prompt", "manuscript")):
                clean[name] = "<redacted>"
            else:
                clean[name] = sanitize_metadata(item, depth=depth + 1)
        return clean
    if isinstance(value, (list, tuple)):
        return [sanitize_metadata(item, depth=depth + 1) for item in list(value)[:40]]
    return _safe_text(value, 500)


def prompt_identity(system_prompt: Any = "", user_prompt: Any = "") -> str:
    """Hash prompt identity without storing the prompt itself."""
    payload = (str(system_prompt or "") + "\n<SEP>\n" + str(user_prompt or "")).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


@dataclass
class ModelUsageEvent:
    provider: str
    model: str
    status: str
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    created_at: str = field(default_factory=_utc_now_iso)
    phase_id: str = ""
    case_id: str = ""
    benchmark_version: str = ""
    profile_id: str = ""
    attempt_index: int = 0
    streaming: bool = False
    thinking_mode: str = ""
    requested_thinking_budget: int | None = None
    prompt_hash: str = ""
    request_started_at: str = ""
    first_token_at: str = ""
    completed_at: str = ""
    latency_ms: int = 0
    prompt_input_tokens: int = 0
    cache_hit_tokens: int = 0
    cache_miss_tokens: int = 0
    output_tokens: int = 0
    reasoning_or_thought_tokens: int = 0
    total_tokens: int = 0
    finish_reason: str = ""
    http_status: int | None = None
    error_class: str = ""
    retry_reason: str = ""
    escalation_reason: str = ""
    provider_service_tier: str = ""
    provider_model_version: str = ""
    provider_response_id: str = ""
    price_snapshot_id: str = ""
    estimated_cost_microunits: int = 0
    currency: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        payload = asdict(self)
        payload["metadata"] = sanitize_metadata(payload.get("metadata") or {})
        return payload


class NoOpModelUsageSink:
    def emit(self, event: ModelUsageEvent | Mapping[str, Any]) -> None:
        return None


class JsonlModelUsageSink:
    """Thread-safe append-only JSONL sink for benchmark artifacts."""

    def __init__(self, path: str | Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._lock = threading.Lock()

    def emit(self, event: ModelUsageEvent | Mapping[str, Any]) -> None:
        if isinstance(event, ModelUsageEvent):
            row = event.to_dict()
        else:
            row = sanitize_metadata(dict(event or {}))
        line = json.dumps(row, ensure_ascii=False, sort_keys=True)
        with self._lock:
            with self.path.open("a", encoding="utf-8") as handle:
                handle.write(line + "\n")


def emit_usage_event(sink: Any, event: ModelUsageEvent) -> None:
    """Best-effort telemetry emission that can never fail model generation."""
    if sink is None:
        return
    try:
        if hasattr(sink, "emit") and callable(sink.emit):
            sink.emit(event)
        elif callable(sink):
            sink(event)
    except Exception:
        # Telemetry is strictly observational in A3-R5.
        return


def normalize_deepseek_usage(raw: Mapping[str, Any] | None) -> dict[str, int]:
    usage = dict(raw or {})
    details = usage.get("completion_tokens_details")
    if not isinstance(details, Mapping):
        details = {}
    return {
        "prompt_input_tokens": _safe_int(usage.get("prompt_tokens")),
        "cache_hit_tokens": _safe_int(usage.get("prompt_cache_hit_tokens")),
        "cache_miss_tokens": _safe_int(usage.get("prompt_cache_miss_tokens")),
        "output_tokens": _safe_int(usage.get("completion_tokens")),
        "reasoning_or_thought_tokens": _safe_int(details.get("reasoning_tokens")),
        "total_tokens": _safe_int(usage.get("total_tokens")),
    }


def normalize_gemini_usage(raw: Mapping[str, Any] | None) -> dict[str, int | str]:
    usage = dict(raw or {})
    return {
        "prompt_input_tokens": _safe_int(usage.get("promptTokenCount")),
        "cache_hit_tokens": _safe_int(usage.get("cachedContentTokenCount")),
        "cache_miss_tokens": max(
            0,
            _safe_int(usage.get("promptTokenCount")) - _safe_int(usage.get("cachedContentTokenCount")),
        ),
        "output_tokens": _safe_int(usage.get("candidatesTokenCount")),
        "reasoning_or_thought_tokens": _safe_int(usage.get("thoughtsTokenCount")),
        "total_tokens": _safe_int(usage.get("totalTokenCount")),
        "provider_service_tier": _safe_text(usage.get("trafficType") or usage.get("serviceTier"), 80),
    }

def normalize_zhipu_usage(raw: Mapping[str, Any] | None) -> dict[str, int]:
    """Normalize Zhipu/GLM Chat Completions usage fields.

    Official GLM responses expose prompt/completion/total token counts and
    prompt_tokens_details.cached_tokens.  reasoning_tokens is consumed only
    when the provider actually returns it; it is never inferred from
    reasoning_content text.
    """
    usage = dict(raw or {})
    prompt_details = usage.get("prompt_tokens_details")
    if not isinstance(prompt_details, Mapping):
        prompt_details = {}
    completion_details = usage.get("completion_tokens_details")
    if not isinstance(completion_details, Mapping):
        completion_details = {}
    prompt_tokens = _safe_int(usage.get("prompt_tokens"))
    cached_tokens = _safe_int(prompt_details.get("cached_tokens"))
    return {
        "prompt_input_tokens": prompt_tokens,
        "cache_hit_tokens": cached_tokens,
        "cache_miss_tokens": max(0, prompt_tokens - cached_tokens),
        "output_tokens": _safe_int(usage.get("completion_tokens")),
        "reasoning_or_thought_tokens": _safe_int(completion_details.get("reasoning_tokens")),
        "total_tokens": _safe_int(usage.get("total_tokens")),
    }