SSE parser in routstr/upstream/base.py not fully spec-compliant (multi-line data:, CRLF, trailing flush)

[sh1ftred]

Summary

The SSE parsing in routstr/upstream/base.py was significantly improved in commit 31ff1fd (buffered \n\n-delimited events instead of re.split(b"data: ", chunk)). That fix correctly resolved the most serious bug — splitting on the data: prefix per raw chunk, which broke events spanning TCP chunk boundaries.

However, the current parser still makes assumptions that diverge from the SSE spec and from how our own JS SDK (@routstr/sdk, client/sse.ts) parses the same streams. These are latent correctness bugs that surface with any upstream that doesn't emit the exact data: {json}\n\n shape.

Affected code

routstr/upstream/base.py — both streaming generators (the chat/completions path ~line 717 and the second generator ~line 1118). Both share this logic:

sse_buffer = b""
async for chunk in response.aiter_bytes():
    sse_buffer += chunk
    while b"\n\n" in sse_buffer:
        event_raw, sse_buffer = sse_buffer.split(b"\n\n", 1)
        event_raw = event_raw.strip()
        if not event_raw:
            continue
        if event_raw == b"data: [DONE]":
            done_seen = True
            continue
        data_prefix = b"data: "
        if event_raw.startswith(data_prefix):
            payload_bytes = event_raw[len(data_prefix):]
        else:
            payload_bytes = event_raw
        obj = json.loads(payload_bytes)
        ...

Bugs

1. Only \n\n is recognized as an event delimiter; \r\n\r\n is not

Per spec, an event is terminated by a blank line, where a line ending may be \n, \r, or \r\n. The parser only splits on b"\n\n". An upstream/proxy emitting \r\n\r\n-terminated events (legal) will never satisfy b"\n\n" in sse_buffer — the buffer grows unbounded and no events are yielded until the stream closes (and even then the tail is dropped, see #3).

2. Multi-line data: events are mis-parsed

A single event may contain multiple data: lines that must be concatenated with \n to reconstruct the payload:

data: {"choices":[{"delta":
data: {"content":"hi"}}]}

The code strips only the first data: prefix, then hands the rest (still containing a literal data: on line 2) to json.loads, which raises. The event falls through to the non-JSON pass-through branch, so usage/cost is not extracted and the model/id rewriting is silently skipped for that chunk.

3. The trailing buffer is never flushed at stream end

If the upstream closes without a final \n\n (common for the last event before [DONE], or providers that omit the trailing blank line), whatever remains in sse_buffer after the async for loop is dropped. The final content chunk and/or usage chunk can be lost — there is no post-loop drain.

4. [DONE] detection is over-strict

if event_raw == b"data: [DONE]" requires exactly data: [DONE] after strip. It misses data:[DONE] (no space — also legal) and [DONE] arriving as a separate data: line, so done_seen can stay False and [DONE] may be re-emitted as a non-JSON pass-through event.

Reference implementation (our JS SDK)

@routstr/sdk (client/sse.ts) already handles all of the above:

• Splits on \r?\n\r?\n (handles \n\n, \r\n\r\n, \r\r).
• Splits each event block on \r?\n and concatenates all data: lines with \n before parsing.
• Flushes the trailing buffer on stream end.
• Treats the post-prefix payload === "[DONE]" rather than matching the full raw line, and tolerates both data: and data:.

const terminator = /\r?\n\r?\n/g;                  // event delimiter
const lines = eventBlock.split(/\r?\n/);
const dataParts: string[] = [];
for (const line of lines) {
  if (!line || line.startsWith(":")) continue;     // skip comments
  if (line.startsWith("data:")) {
    const value = line.startsWith("data: ") ? line.slice(6) : line.slice(5);
    dataParts.push(value);
  }
}
const payload = dataParts.join("\n");              // multi-line data

Suggested fix

For both streaming generators in routstr/upstream/base.py:

1. Split the buffer on a blank-line delimiter that accepts \r\n\r\n as well as \n\n (e.g. normalize \r\n → \n, or split on rb"\r?\n\r?\n"), still buffering the incomplete tail.
2. Within an event, split on line endings, collect all data: lines (strip data: or data:), and join with \n before json.loads.
3. After the async for loop ends, drain any non-empty sse_buffer as a final event.
4. Compare the reconstructed payload to [DONE] after prefix stripping, not the raw data: [DONE] bytes.

Severity / scope

For OpenAI-compatible upstreams emitting exactly data: {json}\n\n with single-line data: (the common case today), the current code works. These bugs bite when an upstream uses \r\n\r\n (#1 — total stall), sends multi-line data: (#2 — lost usage/cost, skipped model/id rewrite), or closes without a trailing blank line (#3 — dropped final chunk).

Acceptance criteria

• Events terminated by \r\n\r\n are parsed and yielded.
• Multi-line data: events are reconstructed and parsed as one JSON payload.
• The final buffered event is flushed when the upstream closes without a trailing blank line.
• [DONE] is detected for data: [DONE], data:[DONE], and as a trailing data: line.
• Regression tests covering each case (the JS SDK client/sse tests cover equivalents and can be ported).