Blame for app/pricing.py

# =============================================================================
# app/pricing.py
# Auto Model Pricing — weil niemand Lust hat, Kosten von Hand zu pflegen.
# Universal MCP Hub (Sandboxed) - based on PyFundaments Architecture
# Copyright 2026 - Volkan Kücükbudak
# Apache License V. 2 + ESOL 1.1
# =============================================================================
# ARCHITECTURE NOTE:
#   This file lives exclusively in app/ and is ONLY started by app/app.py.
#   NO direct access to fundaments/*, .env, or Guardian (main.py).
#   All config comes from app/.pyfun [PRICING] via app/config.py.
#
# WHY THIS EXISTS:
#   Not every provider has a pricing API (Anthropic & Google: none).
#   Manually maintaining cost_input_per_1k in [MODELS] sucks. Solution:
#   two free public sources cover almost everything, .pyfun stays the override.
#
# SOURCES (priority: .pyfun manual > OpenRouter > LiteLLM):
#   1. .pyfun [MODELS]           → your values ALWAYS win (manual override)
#   2. OpenRouter /api/v1/models → public, no key. Pricing per token for all
#                                  OR models + context_length + max_completion_tokens
#   3. LiteLLM community JSON    → github.com/BerriAI/litellm main branch.
#                                  input_cost_per_token / output_cost_per_token
#                                  for Anthropic, Gemini, OpenAI, Mistral, ...
#
# PRINCIPLE (same as everything here):
#   No network = no crash. No match = costs stay as configured.
#   Server always starts — pricing refresh runs in the BACKGROUND.
#
# CACHING:
#   Fetched data lands in db_sync hub_state ("pricing.cache") with TTL
#   from .pyfun (default 24h). Restart within TTL = no external call.
#
# MCP TOOLS (registered in mcp.py):
#   get_model_pricing(model_name)  → merged cost info for one model
#   refresh_pricing()              → force refresh, returns source counts
# =============================================================================
import asyncio
import logging
from typing import Any, Dict, Optional

import httpx

from . import config
from . import db_sync

logger = logging.getLogger("pricing")

# =============================================================================
# Internal State
# =============================================================================
_cache: Dict[str, Dict[str, Any]] = {}   # normalized_name → {cost_in_1k, cost_out_1k, context, max_out, source}
_meta: Dict[str, Any] = {}               # {"fetched_at": iso, "sources": {...}, "model_count": int}
_task: Optional[asyncio.Task] = None     # background refresh task
_initialized: bool = False

# Loose index: alnum-only keys for dot/dash-insensitive matching.
# Why: OpenRouter writes 'claude-opus-4.6', LiteLLM 'claude-opus-4-6',
# your .pyfun 'claude-opus-4-6'. All should find each other.
_loose_index: Dict[str, str] = {}        # loose_key → normalized cache key
_loose_dirty: bool = True

_HUB_STATE_KEY = "pricing.cache"

# Known vendor prefixes used by OpenRouter / LiteLLM IDs
# ("anthropic/claude-...", "gemini/gemini-...", "meta-llama/llama-...", ...)
_VENDOR_PREFIXES = (
    "anthropic/", "google/", "openai/", "meta-llama/", "mistralai/", "mistral/",
    "qwen/", "nvidia/", "microsoft/", "deepseek/", "x-ai/", "xai/", "perplexity/",
    "cohere/", "gryphe/", "thedrummer/", "undi95/", "cognitivecomputations/",
    "gemini/", "groq/", "together_ai/", "together/", "openrouter/", "fireworks_ai/",
    "azure/", "vertex_ai/", "bedrock/", "cloudflare/", "voyage/", "sambanova/",
    "ai21/", "huggingface/", "ollama/", "ollama_chat/",
)


