diff --git a/litellm/_internal_context.py b/litellm/_internal_context.py
index 416563d5093..e97d79a95bc 100644
--- a/litellm/_internal_context.py
+++ b/litellm/_internal_context.py
@@ -44,6 +44,8 @@ _service_caller: Final[ContextVar[str | None]] = ContextVar("service_caller", de
 
 _mirrored_call: Final[ContextVar[bool]] = ContextVar("mirrored_call", default=False)
 
+_resent_on: Final[ContextVar[Callable[[Exception], bool] | None]] = ContextVar("resent_on", default=None)
+
 _request_parent_otel_span: Final[ContextVar[Span | None]] = ContextVar("request_parent_otel_span", default=None)
 
 
@@ -168,6 +170,23 @@ def in_mirrored_call() -> bool:
     return _mirrored_call.get()
 
 
+@contextmanager
+def resent_on(rejection: Callable[[Exception], bool]) -> Generator[None]:
+    """An attempt its caller sends again when it fails with ``rejection``. That failure is not the request's
+    outcome, so it is neither logged as the request's failure nor retried, and its budget reservation and rate-limit
+    slot stay with the attempt sent again."""
+    token: Final = _resent_on.set(rejection)
+    try:
+        yield
+    finally:
+        _restore(_resent_on, token)
+
+
+def is_resent_failure(error: Exception) -> bool:
+    rejection: Final = _resent_on.get()
+    return rejection is not None and rejection(error)
+
+
 def set_request_parent_otel_span(span: Span) -> None:
     """Anchor the proxy request's OTEL span for datastore service spans that no call site passes a parent to.
     Never reset: it dies with the request task and rides the tasks it spawns, like ``request.state``."""
diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py
index f6879c313de..861ffd91bfa 100644
--- a/litellm/completion_extras/litellm_responses_transformation/handler.py
+++ b/litellm/completion_extras/litellm_responses_transformation/handler.py
@@ -182,6 +182,9 @@ class ResponsesToCompletionBridgeHandler:
         model_response: Final = validated_kwargs["model_response"]
         logging_obj: Final = validated_kwargs["logging_obj"]
         custom_llm_provider: Final = validated_kwargs["custom_llm_provider"]
+        from .transformation import serves_anthropic_messages
+
+        streams_reasoning_signatures: Final = serves_anthropic_messages(logging_obj)
         if kwargs.get("stream") is True and "stream" not in optional_params:
             optional_params = {**optional_params, "stream": True}
         logging_obj.record_bridged_chat_request(optional_params, litellm_params)
@@ -262,6 +265,7 @@ class ResponsesToCompletionBridgeHandler:
                     cast("Mapping[str, object]", optional_params)  # cast-ok: optional_params is an untyped dict
                 ),
                 replays_openai_history=custom_llm_provider == "openai",
+                streams_reasoning_signatures=streams_reasoning_signatures,
             )
             streamwrapper: Final = CustomStreamWrapper(
                 completion_stream=completion_stream,
@@ -284,6 +288,9 @@ class ResponsesToCompletionBridgeHandler:
         model_response: Final = validated_kwargs["model_response"]
         logging_obj: Final = validated_kwargs["logging_obj"]
         custom_llm_provider: Final = validated_kwargs["custom_llm_provider"]
+        from .transformation import serves_anthropic_messages
+
+        streams_reasoning_signatures: Final = serves_anthropic_messages(logging_obj)
         if kwargs.get("stream") is True and "stream" not in optional_params:
             optional_params = {**optional_params, "stream": True}
         logging_obj.record_bridged_chat_request(optional_params, litellm_params)
@@ -367,6 +374,7 @@ class ResponsesToCompletionBridgeHandler:
                     cast("Mapping[str, object]", optional_params)  # cast-ok: optional_params is an untyped dict
                 ),
                 replays_openai_history=custom_llm_provider == "openai",
