fix(streaming): surface upstream connection resets instead of empty 200 streams

[mateo-berri]

Relevant issues

Linear ticket

Resolves LIT-4450

Pre-Submission checklist

Please complete all items before asking a LiteLLM maintainer to review your PR

• I have added meaningful tests
• My PR passes all CI/CD checks (e.g., lint, format, unit tests)
• My PR's scope is as isolated as possible; it only solves 1 specific problem
• I have received a Greptile Confidence Score of at least 4/5 before requesting a maintainer review (Greptile reviews automatically once the PR is opened; only comment @greptileai to re-request a review after pushing changes)

Delays in PR merge?

If you're seeing a delay in your PR being merged, ping the LiteLLM Team on Slack (#pr-review).

Screenshots / Proof of Fix

A customer replaying a captured streaming chat/completions request through the proxy saw intermittent HTTP 200 responses whose SSE body carried no content, no tool calls, and no reasoning, just finish_reason: "stop" (66 out of 1000 attempts), while the same request replayed directly against the downstream endpoint never produced one. Datadog tracing on the affected requests showed Connection reset by peer errors on the upstream call inside spans that still returned 200

This cannot be reproduced on demand against a real provider because it needs the upstream to kill the TCP connection mid-response, so the proof uses a realistic e2e test harness around a live proxy: a fault-injecting OpenAI-compatible downstream (fault_downstream.py) that streams a role delta plus N content deltas as chunked SSE and then kills the connection without ever sending a finish chunk, and a probe (probe_litellm_stream.py) that classifies the SSE body the proxy hands back. Two fault shapes are exercised, a hard RST (SO_LINGER 0) and a half-close FIN that omits the terminal chunk; curl pointed directly at the faulting downstream reports the failure plainly (curl: (18) transfer closed with outstanding read data remaining), so anything other than an explicit error out of the proxy is data loss. Everything from the proxy inward is the real production code path: aiohttp transport, openai provider, streaming wrapper. The harness and config are in the details block below

Before, captured on litellm v1.93.0rc1 from PyPI (the swallow also reproduces identically at this branch's base, b200d66): the probe classifies the proxy response as true_empty_stop, the exact shape from the report above. In the FIN shape all 50 real content deltas are consumed and replaced with a single fabricated empty stop plus [DONE]; the RST shape produces the same class

$ python fault_downstream.py --port 9317
$ litellm --config litellm_config.yaml --port 9463
$ python probe_litellm_stream.py --base-url http://127.0.0.1:9463/v1 --fault-fin
{
  "model": "fault-reset-after-partial-200",
  "fault_chunk_count": 50,
  "status": 200,
  "class": "true_empty_stop"
}
--- raw stream ---
data: {"id":"chatcmpl-ec5da108-5d4e-47b9-977a-ab739dc2d471","object":"chat.completion.chunk","created":1784124097,"model":"fault-reset-after-partial-200","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

After, same harness against a proxy running this branch at 26a7f6d (94c579a only merges the base branch in; re-ran the full harness at the branch tip e14da65 with identical results): a connection killed before the first delta reaches the client fails the request outright with a retriable 500

$ python probe_litellm_stream.py --base-url http://127.0.0.1:9464/v1
{
  "model": "fault-reset-after-partial-200",
  "fault_chunk_count": 0,
  "status": 500,
  "class": "http_500"
}
--- raw stream ---
{"error":{"message":"litellm.APIConnectionError: APIConnectionError: OpenAIException - Response payload is not completed: <TransferEncodingError: 400, message='Not enough data to satisfy transfer length header.'>. ConnectionResetError(54, 'Connection reset by peer')","type":null,"param":null,"code":"500"}}

and a connection killed mid-stream (20 paced deltas already forwarded to the client) delivers every real delta and then surfaces the failure as an SSE error chunk, with no fabricated finish_reason: "stop" and no [DONE]

$ curl -sN -w "\nHTTP_STATUS: %{http_code}\n" http://127.0.0.1:9464/v1/chat/completions \
    -H "Content-Type: application/json" \
    -d '{"model":"fault-reset-after-partial-200","stream":true,"fault_chunk_count":20,"fault_delay_ms":50,"fault_fin":true,"messages":[{"role":"user","content":"hi"}]}'
data: {..."delta":{"role":"assistant"}}]}
data: {..."delta":{"content":"chunk-0 "}}]}
...
data: {..."delta":{"content":"chunk-19 "}}]}
data: {"error": {"message": "litellm.APIConnectionError: APIConnectionError: OpenAIException - Response payload is not completed: <TransferEncodingError: 400, message='Not enough data to satisfy transfer length header.'>", "type": null, "param": null, "code": "500"}}

