From 95ef53878954101321792515f5b2cffb4e58c813 Mon Sep 17 00:00:00 2001 From: yucheng Date: Tue, 15 Sep 2026 01:22:06 +0000 Subject: [PATCH 1/6] fix(utils): log converted streams as streams so spend tracking works Deployment hooks such as Headroom downgrade stream=True to a non-streaming provider call and the agentic loop then hands back a CustomStreamWrapper (or MockResponsesAPIStreamingIterator for Responses). wrapper_async still saw kwargs["stream"] is False, so it took the non-streaming success path with a lazy stream object: no standard_logging_object was built, the proxy cost callback raised failed_tracking_spend, and the wrapper's own end-of-stream dispatch was deduped away. Treat a lazy stream result as streaming for logging regardless of the downgraded kwarg. Regression in v1.99.0 via #35017 Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/utils.py | 16 +++-- tests/test_litellm/test_utils.py | 117 +++++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+), 4 deletions(-) diff --git a/litellm/utils.py b/litellm/utils.py index 18df5e2abf7..9031e23b39e 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -846,6 +846,15 @@ def _is_streaming_response_for_correlation(result: object) -> bool: return isinstance(result, CustomStreamWrapper) +def _is_converted_stream_result(result: object) -> bool: + """True if `result` is a lazy stream wrapper the caller must iterate, even when a deployment + hook downgraded `kwargs["stream"]` to False for the provider call.""" + from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper + from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator + + return isinstance(result, (CustomStreamWrapper, BaseResponsesAPIStreamingIterator)) + + # Runs once per call to check if the user wants to send their data anywhere - PostHog/Sentry/Slack/etc. def function_setup( original_function: str, @@ -1946,10 +1955,9 @@ def client(original_function): raise end_time = datetime.datetime.now() - if _is_streaming_request( - kwargs=kwargs, - call_type=call_type, - ): + if _is_streaming_request(kwargs=kwargs, call_type=call_type) or _is_converted_stream_result(result): + logging_obj.stream = True + logging_obj.model_call_details["stream"] = True if "complete_response" in kwargs and kwargs["complete_response"] is True: chunks: Final = [] for idx, chunk in enumerate(result): diff --git a/tests/test_litellm/test_utils.py b/tests/test_litellm/test_utils.py index ae0e08ebfb1..7c33cf40bc8 100644 --- a/tests/test_litellm/test_utils.py +++ b/tests/test_litellm/test_utils.py @@ -32,6 +32,7 @@ from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.get_litellm_params import get_litellm_params from litellm.litellm_core_utils.thread_pool_executor import executor as logging_executor from litellm.proxy.utils import is_valid_api_key +from litellm.types.integrations.custom_logger import HEADROOM_CONVERTED_STREAM_KEY from litellm.types.utils import ( CallTypes, Delta, @@ -44,6 +45,7 @@ from litellm.types.utils import ( from litellm.types.utils import all_litellm_params, bedrock_batch_litellm_params from litellm.types.router import CredentialLiteLLMParams, GenericLiteLLMParams from litellm.utils import ( + CustomStreamWrapper, ProviderConfigManager, TextCompletionStreamWrapper, _check_provider_match, @@ -5307,6 +5309,121 @@ async def test_wrapper_async_restores_originating_task_context_after_success(mon session_id_var.set("") +class _ConvertStreamDeploymentHook(CustomLogger): + """Headroom-style interception: downgrade stream=True to a non-streaming provider call.""" + + async def async_pre_call_deployment_hook( + self, kwargs: dict[str, object], call_type: CallTypes | None + ) -> dict[str, object] | None: + if not kwargs.get("stream"): + return None + return {**kwargs, "stream": False, HEADROOM_CONVERTED_STREAM_KEY: True} + + +class _SuccessKwargsCapture(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.success_kwargs: list[dict[str, object]] = [] + + async def async_log_success_event( + self, kwargs: dict[str, object], response_obj: object, start_time: datetime, end_time: datetime + ) -> None: + self.success_kwargs.append(kwargs) + + +def _install_converted_stream_callbacks(monkeypatch: pytest.MonkeyPatch) -> _SuccessKwargsCapture: + capture: Final = _SuccessKwargsCapture() + monkeypatch.setattr(litellm, "callbacks", [_ConvertStreamDeploymentHook(), capture]) + monkeypatch.setattr(litellm, "success_callback", []) + monkeypatch.setattr(litellm, "_async_success_callback", []) + monkeypatch.setattr(litellm, "failure_callback", []) + monkeypatch.setattr(litellm, "_async_failure_callback", []) + return capture + + +async def _wait_for_success_kwargs(capture: _SuccessKwargsCapture) -> dict[str, object]: + for _ in range(50): + if capture.success_kwargs: + break + await asyncio.sleep(0.05) + (success_kwargs,) = capture.success_kwargs + return success_kwargs + + +@pytest.mark.asyncio +async def test_wrapper_async_logs_converted_chat_stream_with_standard_logging_object( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression LIT-7729: the fake CustomStreamWrapper hit the non-streaming success path, which + built no standard_logging_object and deduped the wrapper's own end-of-stream dispatch.""" + capture: Final = _install_converted_stream_callbacks(monkeypatch) + + response: Final = await litellm.acompletion( + model="gpt-5.6", + messages=[{"role": "user", "content": "hi"}], + stream=True, + mock_response="converted stream body", + num_retries=0, + ) + assert isinstance(response, CustomStreamWrapper) + chunks: Final = [chunk async for chunk in response] + assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks) == "converted stream body" + + success_kwargs: Final = await _wait_for_success_kwargs(capture) + standard_logging_object: Final = success_kwargs["standard_logging_object"] + assert isinstance(standard_logging_object, dict) + assert standard_logging_object["response_cost"] > 0 + assert standard_logging_object["stream"] is True + assert success_kwargs["stream"] is True + + +@pytest.mark.asyncio +@respx.mock +async def test_wrapper_async_logs_converted_responses_stream_with_standard_logging_object( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression LIT-7729, Responses surface: the fake MockResponsesAPIStreamingIterator took the + same non-streaming success path and lost its standard_logging_object.""" + from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator + + capture: Final = _install_converted_stream_callbacks(monkeypatch) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + litellm.in_memory_llm_clients_cache.flush_cache() + respx.post("https://api.openai.com/v1/responses").respond( + json={ + "id": "resp_converted", + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gpt-5.6", + "output": [ + { + "type": "message", + "id": "msg_converted", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "converted stream body", "annotations": []}], + } + ], + "usage": {"input_tokens": 3, "output_tokens": 4, "total_tokens": 7}, + } + ) + + response: Final = await litellm.aresponses( + model="openai/gpt-5.6", input="hi", stream=True, api_key="sk-test", num_retries=0 + ) + assert isinstance(response, BaseResponsesAPIStreamingIterator) + events: Final = [event async for event in response] + assert events[-1].type == "response.completed" + + success_kwargs: Final = await _wait_for_success_kwargs(capture) + standard_logging_object: Final = success_kwargs["standard_logging_object"] + assert isinstance(standard_logging_object, dict) + assert standard_logging_object["response_cost"] > 0 + assert standard_logging_object["stream"] is True + assert success_kwargs["stream"] is True + + def test_function_setup_failure_after_logging_construction_restores_context(monkeypatch): """If function_setup() constructs Logging() (which already mutated trace_id_var/session_id_var in __init__) but then raises before returning, From 8b86362703a865490e464b83c0a439e829211f22 Mon Sep 17 00:00:00 2001 From: yucheng Date: Tue, 15 Sep 2026 01:42:49 +0000 Subject: [PATCH 2/6] fix(caching): replay cache hits for converted streams as streams A deployment hook (Headroom, code interpreter, web search) can downgrade kwargs["stream"] to False while the caller still expects to iterate the result. The cache handler keyed stream replay and callback deferral off the raw flag, so a cache hit returned a plain object to a caller that iterates, and the Responses iterator never persisted the converted stream in the first place. Key both off the conversion marker as well Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/caching_handler.py | 17 ++-- litellm/responses/streaming_iterator.py | 5 +- litellm/utils.py | 10 ++- tests/test_litellm/test_utils.py | 106 +++++++++++++++++++++++- 4 files changed, 126 insertions(+), 12 deletions(-) diff --git a/litellm/caching/caching_handler.py b/litellm/caching/caching_handler.py index 139dcf058d2..901f2ffbad2 100644 --- a/litellm/caching/caching_handler.py +++ b/litellm/caching/caching_handler.py @@ -35,6 +35,7 @@ from litellm.litellm_core_utils.logging_utils import ( _assemble_complete_response_from_streaming_chunks, ) from litellm.types.caching import CachedEmbedding +from litellm.types.integrations.custom_logger import converted_stream_requested from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.rerank import RerankResponse from litellm.types.utils import ( @@ -107,6 +108,12 @@ def _is_chat_completion_cached_dict(cached_result: dict) -> bool: return "choices" in cached_result +def _stream_replay_requested(kwargs: Mapping[str, object]) -> bool: + """True when the caller must receive a stream, including when a deployment hook downgraded + `kwargs["stream"]` to False for the provider call.""" + return kwargs.get("stream", False) is True or converted_stream_requested(kwargs) + + def _should_defer_streaming_cache_hit_callbacks(*, kwargs: dict[str, object]) -> bool: """ When stream=True, do not run success callbacks at cache-hit time. @@ -117,7 +124,7 @@ def _should_defer_streaming_cache_hit_callbacks(*, kwargs: dict[str, object]) -> handlers when the stream finishes; firing them here too would double-count spend and callback records. """ - return kwargs.get("stream", False) is True + return _stream_replay_requested(kwargs) def _prompt_tokens_details_as_mapping(details: "PromptTokensDetailsWrapper") -> Mapping[str, object]: @@ -823,7 +830,7 @@ class LLMCachingHandler: if (call_type == CallTypes.acompletion.value or call_type == CallTypes.completion.value) and isinstance( cached_result, dict ): - if kwargs.get("stream", False) is True: + if _stream_replay_requested(kwargs): cached_result = self._convert_cached_stream_response( cached_result=cached_result, call_type=call_type, @@ -838,7 +845,7 @@ class LLMCachingHandler: if ( call_type == CallTypes.atext_completion.value or call_type == CallTypes.text_completion.value ) and isinstance(cached_result, dict): - if kwargs.get("stream", False) is True: + if _stream_replay_requested(kwargs): cached_result = self._convert_cached_stream_response( cached_result=cached_result, call_type=call_type, @@ -893,7 +900,7 @@ class LLMCachingHandler: elif (call_type == "aresponses" or call_type == "responses") and isinstance(cached_result, dict): use_chat_completion_cache: Final = _is_chat_completion_cached_dict(cached_result) if use_chat_completion_cache: - if kwargs.get("stream", False) is True: + if _stream_replay_requested(kwargs): bridge_call_type: Final = ( CallTypes.acompletion.value if call_type == "aresponses" else CallTypes.completion.value ) @@ -921,7 +928,7 @@ class LLMCachingHandler: ): response_obj._hidden_params["cache_hit"] = True - if kwargs.get("stream", False) is True: + if _stream_replay_requested(kwargs): cached_result = CachedResponsesAPIStreamingIterator( response=response_obj, logging_obj=logging_obj, diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index b39e130242d..38874768ca8 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -33,6 +33,7 @@ from litellm.litellm_core_utils.llm_response_utils.response_metadata import ( from litellm.litellm_core_utils.thread_pool_executor import executor from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig from litellm.responses.utils import ResponseAPILoggingUtils, ResponsesAPIRequestUtils +from litellm.types.integrations.custom_logger import converted_stream_requested from litellm.types.llms.openai import ( PART_UNION_TYPES, ResponseAPIUsage, @@ -626,7 +627,9 @@ class BaseResponsesAPIStreamingIterator: return request_kwargs = getattr(caching_handler, "request_kwargs", None) - if not _is_json_object(request_kwargs) or request_kwargs.get("stream") is not True: + if not _is_json_object(request_kwargs): + return + if request_kwargs.get("stream") is not True and not converted_stream_requested(request_kwargs): return request_kwargs = request_kwargs.copy() preset_cache_key = getattr(caching_handler, "preset_cache_key", None) diff --git a/litellm/utils.py b/litellm/utils.py index 9031e23b39e..8dc6576cc97 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -855,6 +855,11 @@ def _is_converted_stream_result(result: object) -> bool: return isinstance(result, (CustomStreamWrapper, BaseResponsesAPIStreamingIterator)) +def _mark_logging_as_stream(logging_obj: LiteLLMLoggingObject) -> None: + logging_obj.stream = True + logging_obj.model_call_details["stream"] = True + + # Runs once per call to check if the user wants to send their data anywhere - PostHog/Sentry/Slack/etc. def function_setup( original_function: str, @@ -1898,6 +1903,8 @@ def client(original_function): _caching_handler_response.cached_result is not None and _caching_handler_response.final_embedding_cached_response is None ): + if _is_converted_stream_result(_caching_handler_response.cached_result): + _mark_logging_as_stream(logging_obj) return _caching_handler_response.cached_result elif _caching_handler_response.embedding_all_elements_cache_hit is True: @@ -1956,8 +1963,7 @@ def client(original_function): end_time = datetime.datetime.now() if _is_streaming_request(kwargs=kwargs, call_type=call_type) or _is_converted_stream_result(result): - logging_obj.stream = True - logging_obj.model_call_details["stream"] = True + _mark_logging_as_stream(logging_obj) if "complete_response" in kwargs and kwargs["complete_response"] is True: chunks: Final = [] for idx, chunk in enumerate(result): diff --git a/tests/test_litellm/test_utils.py b/tests/test_litellm/test_utils.py index 7c33cf40bc8..8b83ca29dd7 100644 --- a/tests/test_litellm/test_utils.py +++ b/tests/test_litellm/test_utils.py @@ -20,6 +20,8 @@ from jsonschema import validate import litellm from litellm._internal_context import is_internal_call +from litellm.caching.caching import Cache +from litellm.caching.caching_handler import _PENDING_CACHE_WRITES from litellm.constants import DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT from litellm._logging import ( CorrelationContextFilter, @@ -5324,12 +5326,18 @@ class _SuccessKwargsCapture(CustomLogger): def __init__(self) -> None: super().__init__() self.success_kwargs: list[dict[str, object]] = [] + self.stream_event_responses: list[object] = [] async def async_log_success_event( self, kwargs: dict[str, object], response_obj: object, start_time: datetime, end_time: datetime ) -> None: self.success_kwargs.append(kwargs) + async def async_log_stream_event( + self, kwargs: dict[str, object], response_obj: object, start_time: datetime, end_time: datetime + ) -> None: + self.stream_event_responses.append(response_obj) + def _install_converted_stream_callbacks(monkeypatch: pytest.MonkeyPatch) -> _SuccessKwargsCapture: capture: Final = _SuccessKwargsCapture() @@ -5341,13 +5349,23 @@ def _install_converted_stream_callbacks(monkeypatch: pytest.MonkeyPatch) -> _Suc return capture -async def _wait_for_success_kwargs(capture: _SuccessKwargsCapture) -> dict[str, object]: +async def _wait_for_success_kwargs(capture: _SuccessKwargsCapture, count: int = 1) -> dict[str, object]: for _ in range(50): - if capture.success_kwargs: + if len(capture.success_kwargs) >= count and not _PENDING_CACHE_WRITES: break await asyncio.sleep(0.05) - (success_kwargs,) = capture.success_kwargs - return success_kwargs + await asyncio.sleep(0.2) + assert len(capture.success_kwargs) == count + return capture.success_kwargs[-1] + + +def _assert_cache_hit_logged_as_stream(capture: _SuccessKwargsCapture, success_kwargs: dict[str, object]) -> None: + standard_logging_object: Final = success_kwargs["standard_logging_object"] + assert isinstance(standard_logging_object, dict) + assert standard_logging_object["cache_hit"] is True + assert standard_logging_object["stream"] is True + assert success_kwargs["stream"] is True + assert capture.stream_event_responses == [] @pytest.mark.asyncio @@ -5424,6 +5442,86 @@ async def test_wrapper_async_logs_converted_responses_stream_with_standard_loggi assert success_kwargs["stream"] is True +@pytest.mark.asyncio +async def test_wrapper_async_replays_cached_converted_chat_stream_as_stream( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A cache hit for a converted stream must replay as a stream: the caller still iterates the + result even though the deployment hook set kwargs["stream"] to False.""" + capture: Final = _install_converted_stream_callbacks(monkeypatch) + monkeypatch.setattr(litellm, "cache", Cache(type="local")) + request: Final = { + "model": "gpt-5.6", + "messages": [{"role": "user", "content": "replay me from cache"}], + "stream": True, + "mock_response": "converted stream body", + "num_retries": 0, + } + + first: Final = await litellm.acompletion(**request) + first_chunks: Final = [chunk async for chunk in first] + assert "".join(chunk.choices[0].delta.content or "" for chunk in first_chunks) == "converted stream body" + await _wait_for_success_kwargs(capture) + + replay: Final = await litellm.acompletion(**request) + assert isinstance(replay, CustomStreamWrapper) + replay_chunks: Final = [chunk async for chunk in replay] + assert "".join(chunk.choices[0].delta.content or "" for chunk in replay_chunks) == "converted stream body" + + _assert_cache_hit_logged_as_stream(capture, await _wait_for_success_kwargs(capture, count=2)) + + +@pytest.mark.asyncio +@respx.mock +async def test_wrapper_async_replays_cached_converted_responses_stream_as_stream( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Responses surface of the cache-hit replay: the hit must come back as a streaming iterator.""" + from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator + + capture: Final = _install_converted_stream_callbacks(monkeypatch) + monkeypatch.setattr(litellm, "cache", Cache(type="local")) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + litellm.in_memory_llm_clients_cache.flush_cache() + route: Final = respx.post("https://api.openai.com/v1/responses").respond( + json={ + "id": "resp_cached_converted", + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gpt-5.6", + "output": [ + { + "type": "message", + "id": "msg_cached_converted", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "converted stream body", "annotations": []}], + } + ], + "usage": {"input_tokens": 3, "output_tokens": 4, "total_tokens": 7}, + } + ) + request: Final = { + "model": "openai/gpt-5.6", + "input": "replay me from cache", + "stream": True, + "api_key": "sk-test", + "num_retries": 0, + } + + first: Final = await litellm.aresponses(**request) + assert [event async for event in first][-1].type == "response.completed" + await _wait_for_success_kwargs(capture) + + replay: Final = await litellm.aresponses(**request) + assert isinstance(replay, BaseResponsesAPIStreamingIterator) + assert [event async for event in replay][-1].type == "response.completed" + assert route.call_count == 1 + + _assert_cache_hit_logged_as_stream(capture, await _wait_for_success_kwargs(capture, count=2)) + + def test_function_setup_failure_after_logging_construction_restores_context(monkeypatch): """If function_setup() constructs Logging() (which already mutated trace_id_var/session_id_var in __init__) but then raises before returning, From 621db91d906ab454d757fea6b9fb34195ce5e3f1 Mon Sep 17 00:00:00 2001 From: yucheng Date: Tue, 15 Sep 2026 02:12:49 +0000 Subject: [PATCH 3/6] fix(caching): defer cache-hit callbacks by replayed result type, not request flags A converted-stream request whose cache entry is a plain (non-stream) object is replayed as that plain object, so nothing later fires the success callbacks. Decide deferral from the replayed result's type instead of the request kwargs. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/caching_handler.py | 20 +++++--- tests/local_testing/test_caching_handler.py | 24 +++------- .../caching/test_caching_handler.py | 47 +++++++++++++++++++ 3 files changed, 68 insertions(+), 23 deletions(-) diff --git a/litellm/caching/caching_handler.py b/litellm/caching/caching_handler.py index 901f2ffbad2..1ddc0559547 100644 --- a/litellm/caching/caching_handler.py +++ b/litellm/caching/caching_handler.py @@ -114,17 +114,25 @@ def _stream_replay_requested(kwargs: Mapping[str, object]) -> bool: return kwargs.get("stream", False) is True or converted_stream_requested(kwargs) -def _should_defer_streaming_cache_hit_callbacks(*, kwargs: dict[str, object]) -> bool: +def _should_defer_streaming_cache_hit_callbacks(*, cached_result: object) -> bool: """ - When stream=True, do not run success callbacks at cache-hit time. + When the cache hit is replayed as a stream, do not run success callbacks at cache-hit time. Cached chat/text completion replay uses CustomStreamWrapper; cached Responses replay uses CachedResponsesAPIStreamingIterator; cached Anthropic Messages replay uses CachedAnthropicMessagesStreamIterator. All invoke logging success handlers when the stream finishes; firing them here too would double-count - spend and callback records. + spend and callback records. A plain (non-stream) replay logs here, since nothing + else will. """ - return _stream_replay_requested(kwargs) + from litellm.llms.anthropic.experimental_pass_through.messages.response_cache import ( + CachedAnthropicMessagesStreamIterator, + ) + from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator + + return isinstance( + cached_result, (CustomStreamWrapper, BaseResponsesAPIStreamingIterator, CachedAnthropicMessagesStreamIterator) + ) def _prompt_tokens_details_as_mapping(details: "PromptTokensDetailsWrapper") -> Mapping[str, object]: @@ -274,7 +282,7 @@ class LLMCachingHandler: custom_llm_provider=kwargs.get("custom_llm_provider", None), args=args, ) - if not _should_defer_streaming_cache_hit_callbacks(kwargs=kwargs): + if not _should_defer_streaming_cache_hit_callbacks(cached_result=cached_result): # LOG SUCCESS self._async_log_cache_hit_on_callbacks( logging_obj=logging_obj, @@ -390,7 +398,7 @@ class LLMCachingHandler: is_async=False, ) - if not _should_defer_streaming_cache_hit_callbacks(kwargs=kwargs): + if not _should_defer_streaming_cache_hit_callbacks(cached_result=cached_result): logging_obj.handle_sync_success_callbacks_for_async_calls( result=cached_result, start_time=start_time, diff --git a/tests/local_testing/test_caching_handler.py b/tests/local_testing/test_caching_handler.py index f17a058b3fe..a181ef89fe0 100644 --- a/tests/local_testing/test_caching_handler.py +++ b/tests/local_testing/test_caching_handler.py @@ -927,24 +927,14 @@ def test_sync_get_cache_defers_streaming_completion_hit_callbacks(): def test_should_defer_streaming_cache_hit_callbacks_for_any_streaming_request(): - assert ( - _should_defer_streaming_cache_hit_callbacks( - kwargs={"stream": True}, - ) - is True - ) - assert ( - _should_defer_streaming_cache_hit_callbacks( - kwargs={"stream": False}, - ) - is False - ) - assert ( - _should_defer_streaming_cache_hit_callbacks( - kwargs={}, - ) - is False + logging_obj = MagicMock() + logging_obj.model_call_details = {} + stream_replay = CustomStreamWrapper( + completion_stream=iter(()), model="gpt-4o", logging_obj=logging_obj ) + assert _should_defer_streaming_cache_hit_callbacks(cached_result=stream_replay) is True + assert _should_defer_streaming_cache_hit_callbacks(cached_result=ModelResponse()) is False + assert _should_defer_streaming_cache_hit_callbacks(cached_result={"id": "msg_1"}) is False @pytest.mark.asyncio diff --git a/tests/test_litellm/caching/test_caching_handler.py b/tests/test_litellm/caching/test_caching_handler.py index 071b99850f6..12f141353bb 100644 --- a/tests/test_litellm/caching/test_caching_handler.py +++ b/tests/test_litellm/caching/test_caching_handler.py @@ -693,3 +693,50 @@ async def test_cache_hit_records_the_looked_up_key_as_the_preset_cache_key(monke assert handler.preset_cache_key is not None assert logging_obj.litellm_params["preset_cache_key"] == handler.preset_cache_key assert hit.cached_result._hidden_params["cache_key"] == handler.preset_cache_key + + +@pytest.mark.asyncio +async def test_converted_stream_cache_hit_replayed_as_plain_object_logs_at_hit_time(monkeypatch): + """A converted-stream Anthropic Messages request that hits a non-stream cache entry gets a plain dict back, + so the success callbacks must fire now; nothing else will fire them.""" + import litellm + from litellm.caching.caching import Cache + from litellm.types.utils import CallTypes + + async def aanthropic_messages(**kwargs): + return None + + monkeypatch.setattr(litellm, "cache", Cache(type="local")) + kwargs = { + "model": "claude-sonnet-5", + "messages": [{"role": "user", "content": "hello"}], + "max_tokens": 16, + "caching": True, + "stream": False, + "_websearch_interception_converted_stream": True, + } + cached_message = { + "id": "msg_1", + "type": "message", + "role": "assistant", + "content": [{"type": "text", "text": "hi"}], + } + await litellm.cache.async_add_cache(cached_message, **kwargs) + handler = LLMCachingHandler(original_function=aanthropic_messages, request_kwargs=kwargs, start_time=datetime.now()) + logging_obj = _build_logging_obj(CallTypes.aanthropic_messages.value, stream=False) + logging_obj.async_success_handler = AsyncMock() + logging_obj.handle_sync_success_callbacks_for_async_calls = MagicMock() + + hit = await handler._async_get_cache( + model="claude-sonnet-5", + original_function=aanthropic_messages, + logging_obj=logging_obj, + start_time=datetime.now(), + call_type=CallTypes.aanthropic_messages.value, + kwargs=kwargs, + args=(), + ) + + assert hit is not None and hit.cached_result == cached_message + logging_obj.handle_sync_success_callbacks_for_async_calls.assert_called_once() + assert logging_obj.handle_sync_success_callbacks_for_async_calls.call_args.kwargs["cache_hit"] is True From ce45d6a09d5bf4dde0f8b7ce11c463d7c73ddd2d Mon Sep 17 00:00:00 2001 From: yucheng Date: Tue, 15 Sep 2026 02:28:03 +0000 Subject: [PATCH 4/6] style: drop explanatory docstrings from converted-stream helpers and tests Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/caching_handler.py | 2 -- litellm/utils.py | 2 -- tests/test_litellm/caching/test_caching_handler.py | 2 -- tests/test_litellm/test_utils.py | 9 --------- 4 files changed, 15 deletions(-) diff --git a/litellm/caching/caching_handler.py b/litellm/caching/caching_handler.py index 1ddc0559547..a0ddbdb37ec 100644 --- a/litellm/caching/caching_handler.py +++ b/litellm/caching/caching_handler.py @@ -109,8 +109,6 @@ def _is_chat_completion_cached_dict(cached_result: dict) -> bool: def _stream_replay_requested(kwargs: Mapping[str, object]) -> bool: - """True when the caller must receive a stream, including when a deployment hook downgraded - `kwargs["stream"]` to False for the provider call.""" return kwargs.get("stream", False) is True or converted_stream_requested(kwargs) diff --git a/litellm/utils.py b/litellm/utils.py index 8dc6576cc97..35ad48dd062 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -847,8 +847,6 @@ def _is_streaming_response_for_correlation(result: object) -> bool: def _is_converted_stream_result(result: object) -> bool: - """True if `result` is a lazy stream wrapper the caller must iterate, even when a deployment - hook downgraded `kwargs["stream"]` to False for the provider call.""" from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator diff --git a/tests/test_litellm/caching/test_caching_handler.py b/tests/test_litellm/caching/test_caching_handler.py index 12f141353bb..dd826d80208 100644 --- a/tests/test_litellm/caching/test_caching_handler.py +++ b/tests/test_litellm/caching/test_caching_handler.py @@ -697,8 +697,6 @@ async def test_cache_hit_records_the_looked_up_key_as_the_preset_cache_key(monke @pytest.mark.asyncio async def test_converted_stream_cache_hit_replayed_as_plain_object_logs_at_hit_time(monkeypatch): - """A converted-stream Anthropic Messages request that hits a non-stream cache entry gets a plain dict back, - so the success callbacks must fire now; nothing else will fire them.""" import litellm from litellm.caching.caching import Cache from litellm.types.utils import CallTypes diff --git a/tests/test_litellm/test_utils.py b/tests/test_litellm/test_utils.py index 8b83ca29dd7..02ea06ccf80 100644 --- a/tests/test_litellm/test_utils.py +++ b/tests/test_litellm/test_utils.py @@ -5312,8 +5312,6 @@ async def test_wrapper_async_restores_originating_task_context_after_success(mon class _ConvertStreamDeploymentHook(CustomLogger): - """Headroom-style interception: downgrade stream=True to a non-streaming provider call.""" - async def async_pre_call_deployment_hook( self, kwargs: dict[str, object], call_type: CallTypes | None ) -> dict[str, object] | None: @@ -5372,8 +5370,6 @@ def _assert_cache_hit_logged_as_stream(capture: _SuccessKwargsCapture, success_k async def test_wrapper_async_logs_converted_chat_stream_with_standard_logging_object( monkeypatch: pytest.MonkeyPatch, ) -> None: - """Regression LIT-7729: the fake CustomStreamWrapper hit the non-streaming success path, which - built no standard_logging_object and deduped the wrapper's own end-of-stream dispatch.""" capture: Final = _install_converted_stream_callbacks(monkeypatch) response: Final = await litellm.acompletion( @@ -5400,8 +5396,6 @@ async def test_wrapper_async_logs_converted_chat_stream_with_standard_logging_ob async def test_wrapper_async_logs_converted_responses_stream_with_standard_logging_object( monkeypatch: pytest.MonkeyPatch, ) -> None: - """Regression LIT-7729, Responses surface: the fake MockResponsesAPIStreamingIterator took the - same non-streaming success path and lost its standard_logging_object.""" from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator capture: Final = _install_converted_stream_callbacks(monkeypatch) @@ -5446,8 +5440,6 @@ async def test_wrapper_async_logs_converted_responses_stream_with_standard_loggi async def test_wrapper_async_replays_cached_converted_chat_stream_as_stream( monkeypatch: pytest.MonkeyPatch, ) -> None: - """A cache hit for a converted stream must replay as a stream: the caller still iterates the - result even though the deployment hook set kwargs["stream"] to False.""" capture: Final = _install_converted_stream_callbacks(monkeypatch) monkeypatch.setattr(litellm, "cache", Cache(type="local")) request: Final = { @@ -5476,7 +5468,6 @@ async def test_wrapper_async_replays_cached_converted_chat_stream_as_stream( async def test_wrapper_async_replays_cached_converted_responses_stream_as_stream( monkeypatch: pytest.MonkeyPatch, ) -> None: - """Responses surface of the cache-hit replay: the hit must come back as a streaming iterator.""" from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator capture: Final = _install_converted_stream_callbacks(monkeypatch) From 496c2a55133fa8375c4955d30cb90a4e30804f4a Mon Sep 17 00:00:00 2001 From: yucheng Date: Tue, 15 Sep 2026 08:35:13 +0000 Subject: [PATCH 5/6] fix(caching): replay agentic loop follow-up cache hits as plain objects Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/caching_handler.py | 4 +- .../caching/test_caching_handler.py | 42 +++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/litellm/caching/caching_handler.py b/litellm/caching/caching_handler.py index a0ddbdb37ec..50426ea89ea 100644 --- a/litellm/caching/caching_handler.py +++ b/litellm/caching/caching_handler.py @@ -109,7 +109,9 @@ def _is_chat_completion_cached_dict(cached_result: dict) -> bool: def _stream_replay_requested(kwargs: Mapping[str, object]) -> bool: - return kwargs.get("stream", False) is True or converted_stream_requested(kwargs) + if kwargs.get("stream", False) is True: + return True + return converted_stream_requested(kwargs) and not kwargs.get("_agentic_loop_depth") def _should_defer_streaming_cache_hit_callbacks(*, cached_result: object) -> bool: diff --git a/tests/test_litellm/caching/test_caching_handler.py b/tests/test_litellm/caching/test_caching_handler.py index dd826d80208..39018dca41d 100644 --- a/tests/test_litellm/caching/test_caching_handler.py +++ b/tests/test_litellm/caching/test_caching_handler.py @@ -738,3 +738,45 @@ async def test_converted_stream_cache_hit_replayed_as_plain_object_logs_at_hit_t assert hit is not None and hit.cached_result == cached_message logging_obj.handle_sync_success_callbacks_for_async_calls.assert_called_once() assert logging_obj.handle_sync_success_callbacks_for_async_calls.call_args.kwargs["cache_hit"] is True + + +@pytest.mark.asyncio +async def test_agentic_loop_followup_cache_hit_with_converted_stream_marker_replays_as_plain_object(monkeypatch): + import litellm + from litellm.caching.caching import Cache + from litellm.types.utils import CallTypes + + async def acompletion(**kwargs): + return None + + monkeypatch.setattr(litellm, "cache", Cache(type="local")) + kwargs = { + "model": "gpt-5.6", + "messages": [{"role": "user", "content": "run the code"}], + "caching": True, + "stream": False, + "_code_interpreter_interception_converted_stream": True, + "_agentic_loop_depth": 1, + } + await litellm.cache.async_add_cache( + litellm.ModelResponse(choices=[{"message": {"role": "assistant", "content": "done"}}]), **kwargs + ) + handler = LLMCachingHandler(original_function=acompletion, request_kwargs=kwargs, start_time=datetime.now()) + logging_obj = _build_logging_obj(CallTypes.acompletion.value, stream=False) + logging_obj.async_success_handler = AsyncMock() + logging_obj.handle_sync_success_callbacks_for_async_calls = MagicMock() + + hit = await handler._async_get_cache( + model="gpt-5.6", + original_function=acompletion, + logging_obj=logging_obj, + start_time=datetime.now(), + call_type=CallTypes.acompletion.value, + kwargs=kwargs, + args=(), + ) + + assert hit is not None and isinstance(hit.cached_result, litellm.ModelResponse) + assert hit.cached_result.choices[0].message.content == "done" + logging_obj.handle_sync_success_callbacks_for_async_calls.assert_called_once() + assert logging_obj.handle_sync_success_callbacks_for_async_calls.call_args.kwargs["cache_hit"] is True From 5aa5c092d54592bd8cbe41b12a073e7f0eafc0a1 Mon Sep 17 00:00:00 2001 From: yucheng Date: Tue, 15 Sep 2026 08:42:19 +0000 Subject: [PATCH 6/6] refactor(utils): set converted-stream logging flags inline instead of mutating a helper parameter Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/utils.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/litellm/utils.py b/litellm/utils.py index 35ad48dd062..d2c7d8e4b43 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -853,11 +853,6 @@ def _is_converted_stream_result(result: object) -> bool: return isinstance(result, (CustomStreamWrapper, BaseResponsesAPIStreamingIterator)) -def _mark_logging_as_stream(logging_obj: LiteLLMLoggingObject) -> None: - logging_obj.stream = True - logging_obj.model_call_details["stream"] = True - - # Runs once per call to check if the user wants to send their data anywhere - PostHog/Sentry/Slack/etc. def function_setup( original_function: str, @@ -1902,7 +1897,8 @@ def client(original_function): and _caching_handler_response.final_embedding_cached_response is None ): if _is_converted_stream_result(_caching_handler_response.cached_result): - _mark_logging_as_stream(logging_obj) + logging_obj.stream = True + logging_obj.model_call_details["stream"] = True return _caching_handler_response.cached_result elif _caching_handler_response.embedding_all_elements_cache_hit is True: @@ -1961,7 +1957,8 @@ def client(original_function): end_time = datetime.datetime.now() if _is_streaming_request(kwargs=kwargs, call_type=call_type) or _is_converted_stream_result(result): - _mark_logging_as_stream(logging_obj) + logging_obj.stream = True + logging_obj.model_call_details["stream"] = True if "complete_response" in kwargs and kwargs["complete_response"] is True: chunks: Final = [] for idx, chunk in enumerate(result):