fix(router): honor Codex reminders and map classifier failures

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

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,

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"
     )

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",
+        [
+            "<environment_context>cwd=/repo</environment_context>",
+            "<user_instructions>Keep answers concise</user_instructions>",
+        ],
+    )
+    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")])