awoooi/apps/api/src/services/ai_provider_policy.py

"""Single production AI-provider route contract.

The executable production path is deliberately small and ordered.  Legacy
providers may remain registered for replay/shadow metadata, but callers must
not insert them into a production generation chain.
"""

from __future__ import annotations

from collections.abc import Iterable
from typing import Any

PRODUCTION_OLLAMA_ORDER: tuple[str, ...] = (
    "ollama_gcp_a",
    "ollama_gcp_b",
    "ollama_local",
)
PRODUCTION_PROVIDER_ORDER: tuple[str, ...] = (
    *PRODUCTION_OLLAMA_ORDER,
    "gemini",
)
NON_EXECUTABLE_SHADOW_PROVIDERS: frozenset[str] = frozenset(
    {
        "claude",
        "nvidia",
        "nemotron",
        "openclaw",
        "openclaw_nemo",
        "ollama_tool",
    }
)


def production_provider_order(*, include_gemini: bool = True) -> list[str]:
    """Return a new list so callers cannot mutate the global contract."""
    return list(
        PRODUCTION_PROVIDER_ORDER
        if include_gemini
        else PRODUCTION_OLLAMA_ORDER
    )


def normalize_production_execution_order(
    requested: Iterable[str],
    *,
    require_local: bool = False,
) -> tuple[list[str], list[str]]:
    """Normalize a requested route and report rejected provider insertions.

    The legacy logical ``ollama`` alias expands to all three concrete Ollama
    identities.  A production call must contain an Ollama lane; a lone Gemini
    request is rejected because it has no evidence that the free lane ran.
    Gemini is appended only when it was explicitly requested and local-only
    processing is not required.  This also means a disabled/removed Gemini is
    never silently re-added by normalization.
    """
    values = [str(value).strip().lower() for value in requested if str(value).strip()]
    rejected = sorted(
        {
            value
            for value in values
            if value not in PRODUCTION_PROVIDER_ORDER and value != "ollama"
        }
    )
    has_ollama_lane = "ollama" in values or any(
        value in PRODUCTION_OLLAMA_ORDER for value in values
    )
    if not has_ollama_lane:
        return [], rejected

    normalized = list(PRODUCTION_OLLAMA_ORDER)
    if not require_local and "gemini" in values:
        normalized.append("gemini")
    return normalized, rejected


def production_route_readback(
    *,
    gemini_configured: bool,
    gemini_enabled: bool,
    gemini_authentication: dict[str, Any] | None = None,
    cost_guard: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build a public-safe route/status payload without endpoint URLs or keys."""
    authentication = dict(gemini_authentication or {})
    cost_guard_payload = dict(cost_guard or {})
    authenticated = authentication.get("authenticated")
    authentication_verified = bool(
        authentication.get("authentication_verified") is True
    )
    pricing_policy = cost_guard_payload.get("pricing_policy") or {}
    accounting = cost_guard_payload.get("accounting") or {}
    accounting_status = str(accounting.get("status") or "").strip().lower()
    accounting_verified = bool(
        accounting.get("complete") is True
        and accounting.get("degraded") is not True
        and accounting_status == "receipt_backed"
    )
    cost_guard_ready = bool(
        cost_guard_payload
        and cost_guard_payload.get("readback_status") == "verified"
        and accounting_verified
        and pricing_policy.get("supported") is True
        and cost_guard_payload.get("cost_exceeded") is False
    )
    gemini_production_executable = bool(
        gemini_enabled and cost_guard_ready and authentication_verified
    )
    hops = [
        {
            "position": 1,
            "provider": "ollama_gcp_a",
            "identity": "GCP-A",
            "runtime": "Ollama",
            "paid": False,
            "production_executable": True,
        },
        {
            "position": 2,
            "provider": "ollama_gcp_b",
            "identity": "GCP-B",
            "runtime": "Ollama",
            "paid": False,
            "production_executable": True,
        },
        {
            "position": 3,
            "provider": "ollama_local",
            "identity": "host111",
            "runtime": "Ollama",
            "paid": False,
            "production_executable": True,
        },
        {
            "position": 4,
            "provider": "gemini",
            "identity": "Gemini API",
            "runtime": "gemini-2.5-flash-lite",
            "paid": True,
            "production_executable": gemini_production_executable,
            "configured": bool(gemini_configured),
            "authenticated": authenticated,
            "authentication_verified": authentication_verified,
            "verification_status": authentication.get(
                "verification_status", "not_verified"
            ),
        },
    ]
    return {
        "schema_version": "ai_provider_production_route_v1",
        "provider_order": list(PRODUCTION_PROVIDER_ORDER),
        "route_label": "GCP-A -> GCP-B -> host111 Ollama -> Gemini API",
        "hops": hops,
        "legacy_provider_policy": {
            provider: "non_executable_shadow_metadata_only"
            for provider in sorted(NON_EXECUTABLE_SHADOW_PROVIDERS)
        },
        "gemini": {
            "configured": bool(gemini_configured),
            "enabled": bool(gemini_enabled),
            "production_executable": gemini_production_executable,
            "authenticated": authenticated,
            "authentication_verified": authentication_verified,
            "verification_status": authentication.get(
                "verification_status", "not_verified"
            ),
            "verification_source": authentication.get("verification_source"),
            "verified_at": authentication.get("verified_at"),
            "cost_guard_ready": cost_guard_ready,
            "cost_guard": cost_guard_payload,
        },
    }