# =============================================================================
# Normalization — model IDs differ wildly between sources
# =============================================================================
def _normalize(name: str) -> str:
    """
    Normalize a model name for matching across sources.
    'anthropic/claude-haiku-4.5' → 'claude-haiku-4.5'
    'Claude-Haiku-4.5'           → 'claude-haiku-4.5'
    Dots vs dashes are kept — providers are inconsistent there by design.
    """
    n = (name or "").strip().lower()
    for prefix in _VENDOR_PREFIXES:
        if n.startswith(prefix):
            n = n[len(prefix):]
            break
    return n


def _safe_int(value: Any) -> Optional[int]:
    """int() that returns None instead of raising (LiteLLM ships doc samples
    like sample_spec.max_input_tokens = 'max input tokens, if the provider specifies it')."""
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _per_token_to_per_1k(cost: Any) -> Optional[float]:
    """'0.0000015' per token → 0.0015 per 1k tokens. Returns None on garbage."""
    try:
        val = float(cost)
    except (TypeError, ValueError):
        return None
    if val < 0:  # negative = OpenRouter's "we pay you" marker → treat as 0
        val = 0.0
    return round(val * 1000, 6)


def _loose(name: str) -> str:
    """Strip everything except a-z0-9 — kills dot/dash/underscore differences."""
    return "".join(ch for ch in name if ch.isalnum())


def _rebuild_loose_index() -> None:
    """Rebuild the alnum-only lookup index after a cache change."""
    global _loose_index, _loose_dirty
    _loose_index = {_loose(k): k for k in _cache}
    _loose_dirty = False


def _lookup(model_name: str) -> Optional[Dict[str, Any]]:
    """
    Cache lookup with graceful fallback:
        1. exact normalized match ('claude-opus-4-6')
        2. loose alnum match ('claude-opus-4.6' ≈ 'claude-opus-4-6')
    """
    norm = _normalize(model_name)
    hit = _cache.get(norm)
    if hit:
        return hit
    if _loose_dirty:
        _rebuild_loose_index()
    loose_key = _loose_index.get(_loose(norm))
    if loose_key:
        return _cache.get(loose_key)
    return None


# =============================================================================
# Source Fetchers — each returns {normalized_name: entry}
# =============================================================================
async def _fetch_openrouter(url: str):
    """
    OpenRouter public models endpoint — no key required.
    https://openrouter.ai/api/v1/models → data[].{id, pricing{prompt,completion}, context_length}

    Returns (entries, raw_ids): raw_ids = every real OpenRouter model ID (lowercase) —
    used by check_models() to spot stale/removed models in .pyfun [MODELS].
    """
    out: Dict[str, Dict[str, Any]] = {}
    raw_ids = []
    async with httpx.AsyncClient() as client:
        r = await client.get(url, timeout=30)
        r.raise_for_status()
        payload = r.json()

    for m in payload.get("data", []):
        model_id = m.get("id", "")
        if not model_id:
            continue
        raw_ids.append(model_id.lower())
        pricing = m.get("pricing", {}) or {}
        entry: Dict[str, Any] = {"source": "openrouter"}

        cost_in  = _per_token_to_per_1k(pricing.get("prompt"))
        cost_out = _per_token_to_per_1k(pricing.get("completion"))
        if cost_in is not None:
            entry["cost_input_per_1k"] = cost_in
        if cost_out is not None:
            entry["cost_output_per_1k"] = cost_out
        if m.get("context_length"):
            ctx = _safe_int(m["context_length"])
            if ctx:
                entry["context_tokens"] = ctx

        # max_completion_tokens is nested under top_provider in OR payloads
        max_out = _safe_int((m.get("top_provider") or {}).get("max_completion_tokens"))
        if max_out:
            entry["max_output_tokens"] = max_out

        out[_normalize(model_id)] = entry
    return out, raw_ids


