fix(router): honor Fuse task context and fallback policy

litellm/router_strategy/complexity_router/complexity_router.py

@@ -2167,6 +2167,20 @@ class ComplexityRouter(CustomLogger):
             tier=tier, score=None, signals=("classifier-failed:default-model",), cause="default_model_fallback"
         )
 
+    def _classifier_caller_constraints(
+        self, system_prompt: str | None, request_kwargs: Mapping[str, object] | None
+    ) -> str | None:
+        """Exclude Claude Code's environment and skill catalogs from task forecasts."""
+        return (
+            None
+            if any(
+                is_claude_code_user_agent(user_agent)
+                for metadata in (self._iter_metadata_dicts(request_kwargs) if request_kwargs is not None else ())
+                if isinstance(user_agent := metadata.get("user_agent"), str)
+            )
+            else system_prompt
+        )
+
     async def _classify_with_llm(
         self,
         prompt: str,
@@ -2216,15 +2230,7 @@ class ComplexityRouter(CustomLogger):
         )
 
         encrypted_task: Final = _encrypted_classifier_task(request_kwargs, marker_pairs)
-        caller_system_prompt: Final = (
-            None
-            if any(
-                is_claude_code_user_agent(user_agent)
-                for metadata in (self._iter_metadata_dicts(request_kwargs) if request_kwargs is not None else ())
-                if isinstance(user_agent := metadata.get("user_agent"), str)
-            )
-            else system_prompt
-        )
+        caller_system_prompt: Final = self._classifier_caller_constraints(system_prompt, request_kwargs)
         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=caller_system_prompt,
@@ -2335,10 +2341,14 @@ class ComplexityRouter(CustomLogger):
             raise ValueError("llm_v2_config is not set")
         request: Final[Mapping[str, object]] = request_kwargs or MappingProxyType({})
         markers: Final = self._reminder_markers_for_request(request)
-        asks: Final = tuple(reversed(tuple(_iter_human_asks_newest_first(messages or (), markers))))
         encrypted: Final = _encrypted_classifier_task(request_kwargs, markers)
+        asks: Final = (
+            ("The delegated task in the following agent_message.",)
+            if encrypted is not None
+            else tuple(reversed(tuple(_iter_human_asks_newest_first(messages or (), markers))))
+        )
         task_context: Final[LLMV2TaskContext] = {
-            "caller_constraints": system_prompt,
+            "caller_constraints": self._classifier_caller_constraints(system_prompt, request_kwargs),
             "task_and_follow_ups": asks or (prompt,),
         }
         task: Final = json.dumps(task_context)

litellm/router_strategy/complexity_router/config.py

@@ -1593,6 +1593,8 @@ class ComplexityRouterConfig(BaseModel):
             return self
         if v2 is None:
             raise ValueError("llm_v2_config is required when classifier_type is llm_v2")
+        if self.classifier_fallback != "heuristic":
+            raise ValueError("llm_v2 always fails closed to capable_tier; classifier_fallback cannot override it")
         llm: Final = self.classifier_llm_config
         if self.adaptive or self.tier_definitions is not None or self.enable_non_reasoning_tier:
             raise ValueError("llm_v2 requires two built-in tiers and adaptive=false")

tests/test_litellm/router_strategy/test_llm_v2.py

@@ -142,6 +142,7 @@ def test_verdict_rejects_invalid_probabilities(probability: object) -> None:
         ({"classifier_type": "heuristic"}, "requires classifier_type llm_v2"),
         ({"classifier_llm_config": None}, "classifier_llm_config is required"),
         ({"adaptive": True}, "adaptive=false"),
+        ({"classifier_fallback": "default_model", "default_model": "efficient"}, "fails closed"),
         ({"tiers": {"SIMPLE": ["same"], "REASONING": ["same"]}}, "distinct model"),
         ({"tiers": {"SIMPLE": ["a", "b"], "REASONING": ["c"]}}, "one distinct model"),
         ({"tiers": {"SIMPLE": ["a"], "MEDIUM": ["b"], "REASONING": ["c"]}}, "exactly"),
@@ -221,6 +222,21 @@ async def test_json_object_mode_supplies_schema_in_prompt() -> None:
     assert '"required"' in sent["messages"][0]["content"]
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("user_agent", ("claude-cli/2.1.233", "curl/8.7.1"))
+@pytest.mark.parametrize("metadata_key", ("metadata", "litellm_metadata"))
+async def test_caller_constraints_respect_claude_code_prompt_policy(user_agent: str, metadata_key: str) -> None:
+    router, client = _router(_verdict().model_dump_json())
+    outcome: Final = await router.aclassify(
+        "Fix nested behavior", "Caller system context", request_kwargs={metadata_key: {"user_agent": user_agent}}
+    )
+    assert outcome.cause == "llm_v2_classifier"
+    call: Final = client.acompletion.call_args.kwargs
+    payload: Final = json.loads(call["messages"][1]["content"])
+    assert payload["caller_constraints"] == (None if user_agent.startswith("claude") else "Caller system context")
+    assert payload["task_and_follow_ups"] == ["Fix nested behavior"]
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("calibrated", (False, True))
 async def test_routing_metadata_preserves_exact_forecasts_and_redaction(
@@ -365,8 +381,8 @@ async def test_encrypted_task_uses_native_responses_and_preserves_logging_contro
             {"type": "encrypted_content", "encrypted_content": "opaque-task"},
         ],
     }
-    outcome: Final = await router.aclassify(
-        "",
+    result: Final = await router.async_pre_routing_hook(
+        model="v2-router",
         request_kwargs={
             "input": [task],
             "turn_off_message_logging": True,
@@ -374,13 +390,16 @@ async def test_encrypted_task_uses_native_responses_and_preserves_logging_contro
             "litellm_trace_id": "trace",
         },
     )
-    assert outcome.tier == ComplexityTier.REASONING
-    assert outcome.cause == "llm_v2_classifier"
+    assert result is not None and result.model == "capable"
+    assert result.routing_decision is not None
+    assert result.routing_decision["cause"] == "llm_v2_classifier"
     client.acompletion.assert_not_called()
     client.aresponses.assert_awaited_once()
     call: Final = client.aresponses.call_args.kwargs
     assert call["input"][-1] == task
     assert "opaque-task" not in json.dumps(call["input"][:-1])
+    assert "Task: fix a bug" not in json.dumps(call["input"][:-1])
+    assert "The delegated task in the following agent_message." in json.dumps(call["input"][:-1])
     assert call["max_output_tokens"] == 1024
     assert call["text"]["format"]["schema"]["required"] == ["crux", "demands", "verification", "forecasts"]
     assert call["turn_off_message_logging"] is True