fix(utils): isolate callback errors in async_post_call_success_deployment_hook (#42535)

* fix(utils): isolate callback errors in async_post_call_success_deployment_hook

A callback that raises inside async_post_call_success_deployment_hook no longer
fails the completed request. The exception is logged with the callback class and
call_type, the response stays as it was, and later callbacks still run. Guardrail
callbacks are exempt because raising is how a post-call guardrail blocks

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(utils): drop unrelated ruff autofixes from test_utils

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(utils): drop fastapi import from guardrail propagation regression

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(utils): cover every success deployment hook call type with a raising hook

Parametrize the unit regression over video, embedding, responses, image, rerank,
transcription, chat and anthropic messages responses and assert the failure log
names the callback and call type. Run the integration test through a real proxy
for /v1/chat/completions, /v1/embeddings, /v1/responses and /v1/videos

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): move raising success hook cases into the existing callback delivery file

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: yucheng <yucheng@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/utils.py

@@ -1440,11 +1440,22 @@ async def async_post_call_success_deployment_hook(
     modified_response = response
 
     CustomLogger: Final = _get_cached_custom_logger()
+    CustomGuardrail: Final = _get_cached_custom_guardrail()
     for callback in litellm.callbacks:
         if isinstance(callback, CustomLogger):
-            result = await callback.async_post_call_success_deployment_hook(
-                request_data, cast(LLMResponseTypes, modified_response), typed_call_type
-            )
+            try:
+                result = await callback.async_post_call_success_deployment_hook(
+                    request_data, cast(LLMResponseTypes, modified_response), typed_call_type
+                )
+            except Exception:  # noqa: BLE001  # a broken callback must not fail a completed request
+                if isinstance(callback, CustomGuardrail):
+                    raise
+                verbose_logger.exception(
+                    "async_post_call_success_deployment_hook error in %s for call_type=%s",
+                    type(callback).__name__,
+                    typed_call_type,
+                )
+                continue
             if result is not None:
                 modified_response = result
 

tests/integration/contracts.json

@@ -1782,6 +1782,18 @@
     ],
     "tests/integration/mcp/test_mcp_lifecycle.py::test_same_url_server_grants_scope_discovery_and_direct_or_virtual_execution[bearer]": [
       "other.mcp.permissions.same_url_servers_enforce_discovery_and_execution"
+    ],
+    "tests/integration/observability/test_callback_delivery.py::test_response_survives_raising_success_deployment_hook[chat]": [
+      "other.observability.callbacks.raising_success_deployment_hook_keeps_response"
+    ],
+    "tests/integration/observability/test_callback_delivery.py::test_response_survives_raising_success_deployment_hook[embeddings]": [
+      "other.observability.callbacks.raising_success_deployment_hook_keeps_response"
+    ],
+    "tests/integration/observability/test_callback_delivery.py::test_response_survives_raising_success_deployment_hook[responses]": [
+      "other.observability.callbacks.raising_success_deployment_hook_keeps_response"
+    ],
+    "tests/integration/observability/test_callback_delivery.py::test_response_survives_raising_success_deployment_hook[videos]": [
+      "other.observability.callbacks.raising_success_deployment_hook_keeps_response"
     ]
   },
   "browser": {

tests/integration/observability/test_callback_delivery.py

@@ -1,13 +1,15 @@
+import base64
 import json
 import uuid
+from collections.abc import Callable, Mapping
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Final
 
 import pytest
 import yaml
-
-from integration._support.client import Gateway, eventually
+from integration._support.client import Gateway, JsonValue, eventually, object_value, string_value
 from integration._support.database import read_rows
 from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, wire_server
@@ -152,3 +154,163 @@ def test_concurrent_success_and_failure_join_callbacks_and_rows_without_credenti
                     assert rows[0]["prompt_tokens"] == event["prompt_tokens"]
                 else:
                     assert event["prompt_tokens"] == event["completion_tokens"] == rows[0]["completion_tokens"] == 0
+
+
+_RAISING_HOOK: Final = """
+from litellm.integrations.custom_logger import CustomLogger
+
+
+class RaisingHook(CustomLogger):
+    async def async_post_call_success_deployment_hook(self, request_data, response, call_type):
+        raise RuntimeError(f"hook rejected {type(response).__name__} for {call_type}")
+
+
+instance = RaisingHook()
+"""
+
+_VIDEO_JOB: Final = {
+    "id": "video_hook_isolation",
+    "object": "video",
+    "status": "queued",
+    "model": "sora-2",
+    "seconds": "4",
+    "size": "720x1280",
+}
+
+_UPSTREAM_REPLIES: Final[Mapping[str, Mapping[str, JsonValue]]] = {
+    "/v1/chat/completions": {
+        "id": "chatcmpl_hook_isolation",
+        "object": "chat.completion",
+        "created": 1,
+        "model": "gpt-5.6",
+        "choices": [{"index": 0, "message": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}],
+        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+    },
+    "/v1/embeddings": {
+        "object": "list",
+        "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}],
+        "model": "text-embedding-3-small",
+        "usage": {"prompt_tokens": 1, "total_tokens": 1},
+    },
+    "/v1/responses": {
+        "id": "resp_hook_isolation",
+        "object": "response",
+        "created_at": 1,
+        "status": "completed",
+        "model": "gpt-5.6",
+        "output": [
+            {
+                "type": "message",
+                "id": "msg_hook_isolation",
+                "status": "completed",
+                "role": "assistant",
+                "content": [{"type": "output_text", "text": "hi", "annotations": []}],
+            }
+        ],
+        "parallel_tool_calls": False,
+        "tool_choice": "auto",
+        "tools": [],
+        "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
+    },
+    "/v1/videos": _VIDEO_JOB,
+}
+
+
+def _item(value: JsonValue, index: int) -> JsonValue:
+    assert isinstance(value, list), f"Expected a list, received {type(value).__name__}"
+    return value[index]
+
+
+def _chat_text(body: dict[str, JsonValue]) -> str:
+    return string_value(object_value(object_value(_item(body["choices"], 0))["message"])["content"])
+
+
+def _embedding_vector(body: dict[str, JsonValue]) -> JsonValue:
+    return object_value(_item(body["data"], 0))["embedding"]
+
+
+def _responses_text(body: dict[str, JsonValue]) -> str:
+    return string_value(object_value(_item(object_value(_item(body["output"], 0))["content"], 0))["text"])
+
+
+def _video_job(body: dict[str, JsonValue]) -> tuple[str, str]:
+    encoded_id: Final = string_value(body["id"]).removeprefix("video_")
+    decoded: Final = base64.b64decode(encoded_id).decode()
+    return decoded.rsplit("video_id:", 1)[-1], string_value(body["status"])
+
+
+@dataclass(frozen=True, slots=True)
+class _Surface:
+    route: str
+    upstream_model: str
+    body: Callable[[str], dict[str, JsonValue]]
+    observed: Callable[[dict[str, JsonValue]], JsonValue | tuple[str, str]]
+    expected: JsonValue | tuple[str, str]
+
+
+_SURFACES: Final = (
+    pytest.param(
+        _Surface(
+            "/v1/chat/completions",
+            "openai/gpt-5.6",
+            lambda model: {"model": model, "messages": [{"role": "user", "content": "hook isolation"}]},
+            _chat_text,
+            "hi",
+        ),
+        id="chat",
+    ),
+    pytest.param(
+        _Surface(
+            "/v1/embeddings",
+            "openai/text-embedding-3-small",
+            lambda model: {"model": model, "input": "hook isolation"},
+            _embedding_vector,
+            [0.1, 0.2],
+        ),
+        id="embeddings",
+    ),
+    pytest.param(
+        _Surface(
+            "/v1/responses",
+            "openai/gpt-5.6",
+            lambda model: {"model": model, "input": "hook isolation"},
+            _responses_text,
+            "hi",
+        ),
+        id="responses",
+    ),
+    pytest.param(
+        _Surface(
+            "/v1/videos",
+            "openai/sora-2",
+            lambda model: {"model": model, "prompt": "a cat"},
+            _video_job,
+            (_VIDEO_JOB["id"], _VIDEO_JOB["status"]),
+        ),
+        id="videos",
+    ),
+)
+
+
+@pytest.mark.covers("other.observability.callbacks.raising_success_deployment_hook_keeps_response")
+@pytest.mark.parametrize("surface", _SURFACES)
+def test_response_survives_raising_success_deployment_hook(gateway: Gateway, tmp_path: Path, surface: _Surface) -> None:
+    def upstream(request: Request) -> Reply:
+        assert request.target == surface.route, request.target
+        assert b"hook isolation" in request.body or b"a cat" in request.body, request.body[:300]
+        return Reply(body=json.dumps(_UPSTREAM_REPLIES[surface.route]).encode())
+
+    (tmp_path / "raising_hook.py").write_text(_RAISING_HOOK)
+    config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+    config["litellm_settings"].update({"callbacks": ["raising_hook.instance"]})
+    path: Final = tmp_path / "hook.yaml"
+    path.write_text(yaml.safe_dump(config))
+    with (
+        wire_server(upstream) as provider,
+        owned_proxy(gateway, tmp_path, {}, config=path) as candidate,
+        candidate.scenario() as scenario,
+    ):
+        model: Final = scenario.model(model=surface.upstream_model, api_base=provider.url + "/v1")
+        response: Final = candidate.request("POST", surface.route, surface.body(model))
+        assert response.status_code == 200, response.text
+        assert surface.observed(object_value(response.json())) == surface.expected, response.text

tests/test_litellm/test_utils.py

@@ -32,27 +32,35 @@ from litellm._logging import (
 from litellm.caching.caching import Cache
 from litellm.caching.caching_handler import _PENDING_CACHE_WRITES
 from litellm.constants import DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT
+from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
 from litellm.litellm_core_utils.thread_pool_executor import executor as logging_executor
 from litellm.llms.base_llm.base_model_iterator import MockResponseIterator
 from litellm.proxy.utils import is_valid_api_key
 from litellm.types.integrations.custom_logger import HEADROOM_CONVERTED_STREAM_KEY
+from litellm.types.llms.openai import ResponsesAPIResponse
 from litellm.types.router import CredentialLiteLLMParams, GenericLiteLLMParams
 from litellm.types.utils import (
     ADDRESSED_RESPONSE_ID_FIELD,
     CallTypes,
     Choices,
     Delta,
+    EmbeddingResponse,
+    ImageResponse,
     LlmProviders,
+    LLMResponseTypes,
     ModelResponse,
     ModelResponseStream,
     PromptTokensDetailsWrapper,
+    RerankResponse,
     StreamingChoices,
+    TranscriptionResponse,
     Usage,
     all_litellm_params,
     bedrock_batch_litellm_params,
 )
+from litellm.types.videos.main import VideoObject
 from litellm.utils import (
     CustomStreamWrapper,
     ProviderConfigManager,
@@ -4412,6 +4420,111 @@ async def test_converted_chat_stream_hook_skips_unhandled_wrappers(
     assert wrapper.completion_stream is completion_stream
 
 
+class _ChatShapedSuccessDeploymentHook(CustomLogger):
+    async def async_post_call_success_deployment_hook(
+        self, request_data: dict[str, object], response: object, call_type: CallTypes | None
+    ) -> None:
+        raise AttributeError(f"{type(response).__name__!r} object has no attribute 'choices'")
+
+
+class _RecordingSuccessDeploymentHook(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.seen_responses: tuple[object, ...] = ()
+
+    async def async_post_call_success_deployment_hook(
+        self, request_data: dict[str, object], response: object, call_type: CallTypes | None
+    ) -> None:
+        self.seen_responses = (*self.seen_responses, response)
+
+
+_SUCCESS_RESPONSES_BY_CALL_TYPE: Final = (
+    pytest.param(
+        VideoObject(id="video_abc", object="video", status="queued", model="sora-2", seconds="4", size="720x1280"),
+        CallTypes.avideo_generation,
+        id="video",
+    ),
+    pytest.param(EmbeddingResponse(model="text-embedding-3-small"), CallTypes.aembedding, id="embedding"),
+    pytest.param(
+        ResponsesAPIResponse(
+            id="resp_abc", created_at=1, output=[], parallel_tool_calls=False, tool_choice="auto", tools=[], model="gpt-5.6"
+        ),
+        CallTypes.aresponses,
+        id="responses",
+    ),
+    pytest.param(ImageResponse(), CallTypes.aimage_generation, id="image"),
+    pytest.param(RerankResponse(id="rerank_abc"), CallTypes.arerank, id="rerank"),
+    pytest.param(TranscriptionResponse(text="hi"), CallTypes.atranscription, id="transcription"),
+    pytest.param(ModelResponse(model="gpt-5.6"), CallTypes.acompletion, id="chat"),
+    pytest.param(ModelResponse(model="claude-sonnet-4-5"), CallTypes.aanthropic_messages, id="anthropic_messages"),
+)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("response", "call_type"), _SUCCESS_RESPONSES_BY_CALL_TYPE)
+async def test_success_deployment_hook_raising_keeps_response_and_runs_later_hooks(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, response: object, call_type: CallTypes
+) -> None:
+    second_hook: Final = _RecordingSuccessDeploymentHook()
+    monkeypatch.setattr(litellm, "callbacks", [_ChatShapedSuccessDeploymentHook(), second_hook])
+
+    with caplog.at_level(logging.ERROR, logger=verbose_logger.name):
+        result: Final = await async_post_call_success_deployment_hook(
+            request_data={"model": "m"}, response=response, call_type=call_type
+        )
+
+    assert result is response
+    assert second_hook.seen_responses == (response,)
+    failure_logs: Final = tuple(r for r in caplog.records if "async_post_call_success_deployment_hook error" in r.message)
+    assert len(failure_logs) == 1
+    assert "_ChatShapedSuccessDeploymentHook" in failure_logs[0].message
+    assert str(call_type) in failure_logs[0].message
+    assert failure_logs[0].exc_info is not None
+
+
+@pytest.mark.asyncio
+async def test_success_deployment_hook_raising_keeps_earlier_hook_rewrite(monkeypatch: pytest.MonkeyPatch) -> None:
+    rewriter: Final = _RewritingSuccessDeploymentHook()
+    trailing_hook: Final = _RecordingSuccessDeploymentHook()
+    monkeypatch.setattr(litellm, "callbacks", [rewriter, _ChatShapedSuccessDeploymentHook(), trailing_hook])
+    original: Final = ModelResponse(model="gpt-5.6")
+
+    result: Final = await async_post_call_success_deployment_hook(
+        request_data={"model": "gpt-5.6"}, response=original, call_type=CallTypes.acompletion
+    )
+
+    assert isinstance(result, ModelResponse)
+    assert result is not original
+    assert result.choices[0].message.content == "rewritten by deployment hook"
+    assert trailing_hook.seen_responses == (result,)
+
+
+class _GuardrailBlocked(Exception):
+    pass
+
+
+class _BlockingSuccessDeploymentGuardrail(CustomGuardrail):
+    async def async_post_call_success_deployment_hook(
+        self, request_data: dict, response: LLMResponseTypes, call_type: CallTypes | None
+    ) -> LLMResponseTypes | None:
+        raise _GuardrailBlocked("Violated moderation policy")
+
+
+@pytest.mark.asyncio
+async def test_success_deployment_hook_still_propagates_guardrail_block(monkeypatch: pytest.MonkeyPatch) -> None:
+    later_hook: Final = _RewritingSuccessDeploymentHook()
+    monkeypatch.setattr(
+        litellm, "callbacks", [_BlockingSuccessDeploymentGuardrail(guardrail_name="blocking"), later_hook]
+    )
+
+    with pytest.raises(_GuardrailBlocked):
+        await async_post_call_success_deployment_hook(
+            request_data={"model": "gpt-5.6"}, response=ModelResponse(model="gpt-5.6"), call_type=CallTypes.acompletion
+        )
+
+    assert later_hook.seen_responses == ()
+
+
 @pytest.mark.asyncio
 @respx.mock
 async def test_wrapper_async_leaves_success_deployment_hook_off_requested_fake_stream(