""" API usage and token tracking. Writes daily buckets to home/{username}/usage.json: { "2026-05-01": { "gemini_api/gemini-2.0-flash": {"calls": 3, "prompt_tokens": 8400, "completion_tokens": 520}, "local/llama3.2:latest": {"calls": 2, "prompt_tokens": 1200, "completion_tokens": 310} } } Claude CLI and Gemini CLI backends produce no structured token data and are not tracked. """ import asyncio import json import logging from datetime import date from pathlib import Path from config import settings logger = logging.getLogger(__name__) _LOCK = asyncio.Lock() def _usage_path(username: str) -> Path: return settings.home_root() / username / "usage.json" async def record( username: str, backend: str, model_name: str, prompt_tokens: int, completion_tokens: int, ) -> None: """Append one call's token counts to the daily usage log for this user. backend — "gemini_api" | "local" model_name — the exact model string (e.g. "gemini-2.0-flash", "llama3.2:latest") """ path = _usage_path(username) today = date.today().isoformat() key = f"{backend}/{model_name}" async with _LOCK: try: data: dict = json.loads(path.read_text()) if path.exists() else {} except Exception: data = {} entry = data.setdefault(today, {}).setdefault( key, {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0} ) entry["calls"] += 1 entry["prompt_tokens"] += prompt_tokens entry["completion_tokens"] += completion_tokens try: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2)) except Exception as e: logger.warning("Failed to write usage data to %s: %s", path, e) def read_usage(username: str) -> dict: """Return the full usage dict for this user. Empty dict if no file yet.""" path = _usage_path(username) try: return json.loads(path.read_text()) if path.exists() else {} except Exception: return {}