mvoe validation logic to RequestHandlers

Signed-off-by: Uri Shaket <ushaket@redhat.com>

Author: ushaket

src/guidellm/backends/openai/request_handlers.py

@@ -219,14 +219,16 @@ def compile_non_streaming(
         text = choice.get("text", "")
         input_metrics, output_metrics = self.extract_metrics(usage, text)
 
-        return GenerationResponse(
+        compiled = GenerationResponse(
             request_id=request.request_id,
             request_args=arguments.model_dump_json(),
             response_id=response.get("id"),  # use vLLM ID if available
             text=text,
             input_metrics=input_metrics,
             output_metrics=output_metrics,
         )
+        self._validate_compiled_response(compiled)
+        return compiled
 
     def add_streaming_line(self, line: str) -> int | None:
         """
@@ -269,14 +271,26 @@ def compile_streaming(
         text = "".join(self.streaming_texts)
         input_metrics, output_metrics = self.extract_metrics(self.streaming_usage, text)
 
-        return GenerationResponse(
+        compiled = GenerationResponse(
             request_id=request.request_id,
             request_args=arguments.model_dump_json(),
             response_id=self.streaming_response_id,  # use vLLM ID if available
             text=text,
             input_metrics=input_metrics,
             output_metrics=output_metrics,
         )
+        self._validate_compiled_response(compiled)
+        return compiled
+
+    def _validate_compiled_response(self, response: GenerationResponse) -> None:
+        """Raise when endpoint produced a terminal payload with no usable output."""
+        has_text = bool(response.text and response.text.strip())
+        output_tokens = response.output_metrics.total_tokens or 0
+        if not has_text and output_tokens <= 0:
+            raise ValueError(
+                "[UNUSABLE_BACKEND_RESPONSE] backend resolved without a usable "
+                "terminal response payload"
+            )
 
     def extract_line_data(self, line: str) -> dict[str, Any] | None:
         """
@@ -493,14 +507,16 @@ def compile_non_streaming(
         text = choice.get("message", {}).get("content", "")
         input_metrics, output_metrics = self.extract_metrics(usage, text)
 
-        return GenerationResponse(
+        compiled = GenerationResponse(
             request_id=request.request_id,
             request_args=arguments.model_dump_json(),
             response_id=response.get("id"),  # use vLLM ID if available
             text=text,
             input_metrics=input_metrics,
             output_metrics=output_metrics,
         )
+        self._validate_compiled_response(compiled)
+        return compiled
 
     def add_streaming_line(self, line: str) -> int | None:
         """
@@ -543,14 +559,16 @@ def compile_streaming(
         text = "".join(self.streaming_texts)
         input_metrics, output_metrics = self.extract_metrics(self.streaming_usage, text)
 
-        return GenerationResponse(
+        compiled = GenerationResponse(
             request_id=request.request_id,
             request_args=arguments.model_dump_json(),
             response_id=self.streaming_response_id,  # use vLLM ID if available
             text=text,
             input_metrics=input_metrics,
             output_metrics=output_metrics,
         )
+        self._validate_compiled_response(compiled)
+        return compiled
 
 
 @OpenAIRequestHandlerFactory.register(

src/guidellm/scheduler/worker.py

@@ -37,7 +37,7 @@
     ResponseT,
 )
 from guidellm.scheduler.strategies import SchedulingStrategy
-from guidellm.schemas import GenerationResponse, RequestInfo
+from guidellm.schemas import RequestInfo
 from guidellm.utils import (
     InterProcessMessaging,
     wait_for_sync_barrier,
@@ -371,14 +371,7 @@ async def _process_next_request(self, target_start: float):
                     raise RuntimeError("Received invalid request info from backend")
 
             request_info.timings.resolve_end = time.time()
-            if not self._has_usable_terminal_response(response):
-                request_info.error = (
-                    "[UNUSABLE_BACKEND_RESPONSE] backend resolved without a usable "
-                    "terminal response payload"
-                )
-                self._send_update("errored", None, request, request_info)
-            else:
-                self._send_update("completed", response, request, request_info)
+            self._send_update("completed", response, request, request_info)
 
             response = request = request_info = None
         except asyncio.CancelledError:
@@ -402,21 +395,6 @@ async def _process_next_request(self, target_start: float):
             if request_info is not None:
                 self.strategy.request_completed(request_info)
 
-    def _has_usable_terminal_response(self, response: ResponseT | None) -> bool:
-        """
-        Return whether backend resolve produced a usable terminal response.
-
-        A missing terminal response indicates the backend transport resolved but did
-        not provide result payload data that can be aggregated.
-        """
-
-        if isinstance(response, GenerationResponse):
-            has_text = bool(response.text and response.text.strip())
-            output_tokens = response.output_metrics.total_tokens or 0
-            return has_text or output_tokens > 0
-
-        return bool(response)
-
     async def _dequeue_next_request(
         self, target_start: float
     ) -> tuple[RequestT, RequestInfo]:

tests/unit/backends/openai/test_request_handlers.py

@@ -300,9 +300,6 @@ def test_format_ignore_eos(self, valid_instances):
                 10,
                 5,
             ),
-            ({"choices": [{"text": ""}], "usage": {}}, "", None, None),
-            ({"choices": [], "usage": {}}, "", None, None),
-            ({}, "", None, None),
         ],
     )
     def test_non_streaming(
@@ -363,7 +360,6 @@ def test_non_streaming(
                 None,
                 None,
             ),
-            (["", "data: [DONE]"], "", None, None),
         ],
     )
     def test_streaming(
@@ -397,6 +393,47 @@ def test_streaming(
         assert response.output_metrics.text_words == len(expected_text.split())
         assert response.output_metrics.text_characters == len(expected_text)
 
+    @pytest.mark.regression
+    @pytest.mark.parametrize(
+        "response",
+        [
+            {"choices": [{"text": ""}], "usage": {}},
+            {"choices": [], "usage": {}},
+            {},
+        ],
+    )
+    def test_non_streaming_raises_for_unusable_terminal_payload(
+        self, valid_instances, generation_request, response
+    ):
+        """Test unusable non-streaming text response raises.
+
+        ### WRITTEN BY AI ###
+        """
+        instance = valid_instances
+        arguments = instance.format(generation_request)
+
+        with pytest.raises(ValueError, match="UNUSABLE_BACKEND_RESPONSE"):
+            instance.compile_non_streaming(generation_request, arguments, response)
+
+    @pytest.mark.regression
+    def test_streaming_raises_for_unusable_terminal_payload(
+        self, valid_instances, generation_request
+    ):
+        """Test unusable streaming text response raises.
+
+        ### WRITTEN BY AI ###
+        """
+        instance = valid_instances
+        arguments = instance.format(generation_request)
+
+        for line in ["", "data: [DONE]"]:
+            result = instance.add_streaming_line(line)
+            if result is None:
+                break
+
+        with pytest.raises(ValueError, match="UNUSABLE_BACKEND_RESPONSE"):
+            instance.compile_streaming(generation_request, arguments)
+
     @pytest.mark.smoke
     @pytest.mark.parametrize(
         ("line", "expected_output"),
@@ -818,18 +855,6 @@ def test_format_multimodal(self, valid_instances):
                 10,
                 5,
             ),
-            (
-                {"choices": [{"message": {"content": ""}}], "usage": {}},
-                "",
-                None,
-                None,
-            ),
-            (
-                {"choices": [], "usage": {}},
-                "",
-                None,
-                None,
-            ),
         ],
     )
     def test_non_streaming(
@@ -883,12 +908,6 @@ def test_non_streaming(
                 None,
                 None,
             ),
-            (
-                ["", "data: [DONE]"],
-                "",
-                None,
-                None,
-            ),
         ],
     )
     def test_streaming(
@@ -920,6 +939,46 @@ def test_streaming(
         assert response.input_metrics.text_tokens == expected_input_tokens
         assert response.output_metrics.text_tokens == expected_output_tokens
 
+    @pytest.mark.regression
+    @pytest.mark.parametrize(
+        "response",
+        [
+            {"choices": [{"message": {"content": ""}}], "usage": {}},
+            {"choices": [], "usage": {}},
+        ],
+    )
+    def test_non_streaming_raises_for_unusable_terminal_payload(
+        self, valid_instances, generation_request, response
+    ):
+        """Test unusable non-streaming chat response raises.
+
+        ### WRITTEN BY AI ###
+        """
+        instance = valid_instances
+        arguments = instance.format(generation_request)
+
+        with pytest.raises(ValueError, match="UNUSABLE_BACKEND_RESPONSE"):
+            instance.compile_non_streaming(generation_request, arguments, response)
+
+    @pytest.mark.regression
+    def test_streaming_raises_for_unusable_terminal_payload(
+        self, valid_instances, generation_request
+    ):
+        """Test unusable streaming chat response raises.
+
+        ### WRITTEN BY AI ###
+        """
+        instance = valid_instances
+        arguments = instance.format(generation_request)
+
+        for line in ["", "data: [DONE]"]:
+            result = instance.add_streaming_line(line)
+            if result is None:
+                break
+
+        with pytest.raises(ValueError, match="UNUSABLE_BACKEND_RESPONSE"):
+            instance.compile_streaming(generation_request, arguments)
+
 
 class TestAudioRequestHandler:
     """Test cases for AudioRequestHandler.