How to Debug Streaming Failures in Claude, GPT, and Gemini (SSE Drops, Partial Output, Stalls)

Streaming bugs are the hardest LLM bugs to reproduce. The request returns 200, the first words arrive, and then the answer cuts off mid-sentence, hangs forever, or ends cleanly with half the JSON missing. This post gives you a method: find out which of three failure shapes you have, reproduce it outside your stack, add the instrumentation that tells drops from stalls from limits, and fix it at the right layer for Claude, GPT, and Gemini.

The three failure shapes

"Streaming is broken" covers at least three different problems with different fixes. Name the one you have before changing any code.

A fourth case hides inside the first. Because the HTTP status is sent before generation starts, a provider failure partway through arrives as an event in the body, not as a status code. Code that only checks response.status_code never sees it.

What each provider's stream looks like

Debugging starts with knowing what a healthy stream ends with.

Gemini is the one that bites people, because with no [DONE] equivalent a connection that dies after a chunk full of text looks exactly like a finished answer unless you check for finishReason. If you use the OpenAI Responses API instead of chat completions, look for the typed terminal events (response.completed, response.incomplete, response.failed) rather than [DONE].

Step 1: Reproduce with curl, outside your stack

Before reading your own code, take your code out of the picture. -N disables curl's output buffering so you see events as they arrive, and -D - prints the response headers, which include the request ID you will need if you contact support.

# Claude
curl -N -sS -D - https://api.anthropic.com/v1/messages \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -H "content-type: application/json" \
  -d '{"model":"'"$MODEL"'","max_tokens":256,"stream":true,
       "messages":[{"role":"user","content":"Count from 1 to 20, one number per line."}]}'

# OpenAI (chat completions)
curl -N -sS -D - https://api.openai.com/v1/chat/completions \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "content-type: application/json" \
  -d '{"model":"'"$MODEL"'","stream":true,"stream_options":{"include_usage":true},
       "messages":[{"role":"user","content":"Count from 1 to 20, one number per line."}]}'

# Gemini
curl -N -sS -D - "https://generativelanguage.googleapis.com/v1beta/models/$MODEL:streamGenerateContent?alt=sse" \
  -H "x-goog-api-key: $GEMINI_API_KEY" \
  -H "content-type: application/json" \
  -d '{"contents":[{"role":"user","parts":[{"text":"Count from 1 to 20, one number per line."}]}]}'

Set MODEL to the model you actually use. Then read the result:

• Events arrive one by one and finish with the terminal marker. The provider is fine. Repeat the request against your own streaming endpoint with the same curl -N. If that one drops or arrives in a single burst at the end, the bug is in your infrastructure or relay code.
• Everything arrives in one burst after a long wait. Something is buffering. This is the classic reverse proxy or compression problem covered in Step 2.
• The stream fails here too. Check the provider status page, save the request ID from the headers, and move on to the mid-stream error handling in Step 4.

Step 2: Find the layer that is killing it

Between your user and the model there can be six hops, and each one has its own idea of how long an idle connection may live or whether a response may be buffered. Work from the outside in.

The tell for an infrastructure timeout is a failure at a suspiciously round number: exactly 30, 60, or 100 seconds from the start or from the last byte. Put the elapsed time in your logs and look for clusters.

Step 3: Parse the stream strictly

Many SDKs hide the raw stream and quietly end the iterator when the connection closes. To tell a drop from a normal finish you need a parser that knows the difference. The one below follows the SSE rules: events end at a blank line, lines starting with a colon are comments, multiple data: lines join with newlines, and a half-finished event at the end of the connection is an error rather than silently discarded.

from dataclasses import dataclass
from typing import AsyncIterator, Optional


class TruncatedStream(Exception):
    pass


@dataclass
class SSEEvent:
    event: str
    data: str
    id: Optional[str] = None


async def parse_sse(lines: AsyncIterator[str]) -> AsyncIterator[SSEEvent]:
    event, data, last_id = "message", [], None
    async for raw in lines:
        line = raw.rstrip("\r\n")
        if line == "":                       # blank line ends one event
            if data:
                yield SSEEvent(event, "\n".join(data), last_id)
            event, data = "message", []
            continue
        if line.startswith(":"):             # comment / keepalive, not an event
            continue
        field, _, value = line.partition(":")
        value = value.removeprefix(" ")
        if field == "event":
            event = value
        elif field == "data":
            data.append(value)
        elif field == "id":
            last_id = value
    if data:                                 # connection ended mid-event
        raise TruncatedStream("stream ended in the middle of an event")

Step 4: Track state, including errors inside the stream

The tracker reduces each provider's events to the same few facts: the text so far, the stop reason, whether the terminal marker arrived, and whether an error event came through. It is the only code that knows provider-specific event names, so everything downstream stays uniform.

import json


class StreamTracker:
    def __init__(self, provider: str):
        self.provider = provider
        self.text: list[str] = []
        self.stop: Optional[str] = None
        self.done = False                    # terminal marker seen
        self.error: Optional[dict] = None    # error delivered inside the stream
        self.events = 0
        self.last_event: Optional[str] = None

    def feed(self, ev: SSEEvent) -> None:
        self.events += 1
        self.last_event = ev.event
        getattr(self, "_" + self.provider)(ev)

    def _anthropic(self, ev):
        d = json.loads(ev.data)
        t = d.get("type")
        if t == "content_block_delta" and d["delta"].get("type") == "text_delta":
            self.text.append(d["delta"]["text"])
        elif t == "message_delta":
            self.stop = d["delta"].get("stop_reason")
        elif t == "message_stop":
            self.done = True
        elif t == "error":                   # arrives after HTTP 200
            self.error = d.get("error", {})

    def _openai(self, ev):
        if ev.data.strip() == "[DONE]":
            self.done = True
            return
        d = json.loads(ev.data)
        if "error" in d:
            self.error = d["error"]
            return
        for ch in d.get("choices", []):
            piece = (ch.get("delta") or {}).get("content")
            if piece:
                self.text.append(piece)
            if ch.get("finish_reason"):
                self.stop = ch["finish_reason"]

    def _gemini(self, ev):
        d = json.loads(ev.data)
        if "error" in d:
            self.error = d["error"]
            return
        block = (d.get("promptFeedback") or {}).get("blockReason")
        if block:
            self.stop, self.done = "blocked:" + block, True
            return
        for c in d.get("candidates", []):
            for p in (c.get("content") or {}).get("parts", []):
                if "text" in p and not p.get("thought"):
                    self.text.append(p["text"])
            if c.get("finishReason"):        # Gemini has no [DONE] sentinel
                self.stop, self.done = c["finishReason"], True

Two details are worth calling out. Gemini's thought parts are skipped so reasoning summaries do not leak into user-visible text. And a Gemini prompt-level blockReason is treated as a terminal state, because no candidates will follow it.

Tool calls need extra care. Claude streams tool input as input_json_delta fragments of a JSON string that is only valid once the content block closes. OpenAI streams tool_calls deltas keyed by index, with the arguments string arriving in pieces. In both cases, parse the arguments only after the terminal marker, and never execute a tool from a partial argument string. If the stream dropped mid-tool-call, throw the whole turn away.

Step 5: Detect stalls with a progress watchdog

An open connection is not progress. The watchdog below enforces three deadlines: one for the first content token, a shorter one between content events afterwards, and a ceiling for the whole response. On Claude it ignores ping events when judging progress, since pings prove the socket is alive and not that the model is generating.

import asyncio, time


class StallError(Exception):
    pass


def is_progress(provider: str):
    if provider == "anthropic":              # pings do not count as progress
        return lambda ev: ev.event == "content_block_delta"
    return lambda ev: True


async def watch(events, progress, first_s=20.0, idle_s=30.0, total_s=300.0):
    it = events.__aiter__()
    start = last = time.monotonic()
    seen = False
    while True:
        now = time.monotonic()
        limit = first_s if not seen else idle_s
        budget = min(limit - (now - last), total_s - (now - start))
        if budget <= 0:
            raise StallError("no first token" if not seen else "no progress")
        try:
            ev = await asyncio.wait_for(it.__anext__(), timeout=budget)
        except StopAsyncIteration:
            return
        except asyncio.TimeoutError:
            raise StallError("no first token" if not seen else "no progress")
        if progress(ev):
            seen = True
            last = time.monotonic()
        yield ev

Pick the first-token deadline per model. A fast chat model might need 10 to 20 seconds. A reasoning model with a large thinking budget might need a couple of minutes. Seed these numbers from your own p99 time-to-first-token, and revisit them when you change models.

Step 6: Classify every stream when it ends

Now you can turn the tracker and any exception into one of six outcomes, and log it with the numbers you need later. This is the function that turns "streaming is flaky" into a chart you can read.

import httpx

LIMIT = {"max_tokens", "length", "MAX_TOKENS", "model_context_window_exceeded"}
FILTER = {"refusal", "content_filter", "SAFETY", "RECITATION", "BLOCKLIST",
          "PROHIBITED_CONTENT", "SPII"}


def classify(tr: StreamTracker, exc: Optional[BaseException] = None) -> str:
    if isinstance(exc, StallError):
        return "stalled"
    if tr.error:
        return "error_event"                 # provider failed after the 200
    if exc is not None or not tr.done:
        return "dropped"                     # no terminal marker
    if tr.stop in LIMIT:
        return "length_limit"
    if tr.stop in FILTER or (tr.stop or "").startswith("blocked:"):
        return "filtered"
    return "ok"


@dataclass
class Trace:
    provider: str
    status: int = 0
    request_id: Optional[str] = None
    ttfb_s: float = 0.0
    ttft_s: float = 0.0
    duration_s: float = 0.0
    outcome: str = ""


TIMEOUT = httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0)


async def consume(provider, url, headers, body, on_text, trace: Trace):
    tr = StreamTracker(provider)
    t0 = time.monotonic()
    exc = None
    try:
        async with httpx.AsyncClient(timeout=TIMEOUT) as client:
            async with client.stream("POST", url, headers=headers, json=body) as r:
                trace.status = r.status_code
                trace.request_id = r.headers.get("request-id") or r.headers.get("x-request-id")
                trace.ttfb_s = time.monotonic() - t0
                if r.status_code != 200:
                    raise RuntimeError(f"HTTP {r.status_code}: {(await r.aread())[:500]!r}")
                events = parse_sse(r.aiter_lines())
                async for ev in watch(events, is_progress(provider)):
                    before = len(tr.text)
                    tr.feed(ev)
                    if len(tr.text) > before:
                        if not trace.ttft_s:
                            trace.ttft_s = time.monotonic() - t0
                        await on_text(tr.text[-1])
    except (StallError, TruncatedStream, httpx.TransportError) as e:
        exc = e                               # ReadError, RemoteProtocolError, ReadTimeout...
    trace.duration_s = time.monotonic() - t0
    trace.outcome = classify(tr, exc)
    return tr, trace

Note what is caught: httpx.TransportError covers ReadError, ReadTimeout, and RemoteProtocolError, which is what you see when the peer closes mid-chunk ("peer closed connection without sending complete message body"). All of those are drops.

Step 7: Decide what to do for each outcome

The right recovery depends on the outcome and on one more fact: has any text already reached the user?

For plain text, a continuation request that sends the partial answer back and asks the model to pick up where it left off can work, but expect small seams, and never do it for JSON, code that must compile, or anything with tool calls. For those, retry from the top. If you already have a multi-provider layer, the failover rule is the same one covered in our provider-agnostic client post: switch providers only before the first token reaches the user.

For overload and rate-limit errors that arrive mid-stream, apply the same backoff and jitter you would use for non-streaming calls. The Bulletproof Claude API Wrapper post covers that logic in detail.

Stop reasons that look like failures but are not

A large share of "my stream got cut off" reports are actually normal finishes. Check the stop reason before debugging the network.

The most common version of this bug is structured output that "fails to parse" because it was truncated at the output limit. The stream was healthy. Check for max_tokens, length, or MAX_TOKENS, raise the limit, and reduce the size of what you ask for. Reasoning and thinking tokens count against the output budget on several models, so a limit that was generous for plain answers can be too small once thinking is enabled.

Step 8: Fix your own relay

If your backend sits between the provider and the browser, it is the most common place to introduce drops and stalls. A correct relay does four things: it flushes each piece immediately, sends a heartbeat when upstream is quiet, cancels the upstream request when the user disconnects, and ends with an explicit done event.

import asyncio, json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()


async def relay(upstream, req: Request, heartbeat_s: float = 15.0):
    it = upstream.__aiter__()
    task = None
    try:
        while True:
            task = asyncio.ensure_future(it.__anext__())
            while True:
                done, _ = await asyncio.wait({task}, timeout=heartbeat_s)
                if done:
                    break
                if await req.is_disconnected():
                    return                    # user left: stop paying for tokens
                yield ": keepalive\n\n"       # keeps proxies from idling out
            try:
                piece = task.result()
            except StopAsyncIteration:
                yield "event: done\ndata: {}\n\n"
                return
            except Exception as e:
                yield "event: error\ndata: " + json.dumps({"message": str(e)}) + "\n\n"
                return
            yield "data: " + json.dumps({"t": piece}) + "\n\n"
    finally:
        if task and not task.done():
            task.cancel()
        await it.aclose()