+                streams_reasoning_signatures=streams_reasoning_signatures,
             )
             streamwrapper: Final = CustomStreamWrapper(
                 completion_stream=completion_stream,
diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py
index fc7a6048538..a2de41a0c45 100644
--- a/litellm/completion_extras/litellm_responses_transformation/transformation.py
+++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py
@@ -39,7 +39,12 @@ from litellm.litellm_core_utils.allowed_tools_tool_choice import (
     parse_chat_allowed_tools_tool_choice,
     responses_allowed_tools_tool_choice,
 )
-from litellm.litellm_core_utils.core_helpers import drops_unsupported_params, iter_items
+from litellm.litellm_core_utils.core_helpers import (
+    as_str_mapping,
+    attribute_or_none,
+    drops_unsupported_params,
+    iter_items,
+)
 from litellm.litellm_core_utils.hidden_params import get_hidden_params, get_or_create_hidden_params
 from litellm.litellm_core_utils.prompt_templates.prompt_cache_breakpoint_utils import (
     prompt_cache_breakpoint_with_readable_mode,
@@ -48,6 +53,7 @@ from litellm.litellm_core_utils.prompt_templates.prompt_cache_breakpoint_utils i
 )
 from litellm.litellm_core_utils.prompt_templates.reasoning_utils import (
     responses_reasoning_items_from_thinking_blocks,
+    thinking_blocks_from_reasoning_items,
 )
 from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
 from litellm.llms.base_llm.bridges.completion_transformation import (
@@ -69,6 +75,8 @@ from litellm.responses.utils import (
 from litellm.types.llms.openai import (
     ChatCompletionAnnotation,
     ChatCompletionReasoningItem,
+    ChatCompletionRedactedThinkingBlock,
+    ChatCompletionThinkingBlock,
     ChatCompletionToolCallChunk,
     ChatCompletionToolCallFunctionChunk,
     ChatCompletionToolParamFunctionChunk,
@@ -77,7 +85,7 @@ from litellm.types.llms.openai import (
     ResponsesAPIResponse,
     ResponsesAPIStreamEvents,
 )
-from litellm.types.utils import GenericStreamingChunk, ModelResponseStream
+from litellm.types.utils import CallTypes, Delta, GenericStreamingChunk, ModelResponseStream, StreamingChoices
 
 if TYPE_CHECKING:
     from openai.types.responses import ResponseInputImageParam, ResponseOutputItem
@@ -93,8 +101,6 @@ if TYPE_CHECKING:
         AllMessageValues,
         ChatCompletionFileObject,
         ChatCompletionImageObject,
-        ChatCompletionRedactedThinkingBlock,
-        ChatCompletionThinkingBlock,
         ChatCompletionToolReferenceObject,
         OpenAIMessageContentListBlock,
     )
@@ -264,6 +270,10 @@ class _BuiltReasoningItem(TypedDict):
     summary: Sequence[_ReasoningSummaryText]
 
 
+def serves_anthropic_messages(logging_obj: object) -> bool:
+    return attribute_or_none(logging_obj, "call_type") == CallTypes.anthropic_messages.value
+
+
 def _get_reasoning_items(
     msg: "AllMessageValues",
 ) -> list[ChatCompletionReasoningItem]:
@@ -496,6 +506,77 @@ def _as_chat_reasoning_items(
     return cast(list[ChatCompletionReasoningItem], list(reasoning_items))
 
 
+def _with_stream_block_index(
+    block: Mapping[str, object],
+    index: int,
+    summary_streamed: bool,
+) -> ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock:
+    if block.get("type") == "redacted_thinking":
+        data: Final = block.get("data")
+        return ChatCompletionRedactedThinkingBlock(
+            type="redacted_thinking",
+            data=data if isinstance(data, str) else "",
+            index=index,
+        )
+    thinking: Final = block.get("thinking")
+    signature: Final = block.get("signature")
+    return ChatCompletionThinkingBlock(
+        type="thinking",
+        thinking=thinking if isinstance(thinking, str) and not summary_streamed else "",
+        signature=signature if isinstance(signature, str) else None,
+        index=index,
+    )
+
+
+def _stream_output_index(chunk: Mapping[str, object]) -> int | None:
+    output_index: Final = chunk.get("output_index")
+    return output_index if isinstance(output_index, int) and not isinstance(output_index, bool) else None
+
+
+def _reasoning_done_thinking_blocks(
+    item: Mapping[str, object],
+    output_index: int | None,
+    summary_streamed: bool,
+) -> tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...]:
+    index: Final = output_index if output_index is not None else 0
+    return tuple(
+        _with_stream_block_index(block, index, summary_streamed)
+        for block in thinking_blocks_from_reasoning_items((item,))
+    )
+
+
+def _reasoning_done_stream_chunk(
+    chunk: Mapping[str, object],
+    summary_streamed_outputs: frozenset[int | None],
+) -> ModelResponseStream | None:
+    event_type: Final = chunk.get("type")
+    event_name: Final = (
+        event_type.value
+        if isinstance(event_type, ResponsesAPIStreamEvents)
+        else event_type
+        if isinstance(event_type, str)
+        else None
+    )
+    if event_name != ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE:
+        return None
+    item: Final = as_str_mapping(chunk.get("item"))
+    if item is None or item.get("type") != "reasoning":
+        return None
+    output_index: Final = _stream_output_index(chunk)
+    signed_blocks: Final = _reasoning_done_thinking_blocks(item, output_index, output_index in summary_streamed_outputs)
+    if not signed_blocks:
+        return None
+    return ModelResponseStream(
+        choices=[
+            StreamingChoices(
+                index=0,
+                delta=Delta(thinking_blocks=list(signed_blocks)),
+                finish_reason=None,
+            )
+        ]
+    )
+
+
 def _shift_annotation(annotation: ChatCompletionAnnotation, offset: int) -> ChatCompletionAnnotation:
     shifted: Final = {
         key: value + offset if key in ("start_index", "end_index") and isinstance(value, int) else value
@@ -1280,6 +1361,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
         structured_output: bool = False,
         replay_openai_history: bool = False,
         completed: bool = False,
+        keeps_every_reasoning_item: bool = False,
     ) -> list["Choices"]:
         """
         Convert Responses API output items to chat completion choices.
@@ -1291,6 +1373,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
                 first message item becomes content
             replay_openai_history: Record the output's item sequence for the next turn
             completed: The response finished, so its item sequence is whole
+            keeps_every_reasoning_item: Keep each reasoning item before an answer, not only the last
 
         Returns:
             List of Choices objects
@@ -1323,7 +1406,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
         for item in output_items:
             if isinstance(item, ResponseReasoningItem):
                 pending_reasoning_items = (
-                    *(pending_reasoning_items if replay_openai_history else ()),
+                    *(pending_reasoning_items if replay_openai_history or keeps_every_reasoning_item else ()),
                     _build_reasoning_item(
                         item_id=item.id,
                         encrypted_content=getattr(item, "encrypted_content", None),
@@ -1573,6 +1656,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
             )
             == "openai",
             completed=raw_response.status == "completed",
+            keeps_every_reasoning_item=serves_anthropic_messages(logging_obj),
         )
 
         response_is_incomplete: Final = raw_response.status == "incomplete" or (
@@ -1637,6 +1721,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
         json_mode: bool | None = False,
         structured_output: bool = False,
         replays_openai_history: bool = False,
+        streams_reasoning_signatures: bool = False,
     ) -> BaseModelResponseIterator:
         return OpenAiResponsesToChatCompletionStreamIterator(
             streaming_response,
@@ -1644,6 +1729,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
             json_mode,
             structured_output=structured_output,
             replays_openai_history=replays_openai_history,
+            streams_reasoning_signatures=streams_reasoning_signatures,
         )
 
     def _convert_content_str_to_input_text(self, content: str, role: str) -> dict[str, object]:
@@ -2037,15 +2123,18 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
         json_mode: bool | None = False,
         structured_output: bool = False,
         replays_openai_history: bool = False,
+        streams_reasoning_signatures: bool = False,
     ):
         super().__init__(streaming_response, sync_stream, json_mode)
         self._replays_openai_history: Final = replays_openai_history
+        self._streams_reasoning_signatures: Final = streams_reasoning_signatures
         self._chat_completion_id: str | None = None
         self._served_service_tier: str | None = None
         self._tool_call_index_map: dict[int, int] = {}  # mutable-ok: per-stream accumulator state
         self._structured_output: Final = structured_output
         self._first_message_item_done: bool = False
         self._reasoning_summary_streamed: bool = False
+        self._summary_streamed_outputs: frozenset[int | None] = frozenset()
 
     def _handle_string_chunk(
         self, str_line: Union[str, "BaseModel"]
@@ -2263,7 +2352,6 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
                         )
                     ]
                 )
-
         elif event_type == "response.output_text.delta":
             # Content part added to output
             content_part = parsed_chunk.get("delta", None)
@@ -2380,9 +2468,16 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
         """
         verbose_logger.debug("Chat provider: transform_streaming_response called with chunk: %s", chunk)
         self._remember_served_service_tier(chunk)
+        reasoning_chunk: Final = (
+            _reasoning_done_stream_chunk(chunk, self._summary_streamed_outputs)
+            if self._streams_reasoning_signatures
+            else None
+        )
         translated: Final = (
             self._empty_content_chunk()
             if self._is_text_delta_after_structured_answer(chunk)
+            else reasoning_chunk
+            if reasoning_chunk is not None
             else OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(
                 chunk,
                 tool_call_index_map=self._tool_call_index_map,
@@ -2395,6 +2490,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
             self._first_message_item_done = True
         if _responses_event_type(chunk) == "response.reasoning_summary_text.delta" and chunk.get("delta"):
             self._reasoning_summary_streamed = True
+            self._summary_streamed_outputs = self._summary_streamed_outputs | {_stream_output_index(chunk)}
         return self._with_served_service_tier(self._with_stream_scoped_id(translated))
 
     def _is_text_delta_after_structured_answer(self, chunk: Mapping[str, object]) -> bool:
diff --git a/litellm/litellm_core_utils/prompt_templates/reasoning_utils.py b/litellm/litellm_core_utils/prompt_templates/reasoning_utils.py
index 817b1fa5915..ff1f2e017e3 100644
--- a/litellm/litellm_core_utils/prompt_templates/reasoning_utils.py
+++ b/litellm/litellm_core_utils/prompt_templates/reasoning_utils.py
@@ -5,6 +5,9 @@ from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
 from itertools import groupby
 from typing import Final, cast  # noqa: TID251  # Existing casts retained unchanged in this pure move
 
+from pydantic import TypeAdapter
+
+from litellm.litellm_core_utils.core_helpers import as_str_mapping
 from litellm.types.llms.openai import (
     ChatCompletionReasoningItem,
     ChatCompletionReasoningSummaryTextBlock,
@@ -94,6 +97,90 @@ def encrypted_content_of_block(block: Mapping[str, object]) -> str | None:
     return encrypted_content_from_signature(_encrypted_reasoning_field(block))
 
 
+_REASONING_SUMMARY_SEPARATOR: Final = "\n\n"
+
+
+_OBJECT_TUPLE: Final = TypeAdapter(tuple[object, ...])
+
+
+def _object_sequence(value: object) -> tuple[object, ...]:
+    if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
+        return ()
+    return _OBJECT_TUPLE.validate_python(value)
+
+
+def _iter_mappings(value: object) -> tuple[Mapping[str, object], ...]:
+    return tuple(mapping for item in _object_sequence(value) if (mapping := as_str_mapping(item)) is not None)
+
+
+def _reasoning_summary_part_text(part: object) -> str:
+    mapping: Final = as_str_mapping(part)
+    if mapping is not None:
+        text: Final = mapping.get("text")
+        return text if isinstance(text, str) else ""
+    text_attr: Final = getattr(part, "text", None)
+    return text_attr if isinstance(text_attr, str) else ""
+
+
+def _reasoning_item_summary_text(item: Mapping[str, object]) -> str:
+    parts: Final = tuple(_reasoning_summary_part_text(part) for part in _object_sequence(item.get("summary")))
+    return _REASONING_SUMMARY_SEPARATOR.join(part for part in parts if part)
+
+
+def thinking_block_from_reasoning_item(
+    item: object,
+) -> ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock | None:
+    """The chat thinking block that carries one Responses `reasoning` item back to an Anthropic client.
+
+    `encrypted_content` rides the same `litellm_encrypted_reasoning:` signature the Responses
+    adapter uses. A summary becomes thinking text. Bytes with no summary become `redacted_thinking`,
+    because an empty thinking string is not something the client can replay.
+    """
+    mapping: Final = as_str_mapping(item)
+    if mapping is None or mapping.get("type") not in (None, "reasoning"):
+        return None
+    encrypted_content: Final = mapping.get("encrypted_content")
+    if not isinstance(encrypted_content, str) or not encrypted_content:
+        return None
+    signature: Final = encrypted_reasoning_signature(encrypted_content)
+    text: Final = _reasoning_item_summary_text(mapping)
+    if not text:
+        return ChatCompletionRedactedThinkingBlock(type="redacted_thinking", data=signature)
+    return ChatCompletionThinkingBlock(type="thinking", thinking=text, signature=signature)
+
+
+def thinking_blocks_from_reasoning_items(
+    items: object,
+) -> tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...]:
+    return tuple(
+        block for item in _object_sequence(items) if (block := thinking_block_from_reasoning_item(item)) is not None
+    )
+
+
+def _block_encrypted_content(block: object) -> str | None:
+    mapping: Final = as_str_mapping(block)
+    if mapping is None:
+        return None
+    return encrypted_content_of_block(mapping)
+
+
+def thinking_blocks_with_encrypted_reasoning(
+    thinking_blocks: object,
+    reasoning_items: object,
+) -> tuple[Mapping[str, object], ...]:
+    """Thinking blocks already on the message, plus any encrypted reasoning item they do not already carry."""
+    direct: Final = _iter_mappings(thinking_blocks)
+    represented: Final = frozenset(
+        content for block in direct if (content := _block_encrypted_content(block)) is not None
+    )
+    extra: Final = tuple(
+        block
+        for block in thinking_blocks_from_reasoning_items(_iter_mappings(reasoning_items))
+        if _block_encrypted_content(block) not in represented
+    )
+    return (*extra, *direct)
+
+
 def is_encrypted_reasoning_block(block: object) -> bool:
     """A thinking or redacted_thinking block carrying Responses API encrypted reasoning.
 
@@ -195,6 +282,14 @@ def anthropic_content_lists(messages: Sequence[object]) -> Iterator[object]:
     )
 
 
+def _content_carries_encrypted_reasoning(content: object) -> bool:
+    return any(is_encrypted_reasoning_block(block) for block in _object_sequence(content))
+
+
+def messages_carry_encrypted_reasoning(messages: Sequence[object]) -> bool:
+    return any(_content_carries_encrypted_reasoning(content) for content in anthropic_content_lists(messages))
+
+
 def _strip_encrypted_reasoning_from_blocks(
     content: object,
     *,
diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py
index 6839e11f8c7..f7bd86f3a0c 100644
--- a/litellm/litellm_core_utils/streaming_handler.py
+++ b/litellm/litellm_core_utils/streaming_handler.py
@@ -1140,6 +1140,7 @@ class CustomStreamWrapper:
                 and model_response.choices[0].delta.role is not None
             )
             or (getattr(model_response.choices[0].delta, "reasoning_items", None) is not None)
+            or getattr(model_response.choices[0].delta, "thinking_blocks", None)
         ):
             return True
         else:
diff --git a/litellm/llms/anthropic/messages/adapters/handler.py b/litellm/llms/anthropic/messages/adapters/handler.py
index 64c7ac62d03..70b539a14d4 100644
--- a/litellm/llms/anthropic/messages/adapters/handler.py
+++ b/litellm/llms/anthropic/messages/adapters/handler.py
@@ -1,5 +1,8 @@
-from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
+import re
+from collections.abc import AsyncIterator, Callable, Coroutine, Iterator, Mapping, Sequence
+from contextlib import AbstractContextManager, nullcontext
 from dataclasses import dataclass
+from functools import partial
 from typing import (
     TYPE_CHECKING,
     Final,
@@ -11,17 +14,25 @@ from pydantic import TypeAdapter
 from typing_extensions import TypedDict
 
 import litellm
+from litellm._internal_context import resent_on
 from litellm._logging import verbose_logger
 from litellm.litellm_core_utils.asyncify import run_async_function
-from litellm.litellm_core_utils.core_helpers import as_str_mapping, drops_unsupported_params
+from litellm.litellm_core_utils.core_helpers import as_object_list, as_str_mapping, drops_unsupported_params
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
+from litellm.litellm_core_utils.model_param_helper import peek_reasoning_summary_aliases
+from litellm.litellm_core_utils.prompt_templates.reasoning_utils import (
+    is_encrypted_reasoning_block,
+    thinking_blocks_with_encrypted_reasoning,
+)
 from litellm.litellm_core_utils.tool_fields import (
     ROUTES_WITHOUT_MCP_SERVERS,
     drops_mcp_servers,
     tools_without_mcp_servers,
 )
+from litellm.llms.anthropic.common_utils import strip_encrypted_reasoning_blocks_from_anthropic_messages
 from litellm.llms.anthropic.messages.adapters.transformation import (
     AnthropicAdapter,
+    openai_thinking_routes_to_responses,
 )
 from litellm.llms.anthropic.messages.context_management import (
     AnthropicContextManagementError,
@@ -50,9 +61,9 @@ from litellm.types.llms.anthropic_messages.anthropic_response import (
 )
 from litellm.types.llms.openai import OpenAIWebSearchOptions
 from litellm.types.utils import ModelResponse
-from litellm.utils import get_model_info
 
 if TYPE_CHECKING:
+    from litellm.chat_completions.dispatch import ChatResult
     from litellm.router import Router
     from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth
 
@@ -70,22 +81,157 @@ class _CompletionKwargs(TypedDict, total=False, extra_items=object):
     custom_llm_provider: str
 
 
+_ANTHROPIC_MESSAGES: Final = TypeAdapter(list[dict[str, object]])
+_UNVERIFIED_ENCRYPTED_CONTENT: Final = re.compile(r"encrypted content \S+ could not be verified", re.IGNORECASE)
+_PrepareReplay: TypeAlias = "Callable[[_AnthropicMessages], _CompletionKwargs]"
+
+
 @dataclass(frozen=True, slots=True)
 class _ChatTargetReasoning:
     effort: str
     summary: str | None = None
 
 
+def _provider_error_code(body: object) -> str | None:
+    mapping: Final = as_str_mapping(body)
+    if mapping is None:
+        return None
+    nested: Final = as_str_mapping(mapping.get("error"))
+    source: Final = nested if nested is not None else mapping
+    code: Final = source.get("code")
+    return code if isinstance(code, str) else None
+
+
+def _logged_rejection_code(error: litellm.BadRequestError) -> str:
+    if error.code == "invalid_encrypted_content" or _provider_error_code(error.body) == "invalid_encrypted_content":
+        return "invalid_encrypted_content"
+    if "invalid_encrypted_content" in str(error):
+        return "invalid_encrypted_content"
+    return "could_not_be_verified"
+
+
+def _completion_text(completion_kwargs: _CompletionKwargs, key: str) -> str:
+    value: Final = completion_kwargs.get(key)
+    return value if isinstance(value, str) else ""
+
+
+def _message_carries_encrypted_reasoning(message: object) -> bool:
+    mapping: Final = as_str_mapping(message)
+    if mapping is None:
+        return False
+    blocks: Final = thinking_blocks_with_encrypted_reasoning(
+        mapping.get("thinking_blocks"), mapping.get("reasoning_items")
+    )
+    return any(is_encrypted_reasoning_block(block) for block in blocks)
+
+
+def _sends_encrypted_reasoning(completion_kwargs: _CompletionKwargs) -> bool:
+    messages: Final = as_object_list(completion_kwargs.get("messages")) or ()
+    return any(_message_carries_encrypted_reasoning(message) for message in messages)
+
+
+def _rejects_encrypted_reasoning(error: litellm.BadRequestError) -> bool:
+    error_message: Final = str(error)
+    return (
+        "invalid_encrypted_content" in error_message or _UNVERIFIED_ENCRYPTED_CONTENT.search(error_message) is not None
+    )
+
+
+def _is_rejected_reasoning(error: Exception) -> bool:
+    return isinstance(error, litellm.BadRequestError) and _rejects_encrypted_reasoning(error)
+
+
+def resent_without_rejected_reasoning(sent_encrypted_reasoning: bool) -> AbstractContextManager[None]:
+    return resent_on(_is_rejected_reasoning) if sent_encrypted_reasoning else nullcontext()
+
+
+def messages_without_rejected_reasoning(
+    messages: Sequence[Mapping[str, object]],
+    error: litellm.BadRequestError,
+    *,
+    sent_encrypted_reasoning: bool,
+    provider: str,
+    model: str,
+) -> "_AnthropicMessages | None":
+    if not sent_encrypted_reasoning or not _rejects_encrypted_reasoning(error):
+        return None
+    verbose_logger.warning(
+        "Backend rejected replayed encrypted reasoning, retrying once without it. "
+        "Enable the encrypted_content_affinity pre-call check to drop such blocks before the call. "
+        "provider=%s model=%s status=%s code=%s",
+        provider,
+        model,
+        error.status_code,
+        _logged_rejection_code(error),
+    )
+    return _ANTHROPIC_MESSAGES.validate_python(
+        strip_encrypted_reasoning_blocks_from_anthropic_messages([dict(message) for message in messages])
+    )
+
+
+def _messages_without_rejected_reasoning(
+    messages: _AnthropicMessages,
+    error: litellm.BadRequestError,
+    completion_kwargs: _CompletionKwargs,
+    sent_encrypted_reasoning: bool,
+) -> "_AnthropicMessages | None":
+    return messages_without_rejected_reasoning(
+        messages,
+        error,
+        sent_encrypted_reasoning=sent_encrypted_reasoning,
+        provider=_completion_text(completion_kwargs, "custom_llm_provider"),
+        model=_completion_text(completion_kwargs, "model"),
+    )
+
+
+async def _acompletion_without_rejected_reasoning(
+    completion_kwargs: _CompletionKwargs,
+    messages: _AnthropicMessages,
+    prepare_replay: _PrepareReplay,
+) -> tuple[_CompletionKwargs, "ChatResult"]:
+    sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(completion_kwargs)
+    try:
+        with resent_without_rejected_reasoning(sent_encrypted_reasoning):
+            return completion_kwargs, await litellm.acompletion(**completion_kwargs)
+    except litellm.BadRequestError as error:
+        replayed_messages: Final = _messages_without_rejected_reasoning(
+            messages, error, completion_kwargs, sent_encrypted_reasoning
+        )
+        if replayed_messages is None:
+            raise
+        replay_kwargs: Final = prepare_replay(replayed_messages)
+        return replay_kwargs, await litellm.acompletion(**replay_kwargs)
+
+
+def _completion_without_rejected_reasoning(
+    completion_kwargs: _CompletionKwargs,
+    messages: _AnthropicMessages,
+    prepare_replay: _PrepareReplay,
+) -> tuple[_CompletionKwargs, "ChatResult | Coroutine[object, object, ChatResult]"]:
+    sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(completion_kwargs)
+    try:
+        with resent_without_rejected_reasoning(sent_encrypted_reasoning):
+            return completion_kwargs, litellm.completion(**completion_kwargs)
+    except litellm.BadRequestError as error:
+        replayed_messages: Final = _messages_without_rejected_reasoning(
+            messages, error, completion_kwargs, sent_encrypted_reasoning
+        )
+        if replayed_messages is None:
+            raise
+        replay_kwargs: Final = prepare_replay(replayed_messages)
+        return replay_kwargs, litellm.completion(**replay_kwargs)
+
+
+def _is_compaction_block(block: object) -> bool:
+    mapping: Final = as_str_mapping(block)
+    return mapping is not None and mapping.get("type") == "compaction"
+
+
 def _messages_have_compaction_block(messages: _AnthropicMessages) -> bool:
     """Return True when any message carries a ``compaction`` content block."""
-    for msg in messages:
-        content = msg.get("content")
-        if not isinstance(content, list):
-            continue
-        for block in content:
-            if isinstance(block, dict) and block.get("type") == "compaction":
-                return True
-    return False
+    return any(
+        any(map(_is_compaction_block, content)) for msg in messages if isinstance(content := msg.get("content"), list)
+    )
 
 
 def _proxy_router_fallback() -> "Router | None":
@@ -419,34 +565,17 @@ class LiteLLMMessagesToCompletionTransformationHandler:
         If the user provides a `summary` field in the thinking dict, it is passed
         through to the OpenAI reasoning params (opt-in per OpenAI spec).
         """
-        custom_llm_provider = completion_kwargs.get("custom_llm_provider")
-        if custom_llm_provider is None:
-            try:
-                _, inferred_provider, _, _ = litellm.utils.get_llm_provider(
-                    model=cast(str, completion_kwargs.get("model"))
-                )
-                custom_llm_provider = inferred_provider
-            except Exception:
-                custom_llm_provider = None
-
-        if custom_llm_provider != "openai":
-            return
-
-        if not isinstance(thinking, dict) or thinking.get("type") != "enabled":
-            return
-
         model: Final = completion_kwargs.get("model")
-        try:
-            model_info: Final = get_model_info(model=cast(str, model), custom_llm_provider=custom_llm_provider)
-            if model_info and model_info.get("supports_reasoning") is False:
-                # Model doesn't support reasoning/responses API, don't route
-                return
-        except Exception:
-            pass
+        custom_llm_provider: Final = completion_kwargs.get("custom_llm_provider")
+        if not isinstance(model, str) or thinking is None:
+            return
+        if not openai_thinking_routes_to_responses(
+            model, custom_llm_provider if isinstance(custom_llm_provider, str) else None, thinking
+        ):
+            return
 
-        if isinstance(model, str) and model and "responses/" not in model:
-            local_model: Final = model.removeprefix(f"{custom_llm_provider}/")
-            completion_kwargs["model"] = f"{custom_llm_provider}/responses/{local_model}"
+        if "responses/" not in model:
+            completion_kwargs["model"] = f"openai/responses/{model.removeprefix('openai/')}"
 
         auto_summary: Final = is_reasoning_auto_summary_enabled()
 
@@ -677,6 +806,7 @@ class LiteLLMMessagesToCompletionTransformationHandler:
             custom_llm_provider=target_provider,
             api_base=api_base if isinstance(api_base, str) else None,
             deployment_model_info=as_str_mapping(provider_kwargs.get("model_info")),
+            reasoning_summary=peek_reasoning_summary_aliases(provider_kwargs),
             drop_params=drop_params,
         )
 
@@ -804,12 +934,9 @@ class LiteLLMMessagesToCompletionTransformationHandler:
             parent_trace_id=sub_call_trace_id,
         )
 
-        (
-            completion_kwargs,
-            tool_name_mapping,
-        ) = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
+        prepare: Final = partial(
+            LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs,
             max_tokens=max_tokens,
-            messages=effective_messages,
             model=model,
             metadata=metadata,
             stop_sequences=stop_sequences,
@@ -824,8 +951,10 @@ class LiteLLMMessagesToCompletionTransformationHandler:
             output_format=output_format,
             extra_kwargs=kwargs,
         )
-
-        completion_response: Final = await litellm.acompletion(**completion_kwargs)
+        initial_kwargs, tool_name_mapping = prepare(messages=effective_messages)
+        completion_kwargs, completion_response = await _acompletion_without_rejected_reasoning(
+            initial_kwargs, effective_messages, lambda replayed_messages: prepare(messages=replayed_messages)[0]
+        )
         logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
 
         if stream:
@@ -951,12 +1080,9 @@ class LiteLLMMessagesToCompletionTransformationHandler:
         effective_messages: Final = polyfill_result.messages if polyfill_result is not None else messages
         effective_system: Final = polyfill_result.system if polyfill_result is not None else system
 
-        (
-            completion_kwargs,
-            tool_name_mapping,
-        ) = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
+        prepare: Final = partial(
+            LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs,
             max_tokens=max_tokens,
-            messages=effective_messages,
             model=model,
             metadata=metadata,
             stop_sequences=stop_sequences,
@@ -971,8 +1097,10 @@ class LiteLLMMessagesToCompletionTransformationHandler:
             output_format=output_format,
             extra_kwargs=kwargs,
         )
-
-        completion_response: Final = litellm.completion(**completion_kwargs)
+        initial_kwargs, tool_name_mapping = prepare(messages=effective_messages)
+        completion_kwargs, completion_response = _completion_without_rejected_reasoning(
+            initial_kwargs, effective_messages, lambda replayed_messages: prepare(messages=replayed_messages)[0]
+        )
         logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
 
         if stream:
diff --git a/litellm/llms/anthropic/messages/adapters/streaming_iterator.py b/litellm/llms/anthropic/messages/adapters/streaming_iterator.py
index 87be996ebbf..cf96c1c5fda 100644
--- a/litellm/llms/anthropic/messages/adapters/streaming_iterator.py
+++ b/litellm/llms/anthropic/messages/adapters/streaming_iterator.py
@@ -132,14 +132,21 @@ def _delta_payload_field(delta_type: StreamingContentBlockDeltaType) -> str:
             assert_never(delta_type)
 
 
-def _first_thinking_block_index(chunk: "ModelResponseStream") -> int | None:
+def _first_thinking_block(chunk: "ModelResponseStream") -> Mapping[str, object] | None:
     delta: Final = _optional_attr(chunk.choices[0], "delta") if chunk.choices else None
     blocks: Final = _optional_attr_sequence(delta, "thinking_blocks")
-    first_block: Final = as_str_mapping(blocks[0]) if blocks else None
-    index: Final = first_block.get("index") if first_block is not None else None
+    return as_str_mapping(blocks[0]) if blocks else None
+
+
+def _thinking_block_index(block: Mapping[str, object] | None) -> int | None:
+    index: Final = block.get("index") if block is not None else None
     return index if isinstance(index, int) else None
 
 
+def _closes_thinking_block(block: Mapping[str, object] | None) -> bool:
+    return block is not None and _thinking_block_index(block) is not None and bool(block.get("signature"))
+
+
 def _optional_str_attr(obj: object, name: str) -> str | None:
     value: Final = _optional_attr(obj, name)
     return value if isinstance(value, str) else None
@@ -632,6 +639,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
     sent_content_block_finish: bool = False
     current_content_block_type: Literal["text", "tool_use", "thinking", "redacted_thinking"] = "text"
     current_thinking_block_index: int | None = None
+    current_thinking_block_closed: bool = False
     sent_last_message: bool = False
     holding_chunk: ContentBlockDelta | None = None
     holding_stop_reason_chunk: MessageBlockDelta | None = None
@@ -1575,9 +1583,16 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
                 original_name: Final = self.tool_name_mapping.get(truncated_name, truncated_name)
                 tool_block["name"] = original_name
 
-        thinking_block_index: Final = _first_thinking_block_index(chunk)
+        thinking_block: Final = _first_thinking_block(chunk)
+        thinking_block_index: Final = _thinking_block_index(thinking_block)
         previous_thinking_block_index: Final = self.current_thinking_block_index
-        if thinking_block_index is not None or block_type != self.current_content_block_type:
+        follows_closed_thinking_block: Final = self.current_thinking_block_closed and block_type == "thinking"
+        self.current_thinking_block_closed = _closes_thinking_block(thinking_block)
+        if (
+            thinking_block_index is not None
+            or block_type != self.current_content_block_type
+            or follows_closed_thinking_block
+        ):
             self.current_thinking_block_index = thinking_block_index
 
         if block_type != self.current_content_block_type:
@@ -1585,7 +1600,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
             self.current_content_block_start = content_block_start
             return True
 
-        if (
+        if follows_closed_thinking_block or (
             thinking_block_index is not None
             and previous_thinking_block_index is not None
             and thinking_block_index != previous_thinking_block_index
diff --git a/litellm/llms/anthropic/messages/adapters/transformation.py b/litellm/llms/anthropic/messages/adapters/transformation.py
index 5e4fba6d706..0c564a194dc 100644
--- a/litellm/llms/anthropic/messages/adapters/transformation.py
+++ b/litellm/llms/anthropic/messages/adapters/transformation.py
@@ -2,13 +2,14 @@ import copy
 import hashlib
 import json
 from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
+from dataclasses import dataclass
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, TypeVar, cast
 
 from pydantic import JsonValue, TypeAdapter
 
 import litellm
-from litellm.litellm_core_utils.core_helpers import as_str_mapping, normalize_drop_params
+from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_items, normalize_drop_params
 from litellm.litellm_core_utils.hidden_params import get_hidden_params
 from litellm.llms.anthropic.messages.convertible_blocks import tool_result_text_part
 from litellm.llms.anthropic.messages.utils import (
@@ -153,6 +154,7 @@ from litellm.litellm_core_utils.prompt_templates.prompt_cache_breakpoint_utils i
 )
 from litellm.litellm_core_utils.prompt_templates.reasoning_utils import (
     reasoning_content_from_thinking_blocks,
+    thinking_blocks_with_encrypted_reasoning,
 )
 from litellm.litellm_core_utils.prompt_templates.tool_call_json_repair import (
     parse_tool_call_arguments,
@@ -258,27 +260,72 @@ def target_supports_mid_conversation_system(model: str | None, custom_llm_provid
     return supports_mid_conversation_system(model=model, custom_llm_provider=custom_llm_provider)
 
 
-def _is_azure_responses_target(
+@dataclass(frozen=True, slots=True)
+class ChatRouteInputs:
+    tools: Sequence[Mapping[str, object]] | None = None
+    reasoning_effort: str | Mapping[str, object] | None = None
+    api_base: str | None = None
+    thinking: Mapping[str, object] | None = None
+    reasoning_summary: object | None = None
+
+
+_NO_ROUTE_INPUTS: Final = ChatRouteInputs()
+
+
+def _inferred_provider(model: str) -> str | None:
+    try:
+        return litellm.utils.get_llm_provider(model=model)[1]
+    except Exception:
+        return None
+
+
+def openai_thinking_routes_to_responses(
+    model: str | None,
+    custom_llm_provider: str | None,
+    thinking: Mapping[str, object] | None,
+) -> bool:
+    if not model or thinking is None or thinking.get("type") != "enabled":
+        return False
+    provider: Final = custom_llm_provider if custom_llm_provider is not None else _inferred_provider(model)
+    if provider != "openai":
+        return False
+    try:
+        model_info: Final = litellm.get_model_info(model=model, custom_llm_provider=provider)
+    except Exception:
+        return True
+    return model_info.get("supports_reasoning") is not False
+
+
+def _keeps_encrypted_reasoning_for_responses(
     model: str | None,
     custom_llm_provider: str | None,
     deployment_model_info: Mapping[str, object] | None,
+    route: ChatRouteInputs,
 ) -> bool:
-    if not model:
-        return False
-    if custom_llm_provider not in (None, LlmProviders.AZURE.value):
-        return False
-    if custom_llm_provider is None and not model.startswith("azure/"):
-        return False
+    from litellm.responses.bridge_check import chat_completion_ends_in_responses_api
 
-    from litellm.responses.bridge_check import responses_api_bridge_check
-
-    resolved_model, provider, _, _ = litellm.get_llm_provider(model=model, custom_llm_provider=custom_llm_provider)
-    model_info, _ = responses_api_bridge_check(
-        model=resolved_model,
-        custom_llm_provider=provider,
-        deployment_model_info=deployment_model_info,
+    return openai_thinking_routes_to_responses(
+        model, custom_llm_provider, route.thinking
+    ) or chat_completion_ends_in_responses_api(
+        model,
+        custom_llm_provider,
+        deployment_model_info,
+        tools=route.tools,
+        reasoning_effort=route.reasoning_effort,
+        reasoning_summary=route.reasoning_summary,
+        api_base=route.api_base,
     )
-    return model_info.get("mode") == "responses"
+
+
+def _anthropic_thinking_blocks(source: object) -> tuple[Mapping[str, object], ...]:
+    return thinking_blocks_with_encrypted_reasoning(
+        getattr(source, "thinking_blocks", None),
+        getattr(source, "reasoning_items", None),
+    )
+
+
+def _streamed_thinking_blocks(delta: object) -> tuple[Mapping[str, object], ...]:
+    return tuple(block for item in attribute_items(delta, "thinking_blocks") if (block := as_str_mapping(item)))
 
 
 def _chat_tool_param(function_chunk: ChatCompletionToolParamFunctionChunk, tool: object) -> ChatCompletionToolParam:
@@ -317,6 +364,7 @@ class AnthropicAdapter:
         custom_llm_provider: str | None = None,
         api_base: str | None = None,
         deployment_model_info: Mapping[str, object] | None = None,
+        reasoning_summary: object | None = None,
         drop_params: bool = True,
     ) -> tuple[ChatCompletionRequest | None, dict[str, str]]:
         """
@@ -353,6 +401,7 @@ class AnthropicAdapter:
             custom_llm_provider=custom_llm_provider,
             api_base=api_base,
             deployment_model_info=deployment_model_info,
+            reasoning_summary=reasoning_summary,
             drop_params=drop_params,
             report_reasoning_drops=True,
         )
@@ -551,11 +600,12 @@ class LiteLLMAnthropicMessagesAdapter:
         custom_llm_provider: str | None = None,
         deployment_model_info: Mapping[str, object] | None = None,
         preserve_midturn_system: bool = False,
+        route: ChatRouteInputs = _NO_ROUTE_INPUTS,
     ) -> list[AllMessageValues]:
         new_messages: Final[list[AllMessageValues]] = []
         replayable_messages: Final = (
             messages
-            if _is_azure_responses_target(model, custom_llm_provider, deployment_model_info)
+            if _keeps_encrypted_reasoning_for_responses(model, custom_llm_provider, deployment_model_info, route)
             else strip_encrypted_reasoning_blocks_from_anthropic_messages(messages)
         )
         leading_count: Final = next(
@@ -1338,6 +1388,7 @@ class LiteLLMAnthropicMessagesAdapter:
         custom_llm_provider: str | None = None,
         api_base: str | None = None,
         deployment_model_info: Mapping[str, object] | None = None,
+        reasoning_summary: object | None = None,
         preserve_midturn_system: bool = False,
         drop_params: bool = True,
         report_reasoning_drops: bool = False,
@@ -1351,27 +1402,15 @@ class LiteLLMAnthropicMessagesAdapter:
               for tools that exceeded OpenAI's 64-char limit
         """
         # Debug: Processing Anthropic message request
-        new_messages: list[AllMessageValues] = []
         tool_name_mapping: dict[str, str] = {}
 
-        ## CONVERT ANTHROPIC MESSAGES TO OPENAI
         messages_list: Final[list[AllAnthropicPassThroughMessageValues]] = cast(
             list[AllAnthropicPassThroughMessageValues],
             anthropic_message_request["messages"],
         )
-        new_messages = self.translate_anthropic_messages_to_openai(
-            messages=messages_list,
-            model=anthropic_message_request.get("model"),
-            custom_llm_provider=custom_llm_provider,
-            deployment_model_info=deployment_model_info,
-            preserve_midturn_system=preserve_midturn_system,
-        )
-        ## ADD SYSTEM MESSAGE TO MESSAGES
-        self._add_system_message_to_messages(new_messages, anthropic_message_request)
-
         new_kwargs: Final[ChatCompletionRequest] = {
             "model": anthropic_message_request["model"],
-            "messages": new_messages,
+            "messages": [],
         }
         ## CONVERT METADATA (user_id + litellm metadata)
         self._translate_metadata_to_openai(
@@ -1398,6 +1437,24 @@ class LiteLLMAnthropicMessagesAdapter:
             drop_params=drop_params,
             report_reasoning_drops=report_reasoning_drops,
         )
+        ## CONVERT ANTHROPIC MESSAGES TO OPENAI, after tools and thinking pick the route
+        new_messages: Final = self.translate_anthropic_messages_to_openai(
+            messages=messages_list,
+            model=anthropic_message_request.get("model"),
+            custom_llm_provider=custom_llm_provider,
+            deployment_model_info=deployment_model_info,
+            preserve_midturn_system=preserve_midturn_system,
+            route=ChatRouteInputs(
+                tools=new_kwargs.get("tools"),
+                reasoning_effort=new_kwargs.get("reasoning_effort"),
+                api_base=api_base,
+                thinking=anthropic_message_request.get("thinking"),
+                reasoning_summary=reasoning_summary,
+            ),
+        )
+        ## ADD SYSTEM MESSAGE TO MESSAGES
+        self._add_system_message_to_messages(new_messages, anthropic_message_request)
+        new_kwargs["messages"] = new_messages
         ## CONVERT STOP_SEQUENCES
         self._translate_stop_sequences_to_openai(
             anthropic_message_request=anthropic_message_request,
@@ -1502,9 +1559,9 @@ class LiteLLMAnthropicMessagesAdapter:
             for choice in choices
         ):
             new_content.extend(dict(_COMPACTION_BLOCK.validate_python(block)) for block in compaction_blocks)
-            # Handle thinking blocks first
-            if hasattr(choice.message, "thinking_blocks") and choice.message.thinking_blocks:
-                for thinking_block in choice.message.thinking_blocks:
+            signed_thinking_blocks = _anthropic_thinking_blocks(choice.message)
+            if signed_thinking_blocks:
+                for thinking_block in signed_thinking_blocks:
                     if thinking_block.get("type") == "thinking":
                         if is_empty_unsigned_thinking_block(thinking_block):
                             continue
@@ -1816,24 +1873,24 @@ class LiteLLMAnthropicMessagesAdapter:
                 choice.delta
             ) is not None:
                 return "text", TextBlock(type="text", text="")
-            elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "thinking_blocks"):
-                thinking_blocks = choice.delta.thinking_blocks or []
-                if len(thinking_blocks) > 0:
-                    thinking_block = thinking_blocks[0]
-                    if thinking_block["type"] == "redacted_thinking":
-                        return "redacted_thinking", ChatCompletionRedactedThinkingBlock(
-                            type="redacted_thinking", data=thinking_block.get("data") or ""
-                        )
-                    if thinking_block["type"] == "thinking":
-                        thinking = thinking_block.get("thinking") or ""
-                        signature = thinking_block.get("signature") or ""
+            elif isinstance(choice, StreamingChoices) and (thinking_blocks := _streamed_thinking_blocks(choice.delta)):
+                thinking_block = thinking_blocks[0]
+                if thinking_block["type"] == "redacted_thinking":
+                    redacted_data = thinking_block.get("data")
+                    return "redacted_thinking", ChatCompletionRedactedThinkingBlock(
+                        type="redacted_thinking",
+                        data=redacted_data if isinstance(redacted_data, str) else "",
+                    )
+                if thinking_block["type"] == "thinking":
+                    thinking = thinking_block.get("thinking") or ""
+                    signature = thinking_block.get("signature") or ""
 
-                        assert isinstance(thinking, str)
-                        assert isinstance(signature, str)
+                    assert isinstance(thinking, str)
+                    assert isinstance(signature, str)
 
-                        return "thinking", ChatCompletionThinkingBlock(
-                            type="thinking", thinking=thinking if signature else "", signature=signature
-                        )
+                    return "thinking", ChatCompletionThinkingBlock(
+                        type="thinking", thinking=thinking if signature else "", signature=signature
+                    )
             # OpenAI-compatible reasoning backends (e.g. vLLM/SGLang reasoning
             # parsers) populate ``reasoning_content`` without ``thinking_blocks``.
             # ``Delta`` deletes the ``thinking_blocks`` attribute when unset, so the
@@ -1866,19 +1923,17 @@ class LiteLLMAnthropicMessagesAdapter:
                 for tool in choice.delta.tool_calls:
                     if tool.function is not None and tool.function.arguments is not None:
                         partial_json = (partial_json or "") + tool.function.arguments
-            elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "thinking_blocks"):
-                thinking_blocks = choice.delta.thinking_blocks or []
-                if len(thinking_blocks) > 0:
-                    for thinking_block in thinking_blocks:
-                        if thinking_block["type"] == "thinking":
-                            thinking = thinking_block.get("thinking") or ""
-                            signature = thinking_block.get("signature") or ""
+            elif isinstance(choice, StreamingChoices) and (thinking_blocks := _streamed_thinking_blocks(choice.delta)):
+                for thinking_block in thinking_blocks:
+                    if thinking_block["type"] == "thinking":
+                        thinking = thinking_block.get("thinking") or ""
+                        signature = thinking_block.get("signature") or ""
 
-                            assert isinstance(thinking, str)
-                            assert isinstance(signature, str)
+                        assert isinstance(thinking, str)
+                        assert isinstance(signature, str)
 
-                            reasoning_content += thinking
-                            reasoning_signature += signature
+                        reasoning_content += thinking
+                        reasoning_signature += signature
             # Handle reasoning_content when thinking_blocks is not present
             # This handles providers like OpenRouter that return reasoning_content
             elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "reasoning_content"):
diff --git a/litellm/llms/anthropic/messages/responses_adapters/handler.py b/litellm/llms/anthropic/messages/responses_adapters/handler.py
index 85f042f621c..fb1b6d5884b 100644
--- a/litellm/llms/anthropic/messages/responses_adapters/handler.py
+++ b/litellm/llms/anthropic/messages/responses_adapters/handler.py
@@ -19,13 +19,14 @@ from litellm.integrations.anthropic_cache_control_hook import (
 from litellm.litellm_core_utils.asyncify import (
     run_async_function,  # pyright: ignore[reportUnknownVariableType]  # untyped helper
 )
-from litellm.litellm_core_utils.core_helpers import as_str_mapping, drops_unsupported_params
+from litellm.litellm_core_utils.core_helpers import as_object_list, as_str_mapping, drops_unsupported_params
 from litellm.litellm_core_utils.prompt_templates.prompt_cache_breakpoint_utils import (
     cache_control_as_prompt_cache_breakpoint,
     prompt_cache_breakpoint_with_readable_mode,
     with_prompt_cache_breakpoint,
 )
 from litellm.litellm_core_utils.tool_fields import drops_mcp_servers, tools_without_mcp_servers
+from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
 from litellm.responses.utils import auto_reasoning_summary
 from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
 from litellm.types.llms.anthropic import (
@@ -43,7 +44,12 @@ from litellm.types.llms.openai import ResponsesAPIResponse
 from litellm.types.utils import CountContextManagement
 from litellm.utils import ProviderConfigManager
 
-from ..adapters.handler import context_management_explicitly_dropped, request_safeguards_evaluator
+from ..adapters.handler import (
+    context_management_explicitly_dropped,
+    messages_without_rejected_reasoning,
+    request_safeguards_evaluator,
+    resent_without_rejected_reasoning,
+)
 from ..context_management import AnthropicContextManagementError, PolyfillResult, apply_context_management
 from ..context_management.constants import COMPACT_EDIT_TYPE
 from ..context_management.dispatcher import normalize_spec
@@ -391,6 +397,76 @@ def _build_responses_kwargs(
     return responses_kwargs
 
 
+_ResponsesKwargsFor: TypeAlias = Callable[[AnthropicRequestMessages], Mapping[str, object]]
+_ResponsesResult: TypeAlias = ResponsesAPIResponse | BaseResponsesAPIStreamingIterator
+_SyncResponsesResult: TypeAlias = _ResponsesResult | Coroutine[object, object, _ResponsesResult]
+
+
+def _is_encrypted_reasoning_item(item: object) -> bool:
+    mapping: Final = as_str_mapping(item)
+    return mapping is not None and mapping.get("type") == "reasoning" and bool(mapping.get("encrypted_content"))
+
+
+def _sends_encrypted_reasoning(responses_kwargs: Mapping[str, object]) -> bool:
+    return any(map(_is_encrypted_reasoning_item, as_object_list(responses_kwargs.get("input")) or ()))
+
+
+def _kwarg_text(responses_kwargs: Mapping[str, object], key: str) -> str:
+    value: Final = responses_kwargs.get(key)
+    return value if isinstance(value, str) else ""
+
+
+def _messages_without_rejected_reasoning(
+    messages: AnthropicRequestMessages,
+    error: litellm.BadRequestError,
+    responses_kwargs: Mapping[str, object],
+    sent_encrypted_reasoning: bool,
+) -> AnthropicRequestMessages | None:
+    return messages_without_rejected_reasoning(
+        messages,
+        error,
+        sent_encrypted_reasoning=sent_encrypted_reasoning,
+        provider=_kwarg_text(responses_kwargs, "custom_llm_provider"),
+        model=_kwarg_text(responses_kwargs, "model"),
+    )
+
+
+async def _aresponses_without_rejected_reasoning(
+    messages: AnthropicRequestMessages, responses_kwargs_for: _ResponsesKwargsFor
+) -> tuple[Mapping[str, object], _ResponsesResult]:
+    responses_kwargs: Final = responses_kwargs_for(messages)
+    sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(responses_kwargs)
+    try:
+        with resent_without_rejected_reasoning(sent_encrypted_reasoning):
+            return responses_kwargs, await litellm.aresponses(**responses_kwargs)
+    except litellm.BadRequestError as error:
+        replayed_messages: Final = _messages_without_rejected_reasoning(
+            messages, error, responses_kwargs, sent_encrypted_reasoning
+        )
+        if replayed_messages is None:
+            raise
+        replay_kwargs: Final = responses_kwargs_for(replayed_messages)
+        return replay_kwargs, await litellm.aresponses(**replay_kwargs)
+
+
+def _responses_without_rejected_reasoning(
+    messages: AnthropicRequestMessages, responses_kwargs_for: _ResponsesKwargsFor
+) -> tuple[Mapping[str, object], _SyncResponsesResult]:
+    responses_kwargs: Final = responses_kwargs_for(messages)
+    sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(responses_kwargs)
+    try:
+        with resent_without_rejected_reasoning(sent_encrypted_reasoning):
+            return responses_kwargs, litellm.responses(**responses_kwargs)
+    except litellm.BadRequestError as error:
+        replayed_messages: Final = _messages_without_rejected_reasoning(
+            messages, error, responses_kwargs, sent_encrypted_reasoning
+        )
+        if replayed_messages is None:
+            raise
+        replay_kwargs: Final = responses_kwargs_for(replayed_messages)
+        return replay_kwargs, litellm.responses(**replay_kwargs)
+
+
 class LiteLLMMessagesToResponsesAPIHandler:
     """
     Handles Anthropic /v1/messages requests for OpenAI / Azure models by
@@ -422,28 +498,32 @@ class LiteLLMMessagesToResponsesAPIHandler:
         polyfill_result: Final = await _polyfilled(
             model=model, messages=messages, tools=tools, system=system, edits=polyfilled_edits, kwargs=kwargs
         )
-        responses_kwargs: Final = _build_responses_kwargs(
-            max_tokens=max_tokens,
-            messages=polyfill_result.messages if polyfill_result is not None else messages,
-            model=model,
-            context_management=native_context_management,
-            metadata=metadata,
-            output_config=output_config,
-            stop_sequences=stop_sequences,
-            stream=stream,
-            system=system,
-            temperature=temperature,
-            thinking=thinking,
-            tool_choice=tool_choice,
-            tools=tools,
-            top_k=top_k,
-            top_p=top_p,
-            output_format=output_format,
-            extra_kwargs=kwargs,
-        )
+
+        def responses_kwargs_for(sent_messages: AnthropicRequestMessages) -> Mapping[str, object]:
+            return _build_responses_kwargs(
+                max_tokens=max_tokens,
+                messages=sent_messages,
+                model=model,
+                context_management=native_context_management,
+                metadata=metadata,
+                output_config=output_config,
+                stop_sequences=stop_sequences,
+                stream=stream,
+                system=system,
+                temperature=temperature,
+                thinking=thinking,
+                tool_choice=tool_choice,
+                tools=tools,
+                top_k=top_k,
+                top_p=top_p,
+                output_format=output_format,
+                extra_kwargs=kwargs,
+            )
 
         safeguards_evaluator: Final = request_safeguards_evaluator(kwargs, messages)
-        result: Final = await litellm.aresponses(**responses_kwargs)
+        responses_kwargs, result = await _aresponses_without_rejected_reasoning(
+            polyfill_result.messages if polyfill_result is not None else messages, responses_kwargs_for
+        )
 
         if stream:
             wrapper: Final = AnthropicResponsesStreamWrapper(
@@ -526,27 +606,31 @@ class LiteLLMMessagesToResponsesAPIHandler:
             else None
         )
         polyfill_result: Final = ran if isinstance(ran, PolyfillResult) else None
-        responses_kwargs: Final = _build_responses_kwargs(
-            max_tokens=max_tokens,
-            messages=polyfill_result.messages if polyfill_result is not None else messages,
-            model=model,
-            context_management=native_context_management,
-            metadata=metadata,
-            output_config=output_config,
-            stop_sequences=stop_sequences,
-            stream=stream,
-            system=system,
-            temperature=temperature,
-            thinking=thinking,
-            tool_choice=tool_choice,
-            tools=tools,
-            top_k=top_k,
-            top_p=top_p,
-            output_format=output_format,
-            extra_kwargs=kwargs,
-        )
 
-        result: Final = litellm.responses(**responses_kwargs)
+        def responses_kwargs_for(sent_messages: AnthropicRequestMessages) -> Mapping[str, object]:
+            return _build_responses_kwargs(
+                max_tokens=max_tokens,
+                messages=sent_messages,
+                model=model,
+                context_management=native_context_management,
+                metadata=metadata,
+                output_config=output_config,
+                stop_sequences=stop_sequences,
+                stream=stream,
+                system=system,
+                temperature=temperature,
+                thinking=thinking,
+                tool_choice=tool_choice,
+                tools=tools,
+                top_k=top_k,
+                top_p=top_p,
+                output_format=output_format,
+                extra_kwargs=kwargs,
+            )
+
+        responses_kwargs, result = _responses_without_rejected_reasoning(
+            polyfill_result.messages if polyfill_result is not None else messages, responses_kwargs_for
+        )
 
         if stream:
             wrapper: Final = AnthropicResponsesStreamWrapper(
diff --git a/litellm/main.py b/litellm/main.py
index 1d18100d1d5..f3620d7a8c2 100644
--- a/litellm/main.py
+++ b/litellm/main.py
@@ -28,6 +28,7 @@ from functools import partial
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Protocol, Union, cast, get_args
 
+from litellm._internal_context import is_resent_failure
 from litellm._logging import redact_string
 from litellm._uuid import uuid
 
@@ -6634,6 +6635,10 @@ def completion(
         )
 
 
+def _is_not_resent_failure(error: BaseException) -> bool:
+    return isinstance(error, Exception) and not is_resent_failure(error)
+
+
 def completion_with_retries(*args, **kwargs):
     """
     Executes a litellm.completion() with 3 retries
@@ -6653,10 +6658,15 @@ def completion_with_retries(*args, **kwargs):
         retryer = tenacity.Retrying(
             wait=tenacity.wait_exponential(multiplier=1, max=10),
             stop=tenacity.stop_after_attempt(num_retries),
+            retry=tenacity.retry_if_exception(_is_not_resent_failure),
             reraise=True,
         )
     else:
-        retryer = tenacity.Retrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
+        retryer = tenacity.Retrying(
+            stop=tenacity.stop_after_attempt(num_retries),
+            retry=tenacity.retry_if_exception(_is_not_resent_failure),
+            reraise=True,
+        )
     return retryer(original_function, *args, **kwargs)
 
 
@@ -6679,10 +6689,15 @@ async def acompletion_with_retries(*args, **kwargs):
         retryer = tenacity.AsyncRetrying(
             wait=tenacity.wait_exponential(multiplier=1, max=10),
             stop=tenacity.stop_after_attempt(num_retries),
+            retry=tenacity.retry_if_exception(_is_not_resent_failure),
             reraise=True,
         )
     else:
-        retryer = tenacity.AsyncRetrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
+        retryer = tenacity.AsyncRetrying(
+            stop=tenacity.stop_after_attempt(num_retries),
+            retry=tenacity.retry_if_exception(_is_not_resent_failure),
+            reraise=True,
+        )
     return await retryer(original_function, *args, **kwargs)
 
 
@@ -6707,10 +6722,15 @@ def responses_with_retries(*args, **kwargs):
         retryer = tenacity.Retrying(
             wait=tenacity.wait_exponential(multiplier=1, max=10),
             stop=tenacity.stop_after_attempt(num_retries),
+            retry=tenacity.retry_if_exception(_is_not_resent_failure),
             reraise=True,
         )
     else:
-        retryer = tenacity.Retrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
+        retryer = tenacity.Retrying(
+            stop=tenacity.stop_after_attempt(num_retries),
+            retry=tenacity.retry_if_exception(_is_not_resent_failure),
+            reraise=True,
+        )
     return retryer(original_function, *args, **kwargs)
 
 
@@ -6734,10 +6754,15 @@ async def aresponses_with_retries(*args, **kwargs):
         retryer = tenacity.AsyncRetrying(
             wait=tenacity.wait_exponential(multiplier=1, max=10),
             stop=tenacity.stop_after_attempt(num_retries),
+            retry=tenacity.retry_if_exception(_is_not_resent_failure),
             reraise=True,
         )
     else:
-        retryer = tenacity.AsyncRetrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
+        retryer = tenacity.AsyncRetrying(
+            stop=tenacity.stop_after_attempt(num_retries),
+            retry=tenacity.retry_if_exception(_is_not_resent_failure),
+            reraise=True,
+        )
     return await retryer(original_function, *args, **kwargs)
 
 
diff --git a/litellm/responses/bridge_check.py b/litellm/responses/bridge_check.py
index 826de8cfdc9..683a51fc6b6 100644
--- a/litellm/responses/bridge_check.py
+++ b/litellm/responses/bridge_check.py
@@ -4,7 +4,10 @@ from urllib.parse import urlsplit
 
 import litellm
 from litellm._logging import verbose_logger
-from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
+from litellm.litellm_core_utils.get_llm_provider_logic import (
+    get_llm_provider,
+    resolve_model_and_provider_for_metadata,
+)
 from litellm.llms.azure_ai.chat.transformation import AzureAIGPT5Config
 from litellm.llms.azure_ai.common_utils import (
     azure_ai_supports_native_responses,
@@ -291,3 +294,39 @@ def responses_api_bridge_check(
         model = model.replace("responses/", "")
 
     return model_info, model
+
+
+def chat_completion_ends_in_responses_api(
+    model: str | None,
+    custom_llm_provider: str | None,
+    deployment_model_info: Mapping[str, object] | None = None,
+    *,
+    tools: Sequence[Mapping[str, object]] | None = None,
+    reasoning_effort: str | Mapping[str, object] | None = None,
+    reasoning_summary: object | None = None,
+    api_base: str | None = None,
+) -> bool:
+    """Whether chat completion for this model is served by a Responses API.
+
+    Encrypted reasoning can be replayed only there. A normal chat provider cannot
+    verify the block, so the messages adapter strips it. ``chatgpt``, a ``responses/``
+    model, a deployment pinned to ``mode: responses``, and a request whose tools or
+    reasoning effort the bridge sends to Responses keep it.
+    """
+    if not model:
+        return False
+    try:
+        resolved_model, provider = resolve_model_and_provider_for_metadata(model, custom_llm_provider)
+    except Exception as error:  # noqa: BLE001  # unknown model names must not fail message translation
+        verbose_logger.debug("responses replay check failed for %s: %s", model, error)
+        return False
+    model_info, _resolved = responses_api_bridge_check(
+        model=resolved_model,
+        custom_llm_provider=provider,
+        tools=tools,
+        reasoning_effort=reasoning_effort,
+        reasoning_summary=reasoning_summary,
+        api_base=api_base,
+        deployment_model_info=deployment_model_info,
+    )
+    return model_info.get("mode") == "responses"
diff --git a/litellm/utils.py b/litellm/utils.py
index 28e2440fbdd..b9bf872e853 100644
--- a/litellm/utils.py
+++ b/litellm/utils.py
@@ -53,7 +53,7 @@ import litellm.litellm_core_utils
 
 # audio_utils.utils is lazy-loaded - only imported when needed for transcription calls
 import litellm.litellm_core_utils.json_validation_rule
-from litellm._internal_context import is_internal_call
+from litellm._internal_context import is_internal_call, is_resent_failure
 from litellm._lazy_imports import (
     _get_modified_max_tokens,
     get_default_encoding,
@@ -2303,12 +2303,16 @@ def client(original_function):
                         retry_policy=kwargs.get("retry_policy"),
                     )
                     kwargs["retry_policy"] = reset_retry_policy()  # prevent infinite loops
-                litellm.num_retries = None  # set retries to None to prevent infinite loops
+                if not is_resent_failure(e):
+                    litellm.num_retries = None  # set retries to None to prevent infinite loops
                 context_window_fallback_dict: Final = kwargs.get("context_window_fallback_dict", {})
 
                 is_completion_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=False)
                 if (
-                    num_retries and not is_completion_litellm_router_call and not is_agentic_followup_failure(e)
+                    num_retries
+                    and not is_completion_litellm_router_call
+                    and not is_agentic_followup_failure(e)
+                    and not is_resent_failure(e)
                 ):  # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
                     if (
                         isinstance(e, openai.APIError)
@@ -2338,11 +2342,15 @@ def client(original_function):
                         retry_policy=kwargs.get("retry_policy"),
                     )
                     kwargs["retry_policy"] = reset_retry_policy()  # prevent infinite loops
-                litellm.num_retries = None  # set retries to None to prevent infinite loops
+                if not is_resent_failure(e):
+                    litellm.num_retries = None  # set retries to None to prevent infinite loops
 
                 is_responses_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=False)
                 if (
-                    num_retries and not is_responses_litellm_router_call and not is_agentic_followup_failure(e)
+                    num_retries
+                    and not is_responses_litellm_router_call
+                    and not is_agentic_followup_failure(e)
+                    and not is_resent_failure(e)
                 ):  # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
                     if (
                         isinstance(e, openai.APIError)
@@ -2355,7 +2363,7 @@ def client(original_function):
             end_time = datetime.datetime.now()
 
             # LOG FAILURE - handle streaming failure logging in the _next_ object, remove `handle_failure` once it's deprecated
-            if logging_obj:
+            if logging_obj and not is_resent_failure(e):
                 logging_obj.failure_handler(
                     e, traceback_exception, start_time, end_time
                 )  # DO NOT MAKE THREADED - router retry fallback relies on this!
@@ -2620,7 +2628,7 @@ def client(original_function):
             # Reuse the timestamp taken right when the deployment call itself failed, before
             # the failure hook ran, so a slow callback doesn't inflate the reported duration.
             end_time = _deployment_call_end_time if _deployment_call_end_time is not None else datetime.datetime.now()  # noqa: DTZ005  # matches the naive datetimes this whole function already times start_time/end_time with
-            if logging_obj and not _is_litellm_internal_call:
+            if logging_obj and not _is_litellm_internal_call and not is_resent_failure(e):
                 unbind_budget_reservation_from_callbacks(logging_obj.litellm_params)
                 try:
                     logging_obj.failure_handler(
@@ -2645,6 +2653,7 @@ def client(original_function):
                     and not is_acompletion_litellm_router_call
                     and not isinstance(e, ImportError)
                     and not is_agentic_followup_failure(e)
+                    and not is_resent_failure(e)
                 ):  # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
                     try:
                         litellm.num_retries = None  # set retries to None to prevent infinite loops
@@ -2655,8 +2664,9 @@ def client(original_function):
                         elif isinstance(e, openai.APIError):  # generic api error
                             kwargs["retry_strategy"] = "constant_retry"
                         result = await litellm.acompletion_with_retries(*args, **kwargs)
-                    except Exception:
-                        pass
+                    except Exception as retry_error:
+                        if is_resent_failure(retry_error):
+                            raise
                     else:
                         return result
                 elif (
@@ -2679,6 +2689,7 @@ def client(original_function):
                     and not is_aresponses_litellm_router_call
                     and not isinstance(e, ImportError)
                     and not is_agentic_followup_failure(e)
+                    and not is_resent_failure(e)
                 ):  # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
                     try:
                         litellm.num_retries = None  # set retries to None to prevent infinite loops
@@ -2689,8 +2700,9 @@ def client(original_function):
                         elif isinstance(e, openai.APIError):  # generic api error
                             kwargs["retry_strategy"] = "constant_retry"
                         result = await litellm.aresponses_with_retries(*args, **kwargs)
-                    except Exception:
-                        pass
+                    except Exception as retry_error:
+                        if is_resent_failure(retry_error):
+                            raise
                     else:
                         return result
 
diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py
index 19aefb430c7..7ba833aad14 100644
--- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py
+++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py
@@ -3837,6 +3837,386 @@ def test_reasoning_items_streaming_emitted_on_response_completed():
     assert ri["summary"][0]["text"] == summary_text
 
 
+def test_reasoning_output_item_done_emits_signed_thinking_blocks_the_stream_keeps():
+    """A chatgpt /v1/messages stream drops the reasoning item unless this delta carries thinking_blocks.
+
+    response.output_item.done is the only event that has encrypted_content. The chat delta has to
+    expose it as a signed thinking block, and that block alone has to count as stream content.
+    """
+    from unittest.mock import MagicMock
+
+    from litellm.completion_extras.litellm_responses_transformation.transformation import (
+        OpenAiResponsesToChatCompletionStreamIterator,
+    )
+    from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
+    from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
+    from litellm.llms.anthropic.messages.adapters.transformation import LiteLLMAnthropicMessagesAdapter
+
+    encrypted: Final = "gAAAA_round_trip"
+    done: Final = {
+        "type": "response.output_item.done",
+        "output_index": 0,
+        "item": {
+            "type": "reasoning",
+            "id": "rs_done",
+            "encrypted_content": encrypted,
+            "summary": [],
+        },
+    }
+    result: Final = OpenAiResponsesToChatCompletionStreamIterator(
+        streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
+    ).chunk_parser(done)
+    blocks: Final = result.choices[0].delta.thinking_blocks
+    assert blocks is not None
+    assert blocks[0]["type"] == "redacted_thinking"
+    assert blocks[0]["data"] == encrypted_reasoning_signature(encrypted)
+    assert blocks[0]["index"] == 0
+    assert getattr(result.choices[0].delta, "reasoning_content", None) in (None, "")
+
+    wrapper: Final = CustomStreamWrapper(
+        completion_stream=None,
+        model="chatgpt/gpt-6.1-sol",
+        logging_obj=MagicMock(),
+        custom_llm_provider="chatgpt",
+    )
+    assert wrapper.is_chunk_non_empty({}, result, {}) is True
+
+    block_type, start = LiteLLMAnthropicMessagesAdapter().translate_streaming_openai_chunk_to_anthropic_content_block(
+        choices=result.choices
+    )
+    assert block_type == "redacted_thinking"
+    assert start["data"] == encrypted_reasoning_signature(encrypted)
+
+
+_SIGNED_REASONING_STREAM_BODY: Final[Mapping[str, object]] = {
+    **_searched_responses_body("azure/gpt-6-sol"),
+    "output": [
+        {"type": "reasoning", "id": "rs_signed", "summary": [], "encrypted_content": "gAAAA_consumer"},
+        _PONG_MESSAGE,
+    ],
+}
+
+
+def _signed_reasoning_sse() -> str:
+    events: Final = (
+        {
+            "type": "response.created",
+            "response": {**_SIGNED_REASONING_STREAM_BODY, "status": "in_progress", "output": [], "usage": None},
+        },
+        {"type": "response.output_item.done", "output_index": 0, "item": _SIGNED_REASONING_STREAM_BODY["output"][0]},
+        {
+            "type": "response.output_text.delta",
+            "item_id": "msg_logged_call_type",
+            "output_index": 1,
+            "content_index": 0,
+            "delta": "pong",
+        },
+        {"type": "response.completed", "response": _SIGNED_REASONING_STREAM_BODY},
+    )
+    return "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events)
+
+
+_AZURE_BRIDGE_KWARGS: Final[Mapping[str, object]] = {
+    "api_base": "https://example.invalid",
+    "api_version": "2025-04-01-preview",
+    "api_key": "test-key",
+}
+
+
+@pytest.mark.asyncio
+async def test_streamed_reasoning_signature_reaches_messages_clients_and_never_chat_completions_deltas(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
+
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    with respx.mock(assert_all_called=True) as router:
+        router.post(url__startswith="https://example.invalid/openai/responses").mock(
+            side_effect=lambda _request: httpx.Response(
+                200, text=_signed_reasoning_sse(), headers={"content-type": "text/event-stream"}
+            )
+        )
+        chat_stream: Final = await litellm.acompletion(
+            model="azure/gpt-6-sol",
+            messages=[{"role": "user", "content": "ping"}],
+            tools=[_GET_WEATHER_TOOL],
+            stream=True,
+            **_AZURE_BRIDGE_KWARGS,
+        )
+        chat_deltas: Final = [chunk.choices[0].delta async for chunk in chat_stream if chunk.choices]
+        messages_stream: Final = await litellm.anthropic.messages.acreate(
+            model="azure/gpt-6-sol",
+            messages=[{"role": "user", "content": "ping"}],
+            tools=[{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}],
+            max_tokens=64,
+            stream=True,
+            **_AZURE_BRIDGE_KWARGS,
+        )
+        sse: Final = b"".join([chunk async for chunk in messages_stream]).decode()
+
+    assert [getattr(delta, "thinking_blocks", None) for delta in chat_deltas if getattr(delta, "thinking_blocks", None)] == []
+    assert "".join(delta.content or "" for delta in chat_deltas) == "pong"
+    messages_events: Final = [json.loads(line[len("data: ") :]) for line in sse.splitlines() if line.startswith("data: ")]
+    started: Final = [event["content_block"] for event in messages_events if event.get("type") == "content_block_start"]
+    assert [block["type"] for block in started] == ["redacted_thinking", "text"]
+    assert started[0]["data"] == encrypted_reasoning_signature("gAAAA_consumer")
+
+
+def test_reasoning_output_item_done_with_summary_signs_the_thinking_text():
+    from litellm.completion_extras.litellm_responses_transformation.transformation import (
+        OpenAiResponsesToChatCompletionStreamIterator,
+    )
+    from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
+
+    encrypted: Final = "gAAAA_with_summary"
+    done: Final = {
+        "type": "response.output_item.done",
+        "output_index": 2,
+        "item": {
+            "type": "reasoning",
+            "id": "rs_sum",
+            "encrypted_content": encrypted,
+            "summary": [{"type": "summary_text", "text": "Oslo is colder."}],
+        },
+    }
+    result: Final = OpenAiResponsesToChatCompletionStreamIterator(
+        streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
+    ).chunk_parser(done)
+    block: Final = result.choices[0].delta.thinking_blocks[0]
+    assert block["type"] == "thinking"
+    assert block["thinking"] == "Oslo is colder."
+    assert block["signature"] == encrypted_reasoning_signature(encrypted)
+    assert block["index"] == 2
+
+
+def test_streamed_reasoning_summary_reaches_messages_clients_once_with_its_signature():
+    from litellm.completion_extras.litellm_responses_transformation.transformation import (
+        OpenAiResponsesToChatCompletionStreamIterator,
+    )
+    from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
+    from litellm.llms.anthropic.messages.adapters.streaming_iterator import AnthropicStreamWrapper
+    from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
+
+    encrypted: Final = "gAAAA_streamed_summary"
+    events: Final = (
+        {"type": "response.reasoning_summary_part.added", "output_index": 0, "summary_index": 0},
+        {"type": "response.reasoning_summary_text.delta", "output_index": 0, "delta": "Oslo is "},
+        {"type": "response.reasoning_summary_text.delta", "output_index": 0, "delta": "colder."},
+        {
+            "type": "response.output_item.done",
+            "output_index": 0,
+            "item": {
+                "type": "reasoning",
+                "id": "rs_streamed",
+                "encrypted_content": encrypted,
+                "summary": [{"type": "summary_text", "text": "Oslo is colder."}],
+            },
+        },
+        {"type": "response.output_text.delta", "output_index": 1, "delta": "Oslo."},
+    )
+    parser: Final = OpenAiResponsesToChatCompletionStreamIterator(
+        streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
+    )
+    chunks: Final = (
+        *(parser.chunk_parser(event) for event in events),
+        ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(), finish_reason="stop")]),
+    )
+
+    sse_events: Final = list(AnthropicStreamWrapper(completion_stream=iter(chunks), model="chatgpt/gpt-5.5"))
+    deltas: Final = [event["delta"] for event in sse_events if event.get("type") == "content_block_delta"]
+    assert "".join(d.get("thinking", "") for d in deltas if d["type"] == "thinking_delta") == "Oslo is colder."
+    assert [d["signature"] for d in deltas if d["type"] == "signature_delta"] == [
+        encrypted_reasoning_signature(encrypted)
+    ]
+
+
+def test_reasoning_item_repeated_at_stream_end_reaches_messages_clients_once():
+    from litellm.completion_extras.litellm_responses_transformation.transformation import (
+        OpenAiResponsesToChatCompletionStreamIterator,
+    )
+    from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
+    from litellm.llms.anthropic.messages.adapters.streaming_iterator import AnthropicStreamWrapper
+    from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
+
+    encrypted: Final = "gAAAA_completed_repeat"
+    reasoning_item: Final = {"type": "reasoning", "id": "rs_repeat", "summary": [], "encrypted_content": encrypted}
+    events: Final = (
+        {"type": "response.output_item.done", "output_index": 0, "item": reasoning_item},
+        {"type": "response.output_text.delta", "output_index": 1, "delta": "Oslo."},
+    )
+    parser: Final = OpenAiResponsesToChatCompletionStreamIterator(
+        streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
+    )
+    chunks: Final = (
+        *(parser.chunk_parser(event) for event in events),
+        ModelResponseStream(
+            choices=[StreamingChoices(index=0, delta=Delta(reasoning_items=[reasoning_item]), finish_reason=None)]
+        ),
+        ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(), finish_reason="stop")]),
+    )
+
+    sse_events: Final = list(AnthropicStreamWrapper(completion_stream=iter(chunks), model="azure/gpt-5.4-nano"))
+    started: Final = [event["content_block"] for event in sse_events if event.get("type") == "content_block_start"]
+    assert [block["type"] for block in started] == ["redacted_thinking", "text"]
+    assert started[0]["data"] == encrypted_reasoning_signature(encrypted)
+
+
+def test_reasoning_output_item_done_without_encrypted_content_stays_empty():
+    from litellm.completion_extras.litellm_responses_transformation.transformation import (
+        OpenAiResponsesToChatCompletionStreamIterator,
+    )
+
+    done: Final = {
+        "type": "response.output_item.done",
+        "output_index": 0,
+        "item": {
+            "type": "reasoning",
+            "id": "rs_plain",
+            "summary": [{"type": "summary_text", "text": "already streamed"}],
+        },
+    }
+    result: Final = OpenAiResponsesToChatCompletionStreamIterator(
+        streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
+    ).chunk_parser(done)
+    assert getattr(result.choices[0].delta, "thinking_blocks", None) in (None, [])
+
+
+def test_transform_response_hands_encrypted_reasoning_to_messages_clients_only():
+    from unittest.mock import Mock
+
+    from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
+    from litellm.llms.anthropic.messages.adapters.transformation import (
+        LiteLLMAnthropicMessagesAdapter,
+    )
+    from litellm.types.llms.openai import ResponsesAPIResponse
+    from litellm.types.utils import ModelResponse, Usage
+
+    encrypted: Final = "gAAAA_non_stream"
+    reasoning: Final = ResponseReasoningItem(
+        id="rs_tool",
+        summary=[],
+        type="reasoning",
+        content=None,
+        encrypted_content=encrypted,
+        status=None,
+    )
+    function_call: Final = ResponseFunctionToolCall(
+        id="fc_1",
+        type="function_call",
+        status="completed",
+        arguments='{"city":"Oslo"}',
+        call_id="call_1",
+        name="get_weather",
+    )
+    raw: Final = ResponsesAPIResponse.model_construct(
+        id="resp_tool",
+        created_at=1,
+        error=None,
+        incomplete_details=None,
+        model="gpt-6.1-sol",
+        output=[reasoning, function_call],
+        status="completed",
+        usage=None,
+    )
+    model_response: Final = ModelResponse(
+        id="chatcmpl-tool",
+        created=1,
+        model=None,
+        object="chat.completion",
+        system_fingerprint=None,
+        choices=[],
+        usage=Usage(completion_tokens=0, prompt_tokens=0, total_tokens=0),
+    )
+    result: Final = LiteLLMResponsesTransformationHandler().transform_response(
+        model="gpt-6.1-sol",
+        raw_response=raw,
+        model_response=model_response,
+        logging_obj=Mock(),
+        request_data={"model": "gpt-6.1-sol"},
+        messages=[{"role": "user", "content": "Which city?"}],
+        optional_params={},
+        litellm_params={},
+        encoding=Mock(),
+    )
+    message: Final = result.choices[0].message
+    assert message.reasoning_items[0]["encrypted_content"] == encrypted
+    assert "thinking_blocks" not in result.model_dump()["choices"][0]["message"]
+    assert message.tool_calls[0].function.name == "get_weather"
+
+    anthropic_response: Final = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(
+        response=result, tool_name_mapping=None
+    )
+    content: Final = anthropic_response["content"]
+    assert [block["type"] for block in content] == ["redacted_thinking", "tool_use"]
+    assert content[0]["data"] == encrypted_reasoning_signature(encrypted)
+
+
+@pytest.mark.parametrize(
+    ("call_type", "kept_ids"),
+    [
+        pytest.param("anthropic_messages", ["rs_first", "rs_second"], id="messages_client_keeps_every_item"),
+        pytest.param("acompletion", ["rs_second"], id="chat_client_keeps_the_last_item"),
+    ],
+)
+def test_non_streamed_tool_turn_hands_messages_clients_every_reasoning_item(
+    call_type: str, kept_ids: list[str]
+) -> None:
+    from types import SimpleNamespace
+    from unittest.mock import Mock
+
+    from litellm.types.llms.openai import ResponsesAPIResponse
+    from litellm.types.utils import ModelResponse, Usage
+
+    def reasoning(item_id: str) -> ResponseReasoningItem:
+        return ResponseReasoningItem(
+            id=item_id, summary=[], type="reasoning", content=None, encrypted_content=f"gAAAA_{item_id}", status=None
+        )
+
+    def call(index: int, city: str) -> ResponseFunctionToolCall:
+        return ResponseFunctionToolCall(
+            id=f"fc_{index}",
+            type="function_call",
+            status="completed",
+            arguments=json.dumps({"city": city}),
+            call_id=f"call_{index}",
+            name="get_weather",
+        )
+
+    raw: Final = ResponsesAPIResponse.model_construct(
+        id="resp_two_reasoning",
+        created_at=1,
+        error=None,
+        incomplete_details=None,
+        model="gpt-5.4-nano",
+        output=[reasoning("rs_first"), call(1, "Oslo"), reasoning("rs_second"), call(2, "Lyon")],
+        status="completed",
+        usage=None,
+    )
+    result: Final = LiteLLMResponsesTransformationHandler().transform_response(
+        model="gpt-5.4-nano",
+        raw_response=raw,
+        model_response=ModelResponse(
+            id="chatcmpl-two-reasoning",
+            created=1,
+            model=None,
+            object="chat.completion",
+            system_fingerprint=None,
+            choices=[],
+            usage=Usage(completion_tokens=0, prompt_tokens=0, total_tokens=0),
+        ),
+        logging_obj=SimpleNamespace(call_type=call_type),
+        request_data={"model": "gpt-5.4-nano"},
+        messages=[{"role": "user", "content": "Oslo or Lyon?"}],
+        optional_params={},
+        litellm_params={"custom_llm_provider": "azure"},
+        encoding=Mock(),
+    )
+
+    message: Final = result.choices[0].message
+    assert [item["id"] for item in message.reasoning_items] == kept_ids
+    assert [tool_call.function.name for tool_call in message.tool_calls] == ["get_weather", "get_weather"]
+
+
 def test_streaming_function_call_tool_id_for_degenerate_call_id():
     """In streaming, Bedrock Mantle's function_call event carries a unique ``id``
     (``fc_...``) and a non-unique, index-based ``call_id`` (``call_0``). For that
diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py
index 94fd915844a..df50ae9dc44 100644
--- a/tests/unit/litellm_core_utils/test_streaming_handler.py
+++ b/tests/unit/litellm_core_utils/test_streaming_handler.py
@@ -276,6 +276,37 @@ def test_is_chunk_non_empty(initialized_custom_stream_wrapper: CustomStreamWrapp
     )
 
 
+def test_is_chunk_non_empty_keeps_thinking_blocks_without_other_content(
+    initialized_custom_stream_wrapper: CustomStreamWrapper,
+):
+    """Encrypted reasoning arrives as thinking_blocks and nothing else, so that delta has to be yielded."""
+    chunk: Final = ModelResponseStream(
+        id="chunk-thinking",
+        object="chat.completion.chunk",
+        created=1,
+        model="chatgpt/gpt-6.1-sol",
+        choices=[
+            StreamingChoices(
+                index=0,
+                finish_reason=None,
+                delta=Delta(
+                    thinking_blocks=[
+                        {"type": "redacted_thinking", "data": "litellm_encrypted_reasoning:gAAAA"}
+                    ]
+                ),
+            )
+        ],
+    )
+    assert (
+        initialized_custom_stream_wrapper.is_chunk_non_empty(
+            completion_obj={},
+            model_response=chunk,
+            response_obj={},
+        )
+        is True
+    )
+
+
 def test_is_chunk_non_empty_with_annotations(
     initialized_custom_stream_wrapper: CustomStreamWrapper,
 ):
diff --git a/tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py b/tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py
index 420de9cb4f5..bad7b3f1750 100644
--- a/tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py
+++ b/tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py
@@ -2,7 +2,8 @@ import base64
 import functools
 import json
 import logging
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
+from contextlib import nullcontext
 from types import MappingProxyType
 from typing import Any, Final, cast
 
@@ -11,6 +12,7 @@ import pytest
 import respx
 
 import litellm
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
     TOOL_RESULT_IMAGE_PLACEHOLDER,
 )
@@ -34,6 +36,7 @@ from litellm.llms.anthropic.messages.native.mid_conversation_system import (
 from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
 from litellm.types.llms.anthropic import (
     AnthopicMessagesAssistantMessageParam,
+    AnthropicMessagesRequest,
     AnthropicMessagesUserMessageParam,
 )
 from litellm.types.llms.openai import ChatCompletionAssistantToolCall
@@ -616,6 +619,218 @@ def test_translate_anthropic_messages_to_openai_drops_bridge_encrypted_reasoning
     assert [b["signature"] for b in result[2]["thinking_blocks"]] == ["EqQBCkYIAxgCIkA_signed"]
 
 
+def test_chatgpt_messages_keep_encrypted_reasoning_and_replay_it():
+    """chatgpt /v1/messages ends in the Responses API, so the signed block has to survive into the next request."""
+    from litellm.completion_extras.litellm_responses_transformation.transformation import (
+        LiteLLMResponsesTransformationHandler,
+    )
+
+    summary_encrypted: Final = "gAAAA_summary"
+    bytes_encrypted: Final = "gAAAA_bytes"
+    messages: Final = [
+        AnthropicMessagesUserMessageParam(role="user", content="Which city is colder?"),
+        AnthopicMessagesAssistantMessageParam(
+            role="assistant",
+            content=[
+                {
+                    "type": "thinking",
+                    "thinking": "Oslo is colder.",
+                    "signature": encrypted_reasoning_signature(summary_encrypted),
+                },
+                {"type": "redacted_thinking", "data": encrypted_reasoning_signature(bytes_encrypted)},
+                {"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "Oslo"}},
+            ],
+        ),
+        AnthropicMessagesUserMessageParam(
+            role="user",
+            content=[{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}],
+        ),
+    ]
+
+    kept: Final = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
+        messages=messages,
+        model="chatgpt/gpt-6.1-sol",
+        custom_llm_provider="chatgpt",
+    )
+    assistant: Final = kept[1]
+    assert [block["type"] for block in assistant["thinking_blocks"]] == ["thinking", "redacted_thinking"]
+    assert assistant["thinking_blocks"][0]["signature"] == encrypted_reasoning_signature(summary_encrypted)
+    assert assistant["thinking_blocks"][1]["data"] == encrypted_reasoning_signature(bytes_encrypted)
+
+    responses_input, _instructions = LiteLLMResponsesTransformationHandler().convert_chat_completion_messages_to_responses_api(
+        kept
+    )
+    reasoning: Final = tuple(item for item in responses_input if item.get("type") == "reasoning")
+    assert [item["encrypted_content"] for item in reasoning] == [summary_encrypted, bytes_encrypted]
+    types: Final = [item.get("type") for item in responses_input]
+    assert types.index("reasoning") < types.index("function_call")
+
+    stripped: Final = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
+        messages=messages,
+        model="gpt-4o",
+        custom_llm_provider="openai",
+    )
+    assert not stripped[1].get("thinking_blocks")
+
+    prefixed: Final = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
+        messages=messages,
+        model="responses/gpt-5-mini",
+        custom_llm_provider="openai",
+    )
+    assert prefixed[1]["thinking_blocks"][1]["data"] == encrypted_reasoning_signature(bytes_encrypted)
+
+
+def test_translate_openai_reasoning_items_become_signed_thinking_blocks():
+    encrypted: Final = "gAAAA_outbound"
+    openai_choices: Final = [
+        Choices(
+            message=Message(
+                role="assistant",
+                content="Oslo.",
+                reasoning_items=[
+                    {
+                        "type": "reasoning",
+                        "id": "rs_1",
+                        "encrypted_content": encrypted,
+                        "summary": [{"type": "summary_text", "text": "Oslo is colder."}],
+                    }
+                ],
+                tool_calls=[
+                    ChatCompletionAssistantToolCall(
+                        id="call_1",
+                        type="function",
+                        function=Function(name="get_weather", arguments='{"city": "Oslo"}'),
+                    )
+                ],
+            )
+        )
+    ]
+
+    result: Final = LiteLLMAnthropicMessagesAdapter()._translate_openai_content_to_anthropic(choices=openai_choices)
+
+    assert result[0]["type"] == "thinking"
+    assert result[0]["thinking"] == "Oslo is colder."
+    assert result[0]["signature"] == encrypted_reasoning_signature(encrypted)
+    assert result[1]["type"] == "text"
+    assert result[2]["type"] == "tool_use"
+
+
+def test_translate_openai_reasoning_items_without_summary_become_redacted_thinking():
+    encrypted: Final = "gAAAA_bytes_only"
+    openai_choices: Final = [
+        Choices(
+            message=Message(
+                role="assistant",
+                content=None,
+                reasoning_items=[
+                    {"type": "reasoning", "id": "rs_1", "encrypted_content": encrypted, "summary": []}
+                ],
+            )
+        )
+    ]
+
+    result: Final = LiteLLMAnthropicMessagesAdapter()._translate_openai_content_to_anthropic(choices=openai_choices)
+
+    assert result == [
+        {"type": "redacted_thinking", "data": encrypted_reasoning_signature(encrypted)},
+    ]
+
+
+def test_translate_openai_reasoning_items_are_not_duplicated_when_thinking_blocks_already_carry_them():
+    encrypted: Final = "gAAAA_once"
+    signature: Final = encrypted_reasoning_signature(encrypted)
+    openai_choices: Final = [
+        Choices(
+            message=Message(
+                role="assistant",
+                content="Oslo.",
+                thinking_blocks=[{"type": "redacted_thinking", "data": signature}],
+                reasoning_items=[
+                    {"type": "reasoning", "id": "rs_1", "encrypted_content": encrypted, "summary": []}
+                ],
+            )
+        )
+    ]
+
+    result: Final = LiteLLMAnthropicMessagesAdapter()._translate_openai_content_to_anthropic(choices=openai_choices)
+
+    assert [block["type"] for block in result] == ["redacted_thinking", "text"]
+
+
+def _replayed_reasoning_request(tools: list[dict[str, object]]) -> AnthropicMessagesRequest:
+    return AnthropicMessagesRequest(
+        model="azure/gpt-5.4-nano",
+        max_tokens=64,
+        messages=[
+            AnthropicMessagesUserMessageParam(role="user", content="Which city is colder?"),
+            AnthopicMessagesAssistantMessageParam(
+                role="assistant",
+                content=[
+                    {"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAAA_azure_tools")},
+                    {"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "Oslo"}},
+                ],
+            ),
+            AnthropicMessagesUserMessageParam(
+                role="user",
+                content=[{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}],
+            ),
+        ],
+        tools=tools,
+    )
+
+
+@pytest.mark.parametrize(
+    ("tools", "kept"),
+    [
+        pytest.param(
+            [{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}],
+            True,
+            id="function_tools_bridge_to_responses",
+        ),
+        pytest.param([], False, id="no_tools_stay_on_chat"),
+    ],
+)
+def test_encrypted_reasoning_is_kept_when_the_request_tools_route_it_to_responses(
+    tools: list[dict[str, object]], kept: bool
+) -> None:
+    openai_request, _mapping = LiteLLMAnthropicMessagesAdapter().translate_anthropic_to_openai(
+        _replayed_reasoning_request(tools),
+        custom_llm_provider="azure",
+        api_base="https://keep-check.openai.azure.com",
+    )
+
+    assistant: Final = openai_request["messages"][1]
+    assert bool(assistant.get("thinking_blocks")) is kept
+
+
+@pytest.mark.parametrize(
+    ("deployment_params", "kept"),
+    [
+        pytest.param({"reasoning_summary": "auto"}, True, id="deployment_summary_bridges_to_responses"),
+        pytest.param({"extra_body": {"reasoningSummary": "auto"}}, True, id="extra_body_summary_bridges_to_responses"),
+        pytest.param({}, False, id="no_summary_stays_on_chat"),
+    ],
+)
+def test_encrypted_reasoning_is_kept_when_the_deployment_reasoning_summary_routes_it_to_responses(
+    deployment_params: dict[str, object], kept: bool
+) -> None:
+    from litellm.llms.anthropic.messages.adapters.handler import (
+        LiteLLMMessagesToCompletionTransformationHandler,
+    )
+
+    request: Final = _replayed_reasoning_request([])
+    completion_kwargs, _mapping = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
+        max_tokens=64,
+        messages=cast(list[dict[str, object]], request["messages"]),
+        model="azure/gpt-5.1",
+        thinking={"type": "enabled", "budget_tokens": 2048},
+        extra_kwargs={"api_base": "https://keep-check.openai.azure.com", **deployment_params},
+    )
+
+    assistant: Final = cast(list[Mapping[str, object]], completion_kwargs["messages"])[1]
+    assert bool(assistant.get("thinking_blocks")) is kept
+
+
 def test_translate_anthropic_messages_to_openai_sets_reasoning_content():
     """Reasoning-aware chat providers read reasoning_content, so thinking text must land there.
 
@@ -5578,3 +5793,503 @@ async def test_chat_bridge_tool_result_text_block_without_text_reaches_the_model
             ],
         }
     ]
