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):