Streaming Responses and Invalid Model Output

for event in provider_stream:
    buffer += event.partial_json
    args = parse_partial(buffer)
    if "sku" in args:
        send_to_client(run_tool("get_product_spec", args))
send_event("done")

Partial output is not valid output

Pydantic can parse an unfinished JSON text, and the result looks plausible:

Verified: capstone/fde-product-discovery-copilot/lessons/M2-L6/test_m2_l6_streaming.py

def test_a_partial_parse_would_have_hidden_the_problem() -> None:
    from pydantic_core import from_json
    cut = '{"category": "sensors", "quantity": '
    assert from_json(cut, allow_partial=True) == {"category": "sensors"}  # looks fine, quantity silently missing
    assert check_output(cut, Requirements)[0] == "truncated"

With allow_partial=True the unfinished piece is dropped silently. Use partial parsing for a progress display only.

Classify the finished reply, then choose the cure

Figure 2. States of a streamed structured answer

How to read this: only Accepted ever reaches a user or a tool, and every other path is bounded and ends in Failed.

Verified: capstone/fde-product-discovery-copilot/lessons/M2-L6/m2_l6_streaming.py

Outcome = Literal["valid", "fenced", "empty", "truncated", "invalid_json", "schema_violation"]
# What to do with each outcome. Only the first two ever reach a user or a tool.
ACTION: dict[Outcome, str] = {
    "valid": "accept",
    "fenced": "accept",  # a Markdown fence around valid JSON: strip it, it is not an error
    "empty": "retry",
    "truncated": "retry_with_more_tokens",  # the cure is a larger limit or a continuation, not a repair prompt
    "invalid_json": "repair_once",
    "schema_violation": "repair_once",
}

Per the provider's documentation, a reply cut at the token limit has the stop reason max_tokens. Without a stop reason, the function below guesses from where the parse failed.

Verified: capstone/fde-product-discovery-copilot/lessons/M2-L6/m2_l6_streaming.py

def check_output(text: str, model: type[T], stop_reason: str | None = None) -> tuple[Outcome, T | None]:
    """Classify a finished model reply. `stop_reason` ('max_tokens') is the reliable truncation signal when you have it."""
    body = text.strip()
    if not body:
        return "empty", None
    if stop_reason == "max_tokens":
        return "truncated", None
    fenced = body.startswith("```")
    try:
        return ("fenced" if fenced else "valid"), parse_json_model(model, body)
    except json.JSONDecodeError as err:
        inner = body.removeprefix("```json").removeprefix("```").strip()
        ended_early = err.msg.startswith("Unterminated string") or err.pos >= len(inner)
        return ("truncated" if ended_early else "invalid_json"), None
    except ValidationError:
        return "schema_violation", None

In the interview: "The answer stopped halfway and the user did not notice. What happened and what do you change?" A good answer finds the missing terminal event, adds done and error, and gives the interface a visible incomplete state with a retry.

Worked example

Assumptions (illustrative): a 40-word reply streams in chunks of two words, so 20 delta events; the connection drops after the 12th.

Explain it back. Tell a sales engineer, in three sentences, why an answer can stop in the middle and what the interface now does. Say the connection broke, that an unfinished answer is now marked incomplete, and that retrying is safe because nothing was acted on.

Build it

The code is in capstone/fde-product-discovery-copilot/lessons/M2-L6/. Step 1: stream from a source that fails after one delta; the status stays 200.

Verified: capstone/fde-product-discovery-copilot/lessons/M2-L6/test_m2_l6_streaming.py

def test_a_failure_after_the_200_is_only_visible_inside_the_stream() -> None:
    def broken() -> Iterator[str]:
        yield "partial "
        raise RuntimeError("upstream reset")

    response = TestClient(create_demo_app(broken)).get("/stream")
    assert response.status_code == 200  # the status was sent before the failure
    assert [name for name, _ in parse_sse(response.text)] == ["delta", "error"]

Step 2: run the whole folder.

./.venv/Scripts/python -m pytest lessons/M2-L6 -q

Real output (timing varies):

........................                                                 [100%]
24 passed in 5.55s

The count includes the exercise's reference solution; you will see 21.

Exercise

Write stream_events_capped(deltas, max_chars): forward deltas until the next one would pass max_chars, then send one error event with the code too_long and stop.

Acceptance criteria: a short answer ends in done; a long one ends in one error, no done, reporting the characters sent; the source iterator is closed. Done when the three tests pass.

Common mistakes

• Treating the end of the connection as the end of the answer. Only done means finished.
• Sending an error as a new HTTP status. The status line is gone after the first byte. Use an event.
• Acting on a partial parse. It looks valid. Parse once, after the last delta.
• Forwarding exception text. It can hold addresses and keys. Send a fixed message; log the cause.

Knowledge check

You can now

• Serve and read an event stream.
• Guarantee one terminal event.
• Classify a reply and choose the cure.
• Explain why a partial parse must never drive an action.

Further reading

• FastAPI: server-sent events: EventSourceResponse.
• WHATWG: server-sent events: the format.
• Claude API: streaming messages: error events.

Last verified: 2026-11-02

Next

This is the last core lesson of the module on LLM application engineering. The next module starts with Ingestion and normalization.