"""Cost-/Usage-Tracking client for the central Onyva Dev Login. Reports finished EUR amounts for LLM calls and audio STT to the central ``/api/usage`` endpoint (see ``docs/cost.md``). Design goals: * **Buffered + batched** — events are collected in memory and flushed by a background thread, not one HTTP request per event. * **Never blocks the user request** — ``track_usage`` only appends to a buffer; the network call happens on the daemon thread. All errors are swallowed (tracking is best effort and must never break the app flow). * **Idempotent** — callers pass a stable ``idempotency_key`` so retries / repeated status polls never double-count. The ingest key (``usk_…``) comes from a **per-tenant Gitea secret** ``COST_INGEST_KEY_`` and reaches the container as ``COST_INGEST_KEY`` — kept separate from the existing ``TENANT_SECRETS_`` blob (see ``infra/deploy.py``). """ from __future__ import annotations import os import threading import time import requests # ── config ─────────────────────────────────────────────────────────────────── # Same base URL as the auth server. Defaults to OIDC_ISSUER, then the public login. COST_API_BASE = os.getenv( "COST_API_BASE", os.getenv("OIDC_ISSUER", "https://login.onyva.dev"), ).strip().rstrip("/") # Per-tenant ingest key (usk_…). Empty → tracking is inert (see _enabled()). COST_INGEST_KEY = os.getenv("COST_INGEST_KEY", "").strip() COST_TRACKING_ENABLED = os.getenv("COST_TRACKING_ENABLED", "true").lower() == "true" _FLUSH_INTERVAL = float(os.getenv("COST_FLUSH_INTERVAL", "5")) _HTTP_TIMEOUT = float(os.getenv("COST_HTTP_TIMEOUT", "5")) _buffer: list[dict] = [] _lock = threading.Lock() _worker_started = False def _enabled() -> bool: # No key → nothing to authenticate with → stay completely inert (also covers # local dev where the key is simply unset). return COST_TRACKING_ENABLED and bool(COST_INGEST_KEY) # ── pricing (EUR) ───────────────────────────────────────────────────────────── # The app reports finished EUR amounts; the server stores them 1:1. Rates are # overridable via env so they can be tuned without a redeploy. # # Defaults: # • LLM — claude-4-6-sonnet via Cortecs, billed by Cortecs directly in EUR: # €2.87 / €14.31 per 1M tokens (list price, 17 Jul 2026). No USD # conversion applies — unlike Bedrock, Cortecs quotes EUR natively. # • STT — Mistral Voxtral Mini Transcribe: $0.003/min → €0.002625/min # (converted at 1 USD ≈ €0.875, 24 Jun 2026). Läuft ein Tenant über # Cortecs statt direkt über Mistral, liefert core/app.py den dortigen # EUR-Tarif als `per_minute` mit (siehe _STT_PROVIDERS). # Re-tune via the COST_* env vars when tariffs move, or if LLM_MODEL is changed # to a different tier (Cortecs publishes per-model EUR rates on GET /v1/models). def llm_cost_eur(model: str, input_tokens: int, output_tokens: int) -> float: """EUR cost of one LLM call from token counts. Rates are €/1M tokens.""" in_per_mtok = float(os.getenv("COST_LLM_INPUT_PER_MTOK", "2.87")) out_per_mtok = float(os.getenv("COST_LLM_OUTPUT_PER_MTOK", "14.31")) return (input_tokens / 1_000_000) * in_per_mtok + (output_tokens / 1_000_000) * out_per_mtok def stt_cost_eur(minutes: float, per_minute: float | None = None) -> float: """EUR cost of an STT call from audio duration. Rate is €/minute. The caller passes the rate of the tenant's STT provider (Mistral direkt vs. Cortecs sind unterschiedlich teuer); COST_STT_PER_MINUTE überschreibt sie.""" default = "0.002625" if per_minute is None else str(per_minute) rate = float(os.getenv("COST_STT_PER_MINUTE") or default) return max(minutes, 0.0) * rate # ── public API ──────────────────────────────────────────────────────────────── def track_usage( cost_point: str, amount_eur: float, account: str | None = None, quantity: float | None = None, idempotency_key: str | None = None, metadata: dict | None = None, ) -> None: """Non-blocking: append a usage event to the buffer. No-op when disabled.""" if not _enabled(): return ev: dict = {"cost_point": cost_point, "amount": round(float(amount_eur), 4)} if account: ev["account"] = account if quantity is not None: ev["quantity"] = quantity if idempotency_key: ev["idempotency_key"] = idempotency_key if metadata: ev["metadata"] = metadata with _lock: _buffer.append(ev) def flush_now() -> None: """Send everything currently buffered. Safe to call from shutdown.""" if not _enabled(): return with _lock: if not _buffer: return batch, _buffer[:] = _buffer[:], [] try: requests.post( f"{COST_API_BASE}/api/usage", headers={"Authorization": f"Bearer {COST_INGEST_KEY}"}, json={"events": batch}, timeout=_HTTP_TIMEOUT, ) except Exception as exc: # best effort — never propagate # Re-buffer so a transient outage doesn't silently drop events. with _lock: _buffer[:0] = batch print(f"[cost] flush failed ({len(batch)} events re-buffered): {exc}") def _loop() -> None: while True: time.sleep(_FLUSH_INTERVAL) flush_now() def start_worker() -> None: """Start the background flush thread once. Inert when tracking is disabled.""" global _worker_started if _worker_started or not _enabled(): return _worker_started = True threading.Thread(target=_loop, daemon=True, name="cost-flush").start() print(f"[cost] tracking enabled → {COST_API_BASE}/api/usage")