The Problem: SSE's Hidden Fragmentation

Building a streaming chat interface with MonkeyCode's free model access, I initially thought the transport layer was straightforward. The server sent Server-Sent Events (SSE), and each event was a self-contained JSON object, a neat package of text tokens. My parser handled these perfectly, decoding them one by one. Then, a longer response arrived. Instead of a smooth stream, my terminal erupted with json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0) on the second event.

The critical realization: SSE, by design, does not guarantee that each event contains a complete, self-contained JSON document. The server wasn't failing; my parser was. It made an assumption that the SSE protocol simply doesn't uphold. When a model generates a response that's too long for a single SSE message, the server might split the JSON payload across multiple events. This fragmentation is the root cause of the decoding errors.

Diagram showing how a single JSON object can be split across multiple SSE events.

Understanding Server-Sent Events (SSE)

SSE is a lightweight protocol that enables a server to push data to a client over a single, long-lived HTTP connection. It's designed for unidirectional communication, ideal for scenarios like live updates, notifications, or, as in this case, streaming model outputs. The protocol defines a simple text-based framing mechanism. Each event is a block of text, terminated by a double newline (`\n\n`). Before the double newline, there can be multiple lines starting with specific field names, such as data:, event:, id:, and retry:. The data: field can contain arbitrary text, including JSON, but it can also be repeated within a single event, concatenating multiple lines into one data payload.

The crucial detail is that there's no inherent mechanism within the SSE specification itself to indicate whether a data: payload is a complete message or a fragment. A server might send a very long JSON string, and to avoid hitting buffer limits or to provide more frequent updates, it could chop that string into smaller pieces, sending each piece as the data field of a separate SSE event. My initial parser treated each incoming SSE event as an independent unit, attempting to decode its data field as a complete JSON document. This worked fine for short responses but failed spectacularly when the JSON was split.

The Fragmentation Scenario

