feat(pricing): live cross-checked prices, refreshed on every install, at first boot and on a set cadence (#16229, #16231)

.pre-commit-config.yaml

@@ -597,7 +597,7 @@ repos:
       - id: name-tests-test  # Ensure test files are named correctly
         args: [--pytest-test-first]
         # Exclude colocated tests (*_test.py convention) and special test files
-        exclude: (_test\.py$|^(tests/|autobot-infrastructure/.*/tests/|autobot-.*/tests/).*(results/|benchmarks/|fixtures/|conftest\.py|mock_llm_interface\.py|ci_cd_integration\.py|integration_test_infrastructure\.py|security_port_audit\.py|frontend_comprehensive_corrected\.py|phase9_comprehensive_test_suite\.py|benchmark_base\.py|api_benchmarks\.py|rag_benchmarks\.py|mocks\.py|system_assessment_report\.py))
+        exclude: (_test\.py$|^(tests/|autobot-infrastructure/.*/tests/|autobot-.*/tests/).*(results/|benchmarks/|fixtures/|conftest\.py|mock_llm_interface\.py|ci_cd_integration\.py|integration_test_infrastructure\.py|security_port_audit\.py|frontend_comprehensive_corrected\.py|phase9_comprehensive_test_suite\.py|benchmark_base\.py|api_benchmarks\.py|rag_benchmarks\.py|mocks\.py|system_assessment_report\.py|_code_sync_import\.py))
 
   # Security checks with bandit
   - repo: https://github.com/PyCQA/bandit

autobot-backend/api/admin_pricing.py

@@ -2,39 +2,33 @@
 # SPDX-License-Identifier: Apache-2.0
 # AutoBot - AI-Powered Automation Platform
 # Author: mrveiss
-"""Admin API: manual pricing override and refresh status (GH#6480).
+"""Admin API: manual pricing override, refresh status, and refresh-now (GH#6480, #16231).
 
 Endpoints:
   PUT  /api/admin/pricing/{provider}/{model}  — emergency override
   DELETE /api/admin/pricing/{provider}/{model} — remove override
   GET  /api/admin/pricing/status              — refresh status per provider
+  POST /api/admin/pricing/refresh             — refresh now, on demand (#16231)
 """
 
 from datetime import datetime, timezone
 
 from fastapi import APIRouter, Depends
-from pydantic import BaseModel, Field
 
+from api.schemas_pricing import PricingOverrideRequest
 from auth_rbac import require_role
 
 router = APIRouter(prefix="/admin")
 
 
-class PricingOverrideRequest(BaseModel):
-    input_per_1m: float = Field(..., gt=0, description="Input cost per 1M tokens (USD)")
-    output_per_1m: float = Field(..., gt=0, description="Output cost per 1M tokens (USD)")
-    cache_read_per_1m: float = Field(0.0, ge=0)
-    cache_write_per_1m: float = Field(0.0, ge=0)
-
-
 @router.put("/pricing/{provider}/{model}")
 async def override_model_pricing(
     provider: str,
     model: str,
     body: PricingOverrideRequest,
     _admin: bool = Depends(require_role("admin", "superadmin")),
 ) -> dict:
-    """Write an emergency pricing override directly to Redis."""
+    """Store an emergency pricing override; it outranks the refreshed price until removed."""
     from autobot_shared.logging_manager import get_logger
     from llm_shared.pricing.redis_store import PricingRedisStore
     from llm_shared.pricing.sources import ModelPricing
@@ -49,9 +43,10 @@ async def override_model_pricing(
         cache_read_per_1m=body.cache_read_per_1m,
         cache_write_per_1m=body.cache_write_per_1m,
         updated_at=datetime.now(tz=timezone.utc),
+        source="override",
     )
     store = PricingRedisStore()
-    ok = await store.set(pricing)
+    ok = await store.set_override(pricing)
     logger.info(
         "admin pricing override: provider=%s model=%s input=%.4f output=%.4f success=%s",
         provider,
@@ -74,11 +69,11 @@ async def delete_model_pricing_override(
     model: str,
     _admin: bool = Depends(require_role("admin", "superadmin")),
 ) -> dict:
-    """Remove a pricing override from Redis (next refresh will re-populate)."""
+    """Remove a pricing override; the refreshed price applies again."""
     from llm_shared.pricing.redis_store import PricingRedisStore
 
     store = PricingRedisStore()
-    deleted = await store.delete(provider, model)
+    deleted = await store.delete_override(model)
     return {"provider": provider, "model": model, "deleted": deleted}
 
 
@@ -90,3 +85,11 @@ async def get_pricing_refresh_status(_admin: bool = Depends(require_role("admin"
     store = PricingRedisStore()
     status = await store.get_refresh_status()
     return {"providers": status}
+
+
+@router.post("/pricing/refresh")
+async def refresh_pricing_now(_admin: bool = Depends(require_role("admin", "superadmin"))) -> dict:
+    """Refresh pricing on demand and report the per-source result (#16231)."""
+    from services.pricing_refresh import refresh_all
+
+    return await refresh_all()

autobot-backend/api/pricing_health.py

@@ -28,6 +28,7 @@ async def _pricing_health_probe(request: Request | None) -> ComponentHealth:
 
         store = PricingRedisStore()
         status = await store.get_refresh_status()
+        crosscheck = await store.get_crosscheck()
 
         if not status:
             return ComponentHealth(
@@ -62,6 +63,9 @@ async def _pricing_health_probe(request: Request | None) -> ComponentHealth:
         else:
             probe_status = "ok"
             detail = f"{len(status)} provider(s) refreshed successfully"
+        if crosscheck:  # #16229: how many models the catalogues were actually compared on
+            compared, disagreed = crosscheck.get("compared", 0), len(crosscheck.get("disagreed", []))
+            detail += f"; cross-check compared {compared}, disagreed {disagreed}"
 
         return ComponentHealth(
             name=KnownProbes.PRICING,

autobot-backend/api/schemas_pricing.py

@@ -0,0 +1,18 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""Request models for the admin pricing API (GH#6480).
+
+Moved out of ``api/admin_pricing.py`` unchanged, so the OpenAPI schema and the
+generated frontend types stay identical (#16229).
+"""
+
+from pydantic import BaseModel, Field
+
+
+class PricingOverrideRequest(BaseModel):
+    input_per_1m: float = Field(..., gt=0, description="Input cost per 1M tokens (USD)")
+    output_per_1m: float = Field(..., gt=0, description="Output cost per 1M tokens (USD)")
+    cache_read_per_1m: float = Field(0.0, ge=0)
+    cache_write_per_1m: float = Field(0.0, ge=0)

autobot-backend/celery_app.py

@@ -12,6 +12,7 @@
 """
 
 import urllib.parse
+from datetime import timedelta
 from pathlib import Path
 
 from celery import Celery
@@ -187,8 +188,11 @@
 # decorators all run at worker/beat startup.
 celery_app.autodiscover_tasks(["tasks", "workers", "llc.scheduler"], related_name=None)
 # GH#6480: pricing refresh lives in services/, outside the discovered packages —
-# import explicitly so workers register pricing.refresh_daily.
+# import explicitly so workers register pricing.refresh_daily and
+# pricing.refresh_if_empty, and connect the worker_ready handler that queues
+# the latter on worker startup (#16231, #16250).
 import services.pricing_refresh  # noqa: F401
+from llm_shared.pricing.redis_store import REFRESH_INTERVAL_HOURS as _PRICING_REFRESH_INTERVAL_HOURS  # noqa: E402
 from utils.celery_schedules import crontab_from_string  # noqa: E402
 
 # =========================================================================
@@ -256,10 +260,12 @@
         "schedule": crontab(hour=2, minute=0),
         "kwargs": {"max_age_days": 7},
     },
-    # GH#6480: daily pricing refresh from provider sources into Redis (02:15 UTC)
+    # GH#6480: pricing refresh from provider sources into Redis, on an
+    # env-configurable cadence (AUTOBOT_PRICING_REFRESH_INTERVAL_HOURS, default
+    # 24h) rather than a fixed crontab (#16231).
     "pricing-refresh-daily": {
         "task": "pricing.refresh_daily",
-        "schedule": crontab(hour=2, minute=15),
+        "schedule": timedelta(hours=_PRICING_REFRESH_INTERVAL_HOURS),
     },
     # GH#4463: weekly cleanup of stale mobile devices (inactive for 90+ days)
     "mobile-devices-cleanup-weekly": {

autobot-backend/conftest.py

@@ -732,14 +732,20 @@ def calculate_cost(self, *_a, **_k) -> float:
     # and autobot_shared.redis_client (real module); the llm_shared stub's empty
     # __path__ otherwise breaks every provider-source import and its colocated
     # services/pricing_refresh_test.py. Dependency order: sources → redis_store
-    # → per-provider sources → package __init__.
-    _real_load_and_bind("llm_shared.pricing.sources", _llm_root / "pricing" / "sources.py")
-    _real_load_and_bind("llm_shared.pricing.redis_store", _llm_root / "pricing" / "redis_store.py")
-    _real_load_and_bind("llm_shared.pricing.anthropic_source", _llm_root / "pricing" / "anthropic_source.py")
-    _real_load_and_bind("llm_shared.pricing.openai_source", _llm_root / "pricing" / "openai_source.py")
-    _real_load_and_bind("llm_shared.pricing.google_source", _llm_root / "pricing" / "google_source.py")
-    _real_load_and_bind("llm_shared.pricing.deepseek_source", _llm_root / "pricing" / "deepseek_source.py")
-    _real_load_and_bind("llm_shared.pricing.vertexai_source", _llm_root / "pricing" / "vertexai_source.py")
+    # → live sources + cross-check (#16229) → per-provider sources → package __init__.
+    _pricing_subs = (
+        "sources",
+        "redis_store",
+        "live_sources",
+        "crosscheck",
+        "anthropic_source",
+        "openai_source",
+        "google_source",
+        "deepseek_source",
+        "vertexai_source",
+    )
+    for _pricing_sub in _pricing_subs:
+        _real_load_and_bind(f"llm_shared.pricing.{_pricing_sub}", _llm_root / "pricing" / f"{_pricing_sub}.py")
     _real_load_and_bind("llm_shared.pricing", _llm_root / "pricing" / "__init__.py")
     # The submodule real-loads above ran before "llm_shared.pricing" existed in
     # sys.modules, so _real_load_and_bind's own parent-bind was a no-op for them
@@ -749,15 +755,7 @@ def calculate_cost(self, *_a, **_k) -> float:
     # via getattr(llm_shared.pricing, "redis_store") instead of raising
     # AttributeError.
     _pricing_pkg = sys.modules["llm_shared.pricing"]
-    for _pricing_sub in (
-        "sources",
-        "redis_store",
-        "anthropic_source",
-        "openai_source",
-        "google_source",
-        "deepseek_source",
-        "vertexai_source",
-    ):
+    for _pricing_sub in _pricing_subs:
         setattr(_pricing_pkg, _pricing_sub, sys.modules[f"llm_shared.pricing.{_pricing_sub}"])
 
     # llm_shared.cache — provide symbols consumed by services.llm_service

autobot-backend/llm_shared/pricing/crosscheck.py

@@ -0,0 +1,116 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""Cross-check LiteLLM's prices against OpenRouter's (#16229).
+
+The catalogues name models differently ("claude-haiku-4-5" against
+"anthropic/claude-haiku-4.5") and cover different populations -- thousands of
+LiteLLM entries against a few hundred OpenRouter models. So the report keeps
+its counts apart, because each one alone reads like a clean result:
+
+- ``compared``: in both catalogues under one vendor family, prices compared
+- ``disagreed``: compared, and different beyond the tolerance
+- ``only_primary`` / ``only_secondary``: listed by one catalogue only
+- ``not_comparable``: reseller routes ("bedrock/...") and variants (":free")
+
+A model only one catalogue lists was never checked. It is NOT an agreement.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+
+from llm_shared.pricing.sources import ModelPricing
+
+#: Vendor spellings that differ between the catalogues for one provider.
+#: Names, not prices; extend it when a low ``compared`` count shows a gap.
+_VENDOR_ALIASES = {"google": "gemini", "mistralai": "mistral", "x-ai": "xai"}
+_COMPARED_FIELDS = ("input_per_1m", "output_per_1m")
+
+
+@dataclass
+class CrossCheckReport:
+    tolerance_percent: float
+    compared: int = 0
+    agreed: int = 0
+    only_primary: int = 0
+    only_secondary: int = 0
+    not_comparable: int = 0
+    disagreed: list[dict] = field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+def match_key(vendor: str, model_id: str) -> tuple[str, str]:
+    """A catalogue-neutral key: the vendor family and a normalised model name."""
+    name = model_id.split(":", 1)[0].lower().replace(".", "-")
+    family = vendor.lower()
+    return (_VENDOR_ALIASES.get(family, family), name)
+
+
+def _index(pricings: dict[str, ModelPricing], *, secondary: bool) -> tuple[dict, int]:
+    """Comparable entries by match key, and how many were not comparable."""
+    index: dict[tuple[str, str], tuple[str, ModelPricing]] = {}
+    skipped = 0
+    for key, pricing in pricings.items():
+        comparable = ":" not in key if secondary else "/" not in key
+        if not comparable:
+            skipped += 1
+            continue
+        index.setdefault(match_key(pricing.provider, pricing.model_id), (key, pricing))
+    return index, skipped
+
+
+def _worst_difference(a: ModelPricing, b: ModelPricing, tolerance: float) -> tuple[str, float] | None:
+    """The field that differs most beyond *tolerance* percent, or None."""
+    worst: tuple[str, float] | None = None
+    for name in _COMPARED_FIELDS:
+        x, y = getattr(a, name), getattr(b, name)
+        base = max(abs(x), abs(y))
+        pct = abs(x - y) / base * 100 if base else 0.0
+        if pct > tolerance and (worst is None or pct > worst[1]):
+            worst = (name, pct)
+    return worst
+
+
+def cross_check(
+    primary: dict[str, ModelPricing], secondary: dict[str, ModelPricing], tolerance_percent: float
+) -> tuple[CrossCheckReport, dict[str, str], list[str]]:
+    """Compare the catalogues.
+
+    Returns the report, a verdict per compared-or-single primary key ("agree",
+    "disagree", "single"), and the secondary keys only the secondary lists.
+    """
+    p_idx, p_skipped = _index(primary, secondary=False)
+    s_idx, s_skipped = _index(secondary, secondary=True)
+    report = CrossCheckReport(tolerance_percent=tolerance_percent, not_comparable=p_skipped + s_skipped)
+    verdicts: dict[str, str] = {}
+    for mkey, (pkey, p) in p_idx.items():
+        match = s_idx.get(mkey)
+        if match is None:
+            report.only_primary += 1
+            verdicts[pkey] = "single"
+            continue
+        report.compared += 1
+        worst = _worst_difference(p, match[1], tolerance_percent)
+        if worst is None:
+            report.agreed += 1
+            verdicts[pkey] = "agree"
+            continue
+        field_name, pct = worst
+        report.disagreed.append(
+            {
+                "model": pkey,
+                "secondary": match[0],
+                "field": field_name,
+                "primary_value": getattr(p, field_name),
+                "secondary_value": getattr(match[1], field_name),
+                "percent": round(pct, 1),
+            }
+        )
+        verdicts[pkey] = "disagree"
+    secondary_only = [skey for mkey, (skey, _s) in s_idx.items() if mkey not in p_idx]
+    report.only_secondary = len(secondary_only)
+    return report, verdicts, secondary_only