fix(neuralnav): tighten DeploymentIntent types and skip extraction when overrides suffice

NeuralNav's LLM hallucinates invalid use_case values
(e.g., "text_summarization" instead of "summarization_short"), causing
422 errors from the /api/v1/extract endpoint.

Two fixes:
- DeploymentIntent now uses Literal types for use_case (9 values),
  experience_class (5 values), and priority fields (low/medium/high),
  matching NeuralNav's schema for defense-in-depth validation.
- recommend() skips the extract_intent() call when both use_case_override
  and user_count_override are provided, since those are the only fields
  consumed from the extracted intent. This avoids the 422 entirely when
  the caller specifies the use case directly.

Signed-off-by: Amit Oren <amoren@redhat.com>

Author: amito

src/rhoai_mcp/composites/neuralnav/client.py

@@ -190,22 +190,31 @@ def recommend(
     ) -> RecommendationResult:
         """Run the full recommendation flow.
 
-        1. Extract intent from text
+        1. Extract intent from text (skipped when overrides cover all needed fields)
         2. Apply overrides
         3. Fetch SLO defaults + workload profile + expected RPS
         4. Apply SLO overrides on top of fetched defaults
         5. Get ranked recommendations with all constraints
         6. Extract top recommendation from each ranking list
         """
-        # Step 1: Extract intent
-        intent = self.extract_intent(text)
-
-        # Step 2: Apply overrides
-        use_case = use_case_override if use_case_override is not None else intent.use_case
-        user_count = user_count_override if user_count_override is not None else intent.user_count
-        gpu_types = (
-            gpu_types_override if gpu_types_override is not None else intent.preferred_gpu_types
-        )
+        # Step 1: Extract intent (skip when all overrides are provided)
+        if (
+            use_case_override is not None
+            and user_count_override is not None
+            and gpu_types_override is not None
+        ):
+            use_case = use_case_override
+            user_count = user_count_override
+            gpu_types = gpu_types_override
+        else:
+            intent = self.extract_intent(text)
+            use_case = use_case_override if use_case_override is not None else intent.use_case
+            user_count = (
+                user_count_override if user_count_override is not None else intent.user_count
+            )
+            gpu_types = (
+                gpu_types_override if gpu_types_override is not None else intent.preferred_gpu_types
+            )
 
         # Step 3: Fetch defaults
         slo_data = self.get_slo_defaults(use_case)

src/rhoai_mcp/composites/neuralnav/models.py

@@ -2,24 +2,48 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Literal
 
 from pydantic import BaseModel, Field
 
+UseCaseType = Literal[
+    "chatbot_conversational",
+    "code_completion",
+    "code_generation_detailed",
+    "translation",
+    "content_generation",
+    "summarization_short",
+    "document_analysis_rag",
+    "long_document_summarization",
+    "research_legal_analysis",
+]
+
+ExperienceClassType = Literal[
+    "instant",
+    "conversational",
+    "interactive",
+    "deferred",
+    "batch",
+]
+
+PriorityType = Literal["low", "medium", "high"]
+
 
 class DeploymentIntent(BaseModel):
     """Extracted deployment intent from natural language."""
 
-    use_case: str = Field(..., description="Primary use case type")
+    use_case: UseCaseType = Field(..., description="Primary use case type")
     user_count: int = Field(..., description="Number of users or scale")
-    experience_class: str = Field(default="conversational", description="User experience class")
+    experience_class: ExperienceClassType = Field(
+        default="conversational", description="User experience class"
+    )
     preferred_gpu_types: list[str] = Field(
         default_factory=list, description="Preferred GPU types (empty = any)"
     )
-    accuracy_priority: str = Field(default="medium", description="Accuracy importance")
-    cost_priority: str = Field(default="medium", description="Cost sensitivity")
-    latency_priority: str = Field(default="medium", description="Latency importance")
-    complexity_priority: str = Field(default="medium", description="Simplicity preference")
+    accuracy_priority: PriorityType = Field(default="medium", description="Accuracy importance")
+    cost_priority: PriorityType = Field(default="medium", description="Cost sensitivity")
+    latency_priority: PriorityType = Field(default="medium", description="Latency importance")
+    complexity_priority: PriorityType = Field(default="medium", description="Simplicity preference")
     domain_specialization: list[str] = Field(
         default_factory=lambda: ["general"], description="Domain requirements"
     )

tests/composites/neuralnav/test_client.py

@@ -327,14 +327,9 @@ def test_recommend_full_flow(self, mock_httpx: MagicMock) -> None:
 
     @patch("rhoai_mcp.composites.neuralnav.client.httpx")
     def test_recommend_with_overrides(self, mock_httpx: MagicMock) -> None:
-        """Overrides replace extracted intent values."""
+        """When both use_case and user_count overrides are provided, extraction is skipped."""
         mock_client = MagicMock()
 
-        extract_resp = MagicMock()
-        extract_resp.status_code = 200
-        extract_resp.json.return_value = SAMPLE_INTENT
-        extract_resp.raise_for_status = MagicMock()
-
         slo_resp = MagicMock()
         slo_resp.status_code = 200
         slo_resp.json.return_value = SAMPLE_SLO_DEFAULTS
