diff --git a/litellm/responses/main.py b/litellm/responses/main.py
index e88cd618a8b..a68cd02e61b 100644
--- a/litellm/responses/main.py
+++ b/litellm/responses/main.py
@@ -4,10 +4,11 @@ from collections.abc import Coroutine, Generator, Iterable, Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from functools import partial
-from typing import TYPE_CHECKING, Any, Final, Literal, Optional, cast
+from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, Optional, TypeAlias, cast
import httpx
from pydantic import BaseModel
+from typing_extensions import assert_never
import litellm
from litellm._logging import verbose_logger
@@ -407,6 +408,37 @@ def _bridges_to_chat_completions(
return responses_api_provider_config is None or use_chat_completions_api is True
+_ResponsesCompatibilityFailure: TypeAlias = Literal["encrypted_task_unsupported"]
+
+
+def _encrypted_task_support_failure(
+ responses_api_provider_config: BaseResponsesAPIConfig | None, use_chat_completions_api: bool
+) -> _ResponsesCompatibilityFailure | None:
+ if (
+ responses_api_provider_config is None
+ or _bridges_to_chat_completions(responses_api_provider_config, use_chat_completions_api)
+ or not responses_api_provider_config.supports_encrypted_agent_messages()
+ ):
+ return "encrypted_task_unsupported"
+ return None
+
+
+def _raise_responses_compatibility_failure(
+ failure: _ResponsesCompatibilityFailure, model: str, custom_llm_provider: str | None
+) -> NoReturn:
+ match failure:
+ case "encrypted_task_unsupported":
+ raise litellm.exception_type(
+ model=model,
+ custom_llm_provider=custom_llm_provider,
+ original_exception=ValueError(
+ "Encrypted task classification requires a compatible native Responses deployment"
+ ),
+ )
+ case _:
+ assert_never(failure)
+
+
def _deployment_passes_through_responses(model_info: object) -> bool:
"""Whether ``model_info.supported_endpoints`` opts the deployment into native ``{api_base}/responses``."""
if not isinstance(model_info, dict):
@@ -1187,12 +1219,16 @@ def responses(
model, custom_llm_provider, deployment_model_info
)
- if require_encrypted_task_support and (
- _bridges_to_chat_completions(responses_api_provider_config, use_chat_completions_api)
- or responses_api_provider_config is None
- or not responses_api_provider_config.supports_encrypted_agent_messages()
+ if (
+ require_encrypted_task_support
+ and (
+ compatibility_failure := _encrypted_task_support_failure(
+ responses_api_provider_config, use_chat_completions_api
+ )
+ )
+ is not None
):
- raise ValueError("Encrypted task classification requires a compatible native Responses deployment")
+ _raise_responses_compatibility_failure(compatibility_failure, model, custom_llm_provider)
local_vars.update(kwargs)
# Map reasoning_effort (from litellm_params/proxy config) to reasoning when not set
diff --git a/litellm/router_strategy/complexity_router/complexity_router.py b/litellm/router_strategy/complexity_router/complexity_router.py
index 6df74b26f60..214872c17c9 100644
--- a/litellm/router_strategy/complexity_router/complexity_router.py
+++ b/litellm/router_strategy/complexity_router/complexity_router.py
@@ -1690,7 +1690,7 @@ class ComplexityRouter(CustomLogger):
if self.config.classifier_type == "custom":
return await self._classify_with_plugin(prompt, system_prompt, request_kwargs, raw_messages)
if self.config.classifier_type in ("heuristic_first", "hybrid") and _encrypted_classifier_task(
- request_kwargs, self._reminder_markers
+ request_kwargs, self._reminder_markers_for_request(request_kwargs or EMPTY_MAPPING)
):
return await self._llm_classifier_outcome(prompt, system_prompt, request_kwargs, messages)
if self.config.classifier_type == "heuristic_first" and self.config.classifier_llm_config is not None:
@@ -2028,7 +2028,7 @@ class ComplexityRouter(CustomLogger):
> 1
)
- encrypted_task: Final = _encrypted_classifier_task(request_kwargs, self._reminder_markers)
+ encrypted_task: Final = _encrypted_classifier_task(request_kwargs, marker_pairs)
user_payload: Final = self._build_classifier_user_payload(
prompt="The delegated task in the following agent_message." if encrypted_task is not None else prompt,
system_prompt=system_prompt,
diff --git a/tests/test_litellm/responses/test_responses_api_bridge_flag.py b/tests/test_litellm/responses/test_responses_api_bridge_flag.py
index 57aa2a6baa2..ed44a9f4545 100644
--- a/tests/test_litellm/responses/test_responses_api_bridge_flag.py
+++ b/tests/test_litellm/responses/test_responses_api_bridge_flag.py
@@ -7,10 +7,14 @@ calls so routed requests do not hit a custom api_base /v1/responses endpoint.
"""
from importlib import import_module
+from typing import Final
from unittest.mock import MagicMock, patch
+import httpx
+import pytest
import litellm
+from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
from litellm.types.utils import Choices, Message, ModelResponse, Usage
@@ -18,6 +22,26 @@ from litellm.types.utils import Choices, Message, ModelResponse, Usage
class TestUseResponsesApiBridgeFlag:
"""Test that bridge opt-in forces the chat completions path."""
+ @pytest.mark.parametrize("model", ["openai/chat_completions/gpt-6-astra", "xai/test-classifier"])
+ def test_encrypted_classifier_rejection_preserves_public_error(self, model: str) -> None:
+ respond: Final = MagicMock(side_effect=AssertionError("Incompatible classifier sent an upstream request"))
+ with httpx.Client(transport=httpx.MockTransport(respond)) as client:
+ with pytest.raises(
+ litellm.APIConnectionError,
+ match="Encrypted task classification requires a compatible native Responses deployment",
+ ) as error:
+ litellm.responses(
+ model=model,
+ input="Delegated task",
+ api_key="test-key",
+ api_base="https://classifier.test/v1",
+ client=HTTPHandler(client=client),
+ _require_encrypted_task_support=True,
+ num_retries=0,
+ )
+ assert error.value.status_code == 500
+ respond.assert_not_called()
+
@patch.object(
import_module("litellm.responses.main").litellm_completion_transformation_handler, "response_api_handler"
)
diff --git a/tests/test_litellm/router_strategy/test_complexity_router.py b/tests/test_litellm/router_strategy/test_complexity_router.py
index 2ea8a8b53fe..cd8f05d1e41 100644
--- a/tests/test_litellm/router_strategy/test_complexity_router.py
+++ b/tests/test_litellm/router_strategy/test_complexity_router.py
@@ -2577,6 +2577,40 @@ async def native_classifier_http() -> AsyncIterator[tuple[AsyncHTTPHandler, Magi
class TestEncryptedTaskClassifier:
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("classifier_type", ["llm", "heuristic_first", "hybrid"])
+ @pytest.mark.parametrize("codex", [True, False])
+ @pytest.mark.parametrize(
+ "reminder",
+ [
+ "cwd=/repo",
+ "Keep answers concise",
+ ],
+ )
+ async def test_encrypted_task_detection_uses_request_reminder_markers(
+ self, classifier_type: str, codex: bool, reminder: str
+ ):
+ router, dependency = _native_classifier_router(classifier_type=classifier_type)
+ task: Final = _encrypted_agent_task()
+ request: Final = {
+ "input": [task, {"role": "user", "content": reminder}],
+ "metadata": {"user_agent": "codex-tui" if codex else "curl/8.7.1"},
+ }
+ original: Final = deepcopy(request)
+
+ result: Final = await router.async_pre_routing_hook(model="encrypted-router", request_kwargs=request)
+
+ assert request == original
+ assert result.model == ("deep-model" if codex else "cheap-model")
+ if codex:
+ assert result.routing_decision["cause"] == "llm_classifier"
+ assert result.routing_decision["tier"] == "REASONING"
+ dependency.aresponses.assert_awaited_once()
+ assert dependency.aresponses.call_args.kwargs["input"][-1] == task
+ dependency.acompletion.assert_not_called()
+ else:
+ dependency.aresponses.assert_not_called()
+
@pytest.mark.asyncio
@pytest.mark.parametrize("classifier_type", ["llm", "heuristic_first", "hybrid"])
@pytest.mark.parametrize("tier,model", [("SIMPLE", "cheap-model"), ("REASONING", "deep-model")])