From 2d3e63095c025c2c435beb9280c65bc6f0f8ea59 Mon Sep 17 00:00:00 2001
From: Erin La <107987318+giterinhub@users.noreply.github.com>
Date: Thu, 23 Jul 2026 17:47:44 +0000
Subject: [PATCH 1/3] feat(genai): add FallbackPolicy for automatic model
 failovers (#1288)

---
 google/genai/__init__.py            |   8 +-
 google/genai/_fallback.py           | 155 ++++++++++++++++++++++++++++
 google/genai/tests/test_fallback.py |  64 ++++++++++++
 3 files changed, 225 insertions(+), 2 deletions(-)
 create mode 100644 google/genai/_fallback.py
 create mode 100644 google/genai/tests/test_fallback.py

diff --git a/google/genai/__init__.py b/google/genai/__init__.py
index 9c98f6ef6..df5b53921 100644
--- a/google/genai/__init__.py
+++ b/google/genai/__init__.py
@@ -19,8 +19,12 @@
 from . import types
 from . import version
 from .client import Client
-
+from google.genai._fallback import FallbackPolicy, APIError
 
 __version__ = version.__version__
 
-__all__ = ['Client']
+__all__ = [
+    'Client',
+    "FallbackPolicy",
+    "APIError",
+]
\ No newline at end of file
diff --git a/google/genai/_fallback.py b/google/genai/_fallback.py
new file mode 100644
index 000000000..e20c3b194
--- /dev/null
+++ b/google/genai/_fallback.py
@@ -0,0 +1,155 @@
+"""Model-agnostic fallback policy and middleware for Google GenAI SDK."""
+
+import copy
+import logging
+import asyncio
+from typing import List, Optional, Callable, Any, TypeVar
+
+try:
+    from google.genai.errors import APIError
+except ImportError:
+    class APIError(Exception):  # type: ignore
+        def __init__(self, code: int, response_json: Any = None):
+            self.code = code
+            self.response_json = response_json or {}
+            super().__init__(f"APIError {code}: {response_json}")
+
+logger = logging.getLogger("google.genai.fallback")
+
+T = TypeVar("T")
+
+def _default_is_retryable(exc: Exception, status_codes: List[int]) -> bool:
+    """Default predicate checking if an exception qualifies for fallback."""
+    if isinstance(exc, APIError):
+        return exc.code in status_codes
+    return False
+
+
+class FallbackPolicy:
+    """Production-grade, model-agnostic fallback middleware.
+    
+    This policy does not hardcode any model names. It accepts user-configured
+    fallback models and/or locations dynamically at runtime.
+
+    Attributes:
+        fallback_models: Sequence of model names to try if the primary model fails.
+        fallback_locations: Sequence of Vertex AI locations (regions) to try if primary fails.
+        retry_status_codes: List of HTTP status codes that trigger a fallback (default: 429, 503, 504).
+        is_retryable: Custom predicate function (exc -> bool) determining if an exception triggers fallback.
+        max_retries: Hard upper bound on total attempts.
+    """
+
+    def __init__(
+        self,
+        fallback_models: Optional[List[str]] = None,
+        fallback_locations: Optional[List[str]] = None,
+        retry_status_codes: Optional[List[int]] = None,
+        is_retryable: Optional[Callable[[Exception], bool]] = None,
+        max_retries: int = 3,
+    ):
+        self.fallback_models = fallback_models or []
+        self.fallback_locations = fallback_locations or []
+        self.retry_status_codes = retry_status_codes or [429, 503, 504]
+        self.is_retryable_custom = is_retryable
+        self.max_retries = max_retries
+
+    def is_eligible_for_fallback(self, exc: Exception) -> bool:
+        """Determines if the encountered error should trigger a fallback attempt."""
+        if self.is_retryable_custom is not None:
+            return self.is_retryable_custom(exc)
+        return _default_is_retryable(exc, self.retry_status_codes)
+
+    def _build_attempts_plan(self, kwargs: dict) -> List[dict]:
+        """Builds a sequence of kwarg dictionary payloads to attempt sequentially."""
+        primary_model = kwargs.get("model")
+        primary_location = kwargs.get("location")
+
+        # Build list of model targets
+        model_targets = [primary_model] if primary_model else []
+        for m in self.fallback_models:
+            if m not in model_targets:
+                model_targets.append(m)
+
+        # Build list of location targets
+        location_targets = [primary_location] if primary_location else []
+        for loc in self.fallback_locations:
+            if loc not in location_targets:
+                location_targets.append(loc)
+
+        plan = []
+        
+        # If model fallback sequence specified
+        if self.fallback_models and not self.fallback_locations:
+            for m in model_targets:
+                payload = copy.deepcopy(kwargs)
+                if m:
+                    payload["model"] = m
+                plan.append(payload)
+
+        # If location fallback sequence specified
+        elif self.fallback_locations and not self.fallback_models:
+            for loc in location_targets:
+                payload = copy.deepcopy(kwargs)
+                if loc:
+                    payload["location"] = loc
+                plan.append(payload)
+
+        # If both model and location fallbacks specified
+        elif self.fallback_models and self.fallback_locations:
+            for m in model_targets:
+                for loc in location_targets:
+                    payload = copy.deepcopy(kwargs)
+                    if m:
+                        payload["model"] = m
+                    if loc:
+                        payload["location"] = loc
+                    plan.append(payload)
+        else:
+            # Default: execute with original kwargs
+            plan.append(copy.deepcopy(kwargs))
+
+        return plan[: self.max_retries]
+
+    def execute_sync(self, func: Callable[..., T], **kwargs: Any) -> T:
+        """Executes a synchronous API call through the fallback sequence."""
+        attempts_plan = self._build_attempts_plan(kwargs)
+        last_exception: Optional[Exception] = None
+
+        for attempt_idx, payload in enumerate(attempts_plan, start=1):
+            try:
+                return func(**payload)
+            except Exception as e:
+                last_exception = e
+                if self.is_eligible_for_fallback(e) and attempt_idx < len(attempts_plan):
+                    logger.warning(
+                        f"[GenAI Fallback] Attempt {attempt_idx}/{len(attempts_plan)} failed "
+                        f"with {type(e).__name__}: {e}. Retrying with next fallback configuration..."
+                    )
+                    continue
+                raise e
+
+        if last_exception:
+            raise last_exception
+        raise RuntimeError("Fallback policy failed without raising an exception.")
+
+    async def execute_async(self, func: Callable[..., Any], **kwargs: Any) -> Any:
+        """Executes an asynchronous API call through the fallback sequence."""
+        attempts_plan = self._build_attempts_plan(kwargs)
+        last_exception: Optional[Exception] = None
+
+        for attempt_idx, payload in enumerate(attempts_plan, start=1):
+            try:
+                return await func(**payload)
+            except Exception as e:
+                last_exception = e
+                if self.is_eligible_for_fallback(e) and attempt_idx < len(attempts_plan):
+                    logger.warning(
+                        f"[GenAI Fallback Async] Attempt {attempt_idx}/{len(attempts_plan)} failed "
+                        f"with {type(e).__name__}: {e}. Retrying with next fallback configuration..."
+                    )
+                    continue
+                raise e
+
+        if last_exception:
+            raise last_exception
+        raise RuntimeError("Fallback policy failed without raising an exception.")
diff --git a/google/genai/tests/test_fallback.py b/google/genai/tests/test_fallback.py
new file mode 100644
index 000000000..50c69d91e
--- /dev/null
+++ b/google/genai/tests/test_fallback.py
@@ -0,0 +1,64 @@
+"""Model-agnostic pytest suite for FallbackPolicy."""
+
+import sys
+import os
+import pytest
+from unittest.mock import MagicMock, AsyncMock
+
+root_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
+if root_dir not in sys.path:
+    sys.path.insert(0, root_dir)
+
+import google
+google.__path__.append(os.path.join(root_dir, "google"))
+import google.genai
+google.genai.__path__.append(os.path.join(root_dir, "google", "genai"))
+
+from google.genai._fallback import FallbackPolicy, APIError
+
+def test_sync_primary_success():
+    policy = FallbackPolicy(fallback_models=["fallback-model-b"])
+    mock_func = MagicMock(return_value="primary_success")
+
+    result = policy.execute_sync(mock_func, model="primary-model-a", contents="Hello")
+    
+    assert result == "primary_success"
+    assert mock_func.call_count == 1
+    assert mock_func.call_args.kwargs["model"] == "primary-model-a"
+
+def test_sync_model_fallback_on_429():
+    policy = FallbackPolicy(fallback_models=["fallback-model-b"])
+    error_429 = APIError(429, {"message": "Rate limit exceeded"})
+    mock_func = MagicMock(side_effect=[error_429, "fallback_success"])
+
+    result = policy.execute_sync(mock_func, model="primary-model-a", contents="Hello")
+
+    assert result == "fallback_success"
+    assert mock_func.call_count == 2
+    assert mock_func.call_args_list[0].kwargs["model"] == "primary-model-a"
+    assert mock_func.call_args_list[1].kwargs["model"] == "fallback-model-b"
+
+def test_sync_region_fallback():
+    policy = FallbackPolicy(fallback_locations=["europe-west9", "europe-west1"])
+    error_503 = APIError(503, {"message": "Region unavailable"})
+    mock_func = MagicMock(side_effect=[error_503, "region_fallback_success"])
+
+    result = policy.execute_sync(
+        mock_func, model="user-configured-model", location="us-central1"
+    )
+
+    assert result == "region_fallback_success"
+    assert mock_func.call_count == 2
+    assert mock_func.call_args_list[0].kwargs["location"] == "us-central1"
+    assert mock_func.call_args_list[1].kwargs["location"] == "europe-west9"
+
+@pytest.mark.asyncio
+async def test_async_fallback_execution():
+    policy = FallbackPolicy(fallback_models=["backup-async-model"])
+    error_504 = APIError(504, {"message": "Gateway Timeout"})
+    mock_async = AsyncMock(side_effect=[error_504, "async_success"])
+
+    result = await policy.execute_async(mock_async, model="primary-async-model")
+    assert result == "async_success"
+    assert mock_async.call_count == 2
+    assert mock_async.call_args_list[1].kwargs["model"] == "backup-async-model"

From 94139beedabe281fe5bc70314eceacbd147a55a2 Mon Sep 17 00:00:00 2001
From: Erin La <107987318+giterinhub@users.noreply.github.com>
Date: Thu, 23 Jul 2026 18:04:17 +0000
Subject: [PATCH 2/3] feat(genai): enhance FallbackPolicy middleware (#1288)

Fixes #1288
Closes https://github.com/googleapis/python-genai/issues/1288

- Add automatic status code extraction across APIError, httpx, requests, and response objects
- Add backoff_factor parameter supporting configurable exponential backoff delays
- Add on_fallback telemetry callback hook for custom audit logging and metrics
- Add comprehensive pytest and unittest test suites covering max retries threshold, non-retryable fast-failing error codes, region+model matrix fallbacks, payload mutability audits, and 100-task high-concurrency async stress tests
---
 google/genai/_fallback.py                    |  95 ++++++-
 google/genai/tests/test_fallback.py          | 199 ++++++++++++-
 google/genai/tests/test_fallback_unittest.py | 280 +++++++++++++++++++
 3 files changed, 554 insertions(+), 20 deletions(-)
 create mode 100644 google/genai/tests/test_fallback_unittest.py

diff --git a/google/genai/_fallback.py b/google/genai/_fallback.py
index e20c3b194..e0dbe6c9a 100644
--- a/google/genai/_fallback.py
+++ b/google/genai/_fallback.py
@@ -1,33 +1,52 @@
 """Model-agnostic fallback policy and middleware for Google GenAI SDK."""
 
+import asyncio
 import copy
 import logging
-import asyncio
-from typing import List, Optional, Callable, Any, TypeVar
+import time
+from typing import Any, Callable, List, Optional, TypeVar
 
 try:
     from google.genai.errors import APIError
 except ImportError:
+
     class APIError(Exception):  # type: ignore
         def __init__(self, code: int, response_json: Any = None):
             self.code = code
             self.response_json = response_json or {}
             super().__init__(f"APIError {code}: {response_json}")
 
+
 logger = logging.getLogger("google.genai.fallback")
 
 T = TypeVar("T")
 
+
+def _extract_status_code(exc: Exception) -> Optional[int]:
+    """Extracts HTTP status code across APIError, httpx, and requests exceptions."""
+    if hasattr(exc, "code") and isinstance(getattr(exc, "code"), int):
+        return getattr(exc, "code")
+    if hasattr(exc, "status_code") and isinstance(getattr(exc, "status_code"), int):
+        return getattr(exc, "status_code")
+    response = getattr(exc, "response", None)
+    if response is not None and hasattr(response, "status_code"):
+        code = getattr(response, "status_code")
+        if isinstance(code, int):
+            return code
+    return None
+
+
 def _default_is_retryable(exc: Exception, status_codes: List[int]) -> bool:
     """Default predicate checking if an exception qualifies for fallback."""
-    if isinstance(exc, APIError):
-        return exc.code in status_codes
+    code = _extract_status_code(exc)
+    if code is not None:
+        return code in status_codes
     return False
 
 
 class FallbackPolicy:
     """Production-grade, model-agnostic fallback middleware.
-    
+
     This policy does not hardcode any model names. It accepts user-configured
     fallback models and/or locations dynamically at runtime.
 
@@ -37,6 +56,8 @@ class FallbackPolicy:
         retry_status_codes: List of HTTP status codes that trigger a fallback (default: 429, 503, 504).
         is_retryable: Custom predicate function (exc -> bool) determining if an exception triggers fallback.
         max_retries: Hard upper bound on total attempts.
+        backoff_factor: Optional exponential backoff multiplier in seconds (default: 0.0).
+        on_fallback: Optional callback function called when a fallback attempt is initiated.
     """
 
     def __init__(
@@ -46,12 +67,18 @@ def __init__(
         retry_status_codes: Optional[List[int]] = None,
         is_retryable: Optional[Callable[[Exception], bool]] = None,
         max_retries: int = 3,
+        backoff_factor: float = 0.0,
+        on_fallback: Optional[Callable[[Exception, int, dict], None]] = None,
     ):
+        if max_retries < 1:
+            raise ValueError("max_retries must be at least 1")
         self.fallback_models = fallback_models or []
         self.fallback_locations = fallback_locations or []
         self.retry_status_codes = retry_status_codes or [429, 503, 504]
         self.is_retryable_custom = is_retryable
         self.max_retries = max_retries
+        self.backoff_factor = max(0.0, backoff_factor)
+        self.on_fallback = on_fallback
 
     def is_eligible_for_fallback(self, exc: Exception) -> bool:
         """Determines if the encountered error should trigger a fallback attempt."""
@@ -77,7 +104,7 @@ def _build_attempts_plan(self, kwargs: dict) -> List[dict]:
                 location_targets.append(loc)
 
         plan = []
-        
+
         # If model fallback sequence specified
         if self.fallback_models and not self.fallback_locations:
             for m in model_targets:
@@ -110,6 +137,42 @@ def _build_attempts_plan(self, kwargs: dict) -> List[dict]:
 
         return plan[: self.max_retries]
 
+    def _notify_and_delay_sync(
+        self, exc: Exception, attempt_idx: int, total_attempts: int, next_payload: dict
+    ) -> None:
+        logger.warning(
+            f"[GenAI Fallback] Attempt {attempt_idx}/{total_attempts} failed "
+            f"with {type(exc).__name__}: {exc}. Retrying with next fallback configuration..."
+        )
+        if self.on_fallback is not None:
+            try:
+                self.on_fallback(exc, attempt_idx, next_payload)
+            except Exception as hook_exc:
+                logger.error(f"[GenAI Fallback] Error in on_fallback hook: {hook_exc}")
+
+        if self.backoff_factor > 0:
+            delay = self.backoff_factor * (2 ** (attempt_idx - 1))
+            time.sleep(delay)
+
+    async def _notify_and_delay_async(
+        self, exc: Exception, attempt_idx: int, total_attempts: int, next_payload: dict
+    ) -> None:
+        logger.warning(
+            f"[GenAI Fallback Async] Attempt {attempt_idx}/{total_attempts} failed "
+            f"with {type(exc).__name__}: {exc}. Retrying with next fallback configuration..."
+        )
+        if self.on_fallback is not None:
+            try:
+                self.on_fallback(exc, attempt_idx, next_payload)
+            except Exception as hook_exc:
+                logger.error(
+                    f"[GenAI Fallback Async] Error in on_fallback hook: {hook_exc}"
+                )
+
+        if self.backoff_factor > 0:
+            delay = self.backoff_factor * (2 ** (attempt_idx - 1))
+            await asyncio.sleep(delay)
+
     def execute_sync(self, func: Callable[..., T], **kwargs: Any) -> T:
         """Executes a synchronous API call through the fallback sequence."""
         attempts_plan = self._build_attempts_plan(kwargs)
@@ -120,10 +183,12 @@ def execute_sync(self, func: Callable[..., T], **kwargs: Any) -> T:
                 return func(**payload)
             except Exception as e:
                 last_exception = e
-                if self.is_eligible_for_fallback(e) and attempt_idx < len(attempts_plan):
-                    logger.warning(
-                        f"[GenAI Fallback] Attempt {attempt_idx}/{len(attempts_plan)} failed "
-                        f"with {type(e).__name__}: {e}. Retrying with next fallback configuration..."
+                if self.is_eligible_for_fallback(e) and attempt_idx < len(
+                    attempts_plan
+                ):
+                    next_payload = attempts_plan[attempt_idx]
+                    self._notify_and_delay_sync(
+                        e, attempt_idx, len(attempts_plan), next_payload
                     )
                     continue
                 raise e
@@ -142,10 +207,12 @@ async def execute_async(self, func: Callable[..., Any], **kwargs: Any) -> Any:
                 return await func(**payload)
             except Exception as e:
                 last_exception = e
-                if self.is_eligible_for_fallback(e) and attempt_idx < len(attempts_plan):
-                    logger.warning(
-                        f"[GenAI Fallback Async] Attempt {attempt_idx}/{len(attempts_plan)} failed "
-                        f"with {type(e).__name__}: {e}. Retrying with next fallback configuration..."
+                if self.is_eligible_for_fallback(e) and attempt_idx < len(
+                    attempts_plan
+                ):
+                    next_payload = attempts_plan[attempt_idx]
+                    await self._notify_and_delay_async(
+                        e, attempt_idx, len(attempts_plan), next_payload
                     )
                     continue
                 raise e
diff --git a/google/genai/tests/test_fallback.py b/google/genai/tests/test_fallback.py
index 50c69d91e..e17f0de38 100644
--- a/google/genai/tests/test_fallback.py
+++ b/google/genai/tests/test_fallback.py
@@ -1,31 +1,47 @@
 """Model-agnostic pytest suite for FallbackPolicy."""
 
-import sys
+import asyncio
+import copy
 import os
+import sys
+from unittest.mock import AsyncMock, MagicMock
 import pytest
-from unittest.mock import MagicMock, AsyncMock
 
 root_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
 if root_dir not in sys.path:
     sys.path.insert(0, root_dir)
 
-import google
+import google  # noqa: E402
+
 google.__path__.append(os.path.join(root_dir, "google"))
-import google.genai
+import google.genai  # noqa: E402
+
 google.genai.__path__.append(os.path.join(root_dir, "google", "genai"))
 
-from google.genai._fallback import FallbackPolicy, APIError
+from google.genai._fallback import APIError, FallbackPolicy  # noqa: E402
+
+
+class CustomStatusException(Exception):
+    def __init__(self, status_code: int):
+        self.status_code = status_code
+
+
+class ResponseStatusException(Exception):
+    def __init__(self, status_code: int):
+        self.response = MagicMock(status_code=status_code)
+
 
 def test_sync_primary_success():
     policy = FallbackPolicy(fallback_models=["fallback-model-b"])
     mock_func = MagicMock(return_value="primary_success")
 
     result = policy.execute_sync(mock_func, model="primary-model-a", contents="Hello")
-    
+
     assert result == "primary_success"
     assert mock_func.call_count == 1
     assert mock_func.call_args.kwargs["model"] == "primary-model-a"
 
+
 def test_sync_model_fallback_on_429():
     policy = FallbackPolicy(fallback_models=["fallback-model-b"])
     error_429 = APIError(429, {"message": "Rate limit exceeded"})
@@ -38,6 +54,7 @@ def test_sync_model_fallback_on_429():
     assert mock_func.call_args_list[0].kwargs["model"] == "primary-model-a"
     assert mock_func.call_args_list[1].kwargs["model"] == "fallback-model-b"
 
+
 def test_sync_region_fallback():
     policy = FallbackPolicy(fallback_locations=["europe-west9", "europe-west1"])
     error_503 = APIError(503, {"message": "Region unavailable"})
@@ -52,6 +69,7 @@ def test_sync_region_fallback():
     assert mock_func.call_args_list[0].kwargs["location"] == "us-central1"
     assert mock_func.call_args_list[1].kwargs["location"] == "europe-west9"
 
+
 @pytest.mark.asyncio
 async def test_async_fallback_execution():
     policy = FallbackPolicy(fallback_models=["backup-async-model"])
@@ -62,3 +80,172 @@ async def test_async_fallback_execution():
     assert result == "async_success"
     assert mock_async.call_count == 2
     assert mock_async.call_args_list[1].kwargs["model"] == "backup-async-model"
+
+
+def test_max_retries_threshold_exceeded():
+    """1. Max Retries Threshold: Re-raises underlying APIError when max_retries reached."""
+    policy = FallbackPolicy(
+        fallback_models=["model-b", "model-c", "model-d"], max_retries=3
+    )
+    error_429 = APIError(429, {"message": "Resource Exhausted"})
+    mock_func = MagicMock(
+        side_effect=[error_429, error_429, error_429, "should_not_reach"]
+    )
+
+    with pytest.raises(APIError) as exc_info:
+        policy.execute_sync(mock_func, model="model-a")
+
+    assert exc_info.value.code == 429
+    assert mock_func.call_count == 3
+    assert mock_func.call_args_list[0].kwargs["model"] == "model-a"
+    assert mock_func.call_args_list[1].kwargs["model"] == "model-b"
+    assert mock_func.call_args_list[2].kwargs["model"] == "model-c"
+
+
+@pytest.mark.parametrize("status_code", [400, 401, 403])
+def test_non_retryable_http_errors_fail_fast(status_code):
+    """2. Non-Retryable HTTP Errors: Fail fast on attempt 1 without attempting fallbacks."""
+    policy = FallbackPolicy(fallback_models=["fallback-model"])
+    error_non_retryable = APIError(
+        status_code, {"message": "Permission or Client Error"}
+    )
+    mock_func = MagicMock(side_effect=error_non_retryable)
+
+    with pytest.raises(APIError) as exc_info:
+        policy.execute_sync(mock_func, model="primary-model")
+
+    assert exc_info.value.code == status_code
+    assert mock_func.call_count == 1
+    assert mock_func.call_args.kwargs["model"] == "primary-model"
+
+
+def test_combined_model_and_region_matrix():
+    """3. Combined Model + Region Matrix: Evaluates combinations in order up to max_retries."""
+    policy = FallbackPolicy(
+        fallback_models=["model-b", "model-c"],
+        fallback_locations=["europe-west9", "europe-west1"],
+        max_retries=10,
+    )
+    error_503 = APIError(503, {"message": "Unavailable"})
+
+    # 3 models x 3 locations = 9 combinations
+    mock_func = MagicMock(side_effect=[error_503] * 8 + ["matrix_success"])
+
+    result = policy.execute_sync(mock_func, model="model-a", location="us-central1")
+
+    assert result == "matrix_success"
+    assert mock_func.call_count == 9
+
+    expected_calls = [
+        ("model-a", "us-central1"),
+        ("model-a", "europe-west9"),
+        ("model-a", "europe-west1"),
+        ("model-b", "us-central1"),
+        ("model-b", "europe-west9"),
+        ("model-b", "europe-west1"),
+        ("model-c", "us-central1"),
+        ("model-c", "europe-west9"),
+        ("model-c", "europe-west1"),
+    ]
+
+    for idx, (expected_model, expected_location) in enumerate(expected_calls):
+        call_kwargs = mock_func.call_args_list[idx].kwargs
+        assert call_kwargs["model"] == expected_model
+        assert call_kwargs["location"] == expected_location
+
+
+@pytest.mark.asyncio
+async def test_high_concurrency_async_stress():
+    """4. High-Concurrency Async Stress Test: 100 concurrent async tasks execute safely."""
+    policy = FallbackPolicy(fallback_models=["backup-model"])
+    error_503 = APIError(503, {"message": "Service Unavailable"})
+
+    async def mock_api(task_id: int, model: str):
+        if model == "primary-model":
+            raise error_503
+        return f"task_{task_id}_success"
+
+    tasks = [
+        policy.execute_async(mock_api, task_id=i, model="primary-model")
+        for i in range(100)
+    ]
+
+    results = await asyncio.gather(*tasks)
+    assert len(results) == 100
+    for i in range(100):
+        assert results[i] == f"task_{i}_success"
+
+
+def test_payload_mutability_audit():
+    """5. Payload Mutability Audit: Complex nested kwargs remain unmutated across retries."""
+    policy = FallbackPolicy(fallback_models=["model-b", "model-c"])
+    error_429 = APIError(429, {"message": "Rate limit"})
+
+    nested_contents = [{"role": "user", "parts": [{"text": "Generate analysis"}]}]
+    nested_config = {
+        "temperature": 0.7,
+        "safety_settings": [
+            {
+                "category": "HARM_CATEGORY_HATE_SPEECH",
+                "threshold": "BLOCK_LOW_AND_ABOVE",
+            }
+        ],
+    }
+
+    contents_copy = copy.deepcopy(nested_contents)
+    config_copy = copy.deepcopy(nested_config)
+
+    def side_effect(**kwargs):
+        # Mutate the kwargs passed to the function to test isolation
+        kwargs["contents"][0]["parts"][0]["text"] = "MUTATED"
+        kwargs["config"]["temperature"] = 999.0
+        if kwargs["model"] != "model-c":
+            raise error_429
+        return "success"
+
+    mock_func = MagicMock(side_effect=side_effect)
+
+    result = policy.execute_sync(
+        mock_func,
+        model="model-a",
+        contents=nested_contents,
+        config=nested_config,
+    )
+
+    assert result == "success"
+    # Original user objects MUST NOT be mutated by retries or inner function calls
+    assert nested_contents == contents_copy
+    assert nested_config == config_copy
+
+
+def test_status_code_extraction_support():
+    """Extracts status code from status_code attribute or response.status_code."""
+    policy = FallbackPolicy(fallback_models=["model-b"])
+
+    mock_func1 = MagicMock(side_effect=[CustomStatusException(429), "success_custom"])
+    result1 = policy.execute_sync(mock_func1, model="model-a")
+    assert result1 == "success_custom"
+
+    mock_func2 = MagicMock(side_effect=[ResponseStatusException(503), "success_resp"])
+    result2 = policy.execute_sync(mock_func2, model="model-a")
+    assert result2 == "success_resp"
+
+
+def test_on_fallback_hook():
+    """Triggers on_fallback hook with exception, attempt index, and next payload."""
+    hook_calls = []
+
+    def on_fallback(exc, attempt, next_payload):
+        hook_calls.append((exc, attempt, next_payload["model"]))
+
+    policy = FallbackPolicy(fallback_models=["model-b"], on_fallback=on_fallback)
+    error_429 = APIError(429, {"message": "Limit"})
+    mock_func = MagicMock(side_effect=[error_429, "hook_success"])
+
+    result = policy.execute_sync(mock_func, model="model-a")
+
+    assert result == "hook_success"
+    assert len(hook_calls) == 1
+    assert hook_calls[0][0] == error_429
+    assert hook_calls[0][1] == 1
+    assert hook_calls[0][2] == "model-b"
diff --git a/google/genai/tests/test_fallback_unittest.py b/google/genai/tests/test_fallback_unittest.py
new file mode 100644
index 000000000..c22c62f50
--- /dev/null
+++ b/google/genai/tests/test_fallback_unittest.py
@@ -0,0 +1,280 @@
+"""Model-agnostic unittest suite for FallbackPolicy."""
+
+import asyncio
+import copy
+import os
+import sys
+import unittest
+from unittest.mock import AsyncMock, MagicMock
+
+root_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
+if root_dir not in sys.path:
+    sys.path.insert(0, root_dir)
+
+import google  # noqa: E402
+
+google.__path__.append(os.path.join(root_dir, "google"))
+import google.genai  # noqa: E402
+
+google.genai.__path__.append(os.path.join(root_dir, "google", "genai"))
+
+from google.genai._fallback import APIError, FallbackPolicy  # noqa: E402
+
+
+class CustomStatusException(Exception):
+    def __init__(self, status_code: int):
+        self.status_code = status_code
+
+
+class ResponseStatusException(Exception):
+    def __init__(self, status_code: int):
+        self.response = MagicMock(status_code=status_code)
+
+
+class TestModelAgnosticFallbackPolicy(unittest.TestCase):
+    def test_sync_primary_success(self):
+        """Primary model succeeds without triggering fallbacks."""
+        policy = FallbackPolicy(fallback_models=["fallback-model-b"])
+        mock_func = MagicMock(return_value="primary_success")
+
+        result = policy.execute_sync(
+            mock_func, model="primary-model-a", contents="Hello"
+        )
+
+        self.assertEqual(result, "primary_success")
+        self.assertEqual(mock_func.call_count, 1)
+        self.assertEqual(mock_func.call_args.kwargs["model"], "primary-model-a")
+
+    def test_sync_model_fallback_on_429(self):
+        """Failover occurs to user-specified fallback model on 429 Rate Limit."""
+        policy = FallbackPolicy(fallback_models=["fallback-model-b"])
+        error_429 = APIError(429, {"message": "Rate limit exceeded"})
+        mock_func = MagicMock(side_effect=[error_429, "fallback_success"])
+
+        result = policy.execute_sync(
+            mock_func, model="primary-model-a", contents="Hello"
+        )
+
+        self.assertEqual(result, "fallback_success")
+        self.assertEqual(mock_func.call_count, 2)
+        self.assertEqual(mock_func.call_args_list[0].kwargs["model"], "primary-model-a")
+        self.assertEqual(
+            mock_func.call_args_list[1].kwargs["model"], "fallback-model-b"
+        )
+
+    def test_sync_region_fallback(self):
+        """Failover occurs across regions when location fallback is configured."""
+        policy = FallbackPolicy(fallback_locations=["europe-west9", "europe-west1"])
+        error_503 = APIError(503, {"message": "Region unavailable"})
+        mock_func = MagicMock(side_effect=[error_503, "region_fallback_success"])
+
+        result = policy.execute_sync(
+            mock_func, model="user-configured-model", location="us-central1"
+        )
+
+        self.assertEqual(result, "region_fallback_success")
+        self.assertEqual(mock_func.call_count, 2)
+        self.assertEqual(mock_func.call_args_list[0].kwargs["location"], "us-central1")
+        self.assertEqual(mock_func.call_args_list[1].kwargs["location"], "europe-west9")
+
+    def test_custom_is_retryable_predicate(self):
+        """Custom is_retryable predicate function determines fallback eligibility."""
+        custom_policy = FallbackPolicy(
+            fallback_models=["backup-model"],
+            is_retryable=lambda exc: getattr(exc, "code", None) == 500,
+        )
+        error_500 = APIError(500, {"message": "Internal Error"})
+        mock_func = MagicMock(side_effect=[error_500, "custom_predicate_success"])
+
+        result = custom_policy.execute_sync(mock_func, model="primary-model")
+        self.assertEqual(result, "custom_predicate_success")
+        self.assertEqual(mock_func.call_count, 2)
+
+    def test_async_fallback_execution(self):
+        """Async API calls correctly execute fallbacks."""
+
+        async def run_test():
+            policy = FallbackPolicy(fallback_models=["backup-async-model"])
+            error_504 = APIError(504, {"message": "Gateway Timeout"})
+            mock_async = AsyncMock(side_effect=[error_504, "async_success"])
+
+            result = await policy.execute_async(mock_async, model="primary-async-model")
+            self.assertEqual(result, "async_success")
+            self.assertEqual(mock_async.call_count, 2)
+            self.assertEqual(
+                mock_async.call_args_list[1].kwargs["model"],
+                "backup-async-model",
+            )
+
+        asyncio.run(run_test())
+
+    def test_max_retries_threshold_exceeded(self):
+        """1. Max Retries Threshold: Re-raises underlying APIError when max_retries reached."""
+        policy = FallbackPolicy(
+            fallback_models=["model-b", "model-c", "model-d"], max_retries=3
+        )
+        error_429 = APIError(429, {"message": "Resource Exhausted"})
+        mock_func = MagicMock(
+            side_effect=[error_429, error_429, error_429, "should_not_reach"]
+        )
+
+        with self.assertRaises(APIError) as cm:
+            policy.execute_sync(mock_func, model="model-a")
+
+        self.assertEqual(cm.exception.code, 429)
+        self.assertEqual(mock_func.call_count, 3)
+        self.assertEqual(mock_func.call_args_list[0].kwargs["model"], "model-a")
+        self.assertEqual(mock_func.call_args_list[1].kwargs["model"], "model-b")
+        self.assertEqual(mock_func.call_args_list[2].kwargs["model"], "model-c")
+
+    def test_non_retryable_http_errors_fail_fast(self):
+        """2. Non-Retryable HTTP Errors: Fail fast on attempt 1 for 400, 401, 403."""
+        policy = FallbackPolicy(fallback_models=["fallback-model"])
+
+        for status_code in [400, 401, 403]:
+            error_non_retryable = APIError(
+                status_code, {"message": f"Error {status_code}"}
+            )
+            mock_func = MagicMock(side_effect=error_non_retryable)
+
+            with self.assertRaises(APIError) as cm:
+                policy.execute_sync(mock_func, model="primary-model")
+
+            self.assertEqual(cm.exception.code, status_code)
+            self.assertEqual(mock_func.call_count, 1)
+            self.assertEqual(mock_func.call_args.kwargs["model"], "primary-model")
+
+    def test_combined_model_and_region_matrix(self):
+        """3. Combined Model + Region Matrix: Evaluates matrix combinations in order up to max_retries."""
+        policy = FallbackPolicy(
+            fallback_models=["model-b", "model-c"],
+            fallback_locations=["europe-west9", "europe-west1"],
+            max_retries=10,
+        )
+        error_503 = APIError(503, {"message": "Unavailable"})
+
+        mock_func = MagicMock(side_effect=[error_503] * 8 + ["matrix_success"])
+
+        result = policy.execute_sync(mock_func, model="model-a", location="us-central1")
+
+        self.assertEqual(result, "matrix_success")
+        self.assertEqual(mock_func.call_count, 9)
+
+        expected_calls = [
+            ("model-a", "us-central1"),
+            ("model-a", "europe-west9"),
+            ("model-a", "europe-west1"),
+            ("model-b", "us-central1"),
+            ("model-b", "europe-west9"),
+            ("model-b", "europe-west1"),
+            ("model-c", "us-central1"),
+            ("model-c", "europe-west9"),
+            ("model-c", "europe-west1"),
+        ]
+
+        for idx, (expected_model, expected_location) in enumerate(expected_calls):
+            call_kwargs = mock_func.call_args_list[idx].kwargs
+            self.assertEqual(call_kwargs["model"], expected_model)
+            self.assertEqual(call_kwargs["location"], expected_location)
+
+    def test_high_concurrency_async_stress(self):
+        """4. High-Concurrency Async Stress Test: 100 concurrent async tasks execute safely."""
+
+        async def run_stress_test():
+            policy = FallbackPolicy(fallback_models=["backup-model"])
+            error_503 = APIError(503, {"message": "Service Unavailable"})
+
+            async def mock_api(task_id: int, model: str):
+                if model == "primary-model":
+                    raise error_503
+                return f"task_{task_id}_success"
+
+            tasks = [
+                policy.execute_async(mock_api, task_id=i, model="primary-model")
+                for i in range(100)
+            ]
+
+            results = await asyncio.gather(*tasks)
+            self.assertEqual(len(results), 100)
+            for i in range(100):
+                self.assertEqual(results[i], f"task_{i}_success")
+
+        asyncio.run(run_stress_test())
+
+    def test_payload_mutability_audit(self):
+        """5. Payload Mutability Audit: Complex nested kwargs remain unmutated across retries."""
+        policy = FallbackPolicy(fallback_models=["model-b", "model-c"])
+        error_429 = APIError(429, {"message": "Rate limit"})
+
+        nested_contents = [{"role": "user", "parts": [{"text": "Generate analysis"}]}]
+        nested_config = {
+            "temperature": 0.7,
+            "safety_settings": [
+                {
+                    "category": "HARM_CATEGORY_HATE_SPEECH",
+                    "threshold": "BLOCK_LOW_AND_ABOVE",
+                }
+            ],
+        }
+
+        contents_copy = copy.deepcopy(nested_contents)
+        config_copy = copy.deepcopy(nested_config)
+
+        def side_effect(**kwargs):
+            kwargs["contents"][0]["parts"][0]["text"] = "MUTATED"
+            kwargs["config"]["temperature"] = 999.0
+            if kwargs["model"] != "model-c":
+                raise error_429
+            return "success"
+
+        mock_func = MagicMock(side_effect=side_effect)
+
+        result = policy.execute_sync(
+            mock_func,
+            model="model-a",
+            contents=nested_contents,
+            config=nested_config,
+        )
+
+        self.assertEqual(result, "success")
+        self.assertEqual(nested_contents, contents_copy)
+        self.assertEqual(nested_config, config_copy)
+
+    def test_status_code_extraction_support(self):
+        """Extracts status code from status_code attribute or response.status_code."""
+        policy = FallbackPolicy(fallback_models=["model-b"])
+
+        mock_func1 = MagicMock(
+            side_effect=[CustomStatusException(429), "success_custom"]
+        )
+        result1 = policy.execute_sync(mock_func1, model="model-a")
+        self.assertEqual(result1, "success_custom")
+
+        mock_func2 = MagicMock(
+            side_effect=[ResponseStatusException(503), "success_resp"]
+        )
+        result2 = policy.execute_sync(mock_func2, model="model-a")
+        self.assertEqual(result2, "success_resp")
+
+    def test_on_fallback_hook(self):
+        """Triggers on_fallback hook with exception, attempt index, and next payload."""
+        hook_calls = []
+
+        def on_fallback(exc, attempt, next_payload):
+            hook_calls.append((exc, attempt, next_payload["model"]))
+
+        policy = FallbackPolicy(fallback_models=["model-b"], on_fallback=on_fallback)
+        error_429 = APIError(429, {"message": "Limit"})
+        mock_func = MagicMock(side_effect=[error_429, "hook_success"])
+
+        result = policy.execute_sync(mock_func, model="model-a")
+
+        self.assertEqual(result, "hook_success")
+        self.assertEqual(len(hook_calls), 1)
+        self.assertEqual(hook_calls[0][0], error_429)
+        self.assertEqual(hook_calls[0][1], 1)
+        self.assertEqual(hook_calls[0][2], "model-b")
+
+
+if __name__ == "__main__":
+    unittest.main()

From 43d311662ec3e405567b5d7d2505cba67fb37654 Mon Sep 17 00:00:00 2001
From: Erin La <107987318+giterinhub@users.noreply.github.com>
Date: Fri, 24 Jul 2026 07:54:11 +0000
Subject: [PATCH 3/3] fix(genai): resolve mypy 3.11 type annotations in
 FallbackPolicy (#1288)

- Narrow return type in _extract_status_code to prevent no-any-return mypy error
- Add Dict[str, Any] type parameters for generic dict annotations
- Explicitly export APIError in __all__ to resolve attr-defined mypy error in __init__.py
---
 google/genai/_fallback.py | 38 ++++++++++++++++++++++++++------------
 1 file changed, 26 insertions(+), 12 deletions(-)

diff --git a/google/genai/_fallback.py b/google/genai/_fallback.py
index e0dbe6c9a..6b63a5e90 100644
--- a/google/genai/_fallback.py
+++ b/google/genai/_fallback.py
@@ -4,7 +4,7 @@
 import copy
 import logging
 import time
-from typing import Any, Callable, List, Optional, TypeVar
+from typing import Any, Callable, Dict, List, Optional, TypeVar
 
 try:
     from google.genai.errors import APIError
@@ -17,6 +17,8 @@ def __init__(self, code: int, response_json: Any = None):
             super().__init__(f"APIError {code}: {response_json}")
 
 
+__all__ = ["FallbackPolicy", "APIError"]
+
 logger = logging.getLogger("google.genai.fallback")
 
 T = TypeVar("T")
@@ -24,15 +26,19 @@ def __init__(self, code: int, response_json: Any = None):
 
 def _extract_status_code(exc: Exception) -> Optional[int]:
     """Extracts HTTP status code across APIError, httpx, and requests exceptions."""
-    if hasattr(exc, "code") and isinstance(getattr(exc, "code"), int):
-        return getattr(exc, "code")
-    if hasattr(exc, "status_code") and isinstance(getattr(exc, "status_code"), int):
-        return getattr(exc, "status_code")
+    if hasattr(exc, "code"):
+        code_val = getattr(exc, "code")
+        if isinstance(code_val, int):
+            return code_val
+    if hasattr(exc, "status_code"):
+        status_val = getattr(exc, "status_code")
+        if isinstance(status_val, int):
+            return status_val
     response = getattr(exc, "response", None)
     if response is not None and hasattr(response, "status_code"):
-        code = getattr(response, "status_code")
-        if isinstance(code, int):
-            return code
+        resp_code = getattr(response, "status_code")
+        if isinstance(resp_code, int):
+            return resp_code
     return None
 
 
@@ -68,7 +74,7 @@ def __init__(
         is_retryable: Optional[Callable[[Exception], bool]] = None,
         max_retries: int = 3,
         backoff_factor: float = 0.0,
-        on_fallback: Optional[Callable[[Exception, int, dict], None]] = None,
+        on_fallback: Optional[Callable[[Exception, int, Dict[str, Any]], None]] = None,
     ):
         if max_retries < 1:
             raise ValueError("max_retries must be at least 1")
@@ -86,7 +92,7 @@ def is_eligible_for_fallback(self, exc: Exception) -> bool:
             return self.is_retryable_custom(exc)
         return _default_is_retryable(exc, self.retry_status_codes)
 
-    def _build_attempts_plan(self, kwargs: dict) -> List[dict]:
+    def _build_attempts_plan(self, kwargs: Dict[str, Any]) -> List[Dict[str, Any]]:
         """Builds a sequence of kwarg dictionary payloads to attempt sequentially."""
         primary_model = kwargs.get("model")
         primary_location = kwargs.get("location")
@@ -138,7 +144,11 @@ def _build_attempts_plan(self, kwargs: dict) -> List[dict]:
         return plan[: self.max_retries]
 
     def _notify_and_delay_sync(
-        self, exc: Exception, attempt_idx: int, total_attempts: int, next_payload: dict
+        self,
+        exc: Exception,
+        attempt_idx: int,
+        total_attempts: int,
+        next_payload: Dict[str, Any],
     ) -> None:
         logger.warning(
             f"[GenAI Fallback] Attempt {attempt_idx}/{total_attempts} failed "
@@ -155,7 +165,11 @@ def _notify_and_delay_sync(
             time.sleep(delay)
 
     async def _notify_and_delay_async(
-        self, exc: Exception, attempt_idx: int, total_attempts: int, next_payload: dict
+        self,
+        exc: Exception,
+        attempt_idx: int,
+        total_attempts: int,
+        next_payload: Dict[str, Any],
     ) -> None:
         logger.warning(
             f"[GenAI Fallback Async] Attempt {attempt_idx}/{total_attempts} failed "