diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py
index 83d6fcc0bee..0806077d130 100644
--- a/litellm/litellm_core_utils/litellm_logging.py
+++ b/litellm/litellm_core_utils/litellm_logging.py
@@ -1854,15 +1854,7 @@ class Logging(LiteLLMLoggingBaseClass):
             self.model_call_details["end_time"] = end_time
             self.model_call_details["cache_hit"] = cache_hit
 
-            if self.call_type == CallTypes.anthropic_messages.value:
-                result = self._handle_anthropic_messages_response_logging(result=result)
-            elif (
-                self.call_type == CallTypes.generate_content.value
-                or self.call_type == CallTypes.agenerate_content.value
-            ):
-                result = self._handle_non_streaming_google_genai_generate_content_response_logging(result=result)
-            elif self.call_type == CallTypes.asend_message.value or self.call_type == CallTypes.send_message.value:
-                result = self._handle_a2a_response_logging(result=result)
+            result = self._normalize_result_for_call_type(result=result)
 
             logging_result = self.normalize_logging_result(result=result)
 
@@ -1910,6 +1902,36 @@ class Logging(LiteLLMLoggingBaseClass):
         except Exception as e:
             raise Exception(f"[Non-Blocking] LiteLLM.Success_Call Error: {str(e)}")
 
+    def _normalize_result_for_call_type(self, result: object) -> object:
+        """
+        Convert a call-type-specific result into the shape the logging pipeline expects.
+
+        A failure here must never skip the callbacks: a streaming call has a single
+        fire-and-forget success event, so raising loses the spend record entirely with
+        no client-visible symptom. Falling back to the untransformed result keeps
+        callbacks running on a degraded payload.
+        """
+        try:
+            if self.call_type == CallTypes.anthropic_messages.value:
+                return self._handle_anthropic_messages_response_logging(result=result)
+            if self.call_type in (
+                CallTypes.generate_content.value,
+                CallTypes.agenerate_content.value,
+            ):
+                return self._handle_non_streaming_google_genai_generate_content_response_logging(result=result)
+            if self.call_type in (
+                CallTypes.asend_message.value,
+                CallTypes.send_message.value,
+            ):
+                return self._handle_a2a_response_logging(result=result)
+            return result
+        except Exception as e:  # noqa: BLE001  # any normalizer failure must degrade, never skip callbacks
+            verbose_logger.exception(
+                f"LiteLLM.LoggingError: failed to normalize {self.call_type} result of type "
+                f"{type(result).__name__} for logging; logging the raw result instead. Error: {e}"
+            )
+            return result
+
     def _is_recognized_call_type_for_logging(
         self,
         logging_result: Any,
diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
index edc257f4c3f..86e7c8787e8 100644
--- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
+++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
@@ -693,6 +693,68 @@ async def test_anthropic_messages_marks_litellm_params_async():
         litellm.callbacks = original_callbacks
 
 
+def _anthropic_messages_logging_obj_for_unnormalizable_result(stream: bool) -> LitellmLogging:
+    logging_obj = LitellmLogging(
+        model="gpt-4o",
+        messages=[{"role": "user", "content": "hi"}],
+        stream=stream,
+        call_type="anthropic_messages",
+        start_time=time.time(),
+        litellm_call_id="normalization-failure",
+        function_id="fn",
+    )
+    logging_obj.update_environment_variables(
+        model="gpt-4o",
+        user="",
+        optional_params={},
+        litellm_params={"api_base": "", "aanthropic_messages": True},
+    )
+    return logging_obj
+
+
+def test_success_handler_helper_fn_falls_back_to_raw_result_on_normalization_failure():
+    """A result the ``anthropic_messages`` normalizer cannot handle (e.g. the raw
+    Responses API payload the OpenAI bridge hands over) used to raise out of
+    ``_success_handler_helper_fn`` before any callback ran, so a streaming call, which
+    dispatches exactly one fire-and-forget success event, lost its spend record with no
+    client-visible symptom. The helper now degrades to the untransformed result."""
+    logging_obj = _anthropic_messages_logging_obj_for_unnormalizable_result(stream=True)
+
+    unnormalizable_result = {"id": "resp_1", "object": "response", "output": []}
+
+    start_time, end_time, result = logging_obj._success_handler_helper_fn(result=unnormalizable_result)
+
+    assert result is unnormalizable_result
+    assert start_time is not None and end_time is not None
+
+
+@pytest.mark.asyncio
+async def test_async_success_handler_fires_callbacks_on_normalization_failure():
+    """The async success path must still invoke callbacks when result normalization
+    fails, otherwise ``/v1/messages`` calls silently log nothing."""
+    import litellm
+    from litellm.integrations.custom_logger import CustomLogger
+
+    class Tracker(CustomLogger):
+        def __init__(self):
+            super().__init__()
+            self.calls = []
+
+        async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
+            self.calls.append(response_obj)
+
+    tracker = Tracker()
+    logging_obj = _anthropic_messages_logging_obj_for_unnormalizable_result(stream=False)
+    original_async_callbacks = list(litellm._async_success_callback or [])
+    litellm._async_success_callback = [tracker]
+    try:
+        await logging_obj.async_success_handler(result={"id": "resp_1", "object": "response", "output": []})
+    finally:
+        litellm._async_success_callback = original_async_callbacks
+
+    assert len(tracker.calls) == 1
+
+
 @pytest.mark.asyncio
 async def test_agenerate_content_marks_litellm_params_async():
     """LIT-4475: the async ``agenerate_content`` entrypoint must plant