Spaces:
Running
Running
Download core/provider_availability.py from sqhcn/SmartPaper: direct link, hf CLI and curl.
- Browser
- Download file 21.9 kB
-
https://huggingface.co/spaces/sqhcn/SmartPaper/resolve/main/core/provider_availability.py
- Command line
-
hf download hf://spaces/sqhcn/SmartPaper/core/provider_availability.py
-
curl -L -o provider_availability.py https://huggingface.co/spaces/sqhcn/SmartPaper/resolve/main/core/provider_availability.py
21.9 kB
| # -*- coding: utf-8 -*- | |
| """Provider-neutral availability taxonomy and benchmark-only health tracker. | |
| A3-R5G scope is deliberately narrow: this module has no network calls, no | |
| production route activation, no Worker/D1 access, and no product entitlement | |
| logic. Provider/model/endpoints and all retry/circuit thresholds remain | |
| versioned policy data supplied by callers. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import asdict, dataclass, field | |
| from datetime import datetime, timezone | |
| from email.utils import parsedate_to_datetime | |
| import re | |
| import threading | |
| import time | |
| from typing import Any, Callable, Mapping | |
| ERROR_KINDS = { | |
| "AUTHENTICATION", | |
| "PERMISSION", | |
| "BALANCE", | |
| "RATE_LIMIT", | |
| "OVERLOAD", | |
| "QUOTA_EXHAUSTED", | |
| "TIMEOUT", | |
| "CONNECTION", | |
| "SERVER_ERROR", | |
| "REQUEST_INVALID", | |
| "CANCELLED", | |
| "UNKNOWN", | |
| } | |
| AVAILABILITY_STATES = { | |
| "UNKNOWN", | |
| "HEALTHY", | |
| "DEGRADED", | |
| "RATE_LIMITED", | |
| "OVERLOADED", | |
| "UNREACHABLE", | |
| "AUTH_BLOCKED", | |
| "BALANCE_BLOCKED", | |
| "CIRCUIT_OPEN", | |
| "DISABLED", | |
| } | |
| CIRCUIT_STATES = {"CLOSED", "OPEN", "HALF_OPEN"} | |
| class HealthKey: | |
| provider_profile_id: str | |
| endpoint_profile_id: str | |
| runtime_egress_profile: str | |
| def as_string(self) -> str: | |
| return "::".join( | |
| ( | |
| str(self.provider_profile_id or "").strip(), | |
| str(self.endpoint_profile_id or "").strip(), | |
| str(self.runtime_egress_profile or "").strip(), | |
| ) | |
| ) | |
| class ProviderErrorSignal: | |
| provider: str | |
| error_kind: str | |
| availability_state: str | |
| retryable: bool | |
| http_status: int | None = None | |
| provider_code: str = "" | |
| retry_after_seconds: float | None = None | |
| safe_reason_code: str = "" | |
| penalizes_health: bool = False | |
| def to_dict(self) -> dict[str, Any]: | |
| return asdict(self) | |
| class HealthSnapshot: | |
| availability_state: str = "UNKNOWN" | |
| circuit_state: str = "CLOSED" | |
| eligible: bool = True | |
| recent_successes: int = 0 | |
| recent_failures: int = 0 | |
| success_rate: float = 0.0 | |
| error_rate: float = 0.0 | |
| latency_ms: float = 0.0 | |
| blocked_until: float | None = None | |
| last_error_kind: str = "" | |
| last_provider_code: str = "" | |
| cause_state: str = "UNKNOWN" | |
| def to_dict(self) -> dict[str, Any]: | |
| return asdict(self) | |
| class _HealthRecord: | |
| availability_state: str = "UNKNOWN" | |
| circuit_state: str = "CLOSED" | |
| blocked_until: float | None = None | |
| last_error_kind: str = "" | |
| last_provider_code: str = "" | |
| consecutive_overloads: int = 0 | |
| events: list[tuple[float, bool, bool, float]] = field(default_factory=list) | |
| def _coerce_status(value: Any) -> int | None: | |
| try: | |
| number = int(value) | |
| except (TypeError, ValueError): | |
| return None | |
| return number if number > 0 else None | |
| def _coerce_positive_float(value: Any) -> float | None: | |
| try: | |
| number = float(value) | |
| except (TypeError, ValueError): | |
| return None | |
| return number if number >= 0 else None | |
| def _header_value(headers: Any, name: str) -> str: | |
| if not isinstance(headers, Mapping): | |
| return "" | |
| target = name.lower() | |
| for key, value in headers.items(): | |
| if str(key).lower() == target: | |
| return str(value or "").strip() | |
| return "" | |
| def _parse_retry_after(value: Any, *, now_epoch: float | None = None) -> float | None: | |
| text = str(value or "").strip() | |
| if not text: | |
| return None | |
| numeric = _coerce_positive_float(text) | |
| if numeric is not None: | |
| return numeric | |
| try: | |
| when = parsedate_to_datetime(text) | |
| if when.tzinfo is None: | |
| when = when.replace(tzinfo=timezone.utc) | |
| now = datetime.fromtimestamp(now_epoch if now_epoch is not None else time.time(), tz=timezone.utc) | |
| return max(0.0, (when.astimezone(timezone.utc) - now).total_seconds()) | |
| except Exception: | |
| return None | |
| def _parse_duration_seconds(value: Any) -> float | None: | |
| text = str(value or "").strip().lower() | |
| match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)s", text) | |
| if match: | |
| return float(match.group(1)) | |
| return _coerce_positive_float(text) | |
| def safe_response_metadata(response: Any, *, provider: str) -> dict[str, Any]: | |
| """Extract only non-sensitive operational metadata from a provider response.""" | |
| status = _coerce_status(getattr(response, "status_code", None)) | |
| retry_after = _parse_retry_after( | |
| _header_value(getattr(response, "headers", None), "Retry-After") | |
| ) | |
| provider_code = "" | |
| provider_reason = "" | |
| try: | |
| payload = response.json() | |
| except Exception: | |
| payload = None | |
| if isinstance(payload, Mapping): | |
| error = payload.get("error") | |
| error_map = error if isinstance(error, Mapping) else payload | |
| raw_code = error_map.get("code") if isinstance(error_map, Mapping) else None | |
| if raw_code is None: | |
| raw_code = payload.get("code") | |
| if raw_code is not None: | |
| provider_code = str(raw_code).strip()[:80] | |
| if isinstance(error_map, Mapping): | |
| raw_reason = error_map.get("status") or error_map.get("type") or error_map.get("reason") | |
| if raw_reason: | |
| provider_reason = str(raw_reason).strip()[:120] | |
| if retry_after is None: | |
| details = error_map.get("details") | |
| if isinstance(details, list): | |
| for detail in details: | |
| if not isinstance(detail, Mapping): | |
| continue | |
| retry_after = _parse_duration_seconds( | |
| detail.get("retryDelay") or detail.get("retry_delay") | |
| ) | |
| if retry_after is not None: | |
| break | |
| return { | |
| "provider": str(provider or "").strip().lower(), | |
| "http_status": status, | |
| "provider_code": provider_code, | |
| "provider_reason": provider_reason, | |
| "retry_after_seconds": retry_after, | |
| } | |
| def attach_safe_error_metadata(error: BaseException, response: Any, *, provider: str) -> BaseException: | |
| metadata = safe_response_metadata(response, provider=provider) | |
| for key, value in metadata.items(): | |
| try: | |
| setattr(error, key, value) | |
| except Exception: | |
| pass | |
| return error | |
| def _signal( | |
| *, | |
| provider: str, | |
| error_kind: str, | |
| availability_state: str, | |
| retryable: bool, | |
| http_status: int | None, | |
| provider_code: str, | |
| retry_after_seconds: float | None, | |
| safe_reason_code: str, | |
| penalizes_health: bool, | |
| ) -> ProviderErrorSignal: | |
| kind = str(error_kind or "UNKNOWN").upper() | |
| state = str(availability_state or "UNKNOWN").upper() | |
| if kind not in ERROR_KINDS: | |
| kind = "UNKNOWN" | |
| if state not in AVAILABILITY_STATES: | |
| state = "UNKNOWN" | |
| return ProviderErrorSignal( | |
| provider=str(provider or "").strip().lower(), | |
| error_kind=kind, | |
| availability_state=state, | |
| retryable=bool(retryable), | |
| http_status=_coerce_status(http_status), | |
| provider_code=str(provider_code or "").strip()[:80], | |
| retry_after_seconds=_coerce_positive_float(retry_after_seconds), | |
| safe_reason_code=str(safe_reason_code or "").strip()[:120], | |
| penalizes_health=bool(penalizes_health), | |
| ) | |
| def classify_provider_error( | |
| error: BaseException | None = None, | |
| *, | |
| provider: str = "", | |
| http_status: int | None = None, | |
| provider_code: Any = None, | |
| provider_reason: Any = None, | |
| retry_after_seconds: float | None = None, | |
| ) -> ProviderErrorSignal: | |
| """Normalize provider/network failures without retaining response bodies.""" | |
| provider_name = str(provider or getattr(error, "provider", "") or "").strip().lower() | |
| status = _coerce_status(http_status if http_status is not None else getattr(error, "http_status", None)) | |
| code = str( | |
| provider_code if provider_code is not None else getattr(error, "provider_code", "") or "" | |
| ).strip() | |
| reason = str( | |
| provider_reason if provider_reason is not None else getattr(error, "provider_reason", "") or "" | |
| ).strip() | |
| retry_after = _coerce_positive_float( | |
| retry_after_seconds if retry_after_seconds is not None else getattr(error, "retry_after_seconds", None) | |
| ) | |
| class_name = error.__class__.__name__.lower() if error is not None else "" | |
| safe_reason = ( | |
| reason if provider_name == "gemini" and reason | |
| else code or reason or (f"http_{status}" if status else class_name or "unknown") | |
| ) | |
| # Provider-specific business codes take precedence over outer HTTP status. | |
| if provider_name == "zhipu": | |
| if code in {"1000", "1001", "1003"}: | |
| return _signal(provider=provider_name, error_kind="AUTHENTICATION", availability_state="AUTH_BLOCKED", retryable=False, http_status=status, provider_code=code, retry_after_seconds=retry_after, safe_reason_code=safe_reason, penalizes_health=False) | |
| if code == "1113": | |
| return _signal(provider=provider_name, error_kind="BALANCE", availability_state="BALANCE_BLOCKED", retryable=False, http_status=status, provider_code=code, retry_after_seconds=retry_after, safe_reason_code=safe_reason, penalizes_health=False) | |
| if code == "1302": | |
| return _signal(provider=provider_name, error_kind="RATE_LIMIT", availability_state="RATE_LIMITED", retryable=True, http_status=status, provider_code=code, retry_after_seconds=retry_after, safe_reason_code=safe_reason, penalizes_health=True) | |
| if code == "1305": | |
| return _signal(provider=provider_name, error_kind="OVERLOAD", availability_state="OVERLOADED", retryable=True, http_status=status, provider_code=code, retry_after_seconds=retry_after, safe_reason_code=safe_reason, penalizes_health=True) | |
| if code == "1308": | |
| return _signal(provider=provider_name, error_kind="QUOTA_EXHAUSTED", availability_state="RATE_LIMITED", retryable=False, http_status=status, provider_code=code, retry_after_seconds=retry_after, safe_reason_code=safe_reason, penalizes_health=False) | |
| if "cancel" in class_name: | |
| return _signal(provider=provider_name, error_kind="CANCELLED", availability_state="UNKNOWN", retryable=False, http_status=status, provider_code=code, retry_after_seconds=retry_after, safe_reason_code=safe_reason, penalizes_health=False) | |
| if "timeout" in class_name or isinstance(error, TimeoutError): | |
| return _signal(provider=provider_name, error_kind="TIMEOUT", availability_state="DEGRADED", retryable=True, http_status=status, provider_code=code, retry_after_seconds=retry_after, safe_reason_code=safe_reason, penalizes_health=True) | |
| if "connection" in class_name or "chunked" in class_name: | |
| return _signal(provider=provider_name, error_kind="CONNECTION", availability_state="DEGRADED", retryable=True, http_status=status, provider_code=code, retry_after_seconds=retry_after, safe_reason_code=safe_reason, penalizes_health=True) | |
| if status == 401: | |
| kind, state, retryable, penalty = "AUTHENTICATION", "AUTH_BLOCKED", False, False | |
| elif status == 403: | |
| kind, state, retryable, penalty = "PERMISSION", "AUTH_BLOCKED", False, False | |
| elif status == 402: | |
| kind, state, retryable, penalty = "BALANCE", "BALANCE_BLOCKED", False, False | |
| elif status == 429: | |
| kind, state, retryable, penalty = "RATE_LIMIT", "RATE_LIMITED", True, True | |
| elif status == 503: | |
| kind, state, retryable, penalty = "OVERLOAD", "OVERLOADED", True, True | |
| elif status == 504: | |
| kind, state, retryable, penalty = "TIMEOUT", "DEGRADED", True, True | |
| elif status is not None and status >= 500: | |
| kind, state, retryable, penalty = "SERVER_ERROR", "DEGRADED", True, True | |
| elif status is not None and 400 <= status < 500: | |
| kind, state, retryable, penalty = "REQUEST_INVALID", "UNKNOWN", False, False | |
| elif "authentication" in class_name or "permission" in class_name: | |
| kind, state, retryable, penalty = "AUTHENTICATION", "AUTH_BLOCKED", False, False | |
| elif "ratelimit" in class_name or "rate_limit" in class_name: | |
| kind, state, retryable, penalty = "RATE_LIMIT", "RATE_LIMITED", True, True | |
| elif "request" in class_name: | |
| kind, state, retryable, penalty = "REQUEST_INVALID", "UNKNOWN", False, False | |
| else: | |
| kind, state, retryable, penalty = "UNKNOWN", "UNKNOWN", False, False | |
| return _signal( | |
| provider=provider_name, | |
| error_kind=kind, | |
| availability_state=state, | |
| retryable=retryable, | |
| http_status=status, | |
| provider_code=code, | |
| retry_after_seconds=retry_after, | |
| safe_reason_code=safe_reason, | |
| penalizes_health=penalty, | |
| ) | |
| def snapshot_from_legacy_health( | |
| value: Any, | |
| *, | |
| latency_degraded_threshold_ms: float = 0.0, | |
| ) -> HealthSnapshot: | |
| """Compatibility bridge for the pre-R5G benchmark health shape.""" | |
| if isinstance(value, Mapping): | |
| available = bool(value.get("available", False)) | |
| rate_limited = bool(value.get("rate_limited", False)) | |
| latency_ms = float(value.get("latency_ms") or 0.0) | |
| error_rate = max(0.0, min(float(value.get("error_rate") or 0.0), 1.0)) | |
| else: | |
| available = bool(value) | |
| rate_limited = False | |
| latency_ms = 0.0 | |
| error_rate = 0.0 | |
| if not available: | |
| state, eligible = "UNREACHABLE", False | |
| elif rate_limited: | |
| state, eligible = "RATE_LIMITED", False | |
| elif latency_degraded_threshold_ms > 0 and latency_ms > latency_degraded_threshold_ms: | |
| state, eligible = "DEGRADED", True | |
| else: | |
| state, eligible = "HEALTHY", True | |
| return HealthSnapshot( | |
| availability_state=state, | |
| circuit_state="CLOSED", | |
| eligible=eligible, | |
| recent_successes=1 if available else 0, | |
| recent_failures=0 if available else 1, | |
| success_rate=1.0 if available else 0.0, | |
| error_rate=error_rate, | |
| latency_ms=latency_ms, | |
| cause_state=state, | |
| ) | |
| class ProviderAvailabilityTracker: | |
| """Thread-safe, in-memory availability tracker for benchmark/testing only.""" | |
| def __init__(self, policy: Mapping[str, Any], *, clock: Callable[[], float] | None = None): | |
| self.policy = dict(policy or {}) | |
| self.clock = clock or time.monotonic | |
| self._lock = threading.RLock() | |
| self._records: dict[HealthKey, _HealthRecord] = {} | |
| def _now(self) -> float: | |
| return float(self.clock()) | |
| def _record(self, key: HealthKey) -> _HealthRecord: | |
| return self._records.setdefault(key, _HealthRecord()) | |
| def _window_seconds(self) -> float: | |
| return max(1.0, float(self.policy.get("failure_window_seconds") or 1.0)) | |
| def _failure_threshold(self) -> int: | |
| return max(1, int(self.policy.get("failure_threshold") or 1)) | |
| def _latency_threshold(self) -> float: | |
| return max(0.0, float(self.policy.get("latency_degraded_threshold_ms") or 0.0)) | |
| def _prune(self, record: _HealthRecord, now: float) -> None: | |
| cutoff = now - self._window_seconds() | |
| record.events[:] = [row for row in record.events if row[0] >= cutoff] | |
| def _penalizing_failures(self, record: _HealthRecord) -> int: | |
| return sum(1 for _, success, penalty, _ in record.events if not success and penalty) | |
| def _cooldown(self, record: _HealthRecord, signal: ProviderErrorSignal) -> float | None: | |
| if signal.retry_after_seconds is not None: | |
| return max(0.0, float(signal.retry_after_seconds)) | |
| if signal.error_kind == "OVERLOAD": | |
| schedule = [max(0.0, float(v)) for v in self.policy.get("overload_cooldown_schedule") or []] | |
| if schedule: | |
| index = min(max(record.consecutive_overloads - 1, 0), len(schedule) - 1) | |
| return schedule[index] | |
| if signal.error_kind in {"RATE_LIMIT", "QUOTA_EXHAUSTED"}: | |
| backoff = self.policy.get("rate_limit_backoff_policy") | |
| if isinstance(backoff, Mapping): | |
| base = max(0.0, float(backoff.get("base_seconds") or 0.0)) | |
| cap = max(base, float(backoff.get("max_seconds") or base)) | |
| attempts = max(1, self._penalizing_failures(record)) | |
| return min(cap, base * (2 ** max(0, attempts - 1))) | |
| return None | |
| def _refresh(self, record: _HealthRecord, now: float) -> None: | |
| self._prune(record, now) | |
| if record.circuit_state == "OPEN" and record.blocked_until is not None and now >= record.blocked_until: | |
| record.circuit_state = "HALF_OPEN" | |
| record.availability_state = "DEGRADED" | |
| record.blocked_until = None | |
| elif ( | |
| record.circuit_state == "CLOSED" | |
| and record.availability_state in {"RATE_LIMITED", "OVERLOADED"} | |
| and record.blocked_until is not None | |
| and now >= record.blocked_until | |
| ): | |
| record.availability_state = "DEGRADED" | |
| record.blocked_until = None | |
| def record_success(self, key: HealthKey, *, latency_ms: float = 0.0) -> HealthSnapshot: | |
| with self._lock: | |
| now = self._now() | |
| record = self._record(key) | |
| self._refresh(record, now) | |
| latency = max(0.0, float(latency_ms or 0.0)) | |
| record.events.append((now, True, False, latency)) | |
| record.circuit_state = "CLOSED" | |
| record.blocked_until = None | |
| record.last_error_kind = "" | |
| record.last_provider_code = "" | |
| record.consecutive_overloads = 0 | |
| threshold = self._latency_threshold() | |
| record.availability_state = "DEGRADED" if threshold > 0 and latency > threshold else "HEALTHY" | |
| return self._snapshot_locked(record, now) | |
| def record_failure( | |
| self, | |
| key: HealthKey, | |
| signal: ProviderErrorSignal, | |
| *, | |
| latency_ms: float = 0.0, | |
| ) -> HealthSnapshot: | |
| with self._lock: | |
| now = self._now() | |
| record = self._record(key) | |
| self._refresh(record, now) | |
| latency = max(0.0, float(latency_ms or 0.0)) | |
| record.events.append((now, False, bool(signal.penalizes_health), latency)) | |
| record.last_error_kind = signal.error_kind | |
| record.last_provider_code = signal.provider_code | |
| record.availability_state = signal.availability_state | |
| if signal.error_kind == "OVERLOAD": | |
| record.consecutive_overloads += 1 | |
| else: | |
| record.consecutive_overloads = 0 | |
| cooldown = self._cooldown(record, signal) | |
| if signal.availability_state in {"AUTH_BLOCKED", "BALANCE_BLOCKED"}: | |
| record.circuit_state = "CLOSED" | |
| record.blocked_until = None | |
| elif signal.error_kind in {"RATE_LIMIT", "QUOTA_EXHAUSTED"}: | |
| record.circuit_state = "CLOSED" | |
| record.blocked_until = now + cooldown if cooldown is not None else None | |
| else: | |
| immediate = { | |
| str(v or "").strip().upper() | |
| for v in self.policy.get("immediate_open_error_kinds") or [] | |
| } | |
| should_open = signal.error_kind in immediate | |
| if signal.penalizes_health and self._penalizing_failures(record) >= self._failure_threshold(): | |
| should_open = True | |
| if should_open: | |
| record.circuit_state = "OPEN" | |
| if cooldown is None: | |
| cooldown = max(0.0, float(self.policy.get("half_open_probe_interval") or 0.0)) | |
| record.blocked_until = now + cooldown if cooldown is not None else None | |
| return self._snapshot_locked(record, now) | |
| def reset(self, key: HealthKey) -> None: | |
| with self._lock: | |
| self._records.pop(key, None) | |
| def snapshot(self, key: HealthKey) -> HealthSnapshot: | |
| with self._lock: | |
| now = self._now() | |
| record = self._record(key) | |
| self._refresh(record, now) | |
| return self._snapshot_locked(record, now) | |
| def _snapshot_locked(self, record: _HealthRecord, now: float) -> HealthSnapshot: | |
| self._prune(record, now) | |
| successes = sum(1 for _, success, _, _ in record.events if success) | |
| failures = sum(1 for _, success, penalty, _ in record.events if not success and penalty) | |
| denominator = successes + failures | |
| success_rate = successes / denominator if denominator else 0.0 | |
| error_rate = failures / denominator if denominator else 0.0 | |
| latencies = [latency for _, _, _, latency in record.events if latency > 0] | |
| latency_ms = sum(latencies) / len(latencies) if latencies else 0.0 | |
| cause_state = record.availability_state | |
| state = "CIRCUIT_OPEN" if record.circuit_state == "OPEN" else cause_state | |
| eligible = ( | |
| record.circuit_state != "OPEN" | |
| and cause_state not in { | |
| "AUTH_BLOCKED", | |
| "BALANCE_BLOCKED", | |
| "RATE_LIMITED", | |
| "OVERLOADED", | |
| "UNREACHABLE", | |
| "DISABLED", | |
| } | |
| ) | |
| return HealthSnapshot( | |
| availability_state=state, | |
| circuit_state=record.circuit_state, | |
| eligible=eligible, | |
| recent_successes=successes, | |
| recent_failures=failures, | |
| success_rate=success_rate, | |
| error_rate=error_rate, | |
| latency_ms=latency_ms, | |
| blocked_until=record.blocked_until, | |
| last_error_kind=record.last_error_kind, | |
| last_provider_code=record.last_provider_code, | |
| cause_state=cause_state, | |
| ) | |