fix(router): classify encrypted delegated tasks with native Responses

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.

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

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,

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": "<system-reminder>Injected reminder</system-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."""