From 93591eccb5aee680b06ff0a28dbc5804b1cabbc2 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 1 Sep 2026 09:45:17 -0700 Subject: [PATCH] =?UTF-8?q?fix(proxy):=20harden=20SSE=20DONE=20tracker=20?= =?UTF-8?q?=E2=80=94=20spec=20multi-line=20data=20joins,=20truthy=20lastOn?= =?UTF-8?q?e,=20EOF-write=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- agent/chat_completion_helpers.py | 4 +++- contributors/emails/jon@cromulent.dk | 1 + hermes_cli/proxy/server.py | 7 +++++-- hermes_cli/proxy/sse_done.py | 26 +++++++++++++++++++++++--- 4 files changed, 32 insertions(+), 6 deletions(-) create mode 100644 contributors/emails/jon@cromulent.dk diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 0175469ed6..e24c71e52d 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -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 diff --git a/contributors/emails/jon@cromulent.dk b/contributors/emails/jon@cromulent.dk new file mode 100644 index 0000000000..aab1990bb0 --- /dev/null +++ b/contributors/emails/jon@cromulent.dk @@ -0,0 +1 @@ +jon-nielsen diff --git a/hermes_cli/proxy/server.py b/hermes_cli/proxy/server.py index 5f8968df4b..880ac4830a 100644 --- a/hermes_cli/proxy/server.py +++ b/hermes_cli/proxy/server.py @@ -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) diff --git a/hermes_cli/proxy/sse_done.py b/hermes_cli/proxy/sse_done.py index 2a5ed9a72e..bf2f69e832 100644 --- a/hermes_cli/proxy/sse_done.py +++ b/hermes_cli/proxy/sse_done.py @@ -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):