HTTP_STATUS: 200

At the router level the failure stays a single upstream call: streaming router.acompletion through the same fault yields all 50 real deltas and then raises litellm.APIConnectionError, and the downstream's request log shows exactly one request for the run, so surfacing the error introduces no hidden retries or duplicate upstream calls

Harness: fault-injecting downstream, probe, proxy config

litellm_config.yaml:

model_list:
  - model_name: fault-reset-after-partial-200
    litellm_params:
      model: fault-reset-after-partial-200
      api_base: http://127.0.0.1:9317/v1
      api_key: dummy
      custom_llm_provider: openai
    model_info:
      mode: chat
      id: fault-reset-after-partial-200

litellm_settings:
  drop_params: true
  request_timeout: 600
  set_verbose: true

fault_downstream.py (fault shape per request body: fault_chunk_count deltas, fault_delay_ms pacing, fault_fin for FIN instead of RST):

#!/usr/bin/env python3
from __future__ import annotations

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse
import argparse
import json
import socket
import struct
import time
import uuid


MODEL_RESET = "fault-reset-after-partial-200"


def now() -> int:
    return int(time.time())


def json_bytes(payload: dict) -> bytes:
    return json.dumps(payload, separators=(",", ":")).encode("utf-8")


def int_body(request_json: dict, name: str, default: int) -> int:
    candidates = [
        request_json.get(name),
        (request_json.get("metadata") or {}).get(name)
        if isinstance(request_json.get("metadata"), dict)
        else None,
    ]
    for value in candidates:
        try:
            if value is not None:
                return int(value)
        except (TypeError, ValueError):
            continue
    return default


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    server_version = "fault-downstream/0.1"

    def do_GET(self) -> None:
        if self.path in {"/", "/healthz", "/readyz"}:
            payload = json_bytes(
                {
                    "ok": True,
                    "models": [MODEL_RESET],
                    "ts": time.time(),
                }
            )
            self.send_response(200)
            self.send_header("content-type", "application/json")
            self.send_header("content-length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)
            self.wfile.flush()
            return
        self.send_error(404, "not found")

    def do_POST(self) -> None:
        parsed = urlparse(self.path)
        request_id = self.headers.get("x-request-id") or uuid.uuid4().hex[:12]
        body = self._read_body()
        request_json = self._parse_json(body)
        model = request_json.get("model")
        stream = bool(request_json.get("stream"))

        self._log(
            {
                "event": "request",
                "request_id": request_id,
                "path": parsed.path,
                "model": model,
                "stream": stream,
                "bytes": len(body),
            }
        )

        if parsed.path != "/v1/chat/completions":
            self._send_json(404, {"error": {"message": "unsupported path"}}, request_id)
            return

        if not stream:
            self._send_json(
                400,
                {"error": {"message": "this repro expects stream=true"}},
                request_id,
            )
            return

        if model == MODEL_RESET:
            self._send_chunks_then_reset(request_json, request_id)
            return

        self._send_json(
            400,
            {"error": {"message": f"unknown model: {model}"}},
            request_id,
        )

    def _read_body(self) -> bytes:
        length = int(self.headers.get("content-length") or "0")
        return self.rfile.read(length) if length else b""

    def _parse_json(self, body: bytes) -> dict:
        try:
            parsed = json.loads(body.decode("utf-8"))
            return parsed if isinstance(parsed, dict) else {}
        except Exception:
            return {}

    def _send_chunks_then_reset(self, request_json: dict, request_id: str) -> None:
        self._start_sse(200, request_id, "reset_after_partial_200")
        self._write_sse(
            {
                "id": f"chatcmpl-{request_id}",
                "object": "chat.completion.chunk",
                "created": now(),
                "model": MODEL_RESET,
                "choices": [
                    {
                        "index": 0,
                        "delta": {"role": "assistant"},
                        "finish_reason": None,
                    }
                ],
            }
        )
        chunk_count = max(0, min(int_body(request_json, "fault_chunk_count", 0), 1000))
        delay_ms = max(0, min(int_body(request_json, "fault_delay_ms", 0), 5000))
        for index in range(chunk_count):
            if delay_ms:
                time.sleep(delay_ms / 1000)
            self._write_sse(
                {
                    "id": f"chatcmpl-{request_id}",
                    "object": "chat.completion.chunk",
                    "created": now(),
                    "model": MODEL_RESET,
                    "choices": [
                        {
                            "index": 0,
                            "delta": {"content": f"chunk-{index} "},
                            "finish_reason": None,
                        }
                    ],
                }
            )
        fin_mode = bool(request_json.get("fault_fin"))
        self._log(
            {
                "event": "fin_after_chunks" if fin_mode else "reset_after_chunks",
                "request_id": request_id,
                "content_chunks": chunk_count,
            }
        )
        if fin_mode:
            self.close_connection = True
            self.connection.shutdown(socket.SHUT_WR)
            return
        self._reset_connection(request_id)

    def _send_json(self, status: int, payload: dict, request_id: str) -> None:
        raw = json_bytes(payload)
        self.send_response(status)
        self.send_header("content-type", "application/json")
        self.send_header("content-length", str(len(raw)))
        self.send_header("x-fault-request-id", request_id)
        self.end_headers()
        self.wfile.write(raw)
        self.wfile.flush()

    def _start_sse(self, status: int, request_id: str, fault_mode: str) -> None:
        self.send_response(status)
        self.send_header("content-type", "text/event-stream")
        self.send_header("cache-control", "no-cache")
        self.send_header("transfer-encoding", "chunked")
        self.send_header("x-fault-request-id", request_id)
        self.send_header("x-fault-mode", fault_mode)
        self.end_headers()

    def _write_sse(self, payload: dict) -> None:
        self._write_chunk(
            ("data: " + json.dumps(payload, separators=(",", ":")) + "\n\n").encode(
                "utf-8"
            )
        )

    def _write_chunk(self, payload: bytes) -> None:
        self.wfile.write(("%x\r\n" % len(payload)).encode("ascii"))
        self.wfile.write(payload)
        self.wfile.write(b"\r\n")
        self.wfile.flush()

    def _reset_connection(self, request_id: str) -> None:
        self._log({"event": "tcp_reset", "request_id": request_id})
        try:
            linger = struct.pack("ii", 1, 0)
            self.connection.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, linger)
        finally:
            self.close_connection = True
            self.connection.close()

    def _log(self, payload: dict) -> None:
        payload = {"ts": time.time(), **payload}
        print(json.dumps(payload, sort_keys=True), flush=True)

    def log_message(self, fmt: str, *args) -> None:
        self._log({"event": "access", "message": fmt % args})


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8080)
    args = parser.parse_args()

    server = ThreadingHTTPServer((args.host, args.port), Handler)
    print(
        json.dumps(
            {
                "event": "startup",
                "host": args.host,
                "port": args.port,
                "models": [MODEL_RESET],
            },
            sort_keys=True,
        ),
        flush=True,
    )
    server.serve_forever()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