async def _fetch_litellm(url: str) -> Dict[str, Dict[str, Any]]:
    """
    LiteLLM community price list — no key required.
    github.com/BerriAI/litellm → model_prices_and_context_window.json
    Keys like 'claude-sonnet-4-5', 'gemini/gemini-2.5-flash', 'gpt-4o'.
    """
    out: Dict[str, Dict[str, Any]] = {}
    async with httpx.AsyncClient() as client:
        r = await client.get(url, timeout=30)
        r.raise_for_status()
        payload = r.json()

    for model_id, info in payload.items():
        if not isinstance(info, dict):
            continue  # skip metadata keys like "additional_vendor_names"
        entry: Dict[str, Any] = {"source": "litellm"}

        cost_in  = _per_token_to_per_1k(info.get("input_cost_per_token"))
        cost_out = _per_token_to_per_1k(info.get("output_cost_per_token"))
        if cost_in is not None:
            entry["cost_input_per_1k"] = cost_in
        if cost_out is not None:
            entry["cost_output_per_1k"] = cost_out
        if info.get("max_input_tokens"):
            ctx = _safe_int(info["max_input_tokens"])
            if ctx:
                entry["context_tokens"] = ctx
        if info.get("max_output_tokens"):
            max_out = _safe_int(info["max_output_tokens"])
            if max_out:
                entry["max_output_tokens"] = max_out

        out[_normalize(model_id)] = entry
    return out


# =============================================================================
# Cache — db_sync hub_state, TTL-gated
# =============================================================================
def _cache_is_fresh() -> bool:
    if not _cache or not _meta.get("fetched_at"):
        return False
    from datetime import datetime, timezone
    try:
        fetched = datetime.fromisoformat(_meta["fetched_at"])
    except (ValueError, TypeError):
        return False
    ttl_hours = float(config.get_pricing().get("PRICING_TTL_HOURS", "24"))
    age = datetime.now(timezone.utc) - fetched
    return age.total_seconds() < ttl_hours * 3600


async def _load_cache_from_db() -> None:
    """Read pricing cache from db_sync — never raises."""
    global _cache, _meta, _loose_dirty
    try:
        stored = await db_sync.read(_HUB_STATE_KEY, default=None)
        if isinstance(stored, dict) and stored.get("models"):
            _cache = stored["models"]
            _meta = stored.get("meta", {})
            _loose_dirty = True
            logger.info(
                f"Pricing cache loaded from db_sync: {len(_cache)} models "
                f"(fetched_at: {_meta.get('fetched_at', 'unknown')})"
            )
    except Exception as e:
        logger.warning(f"Pricing cache load failed (continuing without): {type(e).__name__}")


async def _save_cache_to_db() -> None:
    """Write pricing cache to db_sync — never raises."""
    try:
        await db_sync.write(_HUB_STATE_KEY, {"models": _cache, "meta": _meta})
    except Exception as e:
        logger.warning(f"Pricing cache save failed: {type(e).__name__}")


# =============================================================================
# Refresh — fetch sources, merge, cache
# =============================================================================
async def refresh() -> Dict[str, Any]:
    """
    Fetch all pricing sources, rebuild the auto-pricing cache.
    Never raises — a failing source is just skipped (no fetch = no crash).
    """
    global _cache, _meta, _loose_dirty
    from datetime import datetime, timezone

    cfg       = config.get_pricing()
    or_url    = cfg.get("OPENROUTER_MODELS_URL", "https://openrouter.ai/api/v1/models")
    llm_url   = cfg.get(
        "LITELLM_PRICES_URL",
        "https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json",
    )

    merged: Dict[str, Dict[str, Any]] = {}
    sources: Dict[str, int] = {}
    or_ids: list = []

    # --- Source 1: OpenRouter ---
    try:
        or_models, or_ids = await _fetch_openrouter(or_url)
        merged.update(or_models)
        sources["openrouter"] = len(or_models)
        logger.info(f"Pricing source OpenRouter: {len(or_models)} models.")
    except Exception as e:
        sources["openrouter"] = 0
        logger.warning(f"Pricing source OpenRouter failed: {type(e).__name__} — skipped.")

    # --- Source 2: LiteLLM (fills gaps — never overwrites OpenRouter) ---
    try:
        llm_models = await _fetch_litellm(llm_url)
        for name, entry in llm_models.items():
            if name not in merged:
                merged[name] = entry
        sources["litellm"] = len(llm_models)
        logger.info(f"Pricing source LiteLLM: {len(llm_models)} models.")
    except Exception as e:
        sources["litellm"] = 0
        logger.warning(f"Pricing source LiteLLM failed: {type(e).__name__} — skipped.")

    if merged:
        _cache = merged
        _meta = {
            "fetched_at": datetime.now(timezone.utc).isoformat(),
            "sources": sources,
            "model_count": len(merged),
            # keep the last known OpenRouter ID list if this fetch failed
            "openrouter_ids": or_ids or _meta.get("openrouter_ids", []),
        }
        _loose_dirty = True
        await _save_cache_to_db()

    stale = check_models()
    if stale:
        logger.warning(f"Models in .pyfun not found in OpenRouter catalog (stale/removed?): {stale}")

    return {
        "fetched_at": _meta.get("fetched_at"),
        "sources": sources,
        "models_cached": len(_cache),
        "stale_models": stale,
    }


