mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-08-09 02:54:37 +00:00
Merge pull request #543 from Routstr/pr-sse-parser
fix: buffer-based SSE parser for all supported providers
This commit is contained in:
9qeklajc
+220
-96
@@ -2,10 +2,9 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
import traceback
|
||||
import uuid
|
||||
from collections.abc import AsyncGenerator, AsyncIterator
|
||||
from collections.abc import AsyncGenerator, AsyncIterator, Iterator
|
||||
from typing import Any, Mapping, cast
|
||||
|
||||
import httpx
|
||||
@@ -728,56 +727,140 @@ class BaseUpstreamProvider:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _process_event(raw_event: bytes) -> Iterator[bytes]:
|
||||
"""Process one complete SSE event block (lines up to a blank line).
|
||||
|
||||
Handles arbitrary upstream framing across every supported
|
||||
provider:
|
||||
|
||||
* ``data:`` lines are gathered and concatenated per the SSE
|
||||
spec, so a payload split across network chunks is reassembled
|
||||
before parsing.
|
||||
* Comment/keepalive lines (those beginning with ``:`` such as
|
||||
OpenRouter's ``: OPENROUTER PROCESSING``) are dropped. They
|
||||
carry no JSON and forwarding them downstream breaks naive SSE
|
||||
clients; the keepalive only matters for the upstream hop.
|
||||
* Other SSE fields (``event:``/``id:``/``retry:``) are preserved
|
||||
and kept attached to the event's ``data:`` line, which the
|
||||
OpenAI Responses API and Anthropic-style streams rely on.
|
||||
* ``[DONE]`` is swallowed so it can be re-emitted exactly once at
|
||||
end of stream.
|
||||
"""
|
||||
nonlocal last_model_seen, usage_chunk_data, done_seen
|
||||
|
||||
event = raw_event.strip(b"\r\n")
|
||||
if not event:
|
||||
return
|
||||
|
||||
field_lines: list[bytes] = []
|
||||
data_lines: list[bytes] = []
|
||||
for line in event.split(b"\n"):
|
||||
line = line.rstrip(b"\r")
|
||||
if line.startswith(b"data:"):
|
||||
# Strip the field name and a single optional leading space.
|
||||
data_lines.append(line[len(b"data:") :].lstrip(b" "))
|
||||
elif line.startswith(b":"):
|
||||
# SSE comment / keepalive - drop.
|
||||
continue
|
||||
elif line:
|
||||
# Other SSE field (event:/id:/retry:) - preserve in order.
|
||||
field_lines.append(line)
|
||||
|
||||
if not data_lines:
|
||||
return
|
||||
|
||||
data = b"\n".join(data_lines)
|
||||
if not data.strip():
|
||||
return
|
||||
|
||||
# Re-emit preserved SSE fields immediately before the data line so
|
||||
# event/data framing stays intact (single trailing newline each;
|
||||
# the blank-line terminator is appended to the data line below).
|
||||
prefix = b"".join(fl + b"\n" for fl in field_lines)
|
||||
|
||||
if data.strip() == b"[DONE]":
|
||||
done_seen = True
|
||||
return
|
||||
|
||||
try:
|
||||
obj = json.loads(data)
|
||||
except Exception:
|
||||
obj = None
|
||||
|
||||
if isinstance(obj, dict):
|
||||
self._apply_provider_field(obj)
|
||||
if obj.get("model"):
|
||||
last_model_seen = str(obj.get("model"))
|
||||
if requested_model:
|
||||
obj["model"] = requested_model
|
||||
if (
|
||||
"id" not in obj
|
||||
or not isinstance(obj["id"], str)
|
||||
or obj["id"] == "existing-id"
|
||||
):
|
||||
if not hasattr(self, "_current_stream_id"):
|
||||
self._current_stream_id = f"chatcmpl-{uuid.uuid4()}"
|
||||
obj["id"] = self._current_stream_id
|
||||
if isinstance(obj.get("usage"), dict):
|
||||
# Capture usage for end-of-stream cost reconciliation.
|
||||
# Some models (e.g. Gemini thinking models over the
|
||||
# OpenAI-compat endpoint) attach ``usage`` to the SAME
|
||||
# chunk that carries the final content/finish_reason
|
||||
# rather than sending a separate ``choices: []`` usage
|
||||
# chunk. Only swallow the chunk when it is a pure usage
|
||||
# chunk (no choices); otherwise the content would be
|
||||
# silently dropped and the client would receive no
|
||||
# assistant message at all.
|
||||
if obj.get("choices"):
|
||||
# Capture usage (with model) for the cost trailer,
|
||||
# but with choices stripped so the trailer never
|
||||
# re-emits this chunk's content.
|
||||
usage_chunk_data = {
|
||||
k: v for k, v in obj.items() if k != "choices"
|
||||
}
|
||||
usage_chunk_data["choices"] = []
|
||||
# Forward the content now, without usage, so token
|
||||
# usage is reported exactly once (in the trailer).
|
||||
forward = {k: v for k, v in obj.items() if k != "usage"}
|
||||
yield (
|
||||
prefix
|
||||
+ b"data: "
|
||||
+ json.dumps(forward).encode()
|
||||
+ b"\n\n"
|
||||
)
|
||||
return
|
||||
usage_chunk_data = obj
|
||||
return
|
||||
yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n"
|
||||
else:
|
||||
# Non-JSON data payload (partial fragment already reassembled
|
||||
# by buffering, or a provider control string). Re-prefix each
|
||||
# line so multi-line ``data`` stays valid SSE framing - a bare
|
||||
# second line would otherwise reach the client without its
|
||||
# ``data:`` field and break naive parsers.
|
||||
body = b"".join(
|
||||
b"data: " + ln + b"\n" for ln in data.split(b"\n")
|
||||
)
|
||||
yield prefix + body + b"\n"
|
||||
|
||||
try:
|
||||
# Buffer bytes across network chunks and dispatch only on the SSE
|
||||
# event delimiter (a blank line). ``aiter_bytes`` yields arbitrary
|
||||
# byte boundaries, so a single event's JSON can span chunks and
|
||||
# multiple events can arrive together; buffering makes parsing
|
||||
# boundary-independent for every provider.
|
||||
buffer = b""
|
||||
async for chunk in response.aiter_bytes():
|
||||
# Split chunk into SSE events
|
||||
parts = re.split(b"data: ", chunk)
|
||||
for i, part in enumerate(parts):
|
||||
if not part:
|
||||
continue
|
||||
buffer += chunk.replace(b"\r\n", b"\n")
|
||||
while b"\n\n" in buffer:
|
||||
raw_event, buffer = buffer.split(b"\n\n", 1)
|
||||
for out in _process_event(raw_event):
|
||||
yield out
|
||||
|
||||
stripped_part = part.strip()
|
||||
if not stripped_part:
|
||||
continue
|
||||
|
||||
if stripped_part == b"[DONE]":
|
||||
done_seen = True
|
||||
continue
|
||||
|
||||
try:
|
||||
# Only parse if it looks like a JSON object to avoid SSE control messages or partials
|
||||
if part.strip().startswith(b"{") and part.strip().endswith(
|
||||
b"}"
|
||||
):
|
||||
obj = json.loads(part)
|
||||
if isinstance(obj, dict):
|
||||
self._apply_provider_field(obj)
|
||||
if obj.get("model"):
|
||||
last_model_seen = str(obj.get("model"))
|
||||
if requested_model:
|
||||
obj["model"] = requested_model
|
||||
if (
|
||||
"id" not in obj
|
||||
or not isinstance(obj["id"], str)
|
||||
or obj["id"] == "existing-id"
|
||||
):
|
||||
if not hasattr(self, "_current_stream_id"):
|
||||
self._current_stream_id = (
|
||||
f"chatcmpl-{uuid.uuid4()}"
|
||||
)
|
||||
obj["id"] = self._current_stream_id
|
||||
if isinstance(obj.get("usage"), dict):
|
||||
usage_chunk_data = obj
|
||||
continue
|
||||
yield b"data: " + json.dumps(obj).encode() + b"\n\n"
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
prefix = (
|
||||
b"data: " if (i > 0 or chunk.startswith(b"data: ")) else b""
|
||||
)
|
||||
yield prefix + part
|
||||
# Flush any trailing event that lacked a final blank line.
|
||||
if buffer.strip():
|
||||
for out in _process_event(buffer):
|
||||
yield out
|
||||
|
||||
async with create_session() as session:
|
||||
fresh_key = await session.get(key.__class__, key.hashed_key)
|
||||
@@ -1079,56 +1162,97 @@ class BaseUpstreamProvider:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _process_event(raw_event: bytes) -> Iterator[bytes]:
|
||||
"""Process one complete SSE event block for the Responses API.
|
||||
|
||||
Buffers full events (delimited by a blank line) so parsing is
|
||||
boundary-independent, gathers ``data:`` lines, drops comment/
|
||||
keepalive lines (e.g. OpenRouter's ``: OPENROUTER PROCESSING``),
|
||||
and preserves ``event:``/``id:`` fields attached to their data
|
||||
line so Responses API event framing stays intact.
|
||||
"""
|
||||
nonlocal last_model_seen, usage_chunk_data, done_seen
|
||||
nonlocal reasoning_tokens
|
||||
|
||||
event = raw_event.strip(b"\r\n")
|
||||
if not event:
|
||||
return
|
||||
|
||||
field_lines: list[bytes] = []
|
||||
data_lines: list[bytes] = []
|
||||
for line in event.split(b"\n"):
|
||||
line = line.rstrip(b"\r")
|
||||
if line.startswith(b"data:"):
|
||||
data_lines.append(line[len(b"data:") :].lstrip(b" "))
|
||||
elif line.startswith(b":"):
|
||||
# SSE comment / keepalive - drop.
|
||||
continue
|
||||
elif line:
|
||||
# Preserve event:/id:/retry: (Responses API event names).
|
||||
field_lines.append(line)
|
||||
|
||||
if not data_lines:
|
||||
return
|
||||
|
||||
data = b"\n".join(data_lines)
|
||||
if not data.strip():
|
||||
return
|
||||
|
||||
prefix = b"".join(fl + b"\n" for fl in field_lines)
|
||||
|
||||
if data.strip() == b"[DONE]":
|
||||
done_seen = True
|
||||
return
|
||||
|
||||
try:
|
||||
obj = json.loads(data)
|
||||
except json.JSONDecodeError:
|
||||
obj = None
|
||||
|
||||
if isinstance(obj, dict):
|
||||
self._apply_provider_field(obj)
|
||||
if obj.get("model"):
|
||||
last_model_seen = str(obj.get("model"))
|
||||
if requested_model:
|
||||
obj["model"] = requested_model
|
||||
|
||||
# Track reasoning tokens for Responses API
|
||||
if usage := obj.get("usage", {}):
|
||||
if isinstance(usage, dict) and "reasoning_tokens" in usage:
|
||||
reasoning_tokens += usage.get("reasoning_tokens", 0)
|
||||
|
||||
# Responses API usage is in response.completed/incomplete events
|
||||
chunk_type = obj.get("type", "")
|
||||
if chunk_type in (
|
||||
"response.completed",
|
||||
"response.incomplete",
|
||||
):
|
||||
usage_chunk_data = obj
|
||||
return
|
||||
|
||||
yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n"
|
||||
else:
|
||||
# Re-prefix each line so multi-line ``data`` stays valid SSE
|
||||
# framing for the client.
|
||||
body = b"".join(
|
||||
b"data: " + ln + b"\n" for ln in data.split(b"\n")
|
||||
)
|
||||
yield prefix + body + b"\n"
|
||||
|
||||
try:
|
||||
# Buffer across network chunks; dispatch only on the SSE event
|
||||
# delimiter so parsing is independent of byte boundaries.
|
||||
buffer = b""
|
||||
async for chunk in response.aiter_bytes():
|
||||
# Split chunk into SSE events
|
||||
parts = re.split(b"data: ", chunk)
|
||||
for i, part in enumerate(parts):
|
||||
if not part:
|
||||
continue
|
||||
buffer += chunk.replace(b"\r\n", b"\n")
|
||||
while b"\n\n" in buffer:
|
||||
raw_event, buffer = buffer.split(b"\n\n", 1)
|
||||
for out in _process_event(raw_event):
|
||||
yield out
|
||||
|
||||
stripped_part = part.strip()
|
||||
if not stripped_part:
|
||||
continue
|
||||
|
||||
if stripped_part == b"[DONE]":
|
||||
done_seen = True
|
||||
continue
|
||||
|
||||
try:
|
||||
obj = json.loads(part)
|
||||
if isinstance(obj, dict):
|
||||
self._apply_provider_field(obj)
|
||||
if obj.get("model"):
|
||||
last_model_seen = str(obj.get("model"))
|
||||
if requested_model:
|
||||
obj["model"] = requested_model
|
||||
|
||||
# Track reasoning tokens for Responses API
|
||||
if usage := obj.get("usage", {}):
|
||||
if (
|
||||
isinstance(usage, dict)
|
||||
and "reasoning_tokens" in usage
|
||||
):
|
||||
reasoning_tokens += usage.get(
|
||||
"reasoning_tokens", 0
|
||||
)
|
||||
|
||||
# Responses API usage is in response.completed/incomplete events
|
||||
chunk_type = obj.get("type", "")
|
||||
if chunk_type in (
|
||||
"response.completed",
|
||||
"response.incomplete",
|
||||
):
|
||||
usage_chunk_data = obj
|
||||
continue
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
prefix = (
|
||||
b"data: " if (i > 0 or chunk.startswith(b"data: ")) else b""
|
||||
)
|
||||
yield prefix + part
|
||||
if buffer.strip():
|
||||
for out in _process_event(buffer):
|
||||
yield out
|
||||
|
||||
# Always emit a cost-bearing data chunk
|
||||
async with create_session() as session:
|
||||
|
||||
@@ -174,17 +174,20 @@ class TestmintWallet:
|
||||
"""Fallback method to create a basic test token"""
|
||||
import base64
|
||||
import json
|
||||
import random
|
||||
import time
|
||||
import secrets
|
||||
|
||||
unique_id = int(time.time() * 1000000) + random.randint(1000, 9999)
|
||||
# Use a cryptographically random id/secret so every minted token is
|
||||
# guaranteed unique. Time/PRNG-based ids can collide on hosts with
|
||||
# coarse clock resolution, producing byte-identical tokens that hash to
|
||||
# the same api_key and silently dedupe (flaky concurrency tests).
|
||||
unique_id = secrets.token_hex(16)
|
||||
token_data = {
|
||||
"token": [
|
||||
{
|
||||
"mint": self.mint_url,
|
||||
"proofs": [
|
||||
{
|
||||
"id": f"009a1f293253e41e{unique_id % 100000000:08d}",
|
||||
"id": f"009a1f293253e41e{unique_id[:8]}",
|
||||
"amount": amount,
|
||||
"secret": f"test-secret-{amount}-{unique_id}",
|
||||
"C": "02194603ffa36356f4a56b7df9371fc3192472351453ec7398b8da8117e7c3e104",
|
||||
|
||||
@@ -0,0 +1,339 @@
|
||||
"""Battle-test the streaming SSE parser against real per-provider framing.
|
||||
|
||||
Each test drives the *actual* ``handle_streaming_chat_completion`` generator
|
||||
with a mock upstream response whose ``aiter_bytes`` emits byte sequences that
|
||||
mirror what each supported provider sends on the wire (captured from the
|
||||
providers' own streaming docs):
|
||||
|
||||
* OpenAI / Groq / Fireworks / xAI / Perplexity / Azure - plain
|
||||
``data: {json}\\n\\n`` + ``data: [DONE]``.
|
||||
* OpenRouter - same, but with ``: OPENROUTER PROCESSING`` keepalive comments
|
||||
interleaved (the framing that produced the original
|
||||
``Unexpected token ':'`` client crash).
|
||||
* Gemini (native ``alt=sse``) - ``data:`` payloads framed with CRLF.
|
||||
|
||||
The invariant every provider must satisfy: every line the proxy emits that
|
||||
starts with ``data: `` either equals ``[DONE]`` or is valid JSON, and no SSE
|
||||
comment ever reaches the client. That invariant is exactly what the buggy
|
||||
``re.split(b"data: ")`` parser violated for OpenRouter.
|
||||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import AsyncGenerator
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from routstr.core.db import ApiKey
|
||||
from routstr.upstream import base
|
||||
from routstr.upstream.base import BaseUpstreamProvider
|
||||
|
||||
|
||||
def _make_response(chunks: list[bytes]) -> MagicMock:
|
||||
async def aiter_bytes() -> AsyncGenerator[bytes, None]:
|
||||
for chunk in chunks:
|
||||
yield chunk
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.headers = {"content-type": "text/event-stream"}
|
||||
mock_response.aiter_bytes = aiter_bytes
|
||||
return mock_response
|
||||
|
||||
|
||||
async def _drive(chunks: list[bytes], requested_model: str | None = None) -> list[bytes]:
|
||||
"""Run the real streaming generator over ``chunks`` and collect output bytes."""
|
||||
provider = BaseUpstreamProvider(
|
||||
base_url="https://api.example.com", api_key="test_key"
|
||||
)
|
||||
|
||||
key = MagicMock(spec=ApiKey)
|
||||
key.hashed_key = "test_hash"
|
||||
key.balance = 1000
|
||||
|
||||
base.adjust_payment_for_tokens = AsyncMock(
|
||||
return_value={"total_usd": 0.1, "total_msats": 100}
|
||||
)
|
||||
mock_session = MagicMock()
|
||||
mock_session.get = AsyncMock(return_value=key)
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.__aenter__ = AsyncMock(return_value=mock_session)
|
||||
mock_ctx.__aexit__ = AsyncMock(return_value=None)
|
||||
base.create_session = MagicMock(return_value=mock_ctx)
|
||||
|
||||
streaming_response = await provider.handle_streaming_chat_completion(
|
||||
response=_make_response(chunks),
|
||||
key=key,
|
||||
max_cost_for_model=100,
|
||||
background_tasks=MagicMock(),
|
||||
requested_model=requested_model,
|
||||
)
|
||||
|
||||
out: list[bytes] = []
|
||||
async for chunk in streaming_response.body_iterator:
|
||||
if isinstance(chunk, str):
|
||||
out.append(chunk.encode())
|
||||
else:
|
||||
out.append(bytes(chunk))
|
||||
return out
|
||||
|
||||
|
||||
def _data_payloads(out: list[bytes]) -> list[bytes]:
|
||||
"""Return the raw payload of every ``data: `` line across all emitted bytes."""
|
||||
payloads: list[bytes] = []
|
||||
for chunk in out:
|
||||
for line in chunk.split(b"\n"):
|
||||
if line.startswith(b"data: "):
|
||||
payloads.append(line[len(b"data: ") :])
|
||||
return payloads
|
||||
|
||||
|
||||
def _assert_clean(out: list[bytes]) -> list[dict]:
|
||||
"""Core invariant: every data line is [DONE] or valid JSON; no comments leak."""
|
||||
blob = b"".join(out)
|
||||
# No SSE comment line must ever reach the client.
|
||||
for line in blob.split(b"\n"):
|
||||
assert not line.startswith(b":"), f"comment leaked to client: {line!r}"
|
||||
# The original bug signature: a data line whose value is itself a comment.
|
||||
assert not line.startswith(b"data: :"), f"mangled comment frame: {line!r}"
|
||||
|
||||
objs: list[dict] = []
|
||||
for payload in _data_payloads(out):
|
||||
stripped = payload.strip()
|
||||
if stripped == b"[DONE]":
|
||||
continue
|
||||
obj = json.loads(stripped) # raises if the proxy emitted non-JSON data
|
||||
objs.append(obj)
|
||||
return objs
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_openai_style_plain_stream() -> None:
|
||||
"""OpenAI / Groq / Fireworks / xAI / Perplexity: plain data + [DONE]."""
|
||||
chunks = [
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":"Hello"}}]}\n\n',
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":" world"}}]}\n\n',
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out)
|
||||
assert any(o.get("choices") for o in objs)
|
||||
assert b"data: [DONE]\n\n" in b"".join(out)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_openrouter_keepalive_comments() -> None:
|
||||
"""OpenRouter ``: OPENROUTER PROCESSING`` keepalives must never crash clients.
|
||||
|
||||
This is the exact regression: the old parser emitted
|
||||
``data: : OPENROUTER PROCESSING`` which made downstream
|
||||
``JSON.parse`` throw ``Unexpected token ':'``.
|
||||
"""
|
||||
chunks = [
|
||||
b": OPENROUTER PROCESSING\n\n",
|
||||
b": OPENROUTER PROCESSING\n\n",
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":"Hi"}}]}\n\n',
|
||||
b": OPENROUTER PROCESSING\n\n",
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":"!"}}]}\n\n',
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out)
|
||||
# The keepalive must be gone entirely.
|
||||
assert b"OPENROUTER PROCESSING" not in b"".join(out)
|
||||
# Real content survived.
|
||||
contents = [
|
||||
c["delta"]["content"]
|
||||
for o in objs
|
||||
for c in o.get("choices", [])
|
||||
if "delta" in c
|
||||
]
|
||||
assert "Hi" in contents and "!" in contents
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_openrouter_comment_glued_to_data_chunk() -> None:
|
||||
"""Keepalive packed into the same TCP chunk as data (the harder case)."""
|
||||
chunks = [
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":"a"}}]}\n\n'
|
||||
b": OPENROUTER PROCESSING\n\n"
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":"b"}}]}\n\n',
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out)
|
||||
contents = [
|
||||
c["delta"]["content"]
|
||||
for o in objs
|
||||
for c in o.get("choices", [])
|
||||
if "delta" in c
|
||||
]
|
||||
assert contents == ["a", "b"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_json_split_across_chunk_boundary() -> None:
|
||||
"""A single event's JSON arriving in two TCP reads must reassemble."""
|
||||
chunks = [
|
||||
b'data: {"id":"x","choices":[{"delta":{"con',
|
||||
b'tent":"split"}}]}\n\n',
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out)
|
||||
contents = [
|
||||
c["delta"]["content"]
|
||||
for o in objs
|
||||
for c in o.get("choices", [])
|
||||
if "delta" in c
|
||||
]
|
||||
assert contents == ["split"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_byte_by_byte_fragmentation() -> None:
|
||||
"""Pathological framing: one byte per chunk. Must still parse cleanly."""
|
||||
raw = (
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":"drip"}}]}\n\n'
|
||||
b": OPENROUTER PROCESSING\n\n"
|
||||
b"data: [DONE]\n\n"
|
||||
)
|
||||
chunks = [raw[i : i + 1] for i in range(len(raw))]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out)
|
||||
assert objs and objs[0]["choices"][0]["delta"]["content"] == "drip"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_gemini_crlf_framing() -> None:
|
||||
"""Gemini native (alt=sse) frames events with CRLF."""
|
||||
chunks = [
|
||||
b'data: {"id":"g","choices":[{"delta":{"content":"hej"}}]}\r\n\r\n',
|
||||
b'data: {"id":"g","choices":[{"delta":{"content":"!"}}]}\r\n\r\n',
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out)
|
||||
contents = [
|
||||
c["delta"]["content"]
|
||||
for o in objs
|
||||
for c in o.get("choices", [])
|
||||
if "delta" in c
|
||||
]
|
||||
assert contents == ["hej", "!"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_azure_leading_role_chunk() -> None:
|
||||
"""Azure OpenAI opens with a content-filter / role-only chunk."""
|
||||
chunks = [
|
||||
b'data: {"id":"az","choices":[],"prompt_filter_results":[]}\n\n',
|
||||
b'data: {"id":"az","choices":[{"delta":{"role":"assistant"}}]}\n\n',
|
||||
b'data: {"id":"az","choices":[{"delta":{"content":"ok"}}]}\n\n',
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
_assert_clean(out)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_openrouter_mid_stream_error_event() -> None:
|
||||
"""OpenRouter mid-stream errors arrive as a normal data JSON event."""
|
||||
err = {
|
||||
"id": "x",
|
||||
"object": "chat.completion.chunk",
|
||||
"model": "openai/gpt-4o",
|
||||
"error": {"code": "server_error", "message": "Provider disconnected"},
|
||||
"choices": [{"index": 0, "delta": {"content": ""}, "finish_reason": "error"}],
|
||||
}
|
||||
chunks = [
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":"partial"}}]}\n\n',
|
||||
b"data: " + json.dumps(err).encode() + b"\n\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out)
|
||||
assert any("error" in o for o in objs), "error event must be forwarded intact"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_gemini_combined_content_and_usage_chunk() -> None:
|
||||
"""Gemini thinking models pack usage into the final *content* chunk.
|
||||
|
||||
Regression: the parser swallowed any chunk carrying a ``usage`` dict, so
|
||||
when content + usage arrived together the assistant text was dropped and
|
||||
the client saw "no assistant messages" despite a 200 + token accounting.
|
||||
"""
|
||||
chunks = [
|
||||
b'data: {"id":"g","choices":[{"delta":{"content":"the answer"},'
|
||||
b'"finish_reason":"stop"}],"usage":{"prompt_tokens":3,'
|
||||
b'"completion_tokens":2,"total_tokens":5}}\n\n',
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out)
|
||||
contents = [
|
||||
c["delta"]["content"]
|
||||
for o in objs
|
||||
for c in o.get("choices", [])
|
||||
if "delta" in c
|
||||
]
|
||||
# Content delivered exactly once (not dropped, not duplicated by the trailer).
|
||||
assert contents == ["the answer"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_separate_usage_chunk_not_forwarded_as_content() -> None:
|
||||
"""A pure usage chunk (choices: []) is still swallowed, content intact."""
|
||||
chunks = [
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":"hello"}}]}\n\n',
|
||||
b'data: {"id":"x","choices":[],"usage":{"total_tokens":4}}\n\n',
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out)
|
||||
contents = [
|
||||
c["delta"]["content"]
|
||||
for o in objs
|
||||
for c in o.get("choices", [])
|
||||
if "delta" in c
|
||||
]
|
||||
assert contents == ["hello"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_requested_model_override_applied() -> None:
|
||||
"""Model rewriting still works through the buffered parser."""
|
||||
chunks = [
|
||||
b'data: {"id":"x","model":"upstream-model","choices":[{"delta":{"content":"hi"}}]}\n\n',
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
out = await _drive(chunks, requested_model="routstr-model")
|
||||
objs = _assert_clean(out)
|
||||
# The upstream content chunk carried model "upstream-model"; the parser must
|
||||
# rewrite it to the requested model. (The trailing routstr-generated usage
|
||||
# chunk is excluded - it is not an upstream-forwarded chunk.)
|
||||
content_chunks = [o for o in objs if o.get("choices")]
|
||||
assert content_chunks, "expected at least one forwarded content chunk"
|
||||
assert all(o.get("model") == "routstr-model" for o in content_chunks)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiline_non_json_data_each_line_prefixed() -> None:
|
||||
"""A multi-line non-JSON ``data`` block must keep a ``data:`` prefix per line.
|
||||
|
||||
Two ``data:`` lines in one event reassemble to ``line one\\nline two``, which
|
||||
is not JSON, so it takes the raw-forward path. The parser must re-prefix each
|
||||
line; a bare second line would reach the client without its ``data:`` field
|
||||
and break naive SSE parsers.
|
||||
"""
|
||||
chunks = [
|
||||
b"data: line one\ndata: line two\n\n",
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
blob = b"".join(out)
|
||||
for line in blob.split(b"\n"):
|
||||
stripped = line.strip()
|
||||
if not stripped or stripped == b"[DONE]":
|
||||
continue
|
||||
assert line.startswith(b"data: "), f"bare line leaked to client: {line!r}"
|
||||
assert b"data: line one" in blob and b"data: line two" in blob
|
||||
Reference in New Issue
Block a user