+
+
+_RETRY_API_BASE: Final = "https://encrypted-retry.openai.azure.com"
+_RETRY_URL: Final = r".*encrypted-retry\.openai\.azure\.com/openai/.*responses.*"
+_RETRY_TOOLS: Final = [{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}]
+_RETRY_REJECTION: Final = httpx.Response(
+    400,
+    json={
+        "error": {
+            "message": "The encrypted content gAAA_foreign could not be verified.",
+            "type": "invalid_request_error",
+            "param": None,
+            "code": "invalid_encrypted_content",
+        }
+    },
+)
+_UNRELATED_REJECTION: Final = httpx.Response(
+    400, json={"error": {"message": "max_output_tokens is too small", "type": "invalid_request_error"}}
+)
+_UNRELATED_UNVERIFIED_REJECTION: Final = httpx.Response(
+    400, json={"error": {"message": "The organization could not be verified.", "type": "invalid_request_error"}}
+)
+_RETRY_REPLY: Final = {
+    "id": "resp_retry",
+    "object": "response",
+    "created_at": 1700000000,
+    "status": "completed",
+    "model": "gpt-5.4-nano",
+    "output": [
+        {
+            "type": "message",
+            "id": "msg_retry",
+            "status": "completed",
+            "role": "assistant",
+            "content": [{"type": "output_text", "text": "Oslo", "annotations": []}],
+        }
+    ],
+    "usage": {"input_tokens": 10, "output_tokens": 2, "total_tokens": 12},
+    "parallel_tool_calls": True,
+    "tool_choice": "auto",
+    "tools": [],
+}
+
+
+def _history_with_foreign_reasoning() -> list[dict[str, object]]:
+    return [
+        {"role": "user", "content": "Which city is colder?"},
+        {
+            "role": "assistant",
+            "content": [
+                {"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAA_foreign")},
+                {"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "Oslo"}},
+            ],
+        },
+        {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}]},
+    ]
+
+
+def _sent_reasoning_items(request: httpx.Request) -> list[object]:
+    sent: Final = json.loads(request.read())
+    return [item for item in sent["input"] if item.get("type") == "reasoning"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+async def test_async_messages_retry_once_without_reasoning_the_backend_cannot_decrypt(
+    respx_mock: respx.MockRouter,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
+        side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
+    )
+
+    with caplog.at_level(logging.WARNING, logger="LiteLLM"):
+        response: Final = await litellm.anthropic.messages.acreate(
+            model="azure/gpt-5.4-nano",
+            api_base=_RETRY_API_BASE,
+            api_key="fake-key",
+            api_version="2025-04-01-preview",
+            max_tokens=64,
+            tools=_RETRY_TOOLS,
+            messages=_history_with_foreign_reasoning(),
+        )
+
+    warnings: Final = tuple(
+        record.getMessage()
+        for record in caplog.records
+        if record.levelno >= logging.WARNING and record.name == "LiteLLM"
+    )
+    retry_warnings: Final = tuple(message for message in warnings if "retrying once without it" in message)
+    assert route.call_count == 2
+    assert len(_sent_reasoning_items(route.calls[0].request)) == 1
+    assert _sent_reasoning_items(route.calls[1].request) == []
+    assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
+    assert retry_warnings == (
+        "Backend rejected replayed encrypted reasoning, retrying once without it. "
+        "Enable the encrypted_content_affinity pre-call check to drop such blocks before the call. "
+        "provider=azure model=azure/gpt-5.4-nano status=400 code=invalid_encrypted_content",
+    )
+    assert all(
+        "gAAA_foreign" not in message and "litellm_encrypted_reasoning:" not in message for message in warnings
+    )
+
+
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+def test_sync_messages_retry_once_without_reasoning_the_backend_cannot_decrypt(
+    respx_mock: respx.MockRouter,
+) -> None:
+    route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
+        side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
+    )
+
+    response: Final = litellm.anthropic.messages.create(
+        model="azure/gpt-5.4-nano",
+        api_base=_RETRY_API_BASE,
+        api_key="fake-key",
+        api_version="2025-04-01-preview",
+        max_tokens=64,
+        tools=_RETRY_TOOLS,
+        messages=_history_with_foreign_reasoning(),
+    )
+
+    assert route.call_count == 2
+    assert _sent_reasoning_items(route.calls[1].request) == []
+    assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+async def test_async_messages_with_num_retries_send_the_rejected_reasoning_only_once(
+    respx_mock: respx.MockRouter,
+) -> None:
+    route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
+        side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
+    )
+
+    await litellm.anthropic.messages.acreate(
+        model="azure/gpt-5.4-nano",
+        api_base=_RETRY_API_BASE,
+        api_key="fake-key",
+        api_version="2025-04-01-preview",
+        max_tokens=64,
+        tools=_RETRY_TOOLS,
+        messages=_history_with_foreign_reasoning(),
+        num_retries=2,
+    )
+
+    assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 0]
+
+
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+def test_sync_messages_with_global_num_retries_send_the_rejected_reasoning_once_and_keep_the_setting(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "num_retries", 2)
+    route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
+        side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
+    )
+
+    litellm.anthropic.messages.create(
+        model="azure/gpt-5.4-nano",
+        api_base=_RETRY_API_BASE,
+        api_key="fake-key",
+        api_version="2025-04-01-preview",
+        max_tokens=64,
+        tools=_RETRY_TOOLS,
+        messages=_history_with_foreign_reasoning(),
+    )
+
+    assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 0]
+    assert litellm.num_retries == 2
+
+
+_SERVER_ERROR: Final = httpx.Response(
+    500, json={"error": {"message": "The server had an error while processing your request.", "type": "server_error"}}
+)
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+async def test_async_messages_retry_a_server_error_with_the_reasoning_and_drop_it_only_once_rejected(
+    respx_mock: respx.MockRouter,
+) -> None:
+    route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
+        side_effect=[_SERVER_ERROR, _RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
+    )
+
+    response: Final = await litellm.anthropic.messages.acreate(
+        model="azure/gpt-5.4-nano",
+        api_base=_RETRY_API_BASE,
+        api_key="fake-key",
+        api_version="2025-04-01-preview",
+        max_tokens=64,
+        tools=_RETRY_TOOLS,
+        messages=_history_with_foreign_reasoning(),
+        num_retries=2,
+    )
+
+    assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 1, 0]
+    assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
+
+
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+def test_sync_messages_retry_a_server_error_with_the_reasoning_and_drop_it_only_once_rejected(
+    respx_mock: respx.MockRouter,
+) -> None:
+    route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
+        side_effect=[_SERVER_ERROR, _RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
+    )
+
+    response: Final = litellm.anthropic.messages.create(
+        model="azure/gpt-5.4-nano",
+        api_base=_RETRY_API_BASE,
+        api_key="fake-key",
+        api_version="2025-04-01-preview",
+        max_tokens=64,
+        tools=_RETRY_TOOLS,
+        messages=_history_with_foreign_reasoning(),
+        num_retries=2,
+    )
+
+    assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 1, 0]
+    assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
+
+
+class _FailureLog(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.rejections: tuple[str, ...] = ()
+
+    def _record(self, kwargs: Mapping[str, object]) -> None:
+        error: Final = str(kwargs.get("exception"))
+        self.rejections = (
+            *self.rejections,
+            next((phrase for phrase in _UPSTREAM_REJECTION_PHRASES if phrase in error), error),
+        )
+
+    def log_failure_event(
+        self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self._record(kwargs)
+
+    async def async_log_failure_event(
+        self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self._record(kwargs)
+
+
+_UPSTREAM_REJECTION_PHRASES: Final = ("could not be verified", "max_output_tokens is too small")
+_OUTCOMES_OF_THE_TURN_SENT_AGAIN: Final = (
+    pytest.param(httpx.Response(200, json=_RETRY_REPLY), (), id="served"),
+    pytest.param(_UNRELATED_REJECTION, ("max_output_tokens is too small",), id="rejected"),
+)
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
+async def test_async_messages_log_only_the_outcome_of_the_turn_sent_again(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    sent_again: httpx.Response,
+    logged_rejections: tuple[str, ...],
+) -> None:
+    failure_log: Final = _FailureLog()
+    monkeypatch.setattr(litellm, "callbacks", [failure_log])
+    respx_mock.post(url__regex=_RETRY_URL).mock(side_effect=[_RETRY_REJECTION, sent_again])
+
+    with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
+        await litellm.anthropic.messages.acreate(
+            model="azure/gpt-5.4-nano",
+            api_base=_RETRY_API_BASE,
+            api_key="fake-key",
+            api_version="2025-04-01-preview",
+            max_tokens=64,
+            tools=_RETRY_TOOLS,
+            messages=_history_with_foreign_reasoning(),
+        )
+
+    assert failure_log.rejections == logged_rejections
+
+
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
+def test_sync_messages_log_only_the_outcome_of_the_turn_sent_again(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    sent_again: httpx.Response,
+    logged_rejections: tuple[str, ...],
+) -> None:
+    failure_log: Final = _FailureLog()
+    monkeypatch.setattr(litellm, "callbacks", [failure_log])
+    respx_mock.post(url__regex=_RETRY_URL).mock(side_effect=[_RETRY_REJECTION, sent_again])
+
+    with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
+        litellm.anthropic.messages.create(
+            model="azure/gpt-5.4-nano",
+            api_base=_RETRY_API_BASE,
+            api_key="fake-key",
+            api_version="2025-04-01-preview",
+            max_tokens=64,
+            tools=_RETRY_TOOLS,
+            messages=_history_with_foreign_reasoning(),
+        )
+
+    assert failure_log.rejections == logged_rejections
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+@pytest.mark.parametrize(
+    ("messages", "rejection"),
+    [
+        pytest.param(
+            [{"role": "user", "content": "Which city is colder?"}],
+            _RETRY_REJECTION,
+            id="no_encrypted_reasoning_to_drop",
+        ),
+        pytest.param(_history_with_foreign_reasoning(), _UNRELATED_REJECTION, id="unrelated_rejection"),
+        pytest.param(
+            _history_with_foreign_reasoning(), _UNRELATED_UNVERIFIED_REJECTION, id="unrelated_unverified_rejection"
+        ),
+    ],
+)
+async def test_messages_do_not_retry_other_rejections(
+    respx_mock: respx.MockRouter, messages: list[dict[str, object]], rejection: httpx.Response
+) -> None:
+    route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(return_value=rejection)
+
+    with pytest.raises(litellm.BadRequestError):
+        await litellm.anthropic.messages.acreate(
+            model="azure/gpt-5.4-nano",
+            api_base=_RETRY_API_BASE,
+            api_key="fake-key",
+            api_version="2025-04-01-preview",
+            max_tokens=64,
+            tools=_RETRY_TOOLS,
+            messages=messages,
+        )
+
+    assert route.call_count == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+async def test_messages_do_not_retry_when_the_translated_request_already_dropped_the_reasoning(
+    respx_mock: respx.MockRouter,
+) -> None:
+    route: Final = respx_mock.post(url__regex=_CHAT_BRIDGE_URL).mock(return_value=_RETRY_REJECTION)
+
+    with pytest.raises(litellm.BadRequestError):
+        await litellm.anthropic.messages.acreate(
+            model=_CHAT_BRIDGE_MODEL,
+            api_base=_CHAT_BRIDGE_API_BASE,
+            api_key="fake-key",
+            max_tokens=64,
+            messages=_history_with_foreign_reasoning(),
+        )
+
+    assert route.call_count == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+async def test_openai_thinking_on_the_chat_adapter_replays_the_reasoning_its_responses_route_minted(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", True)
+    route: Final = respx_mock.post(url__regex=r".*openai-thinking\.example/v1/responses").mock(
+        return_value=httpx.Response(200, json=_RETRY_REPLY)
+    )
+
+    await litellm.anthropic.messages.acreate(
+        model="openai/gpt-5.4-nano",
+        api_base="https://openai-thinking.example/v1",
+        api_key="fake-key",
+        max_tokens=2048,
+        thinking={"type": "enabled", "budget_tokens": 1024},
+        messages=[
+            {"role": "user", "content": "Which city is colder?"},
+            {
+                "role": "assistant",
+                "content": [
+                    {"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAA_same_org")},
+                    {"type": "text", "text": "Oslo."},
+                ],
+            },
+            {"role": "user", "content": "And by how much?"},
+        ],
+    )
+
+    assert route.call_count == 1
+    assert [item["encrypted_content"] for item in _sent_reasoning_items(route.calls[0].request)] == ["gAAA_same_org"]
+
+
+def _responses_sse(events: Sequence[Mapping[str, object]]) -> bytes:
+    return b"".join(
+        f"event: {event['type']}\ndata: {json.dumps({**event, 'sequence_number': number})}\n\n".encode()
+        for number, event in enumerate(events)
+    )
+
+
+def _reasoning_item(item_id: str, encrypted: str, summary: str = "") -> dict[str, object]:
+    return {
+        "type": "reasoning",
+        "id": item_id,
+        "summary": [{"type": "summary_text", "text": summary}] if summary else [],
+        "encrypted_content": encrypted,
+    }
+
+
+def _streamed_blocks(frames: str) -> list[tuple[str, str, str | None]]:
+    events: Final = tuple(
+        json.loads(line.removeprefix("data: ")) for line in frames.splitlines() if line.startswith("data: ")
+    )
+    deltas: Final = tuple(event for event in events if event["type"] == "content_block_delta")
+
+    def block_deltas(index: int) -> tuple[Mapping[str, str], ...]:
+        return tuple(delta["delta"] for delta in deltas if delta["index"] == index)
+
+    return [
+        (
+            event["content_block"]["type"],
+            "".join(delta.get("thinking", "") for delta in block_deltas(event["index"])),
+            next(
+                (delta["signature"] for delta in block_deltas(event["index"]) if delta["type"] == "signature_delta"),
+                None,
+            ),
+        )
+        for event in events
+        if event["type"] == "content_block_start"
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_chat_bridge_httpx")
+async def test_streamed_reasoning_summaries_stay_with_their_own_encrypted_item(
+    respx_mock: respx.MockRouter,
+) -> None:
+    response: Final = {**_RETRY_REPLY, "id": "resp_two_summaries", "status": "in_progress", "output": []}
+    message: Final = {
+        "type": "message",
+        "id": "msg_two_summaries",
+        "status": "completed",
+        "role": "assistant",
+        "content": [{"type": "output_text", "text": "Oslo", "annotations": []}],
+    }
+    first: Final = _reasoning_item("rs_1", "gAAA_first", "Oslo is further north.")
+    second: Final = _reasoning_item("rs_2", "gAAA_second", "Check the forecast.")
+    completed: Final = {**response, "status": "completed", "output": [first, second, message]}
+
+    def summary_delta(output_index: int, item_id: str, text: str) -> dict[str, object]:
+        return {
+            "type": "response.reasoning_summary_text.delta",
+            "output_index": output_index,
+            "item_id": item_id,
+            "summary_index": 0,
+            "delta": text,
+        }
+
+    events: Final = (
+        {"type": "response.created", "response": response},
+        {"type": "response.output_item.added", "output_index": 0, "item": _reasoning_item("rs_1", "")},
+        summary_delta(0, "rs_1", "Oslo is further north."),
+        {"type": "response.output_item.done", "output_index": 0, "item": first},
+        {"type": "response.output_item.added", "output_index": 1, "item": _reasoning_item("rs_2", "")},
+        summary_delta(1, "rs_2", "Check the forecast."),
+        {"type": "response.output_item.done", "output_index": 1, "item": second},
+        {"type": "response.output_item.added", "output_index": 2, "item": {**message, "content": []}},
+        {
+            "type": "response.output_text.delta",
+            "output_index": 2,
+            "item_id": "msg_two_summaries",
+            "content_index": 0,
+            "delta": "Oslo",
+        },
+        {"type": "response.output_item.done", "output_index": 2, "item": message},
+        {"type": "response.completed", "response": completed},
+    )
+    respx_mock.post(url__regex=_RETRY_URL).mock(
+        return_value=httpx.Response(200, headers={"content-type": "text/event-stream"}, content=_responses_sse(events))
+    )
+
+    stream: Final = await litellm.anthropic.messages.acreate(
+        model="azure/gpt-5.4-nano",
+        api_base=_RETRY_API_BASE,
+        api_key="fake-key",
+        api_version="2025-04-01-preview",
+        max_tokens=64,
+        tools=_RETRY_TOOLS,
+        stream=True,
+        messages=[{"role": "user", "content": "Which city is colder?"}],
+    )
+    frames: Final = b"".join([chunk async for chunk in stream]).decode()
+
+    assert _streamed_blocks(frames) == [
+        ("thinking", "Oslo is further north.", encrypted_reasoning_signature("gAAA_first")),
+        ("thinking", "Check the forecast.", encrypted_reasoning_signature("gAAA_second")),
+        ("text", "", None),
+    ]
diff --git a/tests/unit/llms/anthropic/messages/responses_adapters/test_responses_adapters_handler.py b/tests/unit/llms/anthropic/messages/responses_adapters/test_responses_adapters_handler.py
index c06c9d721ac..cd6e323daa2 100644
--- a/tests/unit/llms/anthropic/messages/responses_adapters/test_responses_adapters_handler.py
+++ b/tests/unit/llms/anthropic/messages/responses_adapters/test_responses_adapters_handler.py
@@ -3,20 +3,24 @@ import json
 import os
 import sys
 from collections.abc import AsyncIterator, Mapping
+from contextlib import nullcontext
 from itertools import chain
 from typing import Final, cast
 from unittest.mock import AsyncMock, patch
 
+import httpx
 import pytest
 import respx
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
 
 import litellm
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
     TOOL_RESULT_IMAGE_BOUNDARY,
     TOOL_RESULT_IMAGE_PLACEHOLDER,
 )
+from litellm.litellm_core_utils.prompt_templates.reasoning_utils import encrypted_reasoning_signature
 from litellm.llms.anthropic.messages.responses_adapters.handler import (
     LiteLLMMessagesToResponsesAPIHandler,
     _build_responses_kwargs,
@@ -1135,3 +1139,291 @@ async def test_the_sync_path_reports_its_edits_on_the_streamed_message_delta(
     ]
     message_delta: Final = next(event for event in events if event["type"] == "message_delta")
     assert message_delta["context_management"]["applied_edits"][0]["cleared_tool_uses"] == 2
+
+
+REJECTED_REPLAYED_REASONING: Final = {
+    "error": {
+        "message": "The encrypted content gAAA_foreign could not be verified.",
+        "type": "invalid_request_error",
+        "param": None,
+        "code": "invalid_encrypted_content",
+    }
+}
+UNVERIFIED_ORGANIZATION: Final = {
+    "error": {"message": "The organization could not be verified.", "type": "invalid_request_error"}
+}
+
+
+def _history(*, replayed_reasoning: bool) -> list[dict[str, object]]:
+    reasoning: Final = (
+        ({"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAA_foreign")},)
+        if replayed_reasoning
+        else ()
+    )
+    return [
+        {"role": "user", "content": "List the logs."},
+        {"role": "assistant", "content": [*reasoning, {"type": "tool_use", "id": "call_1", "name": "Bash", "input": {}}]},
+        {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call_1", "content": "app.log"}]},
+    ]
+
+
+def _replayed_reasoning_counts(upstream: respx.Route) -> list[int]:
+    return [
+        sum(item.get("type") == "reasoning" for item in json.loads(call.request.content)["input"])
+        for call in upstream.calls
+    ]
+
+
+def _upstream(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, first: httpx.Response, stream: bool):
+    return _upstream_after(respx_mock, monkeypatch, (first,), stream)
+
+
+def _upstream_after(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, failures: tuple[httpx.Response, ...], stream: bool
+) -> respx.Route:
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-replay-test")
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    litellm.in_memory_llm_clients_cache.flush_cache()
+    served: Final = (
+        httpx.Response(200, headers={"Content-Type": "text/event-stream"}, content=RESPONSES_SSE_BODY)
+        if stream
+        else httpx.Response(200, json=TEXT_RESPONSE)
+    )
+    return respx_mock.post("https://api.openai.com/v1/responses").mock(side_effect=[*failures, served])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stream", [False, True])
+async def test_openai_turn_is_sent_again_once_without_replayed_reasoning_the_backend_cannot_decrypt(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, stream: bool
+) -> None:
+    upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), stream)
+
+    response: Final = await LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler(
+        max_tokens=64,
+        messages=_history(replayed_reasoning=True),
+        model="openai/gpt-5.6-luna",
+        tools=[dict(BASH_TOOL)],
+        stream=stream,
+        custom_llm_provider="openai",
+    )
+    events: Final = (
+        [json.loads(chunk.decode().split("data: ", 1)[1]) async for chunk in cast(AsyncIterator[bytes], response)]
+        if stream
+        else [cast(Mapping[str, object], response)]
+    )
+
+    assert _replayed_reasoning_counts(upstream) == [1, 0]
+    assert events[-1]["type"] in ("message", "message_stop")
+
+
+def test_sync_openai_turn_is_sent_again_once_without_replayed_reasoning_the_backend_cannot_decrypt(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), False)
+
+    response: Final = LiteLLMMessagesToResponsesAPIHandler.anthropic_messages_handler(
+        max_tokens=64,
+        messages=_history(replayed_reasoning=True),
+        model="openai/gpt-5.6-luna",
+        tools=[dict(BASH_TOOL)],
+        custom_llm_provider="openai",
+    )
+
+    assert _replayed_reasoning_counts(upstream) == [1, 0]
+    assert cast(Mapping[str, object], response)["content"] == [{"type": "text", "text": "done"}]
+
+
+@pytest.mark.asyncio
+async def test_openai_turn_with_num_retries_sends_the_rejected_reasoning_only_once(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), False)
+
+    await litellm.anthropic.messages.acreate(
+        max_tokens=64,
+        messages=_history(replayed_reasoning=True),
+        model="openai/gpt-5.6-luna",
+        tools=[dict(BASH_TOOL)],
+        num_retries=2,
+    )
+
+    assert _replayed_reasoning_counts(upstream) == [1, 0]
+
+
+def test_sync_openai_turn_with_global_num_retries_sends_the_rejected_reasoning_once_and_keeps_the_setting(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "num_retries", 2)
+    upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), False)
+
+    litellm.anthropic.messages.create(
+        max_tokens=64,
+        messages=_history(replayed_reasoning=True),
+        model="openai/gpt-5.6-luna",
+        tools=[dict(BASH_TOOL)],
+    )
+
+    assert _replayed_reasoning_counts(upstream) == [1, 0]
+    assert litellm.num_retries == 2
+
+
+_SERVER_ERROR: Final = {
+    "error": {"message": "The server had an error while processing your request.", "type": "server_error"}
+}
+
+
+@pytest.mark.asyncio
+async def test_openai_turn_retries_a_server_error_with_the_reasoning_and_drops_it_only_once_rejected(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    upstream: Final = _upstream_after(
+        respx_mock,
+        monkeypatch,
+        (httpx.Response(500, json=_SERVER_ERROR), httpx.Response(400, json=REJECTED_REPLAYED_REASONING)),
+        False,
+    )
+
+    response: Final = await litellm.anthropic.messages.acreate(
+        max_tokens=64,
+        messages=_history(replayed_reasoning=True),
+        model="openai/gpt-5.6-luna",
+        tools=[dict(BASH_TOOL)],
+        num_retries=2,
+    )
+
+    assert _replayed_reasoning_counts(upstream) == [1, 1, 0]
+    assert cast(Mapping[str, object], response)["content"] == [{"type": "text", "text": "done"}]
+
+
+def test_sync_openai_turn_retries_a_server_error_with_the_reasoning_and_drops_it_only_once_rejected(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    upstream: Final = _upstream_after(
+        respx_mock,
+        monkeypatch,
+        (httpx.Response(500, json=_SERVER_ERROR), httpx.Response(400, json=REJECTED_REPLAYED_REASONING)),
+        False,
+    )
+
+    response: Final = litellm.anthropic.messages.create(
+        max_tokens=64,
+        messages=_history(replayed_reasoning=True),
+        model="openai/gpt-5.6-luna",
+        tools=[dict(BASH_TOOL)],
+        num_retries=2,
+    )
+
+    assert _replayed_reasoning_counts(upstream) == [1, 1, 0]
+    assert cast(Mapping[str, object], response)["content"] == [{"type": "text", "text": "done"}]
+
+
+class _FailureLog(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.rejections: tuple[str, ...] = ()
+
+    def _record(self, kwargs: Mapping[str, object]) -> None:
+        error: Final = str(kwargs.get("exception"))
+        self.rejections = (
+            *self.rejections,
+            next((phrase for phrase in ("encrypted content", "organization") if phrase in error), error),
+        )
+
+    def log_failure_event(
+        self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self._record(kwargs)
+
+    async def async_log_failure_event(
+        self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self._record(kwargs)
+
+
+_OUTCOMES_OF_THE_TURN_SENT_AGAIN: Final = (
+    pytest.param(httpx.Response(200, json=TEXT_RESPONSE), (), id="served"),
+    pytest.param(httpx.Response(400, json=UNVERIFIED_ORGANIZATION), ("organization",), id="rejected"),
+)
+
+
+def _rejected_then(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, sent_again: httpx.Response) -> None:
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-replay-test")
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    litellm.in_memory_llm_clients_cache.flush_cache()
+    respx_mock.post("https://api.openai.com/v1/responses").mock(
+        side_effect=[httpx.Response(400, json=REJECTED_REPLAYED_REASONING), sent_again]
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
+async def test_openai_turn_logs_only_the_outcome_of_the_turn_sent_again(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    sent_again: httpx.Response,
+    logged_rejections: tuple[str, ...],
+) -> None:
+    failure_log: Final = _FailureLog()
+    monkeypatch.setattr(litellm, "callbacks", [failure_log])
+    _rejected_then(respx_mock, monkeypatch, sent_again)
+
+    with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
+        await litellm.anthropic.messages.acreate(
+            max_tokens=64,
+            messages=_history(replayed_reasoning=True),
+            model="openai/gpt-5.6-luna",
+            tools=[dict(BASH_TOOL)],
+        )
+
+    assert failure_log.rejections == logged_rejections
+
+
+@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
+def test_sync_openai_turn_logs_only_the_outcome_of_the_turn_sent_again(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    sent_again: httpx.Response,
+    logged_rejections: tuple[str, ...],
+) -> None:
+    failure_log: Final = _FailureLog()
+    monkeypatch.setattr(litellm, "callbacks", [failure_log])
+    _rejected_then(respx_mock, monkeypatch, sent_again)
+
+    with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
+        litellm.anthropic.messages.create(
+            max_tokens=64,
+            messages=_history(replayed_reasoning=True),
+            model="openai/gpt-5.6-luna",
+            tools=[dict(BASH_TOOL)],
+        )
+
+    assert failure_log.rejections == logged_rejections
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "replayed_reasoning, rejection",
+    [
+        pytest.param(False, REJECTED_REPLAYED_REASONING, id="no_replayed_reasoning"),
+        pytest.param(True, UNVERIFIED_ORGANIZATION, id="unrelated_rejection"),
+    ],
+)
+async def test_openai_turn_is_not_sent_again_for_other_rejections(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    replayed_reasoning: bool,
+    rejection: Mapping[str, object],
+) -> None:
+    upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=rejection), False)
+
+    with pytest.raises(litellm.BadRequestError):
+        await LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler(
+            max_tokens=64,
+            messages=_history(replayed_reasoning=replayed_reasoning),
+            model="openai/gpt-5.6-luna",
+            tools=[dict(BASH_TOOL)],
+            custom_llm_provider="openai",
+        )
+
+    assert upstream.call_count == 1
diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py
index 2a5c3c81f18..548f78d148b 100644
--- a/tests/unit/test_main.py
+++ b/tests/unit/test_main.py
@@ -8,7 +8,7 @@ import logging
 import os
 import threading
 import urllib.parse
-from collections.abc import Callable, Mapping
+from collections.abc import Awaitable, Callable, Mapping
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
 from datetime import datetime
@@ -9395,6 +9395,19 @@ async def test_responses_with_retries(sync_mode):
         assert mock_responses.call_args.kwargs["max_retries"] == 0
 
 
+@pytest.mark.parametrize("retry_function", [acompletion_with_retries, aresponses_with_retries])
+@pytest.mark.asyncio
+async def test_async_retry_helpers_stop_when_the_call_is_cancelled(
+    retry_function: Callable[..., Awaitable[object]],
+) -> None:
+    upstream: Final = AsyncMock(side_effect=[asyncio.CancelledError(), MagicMock()])
+
+    with pytest.raises(asyncio.CancelledError):
+        await retry_function(model="gpt-4o", num_retries=3, original_function=upstream)
+
+    assert upstream.await_count == 1
+
+
 def test_azure_embedding_exceptions():
     with pytest.raises(Exception, match="Mock error") as exc_info:
         litellm.embedding(