probe_litellm_stream.py:

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
import time
import urllib.error
import urllib.request
from typing import Any


def sse_payloads(raw: str) -> list[dict[str, Any]]:
    payloads: list[dict[str, Any]] = []
    lines: list[str] = []
    for raw_line in raw.splitlines():
        line = raw_line.rstrip("\r")
        if line.startswith("data:"):
            lines.append(line.split("data:", 1)[1].strip())
            continue
        if line == "" and lines:
            payload = "\n".join(lines)
            lines = []
            if payload == "[DONE]":
                continue
            try:
                parsed = json.loads(payload)
            except json.JSONDecodeError:
                continue
            if isinstance(parsed, dict):
                payloads.append(parsed)
    return payloads


def classify(status: int | None, raw: str, error: str | None) -> str:
    if error:
        return "transport_error"
    if status is not None and status >= 400:
        return f"http_{status}"

    events = sse_payloads(raw)
    deltas: list[dict[str, Any]] = []
    finish_reasons: list[Any] = []
    for event in events:
        for choice in event.get("choices") or []:
            if not isinstance(choice, dict):
                continue
            delta = choice.get("delta")
            if isinstance(delta, dict):
                deltas.append(delta)
            if "finish_reason" in choice:
                finish_reasons.append(choice.get("finish_reason"))

    has_role = any(delta.get("role") for delta in deltas)
    has_content = any(delta.get("content") for delta in deltas)
    has_tool_calls = any(delta.get("tool_calls") for delta in deltas)
    has_empty_stop = any(
        delta == {} and reason == "stop"
        for delta, reason in zip(deltas, finish_reasons, strict=False)
    )

    if has_empty_stop:
        if has_content or has_role or has_tool_calls:
            return "partial_stream_then_synthetic_stop"
        return "true_empty_stop"
    if has_content:
        if '"error"' in raw:
            return "content_then_stream_error"
        return "content_then_incomplete_or_done"
    if has_role:
        if '"error"' in raw:
            return "role_only_then_stream_error"
        return "role_only_then_incomplete_or_done"
    return "other"


