From f4ebcef0a17c2952ab90100fdba702d863c728f9 Mon Sep 17 00:00:00 2001 From: moe-berri Date: Thu, 10 Sep 2026 12:32:09 -0700 Subject: [PATCH 1/3] fix(router): classify encrypted delegated tasks with native Responses --- .../transformation.py | 3 + .../complexity_router/README.md | 10 + .../complexity_router/complexity_router.py | 119 ++++++++++-- .../router_strategy/test_complexity_router.py | 178 ++++++++++++++++++ 4 files changed, 293 insertions(+), 17 deletions(-) diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index c3c18e3d009..5a6debc4af5 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -1201,6 +1201,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): # Cast to Any to match the expected union type for tools list items tools.append(cast(Any, web_search_tool)) + def transform_response_format_to_text_format(self, response_format: object) -> "ResponseText | None": + return self._transform_response_format_to_text_format(response_format) + def _transform_response_format_to_text_format(self, response_format: object) -> "ResponseText | None": """ Transform Chat Completion response_format parameter to Responses API text.format parameter. diff --git a/litellm/router_strategy/complexity_router/README.md b/litellm/router_strategy/complexity_router/README.md index d605b43e42a..79119008aa2 100644 --- a/litellm/router_strategy/complexity_router/README.md +++ b/litellm/router_strategy/complexity_router/README.md @@ -361,6 +361,16 @@ model_list: keep the classifier deployment or provider default, or set a supported value such as `none` or `low` to override that call. +When the current ask is a Responses API `agent_message` containing `encrypted_content`, LLM +classification preserves the encrypted task and uses native Responses. This also bypasses the +local scoring shortcut in `heuristic_first` and `hybrid` modes. The configured classifier must use +a native OpenAI or Azure OpenAI Responses deployment with access to the encrypted content. The +provider handles the encrypted task, and the classifier still chooses the tier dynamically + +Unsupported classifier deployments and provider decryption errors use the existing +`classifier_fallback` policy. No fixed tier is introduced for encrypted tasks. Plaintext asks and +requests carrying only historical encrypted reasoning retain the existing classifier path + Classifier calls have a one-attempt hard deadline. After a timeout, the router opens a process-local circuit for that classifier and sends every session through `classifier_fallback` for `classifier_llm_config.circuit_breaker_cooldown_seconds` (30 seconds by default). When the cooldown diff --git a/litellm/router_strategy/complexity_router/complexity_router.py b/litellm/router_strategy/complexity_router/complexity_router.py index faafcea404a..3babe5b96ed 100644 --- a/litellm/router_strategy/complexity_router/complexity_router.py +++ b/litellm/router_strategy/complexity_router/complexity_router.py @@ -25,7 +25,7 @@ from threading import Lock from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal, NamedTuple, cast -from pydantic import BaseModel, create_model +from pydantic import BaseModel, TypeAdapter, create_model from litellm._logging import verbose_router_logger from litellm.constants import ( @@ -56,6 +56,7 @@ from litellm.types.llms.openai import ( AllMessageValues, ChatCompletionImageObject, ChatCompletionTextObject, + ResponsesAPIResponse, ) from litellm.types.utils import ( AUTOROUTER_CLASSIFIER_CALL_ORIGIN, @@ -364,7 +365,7 @@ def _parent_session_kwargs(request_kwargs: Mapping[str, Any] | None) -> Mapping[ return {k: kwargs[k] for k in ("litellm_session_id", "litellm_trace_id") if kwargs.get(k) is not None} -def _response_cost_or_none(response: ModelResponse) -> float | None: +def _response_cost_or_none(response: ModelResponse | ResponsesAPIResponse) -> float | None: hidden_params: Final = response._hidden_params if not isinstance(hidden_params, dict): return None @@ -486,6 +487,36 @@ def _human_text(content: object, marker_pairs: tuple[tuple[str, str], ...] = _DE return _strip_reminder_blocks(_message_text(content), marker_pairs) +def _encrypted_classifier_task( + request_kwargs: Mapping[str, object] | None, + marker_pairs: tuple[tuple[str, str], ...], +) -> dict[str, object] | None: + from litellm.litellm_core_utils.prompt_templates.factory import resolve_structured_messages + + raw_input: Final = (request_kwargs or EMPTY_MAPPING).get("input") + if not isinstance(raw_input, list) or (request_kwargs or EMPTY_MAPPING).get("messages"): + return None + items: Final = TypeAdapter(tuple[dict[str, object], ...]).validate_python(raw_input) + current: Final = next( + ( + item + for item in reversed(items) + if (messages := resolve_structured_messages(messages=None, request_kwargs={"input": [item]})) + and any(_iter_human_asks_newest_first(messages, marker_pairs)) + ), + None, + ) + if current is None or current.get("type") != "agent_message" or not isinstance(current.get("content"), list): + return None + parts: Final = TypeAdapter(tuple[dict[str, object], ...]).validate_python(current["content"]) + if not any(part.get("type") == "encrypted_content" and part.get("encrypted_content") for part in parts): + return None + return { + **current, + "content": [part for part in parts if part.get("type") in ("input_text", "encrypted_content")], + } + + def _iter_human_asks_newest_first( messages: Sequence[Mapping[str, object]], marker_pairs: tuple[tuple[str, str], ...] = _DEFAULT_REMINDER_MARKERS, @@ -1630,6 +1661,10 @@ class ComplexityRouter(CustomLogger): return self._classify_with_heuristic_v2(prompt) 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 + ): + 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: return await self._classify_heuristic_first(prompt, system_prompt, request_kwargs, messages) if self.config.classifier_type == "hybrid" and self.config.classifier_llm_config is not None: @@ -1970,8 +2005,9 @@ class ComplexityRouter(CustomLogger): > 1 ) + encrypted_task: Final = _encrypted_classifier_task(request_kwargs, self._reminder_markers) user_payload: Final = self._build_classifier_user_payload( - prompt=prompt, + prompt="The delegated task in the following agent_message." if encrypted_task is not None else prompt, system_prompt=system_prompt, prior_turns=prior_turns, messages=messages, @@ -2004,34 +2040,37 @@ class ComplexityRouter(CustomLogger): if llm_config.reasoning_effort is not None: classifier_call_params = MappingProxyType({"reasoning_effort": llm_config.reasoning_effort}) - proxy_server_request: Final = { - "body": { - "model": llm_config.model, - "messages": messages_for_call, - "response_format": response_format, - **classifier_call_params, - } - } + payload: Final = ( + self._native_classifier_payload(llm_config.model, messages_for_call, response_format, encrypted_task) + if encrypted_task is not None + else {"messages": messages_for_call, "response_format": response_format, **classifier_call_params} + ) + proxy_server_request: Final = {"body": {"model": llm_config.model, **payload}} + classify: Final = ( + self.litellm_router_instance.aresponses + if encrypted_task is not None + else self.litellm_router_instance.acompletion + ) classifier_timeout_s: Final[float] = llm_config.timeout_ms / 1000 - response: Final[ModelResponse] = await asyncio.wait_for( - self.litellm_router_instance.acompletion( + response: Final[ModelResponse | ResponsesAPIResponse] = await asyncio.wait_for( + classify( model=llm_config.model, - messages=messages_for_call, stream=False, - response_format=response_format, timeout=classifier_timeout_s, num_retries=0, disable_fallbacks=True, metadata=metadata, proxy_server_request=proxy_server_request, turn_off_message_logging=turn_off_message_logging, - **classifier_call_params, + **payload, **_parent_session_kwargs(request_kwargs), ), timeout=classifier_timeout_s, ) - content: Final = response.choices[0].message.content + content: Final = ( + response.output_text if isinstance(response, ResponsesAPIResponse) else response.choices[0].message.content + ) if not content: raise ValueError("LLM classifier returned empty content") raw_tier: Final = _LabeledTierClassification.model_validate_json(content).tier @@ -2040,6 +2079,52 @@ class ComplexityRouter(CustomLogger): raise ValueError(f"LLM classifier returned an unrecognized tier: {raw_tier!r}") return tier, _response_cost_or_none(response) + def _native_classifier_payload( + self, + model: str, + messages: list[AllMessageValues], # mutable-ok: existing transformation accepts the SDK message list + response_format: Mapping[str, object], + encrypted_task: Mapping[str, object], + ) -> Mapping[str, object]: + from litellm.completion_extras.litellm_responses_transformation.transformation import ( + LiteLLMResponsesTransformationHandler, + ) + from litellm.litellm_core_utils.get_llm_provider_logic import declared_authenticating_provider, get_llm_provider + from litellm.types.router import LiteLLM_Params + + deployments: Final = self._group_deployments(model) + if not deployments: + raise ValueError("Encrypted task classification requires a native OpenAI Responses classifier deployment") + for params in (LiteLLM_Params.model_validate(deployment.get("litellm_params")) for deployment in deployments): + if declared_authenticating_provider(params.model, params.custom_llm_provider): + raise ValueError( + "Encrypted task classification requires a native OpenAI Responses classifier deployment" + ) + _, provider, _, _ = get_llm_provider(model=params.model, litellm_params=params) + if ( + provider not in ("openai", "azure") + or params.use_chat_completions_api + or params.model.startswith("openai/chat_completions/") + ): + raise ValueError( + "Encrypted task classification requires a native OpenAI Responses classifier deployment" + ) + transformation: Final = LiteLLMResponsesTransformationHandler() + input_items, instructions = transformation.convert_chat_completion_messages_to_responses_api(messages) + llm_config: Final = self.config.classifier_llm_config + reasoning: Final = ( + {"reasoning": {"effort": llm_config.reasoning_effort}} + if llm_config is not None and llm_config.reasoning_effort is not None + else {} + ) + return { + "input": [*input_items, encrypted_task], + "instructions": instructions, + "text": transformation.transform_response_format_to_text_format(dict(response_format)), + "store": False, + **reasoning, + } + @staticmethod def _build_classifier_user_payload( prompt: str, diff --git a/tests/test_litellm/router_strategy/test_complexity_router.py b/tests/test_litellm/router_strategy/test_complexity_router.py index 51103297c58..87438ef5c69 100644 --- a/tests/test_litellm/router_strategy/test_complexity_router.py +++ b/tests/test_litellm/router_strategy/test_complexity_router.py @@ -5,6 +5,8 @@ Tests the rule-based complexity scoring and tier assignment logic. """ import asyncio +import copy +import json import logging import sys import time @@ -66,6 +68,7 @@ from litellm.types.router import ( LiteLLM_Params, TaggedPreRoutingStrategy, ) +from litellm.types.llms.openai import ResponsesAPIResponse requires_semantic_router = pytest.mark.skipif( @@ -2482,6 +2485,181 @@ class TestTierLabels: assert set(config.tier_boundaries) == {"simple_medium", "medium_complex", "complex_reasoning"} +def _encrypted_agent_task() -> dict[str, object]: + return { + "type": "agent_message", + "author": "/root", + "recipient": "/root/child", + "content": [ + {"type": "input_text", "text": "Message Type: NEW_TASK\nTask name: /root/child\nPayload:\nHello"}, + {"type": "encrypted_content", "encrypted_content": "opaque-provider-task"}, + ], + } + + +def _native_classifier_response(content: str) -> ResponsesAPIResponse: + response: Final = ResponsesAPIResponse( + id="resp_classifier", + created_at=0, + status="completed", + output=[{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": content}]}], + ) + response._hidden_params = {"response_cost": 0.0001} + return response + + +def _native_classifier_router( + output: str = '{"tier":"REASONING"}', + classifier_type: str = "llm", + deployment_model: str = "openai/gpt-6-astra", + failure: Exception | None = None, +) -> tuple[ComplexityRouter, MagicMock]: + dependency: Final = MagicMock( + aresponses=AsyncMock(return_value=_native_classifier_response(output), side_effect=failure), + acompletion=AsyncMock(return_value=_llm_response('{"tier":"SIMPLE"}')), + get_model_list=MagicMock(return_value=[{"litellm_params": {"model": deployment_model}}]), + ) + return ( + ComplexityRouter( + model_name="encrypted-router", + litellm_router_instance=dependency, + complexity_router_config={ + "tiers": {"SIMPLE": "cheap-model", "REASONING": "deep-model"}, + "classifier_type": classifier_type, + "classifier_llm_config": {"model": "classifier", "timeout_ms": 100, "reasoning_effort": "low"}, + "heuristic_first_max_tier": "SIMPLE" if classifier_type == "heuristic_first" else None, + "hybrid_boundary_margin": 0.01 if classifier_type == "hybrid" else None, + "classifier_fallback": "default_model", + "default_model": "deep-model", + "session_affinity": False, + "deployment_affinity": False, + }, + ), + dependency, + ) + + +class TestEncryptedTaskClassifier: + @pytest.mark.asyncio + @pytest.mark.parametrize("classifier_type", ["llm", "heuristic_first", "hybrid"]) + @pytest.mark.parametrize("tier,model", [("SIMPLE", "cheap-model"), ("REASONING", "deep-model")]) + async def test_encrypted_task_routes_by_native_verdict(self, classifier_type: str, tier: str, model: str): + router, dependency = _native_classifier_router(json.dumps({"tier": tier}), classifier_type) + task: Final = _encrypted_agent_task() + request: Final = { + "input": [ + {"role": "user", "content": "Prior task context"}, + task, + {"type": "function_call_output", "call_id": "call_1", "output": "Tool output"}, + {"role": "user", "content": "Injected reminder"}, + ], + "instructions": "Caller constraints", + "tools": [{"type": "function", "name": "execute"}], + "previous_response_id": "resp_parent", + "litellm_session_id": "parent-session", + "litellm_trace_id": "parent-trace", + "turn_off_message_logging": True, + "litellm_metadata": {"user_api_key_hash": "caller-key-hash"}, + } + original: Final = copy.deepcopy(request) + + result: Final = await router.async_pre_routing_hook(model="encrypted-router", request_kwargs=request) + + assert result.model == model + assert result.routing_decision["tier"] == tier + assert result.routing_decision["cause"] == "llm_classifier" + assert result.routing_decision["classifier_cost"] == 0.0001 + assert result.messages is None + assert request == original + dependency.acompletion.assert_not_called() + call: Final = dependency.aresponses.call_args.kwargs + assert call["input"][-1] == task + assert "opaque-provider-task" not in json.dumps(call["input"][:-1]) + assert "Prior task context" in json.dumps(call["input"][:-1]) + assert "Caller constraints" in json.dumps(call["input"][:-1]) + assert "Caller constraints" not in call["instructions"] + assert "SIMPLE" in call["instructions"] and "REASONING" in call["instructions"] + assert call["text"]["format"]["schema"]["properties"]["tier"]["enum"] == [ + "SIMPLE", "MEDIUM", "COMPLEX", "REASONING" + ] + assert call["text"]["format"]["strict"] is True + assert call["reasoning"] == {"effort": "low"} + assert call["store"] is False + assert call["stream"] is False + assert "tools" not in call and "previous_response_id" not in call + assert "messages" not in call and "response_format" not in call + assert call["timeout"] == 0.1 and call["num_retries"] == 0 and call["disable_fallbacks"] is True + assert call["litellm_session_id"] == "parent-session" + assert call["litellm_trace_id"] == "parent-trace" + assert call["turn_off_message_logging"] is True + assert call["metadata"]["user_api_key_hash"] == "caller-key-hash" + assert call["proxy_server_request"]["body"]["input"] == call["input"] + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "items", + [ + [{"type": "reasoning", "encrypted_content": "opaque-history", "summary": []}, {"role": "user", "content": "hi"}], + [_encrypted_agent_task(), {"role": "user", "content": "hi"}], + [{**_encrypted_agent_task(), "content": [{"type": "input_text", "text": "hi"}]}], + [{"role": "user", "content": "gAAAA is plain text"}], + [{"role": "user", "content": "hi"}, {"type": "function_call_output", "call_id": "call_1", "output": "opaque-provider-task"}], + ], + ids=["historical-reasoning", "older-encrypted-task", "plaintext-agent", "ciphertext-looking-text", "tool-output"], + ) + async def test_other_asks_keep_chat_classifier(self, items: list[dict[str, object]]): + router, dependency = _native_classifier_router() + + result: Final = await router.async_pre_routing_hook(model="encrypted-router", request_kwargs={"input": items}) + + assert result.model == "cheap-model" + assert result.routing_decision["cause"] == "llm_classifier" + dependency.aresponses.assert_not_called() + dependency.acompletion.assert_awaited_once() + + @pytest.mark.asyncio + @pytest.mark.parametrize("output", ["", "not-json", '{"tier":"UNKNOWN"}']) + async def test_invalid_native_verdict_uses_existing_fallback(self, output: str): + router, dependency = _native_classifier_router(output=output) + + result: Final = await router.async_pre_routing_hook( + model="encrypted-router", request_kwargs={"input": [_encrypted_agent_task()]} + ) + + assert result.model == "deep-model" + assert result.routing_decision["cause"] == "default_model_fallback" + dependency.aresponses.assert_awaited_once() + dependency.acompletion.assert_not_called() + + @pytest.mark.asyncio + @pytest.mark.parametrize("deployment_model", ["anthropic/test-classifier", "openai/chat_completions/gpt-6-astra"]) + async def test_incompatible_classifier_does_not_flatten_encryption(self, deployment_model: str): + router, dependency = _native_classifier_router(deployment_model=deployment_model) + + result: Final = await router.async_pre_routing_hook( + model="encrypted-router", request_kwargs={"input": [_encrypted_agent_task()]} + ) + + assert result.model == "deep-model" + assert result.routing_decision["cause"] == "default_model_fallback" + dependency.aresponses.assert_not_called() + dependency.acompletion.assert_not_called() + + @pytest.mark.asyncio + @pytest.mark.parametrize("failure", [ValueError("invalid_encrypted_content"), TimeoutError("classifier timed out")]) + async def test_native_provider_failure_uses_existing_fallback(self, failure: Exception): + router, dependency = _native_classifier_router(failure=failure) + + result: Final = await router.async_pre_routing_hook( + model="encrypted-router", request_kwargs={"input": [_encrypted_agent_task()]} + ) + + assert result.model == "deep-model" + assert result.routing_decision["cause"] == "default_model_fallback" + dependency.aresponses.assert_awaited_once() + dependency.acompletion.assert_not_called() + + class TestLLMClassifier: """Test the LLM-based classifier path (aclassify) and its fallback behavior.""" From 208d554c008da91b74062413ad8d5d454f7746b5 Mon Sep 17 00:00:00 2001 From: moe-berri Date: Thu, 10 Sep 2026 13:04:36 -0700 Subject: [PATCH 2/3] fix(router): validate encrypted classifiers after deployment selection --- .../llms/base_llm/responses/transformation.py | 3 + .../llms/openai/responses/transformation.py | 3 + litellm/responses/main.py | 8 + .../complexity_router/README.md | 3 + .../complexity_router/complexity_router.py | 35 ++-- .../router_strategy/test_complexity_router.py | 151 ++++++++++++++++-- 6 files changed, 168 insertions(+), 35 deletions(-) diff --git a/litellm/llms/base_llm/responses/transformation.py b/litellm/llms/base_llm/responses/transformation.py index 1365941fe2a..14f00aaaa21 100644 --- a/litellm/llms/base_llm/responses/transformation.py +++ b/litellm/llms/base_llm/responses/transformation.py @@ -62,6 +62,9 @@ class BaseResponsesAPIConfig(ABC): """ return False + def supports_encrypted_agent_messages(self) -> bool: + return False + def sign_request( self, headers: dict, diff --git a/litellm/llms/openai/responses/transformation.py b/litellm/llms/openai/responses/transformation.py index 666e9b32011..833ae206024 100644 --- a/litellm/llms/openai/responses/transformation.py +++ b/litellm/llms/openai/responses/transformation.py @@ -110,6 +110,9 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig): def supports_native_file_search(self) -> bool: return True + def supports_encrypted_agent_messages(self) -> bool: + return self.custom_llm_provider in (LlmProviders.OPENAI, LlmProviders.AZURE) + @staticmethod def _is_gpt_5_model(model: str) -> bool: """Return True only for actual OpenAI GPT-5 models. diff --git a/litellm/responses/main.py b/litellm/responses/main.py index 7a210cdd970..e88cd618a8b 100644 --- a/litellm/responses/main.py +++ b/litellm/responses/main.py @@ -1078,6 +1078,7 @@ def responses( litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None) _is_async: Final = kwargs.pop("aresponses", False) is True skip_mcp_handler: Final = kwargs.pop("_skip_mcp_handler", False) + require_encrypted_task_support: Final = kwargs.pop("_require_encrypted_task_support", False) is True use_chat_completions_api = _pop_use_chat_completions_api_kw(kwargs) client_headers: Final = kwargs.get("headers") @@ -1186,6 +1187,13 @@ 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() + ): + raise ValueError("Encrypted task classification requires a compatible native Responses deployment") + local_vars.update(kwargs) # Map reasoning_effort (from litellm_params/proxy config) to reasoning when not set if reasoning is None and "reasoning_effort" in local_vars: diff --git a/litellm/router_strategy/complexity_router/README.md b/litellm/router_strategy/complexity_router/README.md index 267def7cc5b..38dfd143cbc 100644 --- a/litellm/router_strategy/complexity_router/README.md +++ b/litellm/router_strategy/complexity_router/README.md @@ -367,6 +367,9 @@ local scoring shortcut in `heuristic_first` and `hybrid` modes. The configured c a native OpenAI or Azure OpenAI Responses deployment with access to the encrypted content. The provider handles the encrypted task, and the classifier still chooses the tier dynamically +Compatibility is checked after normal deployment selection. A paused incompatible member of the +classifier group does not prevent an eligible compatible deployment from classifying the task + Unsupported classifier deployments and provider decryption errors use the existing `classifier_fallback` policy. No fixed tier is introduced for encrypted tasks. Plaintext asks and requests carrying only historical encrypted reasoning retain the existing classifier path diff --git a/litellm/router_strategy/complexity_router/complexity_router.py b/litellm/router_strategy/complexity_router/complexity_router.py index 77de4df3cd7..6df74b26f60 100644 --- a/litellm/router_strategy/complexity_router/complexity_router.py +++ b/litellm/router_strategy/complexity_router/complexity_router.py @@ -25,7 +25,7 @@ from threading import Lock from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal, NamedTuple, cast -from pydantic import BaseModel, TypeAdapter, create_model +from pydantic import BaseModel, TypeAdapter, ValidationError, create_model from litellm._logging import verbose_router_logger from litellm.constants import ( @@ -504,7 +504,10 @@ def _encrypted_classifier_task( raw_input: Final = (request_kwargs or EMPTY_MAPPING).get("input") if not isinstance(raw_input, list) or (request_kwargs or EMPTY_MAPPING).get("messages"): return None - items: Final = TypeAdapter(tuple[dict[str, object], ...]).validate_python(raw_input) + try: + items: Final = TypeAdapter(tuple[dict[str, object], ...]).validate_python(raw_input) + except ValidationError: + return None current: Final = next( ( item @@ -516,7 +519,10 @@ def _encrypted_classifier_task( ) if current is None or current.get("type") != "agent_message" or not isinstance(current.get("content"), list): return None - parts: Final = TypeAdapter(tuple[dict[str, object], ...]).validate_python(current["content"]) + try: + parts: Final = TypeAdapter(tuple[dict[str, object], ...]).validate_python(current["content"]) + except ValidationError: + return None if not any(part.get("type") == "encrypted_content" and part.get("encrypted_content") for part in parts): return None return { @@ -2058,7 +2064,7 @@ class ComplexityRouter(CustomLogger): classifier_call_params = MappingProxyType({"reasoning_effort": llm_config.reasoning_effort}) payload: Final = ( - self._native_classifier_payload(llm_config.model, messages_for_call, response_format, encrypted_task) + self._native_classifier_payload(messages_for_call, response_format, encrypted_task) if encrypted_task is not None else {"messages": messages_for_call, "response_format": response_format, **classifier_call_params} ) @@ -2098,7 +2104,6 @@ class ComplexityRouter(CustomLogger): def _native_classifier_payload( self, - model: str, messages: list[AllMessageValues], # mutable-ok: existing transformation accepts the SDK message list response_format: Mapping[str, object], encrypted_task: Mapping[str, object], @@ -2106,26 +2111,7 @@ class ComplexityRouter(CustomLogger): from litellm.completion_extras.litellm_responses_transformation.transformation import ( LiteLLMResponsesTransformationHandler, ) - from litellm.litellm_core_utils.get_llm_provider_logic import declared_authenticating_provider, get_llm_provider - from litellm.types.router import LiteLLM_Params - deployments: Final = self._group_deployments(model) - if not deployments: - raise ValueError("Encrypted task classification requires a native OpenAI Responses classifier deployment") - for params in (LiteLLM_Params.model_validate(deployment.get("litellm_params")) for deployment in deployments): - if declared_authenticating_provider(params.model, params.custom_llm_provider): - raise ValueError( - "Encrypted task classification requires a native OpenAI Responses classifier deployment" - ) - _, provider, _, _ = get_llm_provider(model=params.model, litellm_params=params) - if ( - provider not in ("openai", "azure") - or params.use_chat_completions_api - or params.model.startswith("openai/chat_completions/") - ): - raise ValueError( - "Encrypted task classification requires a native OpenAI Responses classifier deployment" - ) transformation: Final = LiteLLMResponsesTransformationHandler() input_items, instructions = transformation.convert_chat_completion_messages_to_responses_api(messages) llm_config: Final = self.config.classifier_llm_config @@ -2139,6 +2125,7 @@ class ComplexityRouter(CustomLogger): "instructions": instructions, "text": transformation.transform_response_format_to_text_format(dict(response_format)), "store": False, + "_require_encrypted_task_support": True, **reasoning, } diff --git a/tests/test_litellm/router_strategy/test_complexity_router.py b/tests/test_litellm/router_strategy/test_complexity_router.py index 780465dad3b..2ea8a8b53fe 100644 --- a/tests/test_litellm/router_strategy/test_complexity_router.py +++ b/tests/test_litellm/router_strategy/test_complexity_router.py @@ -5,8 +5,10 @@ Tests the rule-based complexity scoring and tier assignment logic. """ import asyncio +from collections.abc import AsyncIterator import json from copy import deepcopy +from functools import partial import logging import sys import time @@ -14,6 +16,7 @@ from typing import Dict, Final, List from unittest.mock import AsyncMock, MagicMock, patch import pytest +import httpx from pydantic import ValidationError import litellm @@ -69,6 +72,7 @@ from litellm.types.router import ( TaggedPreRoutingStrategy, ) from litellm.types.llms.openai import ResponsesAPIResponse +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler requires_semantic_router = pytest.mark.skipif( @@ -2520,11 +2524,21 @@ def _native_classifier_router( classifier_type: str = "llm", deployment_model: str = "openai/gpt-6-astra", failure: Exception | None = None, + native_router: Router | None = None, + http_handler: AsyncHTTPHandler | None = None, ) -> tuple[ComplexityRouter, MagicMock]: dependency: Final = MagicMock( - aresponses=AsyncMock(return_value=_native_classifier_response(output), side_effect=failure), + aresponses=( + native_router.factory_function(partial(litellm.aresponses, client=http_handler), call_type="aresponses") + if native_router is not None + else AsyncMock(return_value=_native_classifier_response(output), side_effect=failure) + ), acompletion=AsyncMock(return_value=_llm_response('{"tier":"SIMPLE"}')), - get_model_list=MagicMock(return_value=[{"litellm_params": {"model": deployment_model}}]), + get_model_list=( + native_router.get_model_list + if native_router is not None + else MagicMock(return_value=[{"litellm_params": {"model": deployment_model}}]) + ), ) return ( ComplexityRouter( @@ -2533,7 +2547,11 @@ def _native_classifier_router( complexity_router_config={ "tiers": {"SIMPLE": "cheap-model", "REASONING": "deep-model"}, "classifier_type": classifier_type, - "classifier_llm_config": {"model": "classifier", "timeout_ms": 100, "reasoning_effort": "low"}, + "classifier_llm_config": { + "model": "classifier", + "timeout_ms": 5000 if native_router is not None else 100, + "reasoning_effort": "low", + }, "heuristic_first_max_tier": "SIMPLE" if classifier_type == "heuristic_first" else None, "hybrid_boundary_margin": 0.01 if classifier_type == "hybrid" else None, "classifier_fallback": "default_model", @@ -2546,6 +2564,18 @@ def _native_classifier_router( ) +@pytest.fixture +async def native_classifier_http() -> AsyncIterator[tuple[AsyncHTTPHandler, MagicMock]]: + respond: Final = MagicMock( + return_value=httpx.Response(200, json=_native_classifier_response('{"tier":"REASONING"}').model_dump()) + ) + async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client: + handler: Final = AsyncHTTPHandler() + await handler.client.aclose() + handler.client = client + yield handler, respond + + class TestEncryptedTaskClassifier: @pytest.mark.asyncio @pytest.mark.parametrize("classifier_type", ["llm", "heuristic_first", "hybrid"]) @@ -2587,11 +2617,15 @@ class TestEncryptedTaskClassifier: assert "Caller constraints" not in call["instructions"] assert "SIMPLE" in call["instructions"] and "REASONING" in call["instructions"] assert call["text"]["format"]["schema"]["properties"]["tier"]["enum"] == [ - "SIMPLE", "MEDIUM", "COMPLEX", "REASONING" + "SIMPLE", + "MEDIUM", + "COMPLEX", + "REASONING", ] assert call["text"]["format"]["strict"] is True assert call["reasoning"] == {"effort": "low"} assert call["store"] is False + assert call["_require_encrypted_task_support"] is True assert call["stream"] is False assert "tools" not in call and "previous_response_id" not in call assert "messages" not in call and "response_format" not in call @@ -2606,13 +2640,25 @@ class TestEncryptedTaskClassifier: @pytest.mark.parametrize( "items", [ - [{"type": "reasoning", "encrypted_content": "opaque-history", "summary": []}, {"role": "user", "content": "hi"}], + [ + {"type": "reasoning", "encrypted_content": "opaque-history", "summary": []}, + {"role": "user", "content": "hi"}, + ], [_encrypted_agent_task(), {"role": "user", "content": "hi"}], [{**_encrypted_agent_task(), "content": [{"type": "input_text", "text": "hi"}]}], [{"role": "user", "content": "gAAAA is plain text"}], - [{"role": "user", "content": "hi"}, {"type": "function_call_output", "call_id": "call_1", "output": "opaque-provider-task"}], + [ + {"role": "user", "content": "hi"}, + {"type": "function_call_output", "call_id": "call_1", "output": "opaque-provider-task"}, + ], + ], + ids=[ + "historical-reasoning", + "older-encrypted-task", + "plaintext-agent", + "ciphertext-looking-text", + "tool-output", ], - ids=["historical-reasoning", "older-encrypted-task", "plaintext-agent", "ciphertext-looking-text", "tool-output"], ) async def test_other_asks_keep_chat_classifier(self, items: list[dict[str, object]]): router, dependency = _native_classifier_router() @@ -2639,9 +2685,28 @@ class TestEncryptedTaskClassifier: dependency.acompletion.assert_not_called() @pytest.mark.asyncio - @pytest.mark.parametrize("deployment_model", ["anthropic/test-classifier", "openai/chat_completions/gpt-6-astra"]) - async def test_incompatible_classifier_does_not_flatten_encryption(self, deployment_model: str): - router, dependency = _native_classifier_router(deployment_model=deployment_model) + @pytest.mark.parametrize( + "deployment_model", + ["anthropic/test-classifier", "openai/chat_completions/gpt-6-astra", "xai/test-classifier"], + ) + async def test_incompatible_classifier_does_not_flatten_encryption( + self, deployment_model: str, native_classifier_http: tuple[AsyncHTTPHandler, MagicMock] + ): + handler, respond = native_classifier_http + native: Final = Router( + model_list=[ + { + "model_name": "classifier", + "litellm_params": { + "model": deployment_model, + "api_key": "test-key", + "api_base": "https://classifier.test/v1", + }, + } + ], + num_retries=0, + ) + router, _ = _native_classifier_router(native_router=native, http_handler=handler) result: Final = await router.async_pre_routing_hook( model="encrypted-router", request_kwargs={"input": [_encrypted_agent_task()]} @@ -2649,8 +2714,72 @@ class TestEncryptedTaskClassifier: assert result.model == "deep-model" assert result.routing_decision["cause"] == "default_model_fallback" + respond.assert_not_called() + + @pytest.mark.asyncio + @pytest.mark.parametrize("blocked", [True, False]) + async def test_native_classifier_validates_selected_deployment( + self, blocked: bool, native_classifier_http: tuple[AsyncHTTPHandler, MagicMock] + ): + handler, respond = native_classifier_http + native: Final = Router( + model_list=[ + { + "model_name": "classifier", + "litellm_params": {"model": "anthropic/test-classifier", "api_key": "test-key", "order": 0}, + "model_info": {"id": "incompatible", "blocked": blocked}, + }, + { + "model_name": "classifier", + "litellm_params": { + "model": "openai/gpt-6-astra", + "api_key": "test-key", + "order": 1, + "api_base": "https://classifier.test/v1", + }, + "model_info": {"id": "compatible"}, + }, + ], + num_retries=0, + ) + router, _ = _native_classifier_router(native_router=native, http_handler=handler) + task: Final = _encrypted_agent_task() + + result: Final = await router.async_pre_routing_hook(model="encrypted-router", request_kwargs={"input": [task]}) + + assert result.model == "deep-model" + assert result.routing_decision["cause"] == ("llm_classifier" if blocked else "default_model_fallback") + if blocked: + respond.assert_called_once() + request: Final = respond.call_args.args[0] + assert request.url.path == "/v1/responses" + body: Final = json.loads(request.content) + assert body["input"][-1] == task + assert "_require_encrypted_task_support" not in body + else: + respond.assert_not_called() + + @pytest.mark.asyncio + @pytest.mark.parametrize("classifier_type", ["llm", "heuristic_first", "hybrid"]) + @pytest.mark.parametrize( + "input_items", + [ + ["unsupported-input-item"], + [{**_encrypted_agent_task(), "content": [{"type": "input_text", "text": "hi"}, None]}], + ], + ) + async def test_encrypted_detection_does_not_reject_other_input_shapes( + self, classifier_type: str, input_items: list[object] + ): + router, dependency = _native_classifier_router(classifier_type=classifier_type) + + result: Final = await router.aclassify("hi", request_kwargs={"input": input_items}) + + assert result.cause != "default_model_fallback" + assert result.tier == ComplexityTier.SIMPLE dependency.aresponses.assert_not_called() - dependency.acompletion.assert_not_called() + if classifier_type == "llm": + dependency.acompletion.assert_awaited_once() @pytest.mark.asyncio @pytest.mark.parametrize("failure", [ValueError("invalid_encrypted_content"), TimeoutError("classifier timed out")]) From d266b76a8bcbb6cbec408eeeb8758a7af1a92644 Mon Sep 17 00:00:00 2001 From: moe-berri Date: Thu, 10 Sep 2026 13:48:21 -0700 Subject: [PATCH 3/3] fix(router): honor Codex reminders and map classifier failures --- litellm/responses/main.py | 48 ++++++++++++++++--- .../complexity_router/complexity_router.py | 4 +- .../test_responses_api_bridge_flag.py | 24 ++++++++++ .../router_strategy/test_complexity_router.py | 34 +++++++++++++ 4 files changed, 102 insertions(+), 8 deletions(-) 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")])