# -*- 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 "" 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 "" 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] = "" 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\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")), }