@@ -355,7 +350,8 @@ def test_recommend_with_overrides(self, mock_httpx: MagicMock) -> None:
         ranked_resp.json.return_value = SAMPLE_RANKED_RESPONSE
         ranked_resp.raise_for_status = MagicMock()
 
-        mock_client.post.side_effect = [extract_resp, ranked_resp]
+        # Only one POST (ranked-recommend), extraction is skipped
+        mock_client.post.side_effect = [ranked_resp]
         mock_client.get.side_effect = [slo_resp, workload_resp, rps_resp]
 
         mock_httpx.Client.return_value.__enter__ = MagicMock(return_value=mock_client)
@@ -372,6 +368,8 @@ def test_recommend_with_overrides(self, mock_httpx: MagicMock) -> None:
         # Verify the overridden use_case was used for SLO defaults fetch
         get_calls = mock_client.get.call_args_list
         assert "code_completion" in get_calls[0].args[0]
+        # Extraction was skipped — only one POST call (ranked-recommend)
+        assert mock_client.post.call_count == 1
 
     @patch("rhoai_mcp.composites.neuralnav.client.httpx")
     def test_recommend_api_error(self, mock_httpx: MagicMock) -> None:
@@ -625,6 +623,150 @@ def test_recommend_forwards_constraints(self, mock_httpx: MagicMock) -> None:
         assert payload["percentile"] == "p99"
 
 
+class TestNeuralNavClientRecommendExtractionBypass:
+    """Tests for skipping extraction when overrides are sufficient."""
+
+    @patch("rhoai_mcp.composites.neuralnav.client.httpx")
+    def test_recommend_skips_extraction_when_all_overrides_provided(
+        self, mock_httpx: MagicMock
+    ) -> None:
+        """When all overrides are provided, extraction is skipped."""
+        mock_client = MagicMock()
+
+        slo_resp = MagicMock()
+        slo_resp.status_code = 200
+        slo_resp.json.return_value = SAMPLE_SLO_DEFAULTS
+        slo_resp.raise_for_status = MagicMock()
+
+        workload_resp = MagicMock()
+        workload_resp.status_code = 200
+        workload_resp.json.return_value = SAMPLE_WORKLOAD_PROFILE
+        workload_resp.raise_for_status = MagicMock()
+
+        rps_resp = MagicMock()
+        rps_resp.status_code = 200
+        rps_resp.json.return_value = SAMPLE_EXPECTED_RPS
+        rps_resp.raise_for_status = MagicMock()
+
+        ranked_resp = MagicMock()
+        ranked_resp.status_code = 200
+        ranked_resp.json.return_value = SAMPLE_RANKED_RESPONSE
+        ranked_resp.raise_for_status = MagicMock()
+
+        # Only one POST call: ranked-recommend (no extract call)
+        mock_client.post.side_effect = [ranked_resp]
+        mock_client.get.side_effect = [slo_resp, workload_resp, rps_resp]
+
+        mock_httpx.Client.return_value.__enter__ = MagicMock(return_value=mock_client)
+        mock_httpx.Client.return_value.__exit__ = MagicMock(return_value=False)
+
+        client = NeuralNavClient("http://localhost:8000")
+        result = client.recommend(
+            "I need a chatbot for 1000 users",
+            use_case_override="chatbot_conversational",
+            user_count_override=1000,
+            gpu_types_override=["A100"],
+        )
+
+        # Only one POST call was made (ranked-recommend, not extract)
+        assert mock_client.post.call_count == 1
+        assert result.specification["use_case"] == "chatbot_conversational"
+        assert result.specification["user_count"] == 1000
+
+    @patch("rhoai_mcp.composites.neuralnav.client.httpx")
+    def test_recommend_still_extracts_when_only_use_case_override(
+        self, mock_httpx: MagicMock
+    ) -> None:
+        """When only use_case override is provided, extraction still runs for user_count."""
+        mock_client = MagicMock()
+
+        extract_resp = MagicMock()
+        extract_resp.status_code = 200
+        extract_resp.json.return_value = SAMPLE_INTENT
+        extract_resp.raise_for_status = MagicMock()
+
+        slo_resp = MagicMock()
+        slo_resp.status_code = 200
+        slo_resp.json.return_value = SAMPLE_SLO_DEFAULTS
+        slo_resp.raise_for_status = MagicMock()
+
+        workload_resp = MagicMock()
+        workload_resp.status_code = 200
+        workload_resp.json.return_value = SAMPLE_WORKLOAD_PROFILE
+        workload_resp.raise_for_status = MagicMock()
+
+        rps_resp = MagicMock()
+        rps_resp.status_code = 200
+        rps_resp.json.return_value = SAMPLE_EXPECTED_RPS
+        rps_resp.raise_for_status = MagicMock()
+
+        ranked_resp = MagicMock()
+        ranked_resp.status_code = 200
+        ranked_resp.json.return_value = SAMPLE_RANKED_RESPONSE
+        ranked_resp.raise_for_status = MagicMock()
+
+        mock_client.post.side_effect = [extract_resp, ranked_resp]
+        mock_client.get.side_effect = [slo_resp, workload_resp, rps_resp]
+
+        mock_httpx.Client.return_value.__enter__ = MagicMock(return_value=mock_client)
+        mock_httpx.Client.return_value.__exit__ = MagicMock(return_value=False)
+
+        client = NeuralNavClient("http://localhost:8000")
+        result = client.recommend(
+            "I need a chatbot",
+            use_case_override="code_completion",
+        )
+
+        # Two POST calls: extract + ranked-recommend
+        assert mock_client.post.call_count == 2
+        # Use case override is applied
+        assert result.specification["use_case"] == "code_completion"
+
+    @patch("rhoai_mcp.composites.neuralnav.client.httpx")
+    def test_recommend_skips_extraction_uses_gpu_override(self, mock_httpx: MagicMock) -> None:
+        """When extraction is skipped, gpu_types_override is used."""
+        mock_client = MagicMock()
+
+        slo_resp = MagicMock()
+        slo_resp.status_code = 200
+        slo_resp.json.return_value = SAMPLE_SLO_DEFAULTS
+        slo_resp.raise_for_status = MagicMock()
+
+        workload_resp = MagicMock()
+        workload_resp.status_code = 200
+        workload_resp.json.return_value = SAMPLE_WORKLOAD_PROFILE
+        workload_resp.raise_for_status = MagicMock()
+
+        rps_resp = MagicMock()
+        rps_resp.status_code = 200
+        rps_resp.json.return_value = SAMPLE_EXPECTED_RPS
+        rps_resp.raise_for_status = MagicMock()
+
+        ranked_resp = MagicMock()
+        ranked_resp.status_code = 200
+        ranked_resp.json.return_value = SAMPLE_RANKED_RESPONSE
+        ranked_resp.raise_for_status = MagicMock()
+
+        mock_client.post.side_effect = [ranked_resp]
+        mock_client.get.side_effect = [slo_resp, workload_resp, rps_resp]
+
+        mock_httpx.Client.return_value.__enter__ = MagicMock(return_value=mock_client)
+        mock_httpx.Client.return_value.__exit__ = MagicMock(return_value=False)
+
+        client = NeuralNavClient("http://localhost:8000")
+        client.recommend(
+            "I need a chatbot",
+            use_case_override="chatbot_conversational",
+            user_count_override=1000,
+            gpu_types_override=["H100"],
+        )
+
+        # Verify the GPU override was forwarded
+        ranked_call = mock_client.post.call_args
+        payload = ranked_call.kwargs.get("json") or ranked_call[1].get("json")
+        assert payload["preferred_gpu_types"] == ["H100"]
+
+
 class TestNeuralNavClientRequestErrors:
     """Tests for _request error handling edge cases."""
 

tests/composites/neuralnav/test_models.py

@@ -1,5 +1,10 @@
 """Tests for NeuralNav composite models."""
 
+from typing import get_args
+
+import pytest
+from pydantic import ValidationError
+
 from rhoai_mcp.composites.neuralnav.models import (
     DeploymentConfigResult,
     DeploymentIntent,
@@ -9,6 +14,7 @@
     RecommendationScores,
     SLOTargets,
     TrafficProfile,
+    UseCaseType,
 )
 
 
@@ -39,6 +45,42 @@ def test_full_intent(self) -> None:
         assert intent.preferred_gpu_types == ["H100", "A100-80"]
         assert intent.accuracy_priority == "high"
 
+    def test_invalid_use_case_rejected(self) -> None:
+        """Invalid use_case values are rejected by Pydantic validation."""
+        with pytest.raises(ValidationError, match="use_case"):
+            DeploymentIntent(use_case="summarization", user_count=1000)
+
+    def test_invalid_use_case_text_summarization_rejected(self) -> None:
+        """LLM-hallucinated 'text_summarization' is rejected."""
+        with pytest.raises(ValidationError, match="use_case"):
+            DeploymentIntent(use_case="text_summarization", user_count=1000)
+
+    def test_invalid_experience_class_rejected(self) -> None:
+        """Invalid experience_class values are rejected."""
+        with pytest.raises(ValidationError, match="experience_class"):
+            DeploymentIntent(
+                use_case="chatbot_conversational",
+                user_count=1000,
+                experience_class="realtime",
+            )
+
+    def test_invalid_priority_rejected(self) -> None:
+        """Invalid priority values are rejected."""
+        with pytest.raises(ValidationError, match="accuracy_priority"):
+            DeploymentIntent(
+                use_case="chatbot_conversational",
+                user_count=1000,
+                accuracy_priority="critical",
+            )
+
+    def test_all_valid_use_cases_accepted(self) -> None:
+        """All valid use_case values are accepted."""
+        valid_use_cases = list(get_args(UseCaseType))
+        assert len(valid_use_cases) > 0
+        for uc in valid_use_cases:
+            intent = DeploymentIntent(use_case=uc, user_count=100)
+            assert intent.use_case == uc
+
 
 class TestModelRecommendation:
     """Tests for ModelRecommendation model."""