"""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, "claude", "gemini", ) NON_EXECUTABLE_SHADOW_PROVIDERS: frozenset[str] = frozenset( { "nvidia", "nemotron", "openclaw", "openclaw_nemo", "ollama_tool", } ) PAID_PROVIDER_ORDER: tuple[str, ...] = ("claude", "gemini") SANITIZED_CLOUD_PROVIDER_ORDER: tuple[str, ...] = ( "ollama_gcp_a", "ollama_gcp_b", *PAID_PROVIDER_ORDER, ) def production_provider_order( *, include_gemini: bool = True, include_claude: bool | None = None, ) -> list[str]: """Return a new list so callers cannot mutate the global contract.""" if include_claude is None: include_claude = include_gemini order = list(PRODUCTION_OLLAMA_ORDER) if include_claude: order.append("claude") if include_gemini: order.append("gemini") return 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. Paid providers are appended only when explicitly requested and local-only processing is not required. This means a disabled/removed Claude or Gemini lane 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: normalized.extend( provider for provider in PAID_PROVIDER_ORDER if provider in values ) return normalized, rejected def _paid_provider_readback( *, configured: bool, enabled: bool, authentication: dict[str, Any] | None, cost_guard: dict[str, Any] | None, ) -> dict[str, Any]: authentication_payload = dict(authentication or {}) cost_guard_payload = dict(cost_guard or {}) authenticated = authentication_payload.get("authenticated") authentication_verified = bool( authentication_payload.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 ) return { "configured": bool(configured), "enabled": bool(enabled), "route_ready": bool(enabled and cost_guard_ready and authentication_verified), "authenticated": authenticated, "authentication_verified": authentication_verified, "verification_status": authentication_payload.get( "verification_status", "not_verified" ), "verification_source": authentication_payload.get("verification_source"), "verified_at": authentication_payload.get("verified_at"), "cost_guard_ready": cost_guard_ready, "cost_guard": cost_guard_payload, } def production_route_readback( *, gemini_configured: bool, gemini_enabled: bool, gemini_authentication: dict[str, Any] | None = None, cost_guard: dict[str, Any] | None = None, gemini_execution_mode: str = "production", gemini_canary_percent: int = 0, claude_configured: bool = False, claude_enabled: bool = False, claude_authentication: dict[str, Any] | None = None, claude_cost_guard: dict[str, Any] | None = None, claude_execution_mode: str = "shadow", claude_canary_percent: int = 0, ) -> dict[str, Any]: """Build a public-safe route/status payload without endpoint URLs or keys.""" gemini = _paid_provider_readback( configured=gemini_configured, enabled=gemini_enabled, authentication=gemini_authentication, cost_guard=cost_guard, ) claude = _paid_provider_readback( configured=claude_configured, enabled=claude_enabled, authentication=claude_authentication, cost_guard=claude_cost_guard, ) gemini_mode = str(gemini_execution_mode or "shadow").strip().lower() gemini_percent = max(0, min(100, int(gemini_canary_percent or 0))) gemini["execution_mode"] = gemini_mode gemini["canary_percent"] = gemini_percent gemini["production_executable"] = bool( gemini["route_ready"] and gemini_mode == "production" ) gemini["canary_executable"] = bool( gemini["route_ready"] and gemini_mode == "canary" and gemini_percent > 0 ) execution_mode = str(claude_execution_mode or "shadow").strip().lower() canary_percent = max(0, min(100, int(claude_canary_percent or 0))) claude["execution_mode"] = execution_mode claude["canary_percent"] = canary_percent claude["production_executable"] = bool( claude["route_ready"] and execution_mode == "production" ) claude["canary_executable"] = bool( claude["route_ready"] and execution_mode == "canary" and canary_percent > 0 ) hops = [ { "position": 1, "provider": "ollama_gcp_a", "identity": "GCP-A", "runtime": "Ollama", "data_boundary": "cloud", "transport_security_status": "public_http_stopgap_mesh_pending", "paid": False, "production_executable": True, }, { "position": 2, "provider": "ollama_gcp_b", "identity": "GCP-B", "runtime": "Ollama", "data_boundary": "cloud", "transport_security_status": "public_http_stopgap_mesh_pending", "paid": False, "production_executable": True, }, { "position": 3, "provider": "ollama_local", "identity": "host111", "runtime": "Ollama", "data_boundary": "local", "transport_security_status": "private_lan", "paid": False, "production_executable": True, }, { "position": 4, "provider": "claude", "identity": "Anthropic Claude API", "runtime": "claude-sonnet-5", "paid": True, "production_executable": claude["production_executable"], "canary_executable": claude["canary_executable"], "execution_mode": execution_mode, "configured": bool(claude_configured), "authenticated": claude["authenticated"], "authentication_verified": claude["authentication_verified"], "verification_status": claude["verification_status"], }, { "position": 5, "provider": "gemini", "identity": "Gemini API", "runtime": "gemini-2.5-flash-lite", "paid": True, "production_executable": gemini["production_executable"], "canary_executable": gemini["canary_executable"], "execution_mode": gemini_mode, "configured": bool(gemini_configured), "authenticated": gemini["authenticated"], "authentication_verified": gemini["authentication_verified"], "verification_status": gemini["verification_status"], }, ] return { "schema_version": "ai_provider_production_route_v1", "provider_order": list(PRODUCTION_PROVIDER_ORDER), "route_label": ( "GCP-A -> GCP-B -> host111 Ollama -> " "Anthropic Claude API -> Gemini API" ), "transport_security": { "status": "degraded_mesh_pending", "production_closure": False, "reason": "gcp_ollama_public_http_stopgap", "replacement": "wireguard_mesh_10.77.114.x", "sanitized_generation_gate": ( "durable_same_run_endpoint_bound_transport_receipt_required" ), "receipt_ttl_max_seconds": 120, }, "hops": hops, "legacy_provider_policy": { provider: "non_executable_shadow_metadata_only" for provider in sorted(NON_EXECUTABLE_SHADOW_PROVIDERS) }, "claude": claude, "gemini": gemini, }