fix: retry on empty SSE stream to handle premature load balancer disconnects

Author: Bharatram-altimate-ai

packages/opencode/src/skill/validate/batch_validate.py

@@ -120,26 +120,47 @@ def _parse_sse_stream(response):
 # ---------------------------------------------------------------------------
 # Backend API calls
 # ---------------------------------------------------------------------------
+_EMPTY_STREAM_RETRIES = 3
+_EMPTY_STREAM_BACKOFF = 2  # seconds
+
+
+def _stream_post(url, payload, timeout):
+    """POST to url, stream SSE response. Returns list of trace_result/trace_error events."""
+    import time
+    for attempt in range(1, _EMPTY_STREAM_RETRIES + 1):
+        results = []
+        try:
+            resp = _SESSION.post(url, json=payload, headers=_HEADERS, stream=True, timeout=timeout)
+            resp.raise_for_status()
+            for event in _parse_sse_stream(resp):
+                _log_event(event)
+                if event.get("event") in ("trace_result", "trace_error"):
+                    results.append(event)
+        except Exception as e:
+            print(f"  ERROR: {e}", file=sys.stderr)
+            return results
+
+        if results:
+            return results
+
+        if attempt < _EMPTY_STREAM_RETRIES:
+            print(
+                f"  Warning: stream returned no events (attempt {attempt}/{_EMPTY_STREAM_RETRIES}), "
+                f"retrying in {_EMPTY_STREAM_BACKOFF}s...",
+                file=sys.stderr,
+            )
+            time.sleep(_EMPTY_STREAM_BACKOFF)
+
+    print(f"  Warning: stream returned no events after {_EMPTY_STREAM_RETRIES} attempts.", file=sys.stderr)
+    return []
+
+
 def validate_single_trace(trace_id):
     """Call POST /validate for a single trace. Returns list of result dicts."""
     print(f"Validating trace: {trace_id}...", file=sys.stderr)
-    results = []
-    try:
-        resp = _SESSION.post(
-            f"{BASE_URL}/validate",
-            json={"trace_id": trace_id},
-            headers=_HEADERS,
-            stream=True,
-            timeout=300,
-        )
-        resp.raise_for_status()
-        for event in _parse_sse_stream(resp):
-            _log_event(event)
-            if event.get("event") in ("trace_result", "trace_error"):
-                results.append(event)
-    except Exception as e:
-        print(f"  ERROR: {e}", file=sys.stderr)
-        results.append({"event": "trace_error", "trace_id": trace_id, "error": str(e)})
+    results = _stream_post(f"{BASE_URL}/validate", {"trace_id": trace_id}, timeout=300)
+    if not results:
+        results.append({"event": "trace_error", "trace_id": trace_id, "error": "Stream returned no events"})
     return results
 
 
@@ -149,49 +170,17 @@ def validate_date_range(user_id, from_datetime, to_datetime):
         f"Validating traces for user '{user_id}' from {from_datetime} to {to_datetime}...",
         file=sys.stderr,
     )
-    results = []
-    try:
-        resp = _SESSION.post(
-            f"{BASE_URL}/validate/date-range",
-            json={
-                "user_id": user_id,
-                "from_datetime": from_datetime,
-                "to_datetime": to_datetime,
-            },
-            headers=_HEADERS,
-            stream=True,
-            timeout=600,
-        )
-        resp.raise_for_status()
-        for event in _parse_sse_stream(resp):
-            _log_event(event)
-            if event.get("event") in ("trace_result", "trace_error"):
-                results.append(event)
-    except Exception as e:
-        print(f"  ERROR: {e}", file=sys.stderr)
-    return results
+    return _stream_post(
+        f"{BASE_URL}/validate/date-range",
+        {"user_id": user_id, "from_datetime": from_datetime, "to_datetime": to_datetime},
+        timeout=600,
+    )
 
 
 def validate_session(session_id):
     """Call POST /validate/session. Returns list of result dicts."""
     print(f"Validating all traces in session: {session_id}...", file=sys.stderr)
-    results = []
-    try:
-        resp = _SESSION.post(
-            f"{BASE_URL}/validate/session",
-            json={"session_id": session_id},
-            headers=_HEADERS,
-            stream=True,
-            timeout=600,
-        )
-        resp.raise_for_status()
-        for event in _parse_sse_stream(resp):
-            _log_event(event)
-            if event.get("event") in ("trace_result", "trace_error"):
-                results.append(event)
-    except Exception as e:
-        print(f"  ERROR: {e}", file=sys.stderr)
-    return results
+    return _stream_post(f"{BASE_URL}/validate/session", {"session_id": session_id}, timeout=600)
 
 
 def _log_event(event):