Merge pull request #183 from rostilos/1.5.7-rc

feat: Enhance error handling and result validation in CommandQueueCon…

Author: rostilos

python-ecosystem/inference-orchestrator/src/server/command_queue_consumer.py

@@ -18,6 +18,14 @@ class CommandQueueConsumer:
     using the CommandService. Events and final results are pushed back 
     to a job-specific Redis event queue.
     """
+
+    EMPTY_RESULT_SENTINELS = {
+        "null",
+        "none",
+        "no output generated",
+        "failed to generate summary",
+        "i couldn't generate an answer. please try rephrasing your question.",
+    }
 
     def __init__(self, command_service: CommandService):
         self.command_service = command_service
@@ -114,18 +122,45 @@ def event_callback(event: Dict[str, Any]):
                 result = await self.command_service.process_ask(request_dto, event_callback)
             else:
                 raise ValueError(f"Unknown command type: {command_type}")
-            
+
+            if self._has_error(result):
+                error_message = self._get_result_value(result, "error", "AI command failed")
+                await self._publish_event(event_queue_key, {
+                    "type": "error",
+                    "message": str(error_message)
+                })
+                logger.warning(f"Command Job ID {job_id} failed: {error_message}")
+                return
+
             # Format output correctly depending on command type based on their DTO responses
             final_payload = {}
             if command_type == "summarize":
+                summary = self._get_result_value(result, "summary")
+                if not self._has_usable_text(summary):
+                    await self._publish_event(event_queue_key, {
+                        "type": "error",
+                        "message": "AI service returned an empty summary"
+                    })
+                    logger.warning(f"Command Job ID {job_id} failed: empty summarize result")
+                    return
+
                 final_payload = {
-                    "summary": result.summary if hasattr(result, "summary") else result.get("summary"),
-                    "diagram": result.diagram if hasattr(result, "diagram") else result.get("diagram"),
-                    "diagramType": result.diagramType if hasattr(result, "diagramType") else result.get("diagramType", "MERMAID")
+                    "summary": str(summary),
+                    "diagram": self._string_or_empty(self._get_result_value(result, "diagram")),
+                    "diagramType": self._string_or_empty(self._get_result_value(result, "diagramType", "MERMAID")) or "MERMAID"
                 }
             elif command_type == "ask":
+                answer = self._get_result_value(result, "answer")
+                if not self._has_usable_text(answer):
+                    await self._publish_event(event_queue_key, {
+                        "type": "error",
+                        "message": "AI service returned an empty answer"
+                    })
+                    logger.warning(f"Command Job ID {job_id} failed: empty ask result")
+                    return
+
                 final_payload = {
-                    "answer": result.answer if hasattr(result, "answer") else result.get("answer")
+                    "answer": str(answer)
                 }
 
             event_callback({"type": "final", "result": final_payload})
@@ -156,3 +191,27 @@ async def _publish_event(self, key: str, event: Dict[str, Any]):
             await self._redis.lpush(key, event_json)
         except Exception as e:
             logger.error(f"Failed to publish event to {key}: {e}")
+
+    @staticmethod
+    def _get_result_value(result: Any, key: str, default: Any = None) -> Any:
+        if isinstance(result, dict):
+            return result.get(key, default)
+        if hasattr(result, key):
+            return getattr(result, key)
+        return default
+
+    @classmethod
+    def _has_error(cls, result: Any) -> bool:
+        error = cls._get_result_value(result, "error")
+        return error is not None and str(error).strip() != ""
+
+    @staticmethod
+    def _has_usable_text(value: Any) -> bool:
+        if value is None:
+            return False
+        text = str(value).strip()
+        return bool(text) and text.lower() not in CommandQueueConsumer.EMPTY_RESULT_SENTINELS
+
+    @staticmethod
+    def _string_or_empty(value: Any) -> str:
+        return "" if value is None else str(value)

python-ecosystem/inference-orchestrator/src/service/command/command_service.py

@@ -30,6 +30,14 @@ class CommandService:
     # Hard timeout ceiling for commands (seconds). Configurable via .env
     COMMAND_TIMEOUT_SECONDS = int(os.environ.get("COMMAND_TIMEOUT_SECONDS", "600"))
 
+    EMPTY_RESULT_SENTINELS = {
+        "null",
+        "none",
+        "no output generated",
+        "failed to generate summary",
+        "i couldn't generate an answer. please try rephrasing your question.",
+    }
+
     def __init__(self):
         load_dotenv(interpolate=False)
         self.default_jar_path = os.environ.get(
@@ -112,6 +120,11 @@ async def process_summarize(
                     except Exception as close_err:
                         logger.warning(f"Error closing MCP sessions: {close_err}")
 
+                result = self._normalize_summarize_result(result, supports_mermaid=False)
+                if "error" in result:
+                    self._emit_event(event_callback, {"type": "error", "message": result["error"]})
+                    return result
+
                 self._emit_event(event_callback, {
                     "type": "status",
                     "state": "completed",
@@ -222,6 +235,11 @@ async def process_ask(
                     except Exception as close_err:
                         logger.warning(f"Error closing MCP sessions: {close_err}")
 
+                result = self._normalize_ask_result(result)
+                if "error" in result:
+                    self._emit_event(event_callback, {"type": "error", "message": result["error"]})
+                    return result
+
                 self._emit_event(event_callback, {
                     "type": "status",
                     "state": "completed",
@@ -241,7 +259,49 @@ async def process_ask(
             sanitized_msg = create_user_friendly_error(e)
             self._emit_event(event_callback, {"type": "error", "message": sanitized_msg})
             return {"error": sanitized_msg}
-    
+
+    def _normalize_summarize_result(self, result: Any, supports_mermaid: bool) -> Dict[str, Any]:
+        """Validate summarize output before the queue consumer publishes a final event."""
+        if not isinstance(result, dict):
+            return {"error": "AI service returned an invalid summarize result"}
+        if result.get("error"):
+            return {"error": str(result["error"])}
+
+        summary = result.get("summary")
+        if not self._has_usable_text(summary):
+            return {"error": "AI service returned an empty summary"}
+
+        diagram_type = result.get("diagramType") or ("MERMAID" if supports_mermaid else "ASCII")
+        return {
+            "summary": str(summary),
+            "diagram": self._string_or_empty(result.get("diagram")),
+            "diagramType": str(diagram_type),
+        }
+
+    def _normalize_ask_result(self, result: Any) -> Dict[str, Any]:
+        """Validate ask output before the queue consumer publishes a final event."""
+        if not isinstance(result, dict):
+            return {"error": "AI service returned an invalid ask result"}
+        if result.get("error"):
+            return {"error": str(result["error"])}
+
+        answer = result.get("answer")
+        if not self._has_usable_text(answer):
+            return {"error": "AI service returned an empty answer"}
+
+        return {"answer": str(answer)}
+
+    @classmethod
+    def _has_usable_text(cls, value: Any) -> bool:
+        if value is None:
+            return False
+        text = str(value).strip()
+        return bool(text) and text.lower() not in cls.EMPTY_RESULT_SENTINELS
+
+    @staticmethod
+    def _string_or_empty(value: Any) -> str:
+        return "" if value is None else str(value)
+
     def _build_platform_jvm_props(self, request) -> Dict[str, str]:
         """Build JVM properties for Platform MCP server (API + VCS access)."""
         props = {
@@ -701,11 +761,7 @@ async def _execute_summarize(
                         "diagramType": "MERMAID" if supports_mermaid else "ASCII"
                     }
             else:
-                return {
-                    "summary": "Failed to generate summary",
-                    "diagram": "",
-                    "diagramType": "MERMAID" if supports_mermaid else "ASCII"
-                }
+                return {"error": "AI service returned an empty summary"}
 
         except Exception as e:
             logger.error(f"Summarize agent error: {e}", exc_info=True)
@@ -821,7 +877,7 @@ async def _execute_ask(
                 else:
                     return {"answer": final_result}
             else:
-                return {"answer": "I couldn't generate an answer. Please try rephrasing your question."}
+                return {"error": "AI service returned an empty answer"}
 
         except Exception as e:
             logger.error(f"Ask agent error: {e}", exc_info=True)

python-ecosystem/inference-orchestrator/tests/test_command_queue_consumer.py

@@ -0,0 +1,119 @@
+import asyncio
+import json
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from server.command_queue_consumer import CommandQueueConsumer
+
+
+class FakeRedis:
+    def __init__(self):
+        self.events = []
+
+    async def lpush(self, key, value):
+        self.events.append((key, json.loads(value)))
+
+
+def _ask_request():
+    return {
+        "projectId": 1,
+        "projectVcsWorkspace": "ws",
+        "projectVcsRepoSlug": "repo",
+        "projectWorkspace": "workspace",
+        "projectNamespace": "namespace",
+        "aiProvider": "OPENAI_COMPATIBLE",
+        "aiModel": "model",
+        "aiApiKey": "key",
+        "question": "describe this PR",
+        "pullRequestId": 7,
+    }
+
+
+def _summarize_request():
+    return {
+        "projectId": 1,
+        "projectVcsWorkspace": "ws",
+        "projectVcsRepoSlug": "repo",
+        "projectWorkspace": "workspace",
+        "projectNamespace": "namespace",
+        "aiProvider": "OPENAI_COMPATIBLE",
+        "aiModel": "model",
+        "aiApiKey": "key",
+        "pullRequestId": 7,
+    }
+
+
+def _payload(command_type, request):
+    return json.dumps({
+        "job_id": f"job-{command_type}",
+        "command_type": command_type,
+        "request": request,
+    })
+
+
+def _consumer(command_service):
+    consumer = CommandQueueConsumer(command_service)
+    consumer._redis = FakeRedis()
+    return consumer
+
+
+async def _handle_and_collect_events(consumer, payload):
+    await consumer._handle_job(payload)
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    return [event for _, event in consumer._redis.events]
+
+
+@pytest.mark.asyncio(loop_scope="function")
+async def test_error_result_is_published_as_error_without_final():
+    command_service = MagicMock()
+    command_service.process_ask = AsyncMock(return_value={"error": "provider failed"})
+    consumer = _consumer(command_service)
+
+    events = await _handle_and_collect_events(consumer, _payload("ask", _ask_request()))
+
+    assert any(event["type"] == "error" and event["message"] == "provider failed" for event in events)
+    assert not any(event["type"] == "final" for event in events)
+
+
+@pytest.mark.asyncio(loop_scope="function")
+async def test_empty_ask_answer_is_published_as_error_without_final():
+    command_service = MagicMock()
+    command_service.process_ask = AsyncMock(return_value={"answer": None})
+    consumer = _consumer(command_service)
+
+    events = await _handle_and_collect_events(consumer, _payload("ask", _ask_request()))
+
+    assert any(
+        event["type"] == "error" and event["message"] == "AI service returned an empty answer"
+        for event in events
+    )
+    assert not any(event["type"] == "final" for event in events)
+
+
+@pytest.mark.asyncio(loop_scope="function")
+async def test_successful_ask_answer_is_published_as_final():
+    command_service = MagicMock()
+    command_service.process_ask = AsyncMock(return_value={"answer": "42"})
+    consumer = _consumer(command_service)
+
+    events = await _handle_and_collect_events(consumer, _payload("ask", _ask_request()))
+
+    assert {"type": "final", "result": {"answer": "42"}} in events
+    assert not any(event["type"] == "error" for event in events)
+
+
+@pytest.mark.asyncio(loop_scope="function")
+async def test_empty_summarize_result_is_published_as_error_without_final():
+    command_service = MagicMock()
+    command_service.process_summarize = AsyncMock(return_value={"summary": "No output generated"})
+    consumer = _consumer(command_service)
+
+    events = await _handle_and_collect_events(consumer, _payload("summarize", _summarize_request()))
+
+    assert any(
+        event["type"] == "error" and event["message"] == "AI service returned an empty summary"
+        for event in events
+    )
+    assert not any(event["type"] == "final" for event in events)

python-ecosystem/inference-orchestrator/tests/test_command_service.py

@@ -310,6 +310,50 @@ def test_with_newlines(self, service):
         assert "line1" in result
 
 
+# -- _normalize_*_result -----------------------------------------
+
+class TestNormalizeSummarizeResult:
+    def test_preserves_provider_error(self, service):
+        result = service._normalize_summarize_result({"error": "provider failed"}, supports_mermaid=False)
+        assert result == {"error": "provider failed"}
+
+    def test_rejects_non_dict_result(self, service):
+        result = service._normalize_summarize_result(None, supports_mermaid=False)
+        assert result == {"error": "AI service returned an invalid summarize result"}
+
+    @pytest.mark.parametrize("summary", [None, "", "   ", "null", "No output generated", "none"])
+    def test_rejects_empty_summary_values(self, service, summary):
+        result = service._normalize_summarize_result({"summary": summary}, supports_mermaid=False)
+        assert result == {"error": "AI service returned an empty summary"}
+
+    def test_defaults_missing_diagram_fields(self, service):
+        result = service._normalize_summarize_result({"summary": "Summary", "diagram": None}, supports_mermaid=False)
+        assert result == {
+            "summary": "Summary",
+            "diagram": "",
+            "diagramType": "ASCII",
+        }
+
+
+class TestNormalizeAskResult:
+    def test_preserves_provider_error(self, service):
+        result = service._normalize_ask_result({"error": "provider failed"})
+        assert result == {"error": "provider failed"}
+
+    def test_rejects_non_dict_result(self, service):
+        result = service._normalize_ask_result(None)
+        assert result == {"error": "AI service returned an invalid ask result"}
+
+    @pytest.mark.parametrize("answer", [None, "", "   ", "null", "No output generated", "none"])
+    def test_rejects_empty_answer_values(self, service, answer):
+        result = service._normalize_ask_result({"answer": answer})
+        assert result == {"error": "AI service returned an empty answer"}
+
+    def test_accepts_answer(self, service):
+        result = service._normalize_ask_result({"answer": "The PR updates auth handling."})
+        assert result == {"answer": "The PR updates auth handling."}
+
+
 # ── _create_mcp_client ───────────────────────────────────────────
 
 class TestCreateMcpClient: