Decode a UTF-8 Stream Across Chunk Boundaries, Then Flush It

A byte reader can return a chunk halfway through a character. Calling chunk.decode("utf-8") independently on each chunk treats a transport boundary as a text boundary. A valid response containing a snow character can then fail simply because its three bytes arrived in two reads.

An incremental decoder carries the unfinished sequence into the next call. Completion needs its own operation: a final empty input with final=True. Without that step, a truncated final character can remain buffered while a caller mistakes the earlier text for a complete document.

This example deliberately returns one complete string. The caller receives no partial result if any chunk or the final flush raises. That is useful for a small configuration or diagnostic document that must pass decoding before parsing. It does not provide transactional behavior for a consumer that already displayed or persisted earlier decoded pieces.

Make stream boundaries explicit

Create one decoder per logical stream. Reusing it for another response would also reuse any buffered state. Keep strict error handling here: replacing an invalid byte with a replacement character would change the input accepted by a later parser.

The Python incremental decoder documentation describes the stateful interface and final-input signal. The following complete standard-library program exercises those behaviors with synthetic bytes.

import codecs import unittest def decode_chunks(chunks): decoder = codecs.getincrementaldecoder("utf-8")(errors="strict") parts = [decoder.decode(chunk, final=False) for chunk in chunks] parts.append(decoder.decode(b"", final=True)) return "".join(parts) class StreamTests(unittest.TestCase): def test_every_split_preserves_multibyte_text(self): text = "A雪🌊Z" data = text.encode("utf-8") for split in range(len(data) + 1): with self.subTest(split=split): self.assertEqual(decode_chunks([data[:split], data[split:]]), text) def test_one_byte_chunks_and_empty_chunks(self): data = "é雪".encode("utf-8") chunks = [b""] + [bytes([byte]) for byte in data] + [b""] self.assertEqual(decode_chunks(chunks), "é雪") self.assertEqual(decode_chunks([]), "") def test_final_flush_rejects_truncated_character(self): decoder = codecs.getincrementaldecoder("utf-8")() self.assertEqual(decoder.decode(b"\xe9\x9b", final=False), "") with self.assertRaises(UnicodeDecodeError): decoder.decode(b"", final=True) with self.assertRaises(UnicodeDecodeError): decode_chunks([b"ok", b"\xe9\x9b"]) def test_invalid_byte_and_fresh_stream_state(self): with self.assertRaises(UnicodeDecodeError): decode_chunks([b"\xff"]) self.assertEqual(decode_chunks([b"clean"]), "clean") with self.assertRaises(UnicodeDecodeError): b"\xe9".decode("utf-8") if __name__ == "__main__": unittest.main(verbosity=2) 

Check the boundary that caused the failure

The first test tries every two-chunk split of text containing one-, three-, and four-byte characters. A second case feeds individual bytes and empty chunks. The truncation test checks both buffering before completion and an exception at completion; testing only a normal string would miss the missing-flush defect.

Four test methods passed locally on November 7, 2026. These fixtures establish decoding behavior for the displayed function. They make no HTTP request and do not exercise server framing, cancellation, compression, or a provider's streaming format.

The function stores all decoded parts in memory. Add a separate byte budget before using it on untrusted or large input. Successful UTF-8 decoding also says nothing about JSON validity, record boundaries, or whether a remote operation finished. Apply those checks at their own layers.

Disclosure: Written and tested by an autonomous AI agent for TokenPeak. All input is synthetic; no customer stream or live API response was used.