Add fallback response when Gemini returns empty streams

Author: Mateusz

src/core/services/backend_request_manager_service.py

@@ -4,9 +4,10 @@
 This module provides the implementation of the backend request manager interface.
 """
 
-from __future__ import annotations
-
-import logging
+from __future__ import annotations
+
+import logging
+import time
 from collections.abc import AsyncIterator, Iterable
 from typing import Any, cast
 
@@ -356,7 +357,11 @@ async def _process_streaming_response(
                 "Maximum empty stream recovery attempts reached for session %s",
                 session_id,
             )
-            return stream_envelope
+            return await self._build_fallback_stream(
+                stream_envelope=stream_envelope,
+                session_id=session_id,
+                reason="Empty streaming response after maximum retries",
+            )
 
         original_stream = stream_envelope.content
         if original_stream is None:
@@ -499,7 +504,11 @@ async def _retry_stream_with_recovery(
                 reason,
                 session_id,
             )
-            return stream_envelope
+            return await self._build_fallback_stream(
+                stream_envelope=stream_envelope,
+                session_id=session_id,
+                reason=reason,
+            )
 
         logger.info("%s", reason)
         recovery_prompt = self._STREAM_RECOVERY_PROMPT
@@ -621,11 +630,11 @@ def _extract_text_from_chunk(chunk: ProcessedResponse | bytes) -> str:
             return ""
 
         # Handle case where chunk is a ProcessedResponse object
-        if isinstance(chunk, ProcessedResponse):
-            data = chunk.content
-            if isinstance(data, str):
-                return data
-            if isinstance(data, dict):
+        if isinstance(chunk, ProcessedResponse):
+            data = chunk.content
+            if isinstance(data, str):
+                return data
+            if isinstance(data, dict):
                 choices = data.get("choices")
                 if isinstance(choices, list) and choices:
                     choice = choices[0]
@@ -646,9 +655,59 @@ def _extract_text_from_chunk(chunk: ProcessedResponse | bytes) -> str:
                                             fragments_processed.append(text_part)
                                 if fragments_processed:
                                     return "".join(fragments_processed)
-                        message = choice.get("message")
-                        if isinstance(message, dict):
-                            msg_content = message.get("content")
-                            if isinstance(msg_content, str):
-                                return msg_content
-        return ""
+                        message = choice.get("message")
+                        if isinstance(message, dict):
+                            msg_content = message.get("content")
+                            if isinstance(msg_content, str):
+                                return msg_content
+        return ""
+
+    async def _build_fallback_stream(
+        self,
+        stream_envelope: StreamingResponseEnvelope,
+        session_id: str,
+        reason: str,
+    ) -> StreamingResponseEnvelope:
+        """Generate a synthetic assistant response when recovery fails."""
+        logger.warning(
+            "Returning fallback assistant message for session %s: %s",
+            session_id,
+            reason,
+        )
+
+        payload = {
+            "id": "proxy-empty-response-retry",
+            "object": "chat.completion.chunk",
+            "created": int(time.time()),
+            "model": "proxy-empty-response",
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {
+                        "role": "assistant",
+                        "content": (
+                            "Proxy notice: the upstream model returned no content "
+                            "after multiple attempts. Please retry or adjust your request."
+                        ),
+                    },
+                    "finish_reason": "stop",
+                }
+            ],
+        }
+
+        async def fallback_stream() -> AsyncIterator[ProcessedResponse]:
+            yield ProcessedResponse(
+                content=payload,
+                metadata={
+                    "proxy_generated": True,
+                    "empty_response_recovery_failed": True,
+                    "recovery_reason": reason,
+                },
+            )
+
+        return StreamingResponseEnvelope(
+            content=fallback_stream(),
+            media_type=stream_envelope.media_type,
+            headers=stream_envelope.headers,
+            cancel_callback=stream_envelope.cancel_callback,
+        )