Streaming without corrupting the stream

Bytes arrive when they arrive, not where you want them

A streaming response is a socket handing you whatever has turned up. A chunk is not a character, not a line, and not a message. It is a count of bytes that depends on network timing, and the same request can chunk differently on two consecutive runs. Every bug in this lesson comes from one assumption: that the boundary the transport chose means something. It does not.

The reason this shows up in AI code more than anywhere else is that the payload is text a human reads, so corruption is visible, and the test data is English, so it is invisible until the first user types an accent.

One decoder, not one call per chunk

UTF-8 encodes most characters in more than one byte. é is two, € is three, and a chunk boundary lands between them exactly as often as anywhere else.

raw = "café".encode("utf-8")
print(list(raw))            # [99, 97, 102, 195, 169]
print(raw[:4])              # b'caf\xc3'
raw[:4].decode("utf-8")     # UnicodeDecodeError: unexpected end of data

bytes.decode has to see a whole character, so calling it once per chunk raises on any chunk that ends mid-character. An incremental decoder is the same codec with memory: hand it bytes, it returns the characters it can complete and keeps the rest.

import codecs

decoder = codecs.getincrementaldecoder("utf-8")()
print(repr(decoder.decode(raw[:4])))   # 'caf'   the 0xc3 is held back
print(repr(decoder.decode(raw[4:])))   # 'é'     completed by the next byte

Note the double call in getincrementaldecoder("utf-8")(): the first returns the decoder class for that encoding, the second builds an instance. One instance per stream, and it is stateful, so it is never shared between two streams.

The second buffer is yours

The decoder solves partial characters. It has nothing to say about partial lines, and a protocol that separates messages with a newline needs that solved too. So you keep a text buffer and split it on the separator, and now you have to decide which of the resulting pieces are finished lines and which one is not.

buffer = ""
buffer += 'data: {"delta": "hi"}\ndata: {"del'
parts = buffer.split("\n")
buffer = parts.pop()
print(parts)                # ['data: {"delta": "hi"}']
print(repr(buffer))         # 'data: {"del'

split always returns at least one element, so parts.pop() is safe on any input, and when the buffer ends exactly on a separator the piece it puts back is the empty string. Everything left in parts is a complete line.

That is two pieces of state carried between chunks, holding different things: bytes that are not yet a character, and text that is not yet a line. Conflating them is how the second version of this bug gets written.

Never parse a line you have not finished receiving

Once a line is complete, and only then, it can be parsed. json.loads on a partial object raises, and there is no partial-JSON mode worth having: the fix is to not call it early.

import json

print(json.loads('{"delta": "hi"}'))    # {'delta': 'hi'}
json.loads('{"delta": "h')              # JSONDecodeError: Unterminated string

A protocol usually carries non-payload lines too: blank ones, comments, event names. Skip anything that does not start with the payload prefix rather than trying to parse it and catching the failure, because a caught parse error cannot tell "this was not a payload line" from "this payload is corrupt".

Stopping is part of the protocol

Most streams end with a sentinel rather than by closing, so the loop that reads one ends on a message rather than on the end of the body.

Stopping on the sentinel leaves the connection where it was, so releasing it is a job you still owe. A generator is where this bites: leaving a for loop early does not finish the generator, and whatever it was holding stays held until somebody closes it.

def rows():
    try:
        for index in range(5):
            yield index
    finally:
        print("connection released")


stream = rows()
print(next(stream))         # 0
stream.close()              # connection released

close() raises GeneratorExit at the paused yield, so a finally inside the generator runs. On your side of the loop, the same job belongs in a try / finally, so the close happens on the sentinel path, the exception path, and the ran-out-of-chunks path alike.

Pitfalls

• decoder.decode(b"", final=True) is what tells you the stream ended mid-character: it raises UnicodeDecodeError if anything is still held. Skip that call and a truncated response looks like a complete short one.
• One decoder per stream. Reusing an instance carries the previous stream's dangling bytes into the next one.
• A chunk of size one is legal, and so is one chunk carrying the whole response. Both are worth a test, because they exercise opposite halves of the buffering.
• splitlines() is not split("\n") here. It also breaks on form feed, next line, and the Unicode line separators, so a model that emits one of those silently gains a message boundary.

Interview nuance: the question behind this one is usually "how do you know the stream finished rather than died?" Say the two signals out loud: a protocol-level sentinel means the sender said it was done, and a clean end of body with an empty decoder means the transport agreed. Anything else, a body that stops mid-character or an end with no sentinel, is a truncated response, and the client is the only thing in the system positioned to notice. Serving that as a complete answer is how a truncated generation gets stored, cached, and shown to the next user as fact.

Sources: codecs.getincrementaldecoder · Server-sent events · PEP 342, generator close