def call_once(
    base_url: str,
    model: str,
    timeout: float,
    fault_chunk_count: int,
    fault_fin: bool,
) -> dict[str, Any]:
    url = base_url.rstrip("/") + "/chat/completions"
    body = {
        "model": model,
        "messages": [{"role": "user", "content": "hi"}],
        "stream": True,
        "metadata": {"fault_chunk_count": fault_chunk_count},
        "fault_chunk_count": fault_chunk_count,
        "fault_fin": fault_fin,
    }
    started = time.time()
    headers = {"Content-Type": "application/json"}
    probe_key = os.environ.get("LITELLM_PROBE_KEY")
    if probe_key:
        headers["Authorization"] = f"Bearer {probe_key}"
    request = urllib.request.Request(
        url,
        data=json.dumps(body).encode("utf-8"),
        headers=headers,
        method="POST",
    )
    status: int | None = None
    raw = ""
    error = None
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            status = response.status
            raw = response.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as exc:
        status = exc.code
        raw = exc.read().decode("utf-8", errors="replace")
    except Exception as exc:  # noqa: BLE001 - repro script should report all failures.
        error = repr(exc)

    return {
        "model": model,
        "fault_chunk_count": fault_chunk_count,
        "status": status,
        "class": classify(status, raw, error),
        "elapsed_sec": round(time.time() - started, 3),
        "error": error,
        "raw": raw,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--base-url", default="http://127.0.0.1:4000/v1")
    parser.add_argument("--timeout", type=float, default=10)
    parser.add_argument("--model", action="append")
    parser.add_argument(
        "--fault-chunk-count",
        type=int,
        action="append",
        help="Number of content chunks the downstream sends before resetting.",
    )
    parser.add_argument("--fault-fin", action="store_true")
    args = parser.parse_args()

    models = args.model or ["fault-reset-after-partial-200"]
    chunk_counts = args.fault_chunk_count or [0, 50]
    for model in models:
        for chunk_count in chunk_counts:
            result = call_once(args.base_url, model, args.timeout, chunk_count, args.fault_fin)
            print("=" * 80)
            print(
                json.dumps({k: v for k, v in result.items() if k != "raw"}, indent=2)
            )
            print("--- raw stream ---")
            print(result["raw"])

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Type

🐛 Bug Fix

Changes

AiohttpResponseStream.__aiter__ in the aiohttp transport caught aiohttp.ClientPayloadError, RuntimeError("Connection closed."), and TransferEncodingError during body iteration and returned as if the body had completed normally. For a streaming request this means an upstream TCP reset mid-response looks like a clean end of stream, so CustomStreamWrapper fabricates a final chunk with finish_reason: "stop" (plus a usage chunk when stream_options.include_usage is set) and the client receives a well-formed but empty or truncated HTTP 200 stream with no indication anything failed. No failure is logged and no router retry or fallback ever triggers. The transport now maps all three to httpx.ReadError (chaining the original exception) instead of swallowing them; the connection-release finally from #30192 is unchanged

Those swallows were originally added (#11162, #11561, #16294) because some providers close or reset the connection sloppily after delivering a complete response, and erroring those streams at the tail would be a regression. That concern is handled at the layer that actually knows whether the response was complete: CustomStreamWrapper.__anext__ now treats httpx.ReadError / httpx.RemoteProtocolError as a graceful end of stream when the provider's finish_reason has already been received, and as a real failure (failure logging plus MidStreamFallbackError, so router retries and fallbacks engage) when it has not. Streams that end with [DONE] never read past it, so they are unaffected either way

The Responses API streaming iterator is the one other consumer that reads its SSE stream all the way to EOF (its terminal signal is a response.completed event, not [DONE]), so it relied on the same transport swallow when a provider closes with broken framing after the final event. It now applies the same rule as the wrapper, in both the async and sync iterators: httpx.ReadError / httpx.RemoteProtocolError after a terminal event (response.completed, response.incomplete, response.failed) ends the stream cleanly, and before one it remains a real failure (df60e36)

To let both the end-of-stream path and the new post-finish path share one implementation, the body of the StopAsyncIteration handler in __anext__ moved verbatim into _finalize_completed_stream, and the duplicated failure-logging block into _log_stream_failure_and_raise. No behavior change for existing paths

Tests: the two transport tests that pinned the old swallowing behavior are inverted into regression tests asserting httpx.ReadError is raised (mid-stream, before the first chunk, on RuntimeError("Connection closed."), and on raw TransferEncodingError, with the response still closed and unrelated RuntimeErrors untouched), and two new CustomStreamWrapper tests pin the split: a transport read error after the finish chunk completes the stream cleanly with the provider's own finish_reason, and one before any finish chunk raises MidStreamFallbackError with no fabricated finish_reason ever emitted. Six responses-iterator tests pin the same split on that surface: clean end after response.completed for both ReadError and RemoteProtocolError, a raise before any terminal event, each for the async and sync iterators

One unrelated commit to keep CI deterministic: base commit 4a81ec4 added mcp_rpm_limit to UpdateTeamRequest without documenting it in update_team's docstring, and the api-docs check only catches that when directory walk order lets team_endpoints.py shadow the legacy update_team shim in model_management_endpoints.py, so the code-quality job failed on a coin flip per run. e14da65 adds the missing docstring line plus the dashboard schema types regenerated from it

Final Attestation

• The tests check the right things, including the edge cases, and regressions in the respective real-world customer use-cases are not possible after this PR

[codecov]

Codecov Report

❌ Patch coverage is 96.77419% with 2 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
litellm/litellm_core_utils/streaming_handler.py	95.65%	2 Missing ⚠️

📢 Thoughts on this report? Let us know!

[greptile-apps]

Greptile Summary

This PR fixes a silent data-loss bug where upstream TCP connection resets during streaming were swallowed by the aiohttp transport layer, causing CustomStreamWrapper to fabricate a clean finish_reason: "stop" response rather than surfacing the failure.

• Transport layer fix (aiohttp_transport.py): aiohttp.ClientPayloadError, RuntimeError("Connection closed."), and TransferEncodingError previously returned silently from AiohttpResponseStream.__aiter__, making connection resets look like clean end-of-stream. They now map to httpx.ReadError (chaining the original exception), allowing the caller to distinguish genuine transport failures from normal stream completion.
• Wrapper layer fix (streaming_handler.py): CustomStreamWrapper.__anext__ now catches httpx.ReadError/RemoteProtocolError and applies a split: if the provider's finish_reason was already received (received_finish_reason is not None), the error is treated as sloppy-close and the stream ends cleanly via the extracted _finalize_completed_stream helper; if the finish chunk has not yet been received, it logs the failure and raises MidStreamFallbackError so router retries and fallbacks engage. The same split is applied symmetrically in both async/sync ResponsesAPIStreamingIterator paths (terminal event = response.completed/incomplete/failed).

Confidence Score: 5/5

Safe to merge. The change targets a well-isolated bug path (aiohttp transport swallowing connection resets), the fix is applied at two distinct layers with a clear discriminant for post-finish vs mid-stream errors, and every modified behavior is covered by new or inverted regression tests.

The root-cause fix in the transport layer is minimal and surgical. The wrapper-layer split between post-finish and mid-stream errors relies on received_finish_reason which is set atomically inside chunk_creator, making the invariant sound. All three previously-silenced error shapes are now covered by explicit tests asserting httpx.ReadError, and the two new CustomStreamWrapper tests pin both sides of the split. The ResponsesAPIStreamingIterator changes apply the same pattern symmetrically for async and sync paths.

No files require special attention. The core logic is concentrated in streaming_handler.py and aiohttp_transport.py, both of which are well-tested by the accompanying test changes.

Important Files Changed

Filename	Overview
litellm/llms/custom_httpx/aiohttp_transport.py	Removes the three silent-swallow clauses (ClientPayloadError, RuntimeError('Connection closed.'), TransferEncodingError) and maps them to httpx.ReadError instead; ClientPayloadError is now handled by the existing map_aiohttp_exceptions() context manager which already maps it to httpx.ReadError. The finally connection-release block is unchanged.
litellm/litellm_core_utils/streaming_handler.py	Adds except (httpx.ReadError, httpx.RemoteProtocolError) handler to anext with a post-finish vs mid-stream split; extracts the StopAsyncIteration cleanup block into _finalize_completed_stream and the failure-logging block into _log_stream_failure_and_raise with no behavioral change to existing paths.
litellm/responses/streaming_iterator.py	Applies the same post-finish / mid-stream split to both ResponsesAPIStreamingIterator.anext and SyncResponsesAPIStreamingIterator.next, using self.completed_response is None as the discriminant for whether a terminal event was received.
tests/test_litellm/llms/custom_httpx/test_aiohttp_transport.py	Inverts two previously-swallow tests into regression guards asserting httpx.ReadError is raised; adds three new tests covering RuntimeError('Connection closed.'), unrelated RuntimeError pass-through, and raw TransferEncodingError. Also fixes a 0 or default falsy-zero bug in MockContent.exception_at_chunk.
tests/test_litellm/litellm_core_utils/test_streaming_handler.py	Adds two new wrapper-level tests: ReadError after the provider finish chunk ends the stream cleanly (correct finish_reason delivered, no fabricated stop), and ReadError before any finish chunk raises MidStreamFallbackError with no fabricated finish_reason.
tests/test_litellm/responses/test_streaming_iterator.py	Adds six new tests covering ReadError/RemoteProtocolError after a response.completed event (clean end) and before any terminal event (raises) for both async and sync iterators; refactors the helper to accept an optional trailing error.
litellm/proxy/management_endpoints/team_endpoints.py	Adds the missing mcp_rpm_limit docstring line to update_team, fixing a non-deterministic CI failure in the api-docs check.
ui/litellm-dashboard/src/lib/http/schema.d.ts	Auto-generated schema type updated to include the mcp_rpm_limit parameter description in the update_team endpoint definition.

_{Reviews (5): Last reviewed commit: "docs(proxy): document mcp_rpm_limit in u..." | Re-trigger Greptile}

[codspeed-hq]

Merging this PR will not alter performance

✅ 31 untouched benchmarks

---

_{Comparing litellm_fix_stream_reset_empty_200 (40aeb33) with litellm_internal_staging (582907d)¹}

Footnotes

1. No successful run was found on litellm_internal_staging (ebdf0bb) during the generation of this report, so 582907d was used instead as the comparison base. There might be some changes unrelated to this pull request in this report. ↩

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai