ProductHow it worksPricingBlogDocsLoginFind Your First Bug
A streaming AI response shown as a five-stage lifecycle: partial chunk validity, stream completion, mid-stream error recovery, typing indicator state, and the final assembled output
TestingAIStreaming AI Testing

How to Test Streaming AI Responses

Tom Piaggio
Tom PiaggioCo-Founder at Autonoma

Testing streaming AI responses means verifying the response as a lifecycle, not a value: assert on partial-chunk validity, monotonic buffer growth, clean stream completion within a timeout, graceful recovery when the connection drops mid-stream, correct typing-indicator transitions, and the final assembled output against a semantic invariant rather than an exact string, since chunk boundaries and generated wording are both non-deterministic between runs.

The assertion passed on your laptop. In CI, it hung for thirty seconds and timed out. Nothing about the code changed, the difference was timing: locally the response landed before the test checked anything, in CI it hadn't, and the test had no idea what to do with a response only half in flight.

That's the tell. If your streaming AI feature has a test suite, it's almost certainly testing the finished string, not what a user experiences: a message that grows on screen for several seconds, one token at a time, with a UI making decisions about indicators and error states the whole time.

One team had a typing indicator reported stuck for forty seconds, on a response that finished streaming in under two. The text sat complete in the DOM while the indicator kept spinning underneath it. Nobody caught it, because every existing test waited for the full response and asserted on that. The bug lived entirely in the part no test was watching.

Why a Streamed Response Breaks Conventional Assertions

A conventional assertion assumes the thing you're checking already exists in full by the time you check it. Call an endpoint, get a response body, assert on it. That model breaks immediately against a streaming feature: at the moment your test fires the request, the response doesn't exist yet as a value. It's a sequence of events over a lifecycle: a connection opens, tokens arrive one at a time over EventSource or a WebSocket, the UI repaints on every chunk, and only later does a finished response exist to assert against.

Treat it like a normal HTTP call and you get one of two broken tests. Wait for the whole thing and assert on the final string, and you throw away every bug in the middle: the stuck indicator, the malformed chunk, the connection that drops at 60 percent. Or assert too early, and your test becomes a coin flip on network timing, flaking in CI for reasons unrelated to your code. If you've already worked through testing a chatbot, this is the same non-determinism problem one layer lower, down at the transport, not just the wording.

There's a reason evaluation tooling is no help here either. Evals score the finished response, which is precisely the artifact that exists only after every interesting failure has already happened. A stuck spinner has no representation in the text a model produced. Neither does partial output that vanished on re-render, or a scroll anchor that fought the user for four seconds. Those live in the running application, and the only way to assert on them is to drive the application while it streams. That's the discipline Autonoma is built around, and it's worth keeping in view as you read the assertion targets below, because roughly half of them are invisible to anything that inspects a response body.

The Stream Lifecycle Is a State Machine, Not a Value

Name the states and the problem gets tractable. idle is before the user has asked anything. requested fires the moment the request goes out, and it's also when a typing indicator should appear. first token is the instant the first chunk lands, arguably the most important transition, since it's usually when the indicator should disappear. streaming covers every chunk after, where the buffer grows and the UI repaints. complete is reached when a terminator arrives, a literal data: [DONE] line or a finish_reason field. error branches off streaming at any point the connection drops or a malformed frame arrives, and it's the transition most streaming UIs get wrong.

Stream lifecycle state machine from idle through complete, with an error branch off the streaming state and assertion markers at each transitionidlerequestedassert: indicator ONfirst tokenassert: indicator OFFstreamingassert: chunks valid,buffer grows monotonicallycompleteassert: terminator within timeout,semantic match on bufferconnection drops or errorserrorindicator OFFassert: partial text kept, retry shown

Five states, one error branch. Every assertion target below maps to a transition or a state in this same machine.

That's the point: once you stop asserting on "the response" and start asserting on specific transitions, flaky streaming tests mostly stop being flaky. They fail, or pass, for identifiable reasons tied to a specific state, not vague timing nobody can reproduce.

Partial-Response Correctness: Every Chunk Must Parse, and the Buffer Must Only Grow

Every chunk must be valid and safe to render on its own, and the buffer it produces must never be shorter than the buffer before it.

Most teams skip this because it feels like transport, not the AI feature. It doesn't skip cleanly: a malformed delta, an unescaped fragment inside a chat bubble, or a chunk that silently replaces the buffer instead of appending can render broken UI or rewrite a response mid-stream, invisible against the finished string. The fix: a parse check on every delta plus a running length check, the buffer's length after each chunk must be greater than or equal to its length before.

Stream Completion: The Terminator Must Arrive Inside a Timeout

The stream must reach its terminator, a literal data: [DONE] line or a finish_reason field, within a bounded time, and the client must leave the streaming state once it does.

A stream that never terminates isn't paused, it's broken, and the UI needs to know within a timeout short enough that a user isn't staring at a spinner for a full minute. Assert elapsed time against a fixed threshold on every run, not by confirming once in a debugger that the terminator eventually showed up.

Error Mid-Stream: The UI Must Recover, Not Freeze

When the connection drops or errors partway through, the client must surface the failure, keep whatever partial text already arrived, clear any loading state, and offer a way to retry.

This is the failure mode a happy-path test never exercises, because it never cuts the connection. Deliberately abort the stream partway (an abort switch, or a fulfilled request that stops after a few chunks) and check three things independently: partial text stays on screen, the indicator clears instead of spinning forever, and a retry control appears. Miss any one and you get a frequently reported bug: a UI that looks broken even though the network genuinely failed, and there was nothing the app could do about the failure, only about how it responded.

A stream dropping at roughly 60 percent, contrasting the correct UI recovery path against the common broken pathchunks arriving over timedrop at ~60%chunks that never arrivecorrect recoverypartial text stays visibletyping indicator clearsretry affordance showncommon broken pathindicator spins foreverpartial text gets wipedno retry, user stuckassert the correct branch: retained buffer, cleared indicator, visible retry control

Same drop, two outcomes. The only difference between the two branches is what the client code does after the connection dies, not the failure itself.

The Typing Indicator: On at Request, Off at First Token, Never Stuck

The indicator must appear the moment a request goes out, disappear the moment the first token (or, if you choose that convention instead, the completion event) arrives, and never remain visible after an error.

State your convention explicitly, because "off at first token" and "off at completion" are both defensible, and testing the wrong one produces a false failure. Whichever you pick, the indicator's state is cheap to assert against real DOM, and it's the single most common bug in every streaming chat UI, not because it's hard to build correctly, but because it's easy to forget to clear on the one code path, the error path, nobody tests.

Note the phrase "against real DOM," because it's doing all the work in that sentence. There is no protocol-level assertion for a stuck indicator. The stream behaved perfectly; a component didn't unsubscribe. That is the shape of nearly every bug users actually report about streaming features, and it's why the behavioral layer isn't a nice-to-have here the way it might be for a CRUD form. Autonoma sits at exactly that layer, running end-to-end tests against the deployed application on the pull request, asserting on what the interface is doing mid-stream rather than on what eventually arrived.

Final Assembled Output: Assert Intent, Not Exact Text

Once every chunk lands, the concatenated result must satisfy what the user asked for, checked against invariants and semantic similarity rather than an exact string match.

Chunk boundaries are non-deterministic between runs, and if you're streaming from a real model the wording is too, so an assertion pinned to exact text will flake on a paraphrase that's completely correct. Run the same prompt several times and assert on what has to be true of the assembled buffer: it contains the fact you asked for, ends on a complete sentence, stays under a length bound. Never assert on chunk content or chunk count, both are accidents of network timing, not properties of the response. A streaming response is one instance of a wider pattern, a UI whose structure isn't fixed at request time, the same territory testing generative UI covers once the model generates layout, not just words.

Two Levels of Streaming Tests: A Fake Server and a Real Browser

Level 1 tests the stream protocol against a deterministic fake server: scripted tokens, no live model call, sub-second runs, safe on every commit. A fixture you control can be told to omit its terminator or die at chunk six on command, which a real model API won't reliably do.

Level 2 tests the streaming UI behaviorally, in a real browser, against the actual feature. This catches what Level 1 structurally cannot: an indicator that never clears because an event listener got detached, partial text that vanishes when a component re-renders on error, a scroll anchor that jumps mid-stream. None of those bugs live in the protocol. They live in the running application, so a fake-server unit test will never see them.

Most teams do Level 1 and stop, or skip both and learn about the stuck indicator from a support ticket. Level 1 is the easy half to build and the easy half to keep: it's hermetic, fast, and its fixture changes only when the protocol does. Level 2 is where the effort actually goes, and where suites tend to rot, because the assertions are tied to a real interface that gets redesigned while the streaming logic underneath stays put.

This is the split Autonoma is built to absorb. It runs behavioral end-to-end tests against the real deployed application in a preview environment on the pull request, so the Level 2 checks above, indicator on at request, off at first token, cleared after an error, partial text retained, run against the actual UI on every change rather than on whichever sprint someone had time. The Diffs Agent updates the suite from each pull request's diff, which matters for a streaming feature specifically: swapping a chat component or moving the indicator into a different subtree is a one-line diff that silently invalidates every hand-written selector pointed at it.

Keep Level 1 in your own repo regardless. Protocol invariants are yours, they're cheap, and no external tool should be your first line of defense against a malformed chunk. The division that works is protocol tests you own and maintain, behavioral tests that maintain themselves against the running app.

A Runnable Stream Test

Here's the fixture that makes this testable: a small FastAPI server streaming a scripted token sequence, with a param to cut the stream short and one to withhold the terminator on command. It also serves a minimal HTML page wired to the same stream, so one fixture backs both the tests below and the browser test after them.

import asyncio
import json

from fastapi import FastAPI
from fastapi.responses import HTMLResponse, StreamingResponse

app = FastAPI()

SCRIPTED_TOKENS = [
    "Testing", " a", " streaming", " response", " requires", " watching",
    " the", " whole", " lifecycle,", " not", " just", " the", " final", " string.",
]

INDEX_HTML = """<!doctype html>
<html>
<head><meta charset="utf-8"><title>Streaming Chat Fixture</title></head>
<body>
  <button id="ask">Ask</button>
  <div id="typing-indicator" hidden>Assistant is typing...</div>
  <div id="response"></div>
  <div id="error"></div>
  <button id="retry" hidden>Retry</button>
  <script>
    const askButton = document.getElementById('ask');
    const responseEl = document.getElementById('response');
    const indicator = document.getElementById('typing-indicator');
    const retryButton = document.getElementById('retry');
    const errorEl = document.getElementById('error');

    function runStream() {
      responseEl.textContent = '';
      errorEl.textContent = '';
      retryButton.hidden = true;
      indicator.hidden = false;

      const source = new EventSource('/stream');
      let gotFirstToken = false;

      source.onmessage = (event) => {
        if (event.data === '[DONE]') {
          source.close();
          indicator.hidden = true;
          return;
        }
        if (!gotFirstToken) {
          gotFirstToken = true;
          indicator.hidden = true;
        }
        const payload = JSON.parse(event.data);
        responseEl.textContent += payload.delta;
      };

      source.onerror = () => {
        source.close();
        indicator.hidden = true;
        errorEl.textContent = 'Stream failed. Partial response kept above.';
        retryButton.hidden = false;
      };
    }

    askButton.addEventListener('click', runStream);
    retryButton.addEventListener('click', runStream);
  </script>
</body>
</html>
"""


@app.get("/", response_class=HTMLResponse)
async def index():
    return INDEX_HTML


async def token_stream(abort_at, omit_terminator):
    for index, token in enumerate(SCRIPTED_TOKENS):
        if abort_at is not None and index >= abort_at:
            return
        chunk = {"delta": token, "index": index}
        yield f"data: {json.dumps(chunk)}\n\n"
        await asyncio.sleep(0.02)
    if not omit_terminator:
        yield "data: [DONE]\n\n"


@app.get("/stream")
async def stream(abort_at: int | None = None, omit_terminator: bool = False):
    return StreamingResponse(
        token_stream(abort_at, omit_terminator),
        media_type="text/event-stream",
    )

Point a pytest suite at it and the invariants above become assertions instead of hopes: every chunk parses, the buffer only grows, the terminator arrives inside a timeout, and the assembled result is checked with a semantic, run-it-five-times assertion instead of an exact string.

import json
import time

import httpx
import pytest

BASE_URL = "http://localhost:8000"


def consume_stream(params=None, timeout=5.0):
    buffer = ""
    previous_length = 0
    chunk_count = 0
    terminated = False
    start = time.monotonic()

    with httpx.Client(timeout=timeout) as client:
        with client.stream("GET", f"{BASE_URL}/stream", params=params) as response:
            for line in response.iter_lines():
                if time.monotonic() - start > timeout:
                    pytest.fail("stream exceeded timeout without reaching a terminator")
                if not line or not line.startswith("data: "):
                    continue
                payload = line[len("data: "):]
                if payload == "[DONE]":
                    terminated = True
                    break
                data = json.loads(payload)
                buffer += data["delta"]
                assert len(buffer) >= previous_length, "buffer must never shrink or rewrite earlier text"
                previous_length = len(buffer)
                chunk_count += 1

    elapsed = time.monotonic() - start
    return buffer, chunk_count, terminated, elapsed


def test_every_chunk_parses_and_buffer_grows_monotonically():
    buffer, chunk_count, terminated, _ = consume_stream()
    assert chunk_count > 0
    assert terminated
    assert len(buffer) > 0


def test_stream_terminates_within_timeout():
    _, _, terminated, elapsed = consume_stream(timeout=3.0)
    assert terminated
    assert elapsed < 3.0


@pytest.mark.parametrize("run", range(5))
def test_assembled_output_satisfies_intent_across_runs(run):
    buffer, _, terminated, _ = consume_stream()
    assert terminated
    assert "streaming" in buffer.lower()
    assert buffer.strip().endswith(".")

Then force the failure mode a happy-path test never exercises: cut the connection mid-stream and check the client doesn't fabricate a terminator it never received, and retains whatever partial text arrived before the drop.

import json

import httpx

BASE_URL = "http://localhost:8000"


def test_client_surfaces_error_and_keeps_partial_buffer_on_mid_stream_drop():
    buffer = ""
    terminated = False

    with httpx.Client(timeout=2.0) as client:
        try:
            with client.stream("GET", f"{BASE_URL}/stream", params={"abort_at": 6}) as response:
                for line in response.iter_lines():
                    if not line or not line.startswith("data: "):
                        continue
                    payload = line[len("data: "):]
                    if payload == "[DONE]":
                        terminated = True
                        break
                    buffer += json.loads(payload)["delta"]
        except httpx.ReadTimeout:
            pass

    assert not terminated
    assert len(buffer) > 0

The Level 2 Playwright Test

The fixture's HTML page is deliberately minimal: an Ask button, a response element, a typing indicator, and a retry button that appears only after an error. That's enough for a real-browser test to check what the fake-server tests cannot: indicator state lines up with the lifecycle, response text grows over time instead of appearing all at once, and a request Playwright fulfills with a truncated body leaves partial text on screen, clears the indicator, and shows retry.

import { test, expect } from '@playwright/test';

const BASE_URL = process.env.BASE_URL || 'http://localhost:8000';

test('typing indicator appears on request and clears on first token', async ({ page }) => {
  await page.goto(BASE_URL);
  await page.click('#ask');
  await expect(page.locator('#typing-indicator')).toBeVisible();
  await expect(page.locator('#typing-indicator')).toBeHidden({ timeout: 3000 });
});

test('assembled text grows over time as chunks arrive', async ({ page }) => {
  await page.goto(BASE_URL);
  await page.click('#ask');
  await page.waitForTimeout(60);
  const lengthEarly = await page.locator('#response').evaluate((el) => el.textContent?.length ?? 0);
  await page.waitForTimeout(200);
  const lengthLater = await page.locator('#response').evaluate((el) => el.textContent?.length ?? 0);
  expect(lengthLater).toBeGreaterThan(lengthEarly);
});

test('indicator clears and final text lands after the terminator arrives', async ({ page }) => {
  await page.goto(BASE_URL);
  await page.click('#ask');
  await expect(page.locator('#typing-indicator')).toBeHidden({ timeout: 5000 });
  await page.waitForTimeout(500);
  const finalText = await page.locator('#response').textContent();
  expect(finalText?.length).toBeGreaterThan(0);
});

test('a dropped connection keeps partial text, clears the indicator, and offers retry', async ({ page }) => {
  await page.route('**/stream', async (route) => {
    const partialBody = [
      'data: {"delta":"Testing","index":0}\n\n',
      'data: {"delta":" a","index":1}\n\n',
      'data: {"delta":" streaming","index":2}\n\n',
    ].join('');
    await route.fulfill({ status: 200, contentType: 'text/event-stream', body: partialBody });
  });

  await page.goto(BASE_URL);
  await page.click('#ask');

  await expect(page.locator('#typing-indicator')).toBeHidden({ timeout: 5000 });
  await expect(page.locator('#retry')).toBeVisible();
  await expect(page.locator('#response')).toHaveText('Testing a streaming');
});

Wiring This Into CI Without Flaking

Run the fake-server suite on every commit, not on a schedule. It has no live model call and no network dependency beyond localhost, so there's no excuse to skip it. A job-level timeout of a few minutes catches the failure mode CI is actually vulnerable to: a stream that hangs instead of erroring cleanly, blocking the pipeline instead of failing it. Give the server a moment to boot, keep per-test timeouts short, and treat any test that needs a retry to pass as a bug in the assertion, not a reason to add retries.

name: Streaming tests

on:
  pull_request:
  push:
    branches: [main]

jobs:
  fake-stream-protocol:
    runs-on: ubuntu-latest
    timeout-minutes: 5
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Install dependencies
        run: pip install fastapi uvicorn httpx pytest
      - name: Start fake stream server
        run: |
          uvicorn fake_stream_server:app --port 8000 &
          sleep 1
      - name: Run stream protocol and failure tests
        run: pytest tests/ -v

  streaming-ui-behavior:
    runs-on: ubuntu-latest
    timeout-minutes: 5
    needs: fake-stream-protocol
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - uses: actions/setup-node@v4
        with:
          node-version: "20"
      - name: Install server dependencies
        run: pip install fastapi uvicorn
      - name: Install Playwright
        run: |
          npm install -D @playwright/test
          npx playwright install --with-deps chromium
      - name: Start fake stream server
        run: |
          uvicorn fake_stream_server:app --port 8000 &
          sleep 1
      - name: Run Playwright streaming UI tests
        run: npx playwright test playwright/streaming-ui.spec.ts
        env:
          BASE_URL: http://localhost:8000

That last point matters more for streaming than most other tests: a retry silently papers over the exact race, between buffer state, indicator state, and network state, that streaming introduces. Fix the wait condition. Don't add a retry.

Two things are worth carrying out of this piece. The first is that a streaming response is a lifecycle, and every assertion above is anchored to a named state or transition in it, which is what turns a category of test with a reputation for flaking into one that fails for reasons you can name. The second is that the lifecycle only half exists on the wire. The other half exists in the interface, in indicator state, retained partial text, and a retry control that either appeared or didn't, and no amount of protocol coverage reaches it.

Build the fake-server suite yourself and run it on every commit. Then put a behavioral layer under the interface, whether that's the Playwright spec above maintained by hand or Autonoma running it against your deployed app on each pull request. The stuck-indicator bug from the opening of this piece was never a model problem or a network problem. It was a UI that nothing was watching while it streamed.

Frequently Asked Questions

Test the lifecycle, not the finished string: assert that every chunk parses and the buffer only grows, that the stream reaches its terminator inside a timeout, that a mid-stream connection drop clears the loading state without wiping partial text, that the typing indicator's on and off transitions match the lifecycle, and that the assembled output satisfies a semantic or invariant check rather than an exact match.

Run a deterministic fake SSE server behind your tests so you can force every failure mode on demand: a normal completion, a stream that never sends its terminator, and a connection that drops at an arbitrary chunk. Assert against that fixture in a fast, hermetic pytest or vitest suite, then add a smaller set of real-browser tests against the actual UI to catch bugs, like a stuck indicator or wiped partial text, that only exist once a real component is rendering the stream.

Almost always because the assertion is checking timing instead of a lifecycle transition, or because a retry was added to paper over a race between buffer state, indicator state, and network state that a hard timeout and a correctly ordered assertion would have caught cleanly. Fix the wait condition, don't retry the test.

Never assert on exact text or on individual chunk content, since chunk boundaries and, against a real model, the wording itself are both non-deterministic between runs. Assert on invariants instead: the assembled buffer contains the required fact, ends on a complete sentence, stays under a length bound, and passes a semantic similarity check run across several repetitions of the same prompt rather than once.

Drive the real UI in a browser test and assert on DOM state at each lifecycle transition: visible the moment a request goes out, hidden the moment the first token (or completion, depending on which convention you picked) arrives, and, critically, hidden after an error too, since a stuck-forever indicator on the error path is the most commonly reported bug in streaming chat interfaces.

Because an eval scores the finished response, and a stuck indicator has no representation in the text the model produced. The stream can behave perfectly at the protocol level, with every chunk valid and the terminator arriving inside its timeout, while a component simply failed to unsubscribe on the error path. The same is true of partial text that vanished when a component re-rendered, and of a retry control that never appeared. Those exist only in the running interface, so catching them means driving the live UI while it streams. That's the layer Autonoma covers, running behavioral end-to-end tests against the deployed application rather than reading a transcript of what the model said.

Related articles

A horizontal agent trajectory diagram showing a tool call passing a right-tool checkpoint but failing an argument-accuracy checkpoint

How to Test AI Agents That Take Actions (Tool Calls)

A runnable guide to testing tool-calling agents: right tool, right order, right arguments, mocked vs live calls, failure handling, and non-determinism.

A chatbot test pipeline moving from manual QA through scripted and semantic assertions into an automated CI gate that samples the model N times before allowing a merge

Chatbot Automation Testing: Why Assertions Fail

Chatbot automation testing that survives non-deterministic replies: the migration to a CI gate, n-run sampling, threshold gating, and real GitHub Actions YAML.

Ghost Inspector alternative concept: Quara the frog beside a cracked recorded-test snapshot next to a regenerating test path

Ghost Inspector Alternative: Recorder, Framework, or AI?

Looking for a Ghost Inspector alternative? Compare record-and-playback SaaS, code frameworks, and AI-agent-generated testing by approach, not just by tool.

Diagram showing AI-generated auth code without a baseline: an agent writes login code on one side, while expected auth behavior (valid login, rejected password, protected route redirect) must be defined explicitly on the other

How to Test the Auth Code an AI Agent Wrote

When an AI agent writes your authentication, there is no baseline for correct behavior. Here is how to test AI-generated code for the auth bugs that compile, pass review, and lock users out.