A3-R5G Gemini + GLM two-provider free pool benchmark availability

benchmarks/a3_provider_benchmark.py

@@ -25,6 +25,12 @@ if str(HF) not in sys.path:
 
 from core.model_usage import JsonlModelUsageSink, ModelUsageEvent  # noqa: E402
 from core.pipeline_agent import SSciPipelineAgent  # noqa: E402
+from core.provider_availability import (  # noqa: E402
+    HealthKey,
+    ProviderAvailabilityTracker,
+    classify_provider_error,
+    snapshot_from_legacy_health,
+)
 from core.provider_adapters.gemini_rest_adapter import GeminiRestAdapter  # noqa: E402
 from core.provider_adapters.glm_rest_adapter import ZhipuGLMRestAdapter  # noqa: E402
 
@@ -268,6 +274,7 @@ def run_case(
         "currency": "",
         "billing_window": "no_usage",
     }
+    availability_error = classify_provider_error(error, provider=provider).to_dict() if error else {}
     return {
         "case_id": str(case.get("case_id") or ""),
         "stage": str(case.get("stage") or ""),
@@ -284,6 +291,7 @@ def run_case(
         "gate_results": evaluation["gate_results"],
         "error_class": error.__class__.__name__ if error else "",
         "error_message": str(error)[:500] if error else "",
+        "availability_error": availability_error,
         "output_hash": _output_hash(output) if output else "",
         "output_preview": _result_preview(output),
         "usage": usage,
@@ -362,18 +370,36 @@ def _build_external_adapter(
     raise ValueError(f"unsupported external adapter type for {provider}: {adapter_type or '<empty>'}")
 
 
+def _profile_health_key(
+    *,
+    fixture: Mapping[str, Any],
+    profile_id: str,
+    profile: Mapping[str, Any],
+    runtime_egress_profile: str,
+) -> HealthKey:
+    provider = str(profile.get("provider") or "").strip().lower()
+    adapter_config = _provider_config(fixture, provider)
+    endpoint_profile_id = str(adapter_config.get("endpoint_profile_id") or provider or "unknown").strip()
+    return HealthKey(
+        provider_profile_id=str(profile_id or "").strip(),
+        endpoint_profile_id=endpoint_profile_id,
+        runtime_egress_profile=str(runtime_egress_profile or "benchmark_default").strip(),
+    )
+
+
 def select_profile_by_health(
     *,
     fixture: Mapping[str, Any],
     policy_id: str,
-    health: Mapping[str, Any],
+    health: Mapping[str, Any] | None = None,
+    tracker: ProviderAvailabilityTracker | None = None,
+    runtime_egress_profile: str | None = None,
 ) -> str:
-    """Config-driven route selection for benchmark simulations.
+    """Select a benchmark profile from versioned policy and server health.
 
-    The function deliberately consumes provider health observations rather than
-    country codes or browser-supplied location.  Production activation is out
-    of scope for A3-R5C; this helper only proves that future CN/non-CN switching
-    can be policy-driven without geography hard-coded in application logic.
+    R5G keeps backward compatibility with the original provider-keyed health
+    shape while delegating eligibility to the provider-neutral availability
+    snapshot contract.  No country/browser location participates in routing.
     """
     policies = fixture.get("routing_policies")
     if not isinstance(policies, Mapping):
@@ -381,41 +407,54 @@ def select_profile_by_health(
     policy = policies.get(policy_id)
     if not isinstance(policy, Mapping):
         raise ValueError(f"unknown routing policy: {policy_id}")
+
     allowed_cost_classes = {str(v).strip().lower() for v in policy.get("allowed_cost_classes") or []}
     profiles = fixture.get("profiles") if isinstance(fixture.get("profiles"), Mapping) else {}
     max_latency_ms = float(policy.get("max_latency_ms") or 0)
     max_error_rate = float(policy.get("max_error_rate") or 0)
-    skip_rate_limited = bool(policy.get("skip_rate_limited", True))
-    for profile_id in policy.get("candidate_profiles") or []:
-        profile_id = str(profile_id or "").strip()
+    runtime_profile = str(
+        runtime_egress_profile
+        or policy.get("runtime_egress_profile")
+        or "benchmark_default"
+    ).strip()
+    legacy_health = health if isinstance(health, Mapping) else {}
+
+    for raw_profile_id in policy.get("candidate_profiles") or []:
+        profile_id = str(raw_profile_id or "").strip()
         profile = profiles.get(profile_id) if isinstance(profiles, Mapping) else None
         if not isinstance(profile, Mapping):
             continue
+
+        # Cost/entitlement boundary is enforced before availability.
         cost_class = str(profile.get("monetary_cost_class") or "").strip().lower()
         if allowed_cost_classes and cost_class not in allowed_cost_classes:
             continue
-        provider = str(profile.get("provider") or "").strip().lower()
-        provider_health = health.get(provider)
-        if isinstance(provider_health, Mapping):
-            available = bool(provider_health.get("available", False))
-            latency_ms = float(provider_health.get("latency_ms") or 0)
-            error_rate = float(provider_health.get("error_rate") or 0)
-            rate_limited = bool(provider_health.get("rate_limited", False))
+
+        key = _profile_health_key(
+            fixture=fixture,
+            profile_id=profile_id,
+            profile=profile,
+            runtime_egress_profile=runtime_profile,
+        )
+        if tracker is not None:
+            snapshot = tracker.snapshot(key)
         else:
-            available = bool(provider_health)
-            latency_ms = 0.0
-            error_rate = 0.0
-            rate_limited = False
-        if not available:
-            continue
-        if skip_rate_limited and rate_limited:
+            provider = str(profile.get("provider") or "").strip().lower()
+            raw_health = legacy_health.get(key.as_string(), legacy_health.get(provider))
+            snapshot = snapshot_from_legacy_health(
+                raw_health,
+                latency_degraded_threshold_ms=max_latency_ms,
+            )
+
+        if not snapshot.eligible:
             continue
-        if max_latency_ms > 0 and latency_ms > max_latency_ms:
+        if max_latency_ms > 0 and snapshot.latency_ms > max_latency_ms:
             continue
-        if max_error_rate > 0 and error_rate > max_error_rate:
+        if max_error_rate > 0 and snapshot.error_rate > max_error_rate:
             continue
         return profile_id
-    raise RuntimeError(f"no_healthy_provider_for_policy:{policy_id}")
+
+    raise RuntimeError(f"no_eligible_provider_for_policy:{policy_id}")
 
 
 def run_benchmark(

benchmarks/fixtures/a3_benchmark_cases.json

@@ -1,6 +1,6 @@
 {
-  "benchmark_version": "a3-provider-benchmark-v2",
-  "notes": "Synthetic B1/B2 benchmark cases only; no live user prompts, credentials, payment, quota, or manuscript data. Provider/model/endpoints and environment-routing preferences are versioned configuration. Runtime geography is not hard-coded; any future production switching must use server-authoritative provider health/reachability and frozen task policy.",
+  "benchmark_version": "a3-provider-benchmark-v3",
+  "notes": "Synthetic B1/B2 benchmark cases only; no live user prompts, credentials, payment, quota, or manuscript data. A3-R5G Free Pool is limited to Gemini free + Zhipu GLM free. Provider/model/endpoints and all availability thresholds are versioned configuration. Runtime geography is not hard-coded; selection uses server-observed health under a frozen cost policy, and paid fallback is forbidden for the Free pool.",
   "profiles": {
     "gemini_free_mechanical": {
       "provider": "gemini",
@@ -529,27 +529,30 @@
   "provider_adapters": {
     "deepseek": {
       "adapter_type": "pipeline_deepseek",
-      "api_key_env": "DEEPSEEK_API_KEY"
+      "api_key_env": "DEEPSEEK_API_KEY",
+      "endpoint_profile_id": "deepseek_api_default"
     },
     "gemini": {
       "adapter_type": "gemini_rest",
       "api_key_env": "GEMINI_API_KEY",
       "base_url_env": "GEMINI_API_BASE_URL",
       "base_url": "https://generativelanguage.googleapis.com/v1beta",
-      "timeout_seconds": 120
+      "timeout_seconds": 120,
+      "endpoint_profile_id": "google_gemini_developer_api_v1beta"
     },
     "zhipu": {
       "adapter_type": "zhipu_glm_rest",
       "api_key_env": "ZHIPU_API_KEY",
       "base_url_env": "ZHIPU_API_BASE_URL",
       "base_url": "https://open.bigmodel.cn/api/paas/v4",
-      "timeout_seconds": 120
+      "timeout_seconds": 120,
+      "endpoint_profile_id": "zhipu_bigmodel_api_v4"
     }
   },
   "routing_policies": {
     "free_auto_v1": {
       "status": "benchmark_design_only",
-      "selection_strategy": "server_health_then_config_priority",
+      "selection_strategy": "cost_gate_then_availability_then_config_priority",
       "candidate_profiles": [
         "zhipu_glm47_free_no_thinking",
         "gemini_free_mechanical"
@@ -562,11 +565,17 @@
       "skip_rate_limited": true,
       "decision_inputs": [
         "server_reachability",
+        "provider_business_code",
         "recent_provider_health",
         "rate_limit_state",
+        "circuit_state",
         "latency"
       ],
-      "client_region_authoritative": false
+      "client_region_authoritative": false,
+      "availability_policy_id": "benchmark_free_pool_availability_v1",
+      "runtime_egress_profile": "benchmark_default",
+      "price_contract": "input_and_output_zero_required",
+      "paid_fallback_allowed": false
     },
     "standard_resilience_v1": {
       "status": "benchmark_design_only",
@@ -613,5 +622,40 @@
       ],
       "client_region_authoritative": false
     }
+  },
+  "availability_policies": {
+    "benchmark_free_pool_availability_v1": {
+      "status": "synthetic_benchmark_only",
+      "failure_window_seconds": 300,
+      "failure_threshold": 3,
+      "immediate_open_error_kinds": [
+        "OVERLOAD"
+      ],
+      "overload_cooldown_schedule": [
+        30,
+        120,
+        300
+      ],
+      "rate_limit_backoff_policy": {
+        "base_seconds": 15,
+        "max_seconds": 120
+      },
+      "half_open_probe_interval": 30,
+      "max_provider_attempts_per_node": 2,
+      "max_cross_provider_failovers_per_node": 1,
+      "latency_degraded_threshold_ms": 60000,
+      "production_tuning_authorized": false
+    }
+  },
+  "free_provider_pool": {
+    "status": "benchmark_only",
+    "providers": [
+      "zhipu",
+      "gemini"
+    ],
+    "policy_id": "free_auto_v1",
+    "price_contract": "input_and_output_zero_required",
+    "paid_fallback_allowed": false,
+    "production_activation": false
   }
 }

core/provider_adapters/gemini_rest_adapter.py

@@ -15,6 +15,7 @@ from typing import Any, Iterable, Mapping
 
 import requests
 
+from core.provider_availability import attach_safe_error_metadata
 from core.model_usage import (
     ModelUsageEvent,
     emit_usage_event,
@@ -186,16 +187,20 @@ class GeminiRestAdapter:
     def _raise_for_status(self, response: Any) -> None:
         status = int(getattr(response, "status_code", 0) or 0)
         detail = self._safe_error_detail(response)
+
+        def _raise(error: GeminiAPIError) -> None:
+            raise attach_safe_error_metadata(error, response, provider="gemini")
+
         if status in (401, 403):
-            raise GeminiAuthenticationError(f"Gemini authentication failed ({status}): {detail}")
+            _raise(GeminiAuthenticationError(f"Gemini authentication failed ({status}): {detail}"))
         if status == 429:
-            raise GeminiRateLimitError(f"Gemini rate limit reached (429): {detail}")
+            _raise(GeminiRateLimitError(f"Gemini rate limit reached (429): {detail}"))
         if 400 <= status < 500:
-            raise GeminiRequestError(f"Gemini request rejected ({status}): {detail}")
+            _raise(GeminiRequestError(f"Gemini request rejected ({status}): {detail}"))
         if status >= 500:
-            raise GeminiAPIError(f"Gemini provider error ({status}): {detail}")
+            _raise(GeminiAPIError(f"Gemini provider error ({status}): {detail}"))
         if status != 200:
-            raise GeminiAPIError(f"Gemini unexpected HTTP status ({status}): {detail}")
+            _raise(GeminiAPIError(f"Gemini unexpected HTTP status ({status}): {detail}"))
 
     def _event(
         self,

core/provider_adapters/glm_rest_adapter.py

@@ -16,6 +16,7 @@ from typing import Any, Iterable, Mapping
 
 import requests
 
+from core.provider_availability import attach_safe_error_metadata
 from core.model_usage import (
     ModelUsageEvent,
     emit_usage_event,
@@ -181,16 +182,23 @@ class ZhipuGLMRestAdapter:
     def _raise_for_status(self, response: Any) -> None:
         status = int(getattr(response, "status_code", 0) or 0)
         detail = self._safe_error_detail(response)
+
+        def _raise(error: ZhipuAPIError) -> None:
+            raise attach_safe_error_metadata(error, response, provider="zhipu")
+
         if status in (401, 403):
-            raise ZhipuAuthenticationError(f"Zhipu authentication failed ({status}): {detail}")
+            _raise(ZhipuAuthenticationError(f"Zhipu authentication failed ({status}): {detail}"))
         if status == 429:
-            raise ZhipuRateLimitError(f"Zhipu rate limit reached (429): {detail}")
+            # Preserve the existing public exception type for compatibility.
+            # A3-R5G classifies safe provider business codes (e.g. 1302 vs
+            # 1305) before the generic HTTP 429 meaning.
+            _raise(ZhipuRateLimitError(f"Zhipu rate limit reached (429): {detail}"))
         if 400 <= status < 500:
-            raise ZhipuRequestError(f"Zhipu request rejected ({status}): {detail}")
+            _raise(ZhipuRequestError(f"Zhipu request rejected ({status}): {detail}"))
         if status >= 500:
-            raise ZhipuAPIError(f"Zhipu provider error ({status}): {detail}")
+            _raise(ZhipuAPIError(f"Zhipu provider error ({status}): {detail}"))
         if status != 200:
-            raise ZhipuAPIError(f"Zhipu unexpected HTTP status ({status}): {detail}")
+            _raise(ZhipuAPIError(f"Zhipu unexpected HTTP status ({status}): {detail}"))
 
     def _event(
         self,

core/provider_availability.py

@@ -0,0 +1,542 @@
+# -*- 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"}
+
+
+@dataclass(frozen=True)
+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(),
+            )
+        )
+
+
+@dataclass(frozen=True)
+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)
+
+
+@dataclass(frozen=True)
+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)
+
+
+@dataclass
+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,
+        )