fix(proxy): harden SSE DONE tracker — spec multi-line data joins, truthy lastOne, EOF-write guard

Follow-ups on the salvaged cluster:
- sse_done.py: dispatch SSE events at blank-line boundaries and join
  consecutive data: lines per the SSE spec (a split JSON event no longer
  reads as two malformed fragments that disable synthesis)
- accept integer/string-truthy lastOne sentinels (1 / "true") in both the
  proxy tracker and the agent stream reader
- server.py: guard the [DONE] append against client hangup at EOF and
  widen the interrupt tuple with OSError
- contributor email mappings for loulanyue and jon-nielsen
This commit is contained in:
Teknium
2026-09-01 09:45:17 -07:00
parent d304422b3d
commit 93591eccb5
4 changed files with 32 additions and 6 deletions

View File

@@ -4309,7 +4309,9 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
extra = getattr(chunk, "model_extra", None)
if isinstance(extra, dict):
last_one = extra.get("lastOne")
if last_one is True and finish_reason is None:
# Integer/string-truthy sentinels included — relabelled
# upstreams have been seen sending 1 / "true".
if last_one in (True, 1, "true") and finish_reason is None:
finish_reason = "stop"
continue

View File

@@ -0,0 +1 @@
jon-nielsen

View File

@@ -248,8 +248,11 @@ def create_app(adapter: UpstreamAdapter) -> "web.Application":
done_tracker.feed(chunk)
await resp.write(chunk)
if done_tracker is not None and done_tracker.should_append_done():
await resp.write(DONE_SSE_FRAME)
except (aiohttp.ClientError, asyncio.CancelledError) as exc:
try:
await resp.write(DONE_SSE_FRAME)
except Exception as exc: # client hung up at EOF — harmless
logger.debug("proxy: DONE append skipped: %s", exc)
except (aiohttp.ClientError, asyncio.CancelledError, OSError) as exc:
if done_tracker is not None:
done_tracker.mark_interrupted()
logger.warning("proxy: streaming interrupted: %s", exc)

View File

@@ -41,6 +41,7 @@ class SseDoneTracker:
saw_malformed_event: bool = False
interrupted: bool = False
_buf: bytearray = field(default_factory=bytearray, repr=False)
_data_lines: list = field(default_factory=list, repr=False)
def feed(self, chunk: bytes) -> None:
"""Observe a forwarded chunk (bytes are not modified)."""
@@ -68,10 +69,12 @@ class SseDoneTracker:
or self.saw_malformed_event
):
return False
# Flush any trailing line without a final newline (rare but valid).
# Flush any trailing line without a final newline (rare but valid),
# then dispatch a final event that never saw its blank-line boundary.
if self._buf:
self._consume_line(bytes(self._buf))
self._buf.clear()
self._dispatch_event()
if self.saw_done or self.saw_error_event or self.saw_malformed_event:
return False
return self.saw_terminal_finish or self.saw_last_one
@@ -80,9 +83,24 @@ class SseDoneTracker:
# Strip CR from CRLF-delimited SSE.
if line.endswith(b"\r"):
line = line[:-1]
if not line:
# Blank line = SSE event boundary: dispatch accumulated data.
self._dispatch_event()
return
if not line.startswith(b"data:"):
return
payload = line[5:].strip()
# Per the SSE spec one event may span several consecutive ``data:``
# lines whose payloads are joined with "\n" at dispatch time.
# Parsing each line independently would misread a split JSON event
# as two malformed fragments.
self._data_lines.append(line[5:].strip())
def _dispatch_event(self) -> None:
if not self._data_lines:
return
payload = b"\n".join(self._data_lines)
self._data_lines = []
payload = payload.strip()
if payload == b"[DONE]":
self.saw_done = True
return
@@ -103,7 +121,9 @@ class SseDoneTracker:
if event.get("error") is not None:
self.saw_error_event = True
return
if event.get("lastOne") is True:
# Accept integer-truthy sentinels too — relabelled upstreams have
# been observed sending ``"lastOne": 1`` / ``"true"``.
if event.get("lastOne") in (True, 1, "true"):
self.saw_last_one = True
for choice in event.get("choices") or []:
if not isinstance(choice, dict):