@app.post("/chat")
async def chat(req: Request):
    body = await req.json()
    return StreamingResponse(
        relay(stream_text(body), req),        # stream_text: your async generator of str
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

The explicit done event is what gives your own frontend the same terminal-marker rule you applied to the provider. Without it, a browser cannot tell a finished answer from one cut off by a network blip.

On the client, avoid EventSource for chat. It only issues GET requests and reconnects automatically, which replays a request you did not want replayed. Read a fetch stream and check for the done event:

async function streamChat(prompt, onText) {
  const res = await fetch("/chat", {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({ prompt }),
  });
  if (!res.ok || !res.body) throw new Error("HTTP " + res.status);

  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  let buf = "", sawDone = false;

  for (;;) {
    const { value, done } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });
    let i;
    while ((i = buf.indexOf("\n\n")) !== -1) {
      const block = buf.slice(0, i);
      buf = buf.slice(i + 2);
      let name = "message", data = "";
      for (const line of block.split("\n")) {
        if (line.startsWith(":")) continue;
        if (line.startsWith("event: ")) name = line.slice(7);
        if (line.startsWith("data: ")) data += line.slice(6);
      }
      if (name === "done") sawDone = true;
      else if (name === "error") throw new Error(JSON.parse(data).message);
      else if (data) onText(JSON.parse(data).t);
    }
  }
  if (!sawDone) throw new Error("stream ended without a done event");
}

What to log on every stream

You can debug almost any streaming incident in minutes if each request leaves one structured log line with these fields:

• Provider request ID from the response header (request-id on Claude, x-request-id on OpenAI). It is the first thing any provider support team will ask for.
• Time to first byte and time to first content token. The gap between them points at buffering or provider queueing.
• Event count, byte count, and last event type seen. A stream whose last event was content_block_delta was cut mid-answer. One whose last event was message_delta lost only the closing event.
• Terminal marker seen, stop reason, and outcome from the classifier above.
• Total duration. Failures clustering at the same duration are timeouts.
• Client type and network path, so you can separate mobile Safari from a server-to-server job.

Chart the outcome by hour and by route. A sudden rise in dropped after a deploy points at your relay or proxy configuration. A rise in error_event that matches a provider incident points at the provider. A rise in length_limit after a prompt change points at your prompt.

Symptom-to-cause cheat sheet

Test the failures before they happen in production

Streaming code is easy to get wrong because the happy path always works. Test the unhappy paths with scripted streams, no network required:

import pytest

def delta(text):
    return SSEEvent("content_block_delta",
        json.dumps({"type": "content_block_delta", "index": 0,
                    "delta": {"type": "text_delta", "text": text}}))


async def lines(*chunks):
    for c in chunks:
        yield c


@pytest.mark.asyncio
async def test_half_event_is_truncation():
    with pytest.raises(TruncatedStream):
        async for _ in parse_sse(lines("event: message_start", 'data: {"type":"mess')):
            pass


def test_missing_message_stop_is_a_drop():
    tr = StreamTracker("anthropic")
    tr.feed(delta("Hel"))
    assert classify(tr) == "dropped"


def test_error_event_after_200():
    tr = StreamTracker("anthropic")
    tr.feed(delta("Hel"))
    tr.feed(SSEEvent("error", '{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}'))
    assert classify(tr) == "error_event"


def test_length_stop_is_not_a_failure_to_retry():
    tr = StreamTracker("openai")
    tr.feed(SSEEvent("message", '{"choices":[{"delta":{"content":"x"},"finish_reason":"length"}]}'))
    tr.feed(SSEEvent("message", "[DONE]"))
    assert classify(tr) == "length_limit"


async def hangs_after(n):
    for _ in range(n):
        yield delta("hi ")
    await asyncio.sleep(3600)


@pytest.mark.asyncio
async def test_stall_is_detected():
    with pytest.raises(StallError):
        async for _ in watch(hangs_after(2), lambda e: True,
                             first_s=0.2, idle_s=0.2, total_s=5):
            pass

Add one more layer if you can: a tiny local server that serves a recorded real stream, with switches to truncate it after N events, delay a chunk by 60 seconds, or append an error event. Point your staging environment at it and confirm that your proxy, relay, and frontend each behave correctly.

Pitfalls to avoid

• Trusting the SDK iterator to raise on a drop. Some paths end quietly. Check the terminal marker yourself.
• One timeout for everything. A single "request timeout" cannot distinguish a slow first token from a stalled stream. Use first-token, inter-chunk, and total deadlines.
• Retrying after text is on screen. The user sees the answer start over, or worse, get stitched. Decide on a policy and make it visible in the UI.
• Executing a tool from partial arguments. Parse and act only after the terminal marker.
• Forgetting to cancel upstream. A user who closes the tab should not cost you the remaining tokens.
• Treating every 200 as success. Check the terminal marker, the stop reason, and any error event.
• Debugging through three proxies at once. Always reproduce with curl against the provider directly first.

If your stream failures cluster around provider load, the failover design in the Multi-Provider Load Balancer post is the next layer to add.

Frequently asked questions

Keep reading