# =============================================================================
# Initialization — called by app/app.py, NON-BLOCKING
# =============================================================================
async def initialize() -> None:
    """
    Prepare auto-pricing. Loads the db_sync cache if fresh; otherwise kicks
    off a BACKGROUND refresh (asyncio.create_task) — the server never waits
    for external pricing APIs. No network = no crash.
    """
    global _initialized, _task
    if _initialized:
        return

    cfg = config.get_pricing()
    enabled = cfg.get("PRICING_ENABLED", "true").lower() == "true"

    if not enabled:
        logger.info("Auto-pricing disabled in .pyfun [PRICING] — .pyfun [MODELS] values only.")
        _initialized = True
        return

    await _load_cache_from_db()

    if _cache_is_fresh():
        logger.info("Auto-pricing cache fresh — no external fetch needed.")
    else:
        logger.info("Auto-pricing cache stale/empty — background refresh started.")
        _task = asyncio.create_task(_background_refresh())

    _initialized = True


async def _background_refresh() -> None:
    """Runs refresh() and swallows everything — background task must never crash the app."""
    try:
        await refresh()
    except Exception as e:
        logger.warning(f"Pricing background refresh failed (non-fatal): {type(e).__name__}")


# =============================================================================
# Public API — merged view: .pyfun manual values WIN over auto-fetched
# =============================================================================
async def get_pricing(model_name: str) -> Dict[str, Any]:
    """
    Returns merged pricing info for one model.

    Merge order (highest priority first):
        1. .pyfun [MODELS.<model>]   → your manual values always win
        2. auto-fetched cache        → OpenRouter / LiteLLM
        3. nothing                   → costs omitted, source: "unknown"

    Returns dict with: model, provider (if known from .pyfun),
    cost_input_per_1k, cost_output_per_1k (USD), context_tokens,
    max_output_tokens, source ("pyfun" | "openrouter" | "litellm" | "unknown"),
    pricing_note.
    """
    from . import models as models_mod

    result: Dict[str, Any] = {"model": model_name}
    notes = []

    # --- Layer 1: .pyfun manual config (models.py reads [MODELS]) ---
    manual = models_mod.get(model_name)
    if manual:
        result["provider"] = manual.get("provider", "")
        for key in ("cost_input_per_1k", "cost_output_per_1k",
                    "context_tokens", "max_output_tokens",
                    "requests_per_min", "requests_per_day", "capabilities"):
            if manual.get(key):
                value = manual[key]
                if key.startswith("cost_"):
                    # .pyfun-Werte sind Strings ("0.015") — Auto-Werte sind float.
                    # Normalisieren, damit die API immer Zahlen liefert.
                    try:
                        value = round(float(value), 6)
                    except (TypeError, ValueError):
                        pass
                result[key] = value
        if manual.get("cost_input_per_1k") or manual.get("cost_output_per_1k"):
            result["source"] = "pyfun"
            notes.append(".pyfun manual values (override)")
        else:
            notes.append(".pyfun entry has no costs — auto values used")

    # --- Layer 2: auto-fetched cache fills the gaps ---
    auto = _lookup(model_name)
    if auto:
        for key in ("cost_input_per_1k", "cost_output_per_1k",
                    "context_tokens", "max_output_tokens"):
            if key not in result and auto.get(key) is not None:
                result[key] = auto[key]
        if "source" not in result:
            result["source"] = auto.get("source", "auto")
            notes.append(f"auto-fetched from {auto.get('source', 'unknown')}")
    elif "source" not in result:
        result["source"] = "unknown"
        notes.append("no auto source matched this model name")

    # --- Stale hint ---
    if _cache and not _cache_is_fresh():
        notes.append("cache stale — refresh_pricing() to update")

    result["pricing_note"] = "; ".join(notes) if notes else "ok"
    return result


async def catalog(query: str = "", free_only: bool = False, limit: int = 50) -> Dict[str, Any]:
    """
    Real model IDs of the live OpenRouter catalog (for picking valid [MODELS] entries).
    Fetches the catalog once if it is not loaded yet. Read-only, no key needed.
    """
    if not _meta.get("openrouter_ids"):
        await refresh()
    ids = _meta.get("openrouter_ids") or []
    q = (query or "").strip().lower()
    hits = sorted(i for i in ids if (not free_only or i.endswith(":free")) and (not q or q in i))
    limit = max(1, min(int(limit or 50), 200))
    rows = []
    for i in hits[:limit]:
        info = _lookup(i) or {}
        rows.append({"id": i, "context_tokens": info.get("context_tokens"),
                     "cost_input_per_1k": info.get("cost_input_per_1k"),
                     "cost_output_per_1k": info.get("cost_output_per_1k")})
    return {"in_catalog": len(ids), "matches": len(hits), "shown": len(rows), "models": rows}


def check_models() -> list:
    """
    .pyfun [MODELS] entries with provider "openrouter" whose ID is NOT in the live
    OpenRouter catalog (free models come and go). Empty list if the catalog is unknown.
    """
    ids = set(_meta.get("openrouter_ids") or [])
    if not ids:
        return []
    return sorted(
        name for name, cfg in config.get_models().items()
        if cfg.get("provider") == "openrouter" and name.lower() not in ids
    )


def max_output_for(model_name: str) -> Optional[int]:
    """
    Max output tokens for a model (sync, for clamping): manual .pyfun value wins,
    then auto-fetched. None if unknown → no clamp.
    """
    from . import models as models_mod
    manual = _safe_int((models_mod.get(model_name) or {}).get("max_output_tokens"))
    if manual:
        return manual
    auto = _lookup(model_name) if _cache else None
    return _safe_int(auto.get("max_output_tokens")) if auto else None


async def cost_for(model_name: str, tokens_in: Optional[int], tokens_out: Optional[int]) -> Optional[float]:
    """
    Estimated USD cost of one call from merged pricing (manual > auto).
    None if no price is known. Anthropic cache read/write surcharges are not modeled.
    """
    info = await get_pricing(model_name)
    c_in, c_out = info.get("cost_input_per_1k"), info.get("cost_output_per_1k")
    if c_in is None and c_out is None:
        return None
    try:
        return round(((tokens_in or 0) / 1000) * float(c_in or 0)
                     + ((tokens_out or 0) / 1000) * float(c_out or 0), 8)
    except (TypeError, ValueError):
        return None


def is_ready() -> bool:
    """True if pricing cache holds at least one model."""
    return bool(_cache)


def stats() -> Dict[str, Any]:
    """Cache stats for list_active_tools / debugging."""
    return {
        "models_cached": len(_cache),
        "fetched_at": _meta.get("fetched_at"),
        "sources": _meta.get("sources", {}),
        "fresh": _cache_is_fresh(),
        "stale_models": check_models(),
    }


# =============================================================================
# Direct execution guard
# =============================================================================
if __name__ == "__main__":
    print("WARNING: Run via main.py → app.py, not directly.")