feat: record and replay provider exceptions in inferencing integration tests (#4880)

## Summary
- Add exception serialization/deserialization to the API recording
system for accurate error replay during integration tests
- Exceptions are categorized as `llama_stack`, `provider_sdk`,
`builtin`, or `unknown`, with provider-specific reconstruction for
OpenAI and Ollama SDKs
- Patch `AsyncResponses.create` for Responses API error recording
- Add pluggable `testing/providers/` module for extending to new
provider SDKs

## Stack
This is PR 3/4 in the error message consistency series (split from
#3913). Depends on #4878 (included in this branch).

Incremental diff (PR 3 only): `api_recorder.py`, `exception_utils.py`,
`testing/providers/`

1. Error types foundation (#4878) — merge first
2. Responses API error handling (#4879)
3. **This PR** — Record/replay error support
4. Integration tests (#error-integration-tests)

## Test plan
- [x] All pre-commit hooks pass
- [x] CI passes

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: iamemilio

src/llama_stack/core/exceptions/translation.py

@@ -47,8 +47,8 @@ def translate_exception(exc: Exception) -> HTTPException:
         status_code = getattr(exc, "status_code", httpx.codes.INTERNAL_SERVER_ERROR)
         detail = str(exc)
         return HTTPException(status_code=status_code, detail=detail)
-    else:
-        return HTTPException(
-            status_code=httpx.codes.INTERNAL_SERVER_ERROR,
-            detail="Internal server error: An unexpected error occurred.",
-        )
+
+    return HTTPException(
+        status_code=httpx.codes.INTERNAL_SERVER_ERROR,
+        detail="Internal server error: An unexpected error occurred.",
+    )

src/llama_stack/testing/api_recorder.py

@@ -20,6 +20,7 @@
 
 from llama_stack.core.id_generation import reset_id_override, set_id_override
 from llama_stack.log import get_logger
+from llama_stack.testing.exception_utils import deserialize_exception, serialize_exception
 
 logger = get_logger(__name__, category="testing")
 
@@ -856,9 +857,20 @@ async def _patched_inference_method(original_method, self, client_type, endpoint
             recording = storage.find_recording(request_hash)
 
         if recording:
-            response_body = recording["response"]["body"]
+            response_data = recording["response"]
 
-            if recording["response"].get("is_streaming", False):
+            # Handle recorded exceptions
+            if response_data.get("is_exception", False):
+                exc_data = response_data.get("exception_data")
+                if exc_data:
+                    raise deserialize_exception(exc_data)
+                else:
+                    # Legacy format or unknown exception
+                    raise Exception(response_data.get("exception_message", "Unknown error"))
+
+            response_body = response_data["body"]
+
+            if response_data.get("is_streaming", False):
 
                 async def replay_stream():
                     for chunk in response_body:
@@ -889,15 +901,6 @@ async def replay_stream():
             )
 
     if mode == APIRecordingMode.RECORD or (mode == APIRecordingMode.RECORD_IF_MISSING and not recording):
-        if endpoint in ("/v1/models", "/v1/openai/v1/models"):
-            response = original_method(self, *args, **kwargs)
-        else:
-            response = await original_method(self, *args, **kwargs)
-
-        # we want to store the result of the iterator, not the iterator itself
-        if endpoint in ("/v1/models", "/v1/openai/v1/models"):
-            response = [m async for m in response]
-
         request_data = {
             "method": method,
             "url": url,
@@ -907,15 +910,49 @@ async def replay_stream():
             "model": body.get("model", ""),
         }
 
+        try:
+            if endpoint in ("/v1/models", "/v1/openai/v1/models"):
+                response = original_method(self, *args, **kwargs)
+            else:
+                response = await original_method(self, *args, **kwargs)
+
+            # we want to store the result of the iterator, not the iterator itself
+            if endpoint in ("/v1/models", "/v1/openai/v1/models"):
+                response = [m async for m in response]
+
+        except Exception as exc:
+            # Record the exception
+            response_data = {
+                "body": None,
+                "is_streaming": False,
+                "is_exception": True,
+                "exception_data": serialize_exception(exc),
+                "exception_message": str(exc),
+            }
+            storage.store_recording(request_hash, request_data, response_data)
+            raise  # Re-raise so recording mode still fails as expected
+
         # Determine if this is a streaming request based on request parameters
         is_streaming = body.get("stream", False)
 
         if is_streaming:
             # For streaming responses, we need to collect all chunks immediately before yielding
             # This ensures the recording is saved even if the generator isn't fully consumed
             chunks: list[Any] = []
-            async for chunk in response:
-                chunks.append(chunk)
+            try:
+                async for chunk in response:
+                    chunks.append(chunk)
+            except Exception as exc:
+                # Exception during streaming - record what we got plus the exception
+                response_data = {
+                    "body": chunks,
+                    "is_streaming": True,
+                    "is_exception": True,
+                    "exception_data": serialize_exception(exc),
+                    "exception_message": str(exc),
+                }
+                storage.store_recording(request_hash, request_data, response_data)
+                raise
 
             # Store the recording immediately
             response_data = {"body": chunks, "is_streaming": True}
@@ -946,6 +983,7 @@ def patch_inference_clients():
     from openai.resources.completions import AsyncCompletions
     from openai.resources.embeddings import AsyncEmbeddings
     from openai.resources.models import AsyncModels
+    from openai.resources.responses import AsyncResponses
 
     from llama_stack.providers.remote.tool_runtime.tavily_search.tavily_search import TavilySearchToolRuntimeImpl
 
@@ -955,6 +993,7 @@ def patch_inference_clients():
         "completions_create": AsyncCompletions.create,
         "embeddings_create": AsyncEmbeddings.create,
         "models_list": AsyncModels.list,
+        "responses_create": AsyncResponses.create,
         "ollama_generate": OllamaAsyncClient.generate,
         "ollama_chat": OllamaAsyncClient.chat,
         "ollama_embed": OllamaAsyncClient.embed,
@@ -990,11 +1029,17 @@ async def _iter():
 
         return _iter()
 
+    async def patched_responses_create(self, *args, **kwargs):
+        return await _patched_inference_method(
+            _original_methods["responses_create"], self, "openai", "/v1/responses", *args, **kwargs
+        )
+
     # Apply OpenAI patches
     AsyncChatCompletions.create = patched_chat_completions_create
     AsyncCompletions.create = patched_completions_create
     AsyncEmbeddings.create = patched_embeddings_create
     AsyncModels.list = patched_models_list
+    AsyncResponses.create = patched_responses_create
 
     # Create patched methods for Ollama client
     async def patched_ollama_generate(self, *args, **kwargs):
@@ -1068,6 +1113,7 @@ def unpatch_inference_clients():
     from openai.resources.completions import AsyncCompletions
     from openai.resources.embeddings import AsyncEmbeddings
     from openai.resources.models import AsyncModels
+    from openai.resources.responses import AsyncResponses
 
     from llama_stack.providers.remote.tool_runtime.tavily_search.tavily_search import TavilySearchToolRuntimeImpl
 
@@ -1076,6 +1122,7 @@ def unpatch_inference_clients():
     AsyncCompletions.create = _original_methods["completions_create"]
     AsyncEmbeddings.create = _original_methods["embeddings_create"]
     AsyncModels.list = _original_methods["models_list"]
+    AsyncResponses.create = _original_methods["responses_create"]
 
     # Restore Ollama client methods if they were patched
     OllamaAsyncClient.generate = _original_methods["ollama_generate"]

src/llama_stack/testing/exception_utils.py

@@ -0,0 +1,134 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+"""Shared exception handling utilities for recording/replaying exceptions.
+
+This module provides utilities for serializing and deserializing exceptions
+during API recording/replay. The exception types handled here use the shared
+mapping from mapping.py, ensuring consistency between runtime
+exception handling and test replay.
+"""
+
+from typing import Any, Protocol, TypeGuard
+
+import httpx
+
+from llama_stack.core.exceptions.mapping import EXCEPTION_TYPES_BY_NAME
+from llama_stack.testing.providers import GenericProviderError, create_provider_error, detect_provider
+from llama_stack_api.common.errors import LlamaStackError
+
+__all__ = [
+    "GenericProviderError",
+    "GenericLlamaStackError",
+    "ProviderSDKException",
+    "deserialize_exception",
+    "is_provider_sdk_exception",
+    "serialize_exception",
+]
+
+
+class ProviderSDKException(Protocol):
+    """Protocol for provider SDK exceptions with status_code attribute."""
+
+    status_code: int
+    body: dict | None
+
+
+class GenericLlamaStackError(LlamaStackError):
+    """A generic LlamaStackError for replay when exact type can't be reconstructed."""
+
+    def __init__(self, status_code_value: int, message: str = ""):
+        super().__init__(message)
+        # Override the class variable with an instance attribute
+        self.status_code = httpx.codes(status_code_value)
+
+
+def is_provider_sdk_exception(exc: Exception) -> TypeGuard[ProviderSDKException]:
+    """Check if exception is a provider SDK exception (e.g., OpenAI APIStatusError).
+
+    Provider SDK exceptions have a status_code attribute that indicates the HTTP
+    status code from the upstream provider. This matches the duck-typing used
+    in server.translate_exception().
+    """
+    return hasattr(exc, "status_code") and isinstance(getattr(exc, "status_code", None), int)
+
+
+def serialize_exception(exc: Exception) -> dict[str, Any]:
+    """Serialize an exception for recording.
+
+    Categories:
+    - llama_stack: LlamaStackError subclasses (internal errors)
+    - provider_sdk: Exceptions with status_code attr (OpenAI, etc.)
+    - builtin: Python built-in exceptions handled by translate_exception
+    - unknown: Everything else (will replay as generic Exception)
+    """
+    exc_type = type(exc).__name__
+    message = str(exc)
+
+    # Check categories in order of specificity
+    if isinstance(exc, LlamaStackError):
+        return {
+            "category": "llama_stack",
+            "type": exc_type,
+            "message": message,
+            "status_code": int(exc.status_code),
+        }
+    elif is_provider_sdk_exception(exc):
+        error_message = getattr(exc, "error", message)
+        return {
+            "category": "provider_sdk",
+            "provider": detect_provider(exc),
+            "type": exc_type,
+            "message": error_message,
+            "status_code": exc.status_code,
+            "body": getattr(exc, "body", None),
+        }
+    elif exc_type in EXCEPTION_TYPES_BY_NAME:
+        return {
+            "category": "builtin",
+            "type": exc_type,
+            "message": message,
+        }
+    else:
+        return {
+            "category": "unknown",
+            "type": exc_type,
+            "message": message,
+        }
+
+
+def deserialize_exception(data: dict[str, Any]) -> Exception:
+    """Reconstruct an exception from recorded data.
+
+    The reconstructed exception will have the same interface that
+    server.translate_exception() expects, ensuring consistent behavior
+    between live and replay modes.
+    """
+    category = data.get("category", "unknown")
+    exc_type = data.get("type", "Exception")
+    message = data.get("message", "Unknown error")
+
+    if category == "llama_stack":
+        status_code = data.get("status_code", 500)
+        return GenericLlamaStackError(status_code, message)
+
+    elif category == "provider_sdk":
+        provider = data.get("provider", "unknown")
+        return create_provider_error(
+            provider=provider,
+            status_code=data.get("status_code", 500),
+            body=data.get("body"),
+            message=message,
+        )
+
+    elif category == "builtin":
+        if exc_type in EXCEPTION_TYPES_BY_NAME:
+            return EXCEPTION_TYPES_BY_NAME[exc_type](message)
+        return Exception(message)
+
+    else:
+        # Unknown category - return generic exception
+        return Exception(message)

src/llama_stack/testing/providers/__init__.py

@@ -0,0 +1,66 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+"""Provider-specific exception handling for test recording/replay.
+
+Serializes provider SDK exceptions in record mode and reconstructs them in replay
+mode so integration tests see the original exception types.
+
+To add a provider:
+1. Create providers/<name>.py
+2. Define PROVIDER = ProviderConfig(
+       name="mycloud",           # Registry key stored in recordings
+       sdk_module=mycloud_sdk,   # The SDK module (``import mycloud_sdk``)
+       create_error=create_error,  # (status_code, body, message) -> Exception
+   )
+3. Import the module below and add its PROVIDER to the build_providers call
+"""
+
+from . import ollama, openai
+from ._config import ProviderConfig, _validate_provider
+
+
+def build_providers(*configs: ProviderConfig) -> dict[str, ProviderConfig]:
+    """Build PROVIDERS dict from registered provider configs. Validates on load."""
+    result: dict[str, ProviderConfig] = {}
+    for config in configs:
+        _validate_provider(config)
+        if config.name in result:
+            raise ValueError(f"Duplicate provider name: {config.name}") from None
+        result[config.name] = config
+    return result
+
+
+class GenericProviderError(Exception):
+    """Generic provider error for replay when provider-specific type can't be reconstructed."""
+
+    def __init__(self, status_code: int, body: dict | None = None, message: str = ""):
+        super().__init__(message)
+        self.status_code = status_code
+        self.body = body
+
+
+PROVIDERS: dict[str, ProviderConfig] = build_providers(
+    openai.PROVIDER,
+    ollama.PROVIDER,
+)
+
+
+def detect_provider(exc: object) -> str:
+    """Detect the provider from an exception's module."""
+    module = type(exc).__module__
+    for config in PROVIDERS.values():
+        if module.startswith(config._module_prefix):
+            return config.name
+    return "unknown"
+
+
+def create_provider_error(provider: str, status_code: int, body: dict | None, message: str) -> Exception:
+    """Reconstruct a provider-specific error from recorded data."""
+    if provider in PROVIDERS:
+        return PROVIDERS[provider].create_error(status_code, body, message)
+
+    return GenericProviderError(status_code, body, message)