Consider a JSON response that should be a single object: {"message": "This is a very long response..."}. If this string is too large for a single SSE event, the server might split it. For instance:

  • Event 1: data: {"message": "This is a very
  • Event 2: data: long response..."}

My parser received Event 1, saw data: {"message": "This is a very, and tried to decode it as JSON. It failed immediately because it's not valid JSON. It then received Event 2, saw data: long response..."}, and tried to decode that. It also failed. The problem isn't that the data is malformed; it's that the data is incomplete in each individual event.

Building a Reassembly Parser

The solution requires a parser that can buffer incoming SSE events, identify when a JSON payload is fragmented, and reassemble the fragments before attempting to decode. Here’s a conceptual approach and a Python implementation:

Buffering Incoming Events

We need a mechanism to collect the data fields from successive SSE events. A simple list or string buffer will suffice. As each SSE event arrives, we append its data payload to this buffer.

Detecting Fragmentation

The challenge is knowing when a JSON payload is complete. Since SSE doesn't provide explicit markers for JSON fragments, we must infer it. A common strategy is to attempt to parse the buffered data. If parsing succeeds, we've likely received a complete JSON object. If it fails with a JSONDecodeError, it suggests the JSON is incomplete. However, a JSONDecodeError can also occur if the JSON is malformed but complete. We need a more robust check.

A more reliable approach is to look for the structural elements of JSON. A complete JSON document must start with a `{` (for objects) or `[` (for arrays) and end with a matching `}` or `]`. If our buffered data starts with `{` and ends with `}`, or starts with `[` and ends with `]`, it's a strong candidate for a complete JSON document. We should also consider that the data field might contain multiple JSON objects, or text that isn't JSON at all, separated by newlines. A robust parser must handle these cases.

Reassembly Logic

The reassembly process involves:

  1. Initialize an empty buffer (e.g., a string).
  2. For each incoming SSE event:
    • Append the event's data to the buffer.
    • Attempt to parse the current buffer as JSON.
    • If parsing succeeds:
      • Process the valid JSON object.
      • Clear the buffer.
    • If parsing fails:
      • Check if the buffer contains a potential partial JSON structure (e.g., starts with `{` and hasn't ended with `}`). If it does, keep buffering.
      • If it seems like the end of a JSON document has been reached (e.g., buffer ends with `}` or `]`), and parsing still fails, it might indicate an issue with the JSON itself or a more complex fragmentation pattern. In a simple case, we might assume it's complete and log an error or discard.
      • If the buffer contains multiple lines or structures, we need a more sophisticated strategy to identify and extract complete JSON objects within the buffer.

Python Implementation Example

Here’s a simplified Python example using the `sseclient` library (or similar logic if building from scratch) and Python's built-in `json` module:


import json

class JsonFragmentParser:
    def __init__(self):
        self.buffer = ""

    def feed(self, data_chunk):
        self.buffer += data_chunk
        try:
            # Attempt to parse the current buffer
            parsed_json = json.loads(self.buffer)
            # If successful, we have a complete JSON object
            self.buffer = ""  # Clear buffer for next message
            return parsed_json
        except json.JSONDecodeError as e:
            # Check if the error might be due to incomplete JSON structure
            # A common heuristic: if it starts with '{' or '[' and ends before '}' or ']', it's likely fragmented.
            # This is a simplification; real-world scenarios might need more complex state tracking.
            if (self.buffer.startswith('{') and not self.buffer.endswith('}')) or \
               (self.buffer.startswith('[') and not self.buffer.endswith(']')):
                # It's likely a fragment, continue buffering
                return None
            else:
                # It's either malformed JSON, or an unexpected structure.
                # For this example, we'll clear the buffer and indicate failure.
                print(f"JSON Decode Error (potentially malformed or complex fragmentation): {e}")
                print(f"Buffer content: {self.buffer}")
                self.buffer = ""
                return None # Indicate failure or non-JSON data
        except Exception as e:
            print(f"An unexpected error occurred: {e}")
            self.buffer = ""
            return None

# Example Usage (simulating SSE events):
# parser = JsonFragmentParser()
# sse_events = [
#     '{"message": "Hello, ", "role": "assistant"}\n\n',
#     '"partial": "response"}\n\n', # This is the fragmented part
#     '{"message": "World!", "role": "assistant"}\n\n'
# ]
# for event_data in sse_events:
#     # In a real SSE client, you'd extract the 'data' field from the event
#     # For simplicity, assuming event_data is the raw data field content
#     # and assuming it's already stripped of SSE framing like 'data: '
#     
#     # Simulate extracting data payload
#     data_payload = event_data.strip().replace('data: ', '')
#     if data_payload:
#         decoded_json = parser.feed(data_payload)
#         if decoded_json:
#             print("Decoded JSON:", decoded_json)

# A more accurate simulation of fragmented JSON:
# Assume the server sends:
# Event 1: data: {"id": 1, "text": "This is "
# Event 2: data: "a fragmented JSON message."}

# parser = JsonFragmentParser()
# print("Feeding chunk 1:")
# result1 = parser.feed('{"id": 1, "text": "This is "')
# print(f"Result 1: {result1}") # Should be None
# print(f"Buffer after chunk 1: '{parser.buffer}'")

# print("\nFeeding chunk 2:")
# result2 = parser.feed('"a fragmented JSON message."')
# print(f"Result 2: {result2}") # Should be the full JSON object
# print(f"Buffer after chunk 2: '{parser.buffer}'")

# Example with multiple complete JSONs:
# parser = JsonFragmentParser()
# print("\nFeeding complete JSON 1:")
# result3 = parser.feed('{"status": "ok"}')
# print(f"Result 3: {result3}") # Should be {"status": "ok"}
# print(f"Buffer after JSON 1: '{parser.buffer}'")

# print("\nFeeding complete JSON 2:")
# result4 = parser.feed('{"status": "done"}')
# print(f"Result 4: {result4}") # Should be {"status": "done"}
# print(f"Buffer after JSON 2: '{parser.buffer}'")