fix(llm-error): retry streamed 200-OK responses that carry no generation chunk (#6803)

* fix(llm-error): retry streamed 200-OK responses that carry no generation chunk

The non-streaming empty-generations glitch (IndexError) is already
classified as transient and retried. The streaming path reports the same
upstream glitch as ValueError("No generations found in stream.") from
langchain_core's generate_from_stream; it fell through to "generic" and
failed after one attempt regardless of llm_call.retry_max_attempts.

Classify that ValueError as transient so it uses the configured retry
budget. Unrelated ValueErrors stay generic.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* style(llm-error): keep comments adjacent to their constants

---------

Co-authored-by: huangzz <110116@nd.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py

@@ -165,6 +165,10 @@ _REASON_RETRY_BUDGETS: dict[str, int] = {
     "burst_rate": 2,
 }
 
+# Lower-cased message of the ``ValueError`` langchain_core raises when a
+# streamed response (HTTP 200) yields no generation chunk at all.
+_EMPTY_STREAM_MARKER = "no generations found in stream"
+
 # Exception class names that indicate the upstream stream-chunk watchdog
 # fired because the model stalled mid-flight. These deserve a more specific
 # user-facing message than the generic "temporarily unavailable" copy,
@@ -637,6 +641,13 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
         # provider failures rather than failing the whole run.
         if isinstance(exc, IndexError):
             return True, "transient"
+        # The streaming counterpart of the empty-``generations`` glitch above:
+        # ``langchain_core.language_models.chat_models.generate_from_stream``
+        # raises ``ValueError("No generations found in stream.")`` when the
+        # provider answers ``200 OK`` but the stream carries no chunk at all.
+        # Same upstream-payload glitch, same retry/backoff path.
+        if isinstance(exc, ValueError) and _EMPTY_STREAM_MARKER in lowered:
+            return True, "transient"
         if status_code in _RETRIABLE_STATUS_CODES:
             return True, "transient"
         if _matches_any(lowered, _BUSY_PATTERNS):

backend/tests/test_llm_error_handling_middleware.py

@@ -1669,6 +1669,75 @@ def test_async_index_error_exhausted_returns_user_fallback(
     assert "temporarily unavailable" in str(result.content)
 
 
+def test_classify_error_empty_stream_value_error_is_retriable_transient() -> None:
+    """The streaming path reports an HTTP-200-but-empty response as
+    ``ValueError("No generations found in stream.")`` (raised by
+    ``langchain_core...generate_from_stream``). It is the same upstream glitch
+    as the non-streaming empty-``generations`` IndexError and must be retried.
+    """
+    from langchain_core.language_models.chat_models import generate_from_stream
+
+    middleware = _build_middleware()
+    with pytest.raises(ValueError) as excinfo:
+        generate_from_stream(iter(()))
+    retriable, reason = middleware._classify_error(excinfo.value)
+    assert retriable is True
+    assert reason == "transient"
+
+
+def test_classify_error_unrelated_value_error_stays_generic() -> None:
+    middleware = _build_middleware()
+    retriable, reason = middleware._classify_error(ValueError("invalid literal for int()"))
+    assert retriable is False
+    assert reason == "generic"
+
+
+def test_async_empty_stream_retries_with_configured_budget_then_succeeds(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=10, retry_cap_delay_ms=10)
+    attempts = 0
+
+    async def fake_sleep(_delay: float) -> None:
+        return None
+
+    async def handler(_request) -> AIMessage:
+        nonlocal attempts
+        attempts += 1
+        if attempts < 3:
+            raise ValueError("No generations found in stream.")
+        return AIMessage(content="ok")
+
+    monkeypatch.setattr("asyncio.sleep", fake_sleep)
+
+    result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler))
+
+    assert isinstance(result, AIMessage)
+    assert result.content == "ok"
+    assert attempts == 3
+
+
+def test_sync_empty_stream_exhausted_uses_configured_attempts(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=10, retry_cap_delay_ms=10)
+    attempts = 0
+
+    def handler(_request) -> AIMessage:
+        nonlocal attempts
+        attempts += 1
+        raise ValueError("No generations found in stream.")
+
+    monkeypatch.setattr("time.sleep", lambda _delay: None)
+
+    result = middleware.wrap_model_call(SimpleNamespace(), handler)
+
+    assert attempts == 2
+    assert isinstance(result, AIMessage)
+    assert result.additional_kwargs["deerflow_error_fallback"] is True
+    assert result.additional_kwargs["error_reason"] == "transient"
+
+
 # ---------- Process-wide concurrency limiter ----------