fix(proxy): append SSE [DONE] when Nous streams omit the sentinel

Complete Portal streams can finish with finish_reason/lastOne and clean
EOF without data: [DONE], which strict OpenAI clients treat as truncation.
Normalize at the hermes proxy boundary after clean EOF only.

Co-authored-by: Cursor <cursoragent@cursor.com>

agent/chat_completion_helpers.py

@@ -4301,6 +4301,16 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
                         ),
                         raw_text=f"{_err_type}: {_err_msg}",
                     )
+                # Nous Portal usage frames often have choices=[] plus
+                # lastOne=true and no [DONE]. Treat that as a clean
+                # terminal, not a mid-stream drop (#90848).
+                last_one = getattr(chunk, "lastOne", None)
+                if last_one is None:
+                    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:
+                    finish_reason = "stop"
                 continue
 
             delta = chunk.choices[0].delta

hermes_cli/proxy/server.py

@@ -3,10 +3,18 @@
 Listens on ``http://<host>:<port>/v1/<path>`` and forwards each request to
 ``<upstream-base-url>/<path>`` with the client's ``Authorization`` header
 replaced by a freshly-resolved bearer from the configured adapter. The
-response is streamed back unmodified, preserving SSE.
+response body is streamed through unchanged (SSE deltas preserved).
 
-The server is intentionally minimal: it does NOT mediate, log, transform,
-or rewrite request/response bodies. It's a credential-attaching forwarder.
+One narrow SSE compatibility shim applies after a *clean* upstream EOF:
+when a ``text/event-stream`` response carries a terminal ``finish_reason``
+or ``lastOne: true`` but omits the OpenAI ``data: [DONE]`` sentinel, the
+proxy appends a single ``[DONE]`` frame. It never rewrites earlier frames,
+never duplicates an upstream ``[DONE]``, and never synthesizes ``[DONE]``
+after an error event or a mid-stream interrupt (see
+:mod:`hermes_cli.proxy.sse_done`, issue #90848).
+
+Otherwise the server does not mediate, log, or rewrite request/response
+bodies — it is a credential-attaching forwarder.
 """
 
 from __future__ import annotations
@@ -26,6 +34,11 @@ except ImportError:
     AIOHTTP_AVAILABLE = False
 
 from hermes_cli.proxy.adapters.base import UpstreamAdapter, UpstreamCredential
+from hermes_cli.proxy.sse_done import (
+    DONE_SSE_FRAME,
+    SseDoneTracker,
+    content_type_is_sse,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -222,11 +235,23 @@ def create_app(adapter: UpstreamAdapter) -> "web.Application":
         )
         await resp.prepare(request)
 
+        # Track SSE terminal markers so we can append a missing [DONE]
+        # after clean EOF without rewriting any earlier frames.
+        done_tracker: Optional[SseDoneTracker] = None
+        if content_type_is_sse(upstream_resp.headers):
+            done_tracker = SseDoneTracker()
+
         try:
             async for chunk in upstream_resp.content.iter_any():
                 if chunk:
+                    if done_tracker is not None:
+                        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:
+            if done_tracker is not None:
+                done_tracker.mark_interrupted()
             logger.warning("proxy: streaming interrupted: %s", exc)
         finally:
             upstream_resp.release()

hermes_cli/proxy/sse_done.py

@@ -0,0 +1,132 @@
+"""SSE ``[DONE]`` sentinel normalization for OpenAI-compatible proxies.
+
+Some upstreams (notably Nous Portal for certain free models) deliver a
+complete chat-completions stream — content deltas, a non-null
+``finish_reason``, and often a ``lastOne: true`` usage frame — then close
+the connection without the conventional OpenAI terminal event::
+
+    data: [DONE]
+
+Strict OpenAI-compatible clients treat that shape as a truncated stream.
+This module watches the forwarded SSE byte stream and reports whether the
+proxy should append a single ``data: [DONE]`` frame after a *clean*
+upstream EOF.
+
+Rules (issue #90848):
+- Retain every original delta unchanged (this helper never rewrites bytes).
+- Append ``[DONE]`` only after a complete terminal choice
+  (``finish_reason`` non-null) **or** an upstream ``lastOne: true`` marker.
+- Never synthesize ``[DONE]`` after an error event, or when the stream was
+  interrupted before clean EOF.
+- Never emit a second ``[DONE]`` when the upstream already sent one.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+
+
+DONE_SSE_FRAME = b"data: [DONE]\n\n"
+
+
+@dataclass
+class SseDoneTracker:
+    """Incremental scanner over forwarded SSE chunks."""
+
+    saw_done: bool = False
+    saw_terminal_finish: bool = False
+    saw_last_one: bool = False
+    saw_error_event: bool = False
+    saw_malformed_event: bool = False
+    interrupted: bool = False
+    _buf: bytearray = field(default_factory=bytearray, repr=False)
+
+    def feed(self, chunk: bytes) -> None:
+        """Observe a forwarded chunk (bytes are not modified)."""
+        if not chunk:
+            return
+        self._buf.extend(chunk)
+        while True:
+            nl = self._buf.find(b"\n")
+            if nl < 0:
+                break
+            line = bytes(self._buf[:nl])
+            del self._buf[: nl + 1]
+            self._consume_line(line)
+
+    def mark_interrupted(self) -> None:
+        """Upstream stream ended via error/cancel — do not synthesize DONE."""
+        self.interrupted = True
+
+    def should_append_done(self) -> bool:
+        """True when a single terminal ``[DONE]`` should be appended."""
+        if (
+            self.interrupted
+            or self.saw_done
+            or self.saw_error_event
+            or self.saw_malformed_event
+        ):
+            return False
+        # Flush any trailing line without a final newline (rare but valid).
+        if self._buf:
+            self._consume_line(bytes(self._buf))
+            self._buf.clear()
+        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
+
+    def _consume_line(self, line: bytes) -> None:
+        # Strip CR from CRLF-delimited SSE.
+        if line.endswith(b"\r"):
+            line = line[:-1]
+        if not line.startswith(b"data:"):
+            return
+        payload = line[5:].strip()
+        if payload == b"[DONE]":
+            self.saw_done = True
+            return
+        if not payload:
+            return
+        try:
+            text = payload.decode("utf-8")
+        except UnicodeDecodeError:
+            self.saw_malformed_event = True
+            return
+        try:
+            event = json.loads(text)
+        except json.JSONDecodeError:
+            self.saw_malformed_event = True
+            return
+        if not isinstance(event, dict):
+            return
+        if event.get("error") is not None:
+            self.saw_error_event = True
+            return
+        if event.get("lastOne") is True:
+            self.saw_last_one = True
+        for choice in event.get("choices") or []:
+            if not isinstance(choice, dict):
+                continue
+            if choice.get("finish_reason") is not None:
+                self.saw_terminal_finish = True
+            # OpenAI error-shaped finish reasons should not unlock DONE.
+            fr = choice.get("finish_reason")
+            if isinstance(fr, str) and fr.lower() in {"error", "provider_error"}:
+                self.saw_error_event = True
+
+
+def content_type_is_sse(headers) -> bool:
+    """Return True when response headers advertise an SSE body."""
+    try:
+        value = headers.get("Content-Type") or headers.get("content-type") or ""
+    except Exception:
+        value = ""
+    return "text/event-stream" in str(value).lower()
+
+
+__all__ = [
+    "DONE_SSE_FRAME",
+    "SseDoneTracker",
+    "content_type_is_sse",
+]

tests/hermes_cli/test_proxy.py

@@ -365,6 +365,139 @@ def test_server_strips_client_auth_header():
     asyncio.run(run())
 
 
+def _build_sse_upstream(
+    frames: list[bytes],
+    *,
+    path: str = "/v1/chat/completions",
+) -> "web.Application":
+    async def sse(request):
+        _ = await request.read()
+        resp = web.StreamResponse(
+            status=200, headers={"Content-Type": "text/event-stream"},
+        )
+        await resp.prepare(request)
+        for chunk in frames:
+            await resp.write(chunk)
+        await resp.write_eof()
+        return resp
+
+    app = web.Application()
+    app.router.add_route("*", path, sse)
+    return app
+
+
+def test_proxy_appends_done_when_upstream_omits_sentinel():
+    """#90848: complete Portal-shaped SSE without [DONE] gets one appended."""
+    async def run():
+        frames = [
+            b'data: {"choices":[{"delta":{"content":"LONGCAT_OK"}}]}\n\n',
+            b'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n',
+            b'data: {"choices":[],"lastOne":true,"usage":{"prompt_tokens":1}}\n\n',
+        ]
+        upstream_runner, upstream_base = await _start_runner(
+            _build_sse_upstream(frames)
+        )
+        adapter = FakeAdapter(f"{upstream_base}/v1", bearer="ours")
+        proxy_runner, proxy_base = await _start_runner(create_app(adapter))
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"{proxy_base}/v1/chat/completions",
+                    json={"stream": True},
+                ) as resp:
+                    body = await resp.read()
+            text = body.decode("utf-8")
+            assert 'data: {"choices":[{"delta":{"content":"LONGCAT_OK"}}]}' in text
+            assert '"finish_reason":"stop"' in text
+            assert '"lastOne":true' in text
+            assert text.count("data: [DONE]") == 1
+            assert text.rstrip().endswith("data: [DONE]")
+        finally:
+            await proxy_runner.cleanup()
+            await upstream_runner.cleanup()
+
+    asyncio.run(run())
+
+
+def test_proxy_does_not_duplicate_existing_done():
+    async def run():
+        frames = [
+            b'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n',
+            b"data: [DONE]\n\n",
+        ]
+        upstream_runner, upstream_base = await _start_runner(
+            _build_sse_upstream(frames)
+        )
+        adapter = FakeAdapter(f"{upstream_base}/v1", bearer="ours")
+        proxy_runner, proxy_base = await _start_runner(create_app(adapter))
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"{proxy_base}/v1/chat/completions",
+                    json={"stream": True},
+                ) as resp:
+                    body = await resp.read()
+            assert body.decode("utf-8").count("data: [DONE]") == 1
+        finally:
+            await proxy_runner.cleanup()
+            await upstream_runner.cleanup()
+
+    asyncio.run(run())
+
+
+def test_proxy_does_not_append_done_after_error_event():
+    async def run():
+        frames = [
+            b'data: {"choices":[{"delta":{"content":"partial"}}]}\n\n',
+            b'data: {"error":{"message":"boom","type":"api_error"}}\n\n',
+        ]
+        upstream_runner, upstream_base = await _start_runner(
+            _build_sse_upstream(frames)
+        )
+        adapter = FakeAdapter(f"{upstream_base}/v1", bearer="ours")
+        proxy_runner, proxy_base = await _start_runner(create_app(adapter))
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"{proxy_base}/v1/chat/completions",
+                    json={"stream": True},
+                ) as resp:
+                    body = await resp.read()
+            assert "data: [DONE]" not in body.decode("utf-8")
+        finally:
+            await proxy_runner.cleanup()
+            await upstream_runner.cleanup()
+
+    asyncio.run(run())
+
+
+def test_proxy_does_not_append_done_after_malformed_trailing_frame():
+    async def run():
+        frames = [
+            b'data: {"choices":[{"delta":{"content":"partial"}}]}\n\n',
+            b'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n',
+            b'data: {"choices": [MALFORMED]}\n\n',
+        ]
+        upstream_runner, upstream_base = await _start_runner(
+            _build_sse_upstream(frames)
+        )
+        adapter = FakeAdapter(f"{upstream_base}/v1", bearer="ours")
+        proxy_runner, proxy_base = await _start_runner(create_app(adapter))
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"{proxy_base}/v1/chat/completions",
+                    json={"stream": True},
+                ) as resp:
+                    body = await resp.read()
+            assert "data: [DONE]" not in body.decode("utf-8")
+        finally:
+            await proxy_runner.cleanup()
+            await upstream_runner.cleanup()
+
+    asyncio.run(run())
+
+
 # ---------------------------------------------------------------------------
 # CLI handlers
 # ---------------------------------------------------------------------------

tests/hermes_cli/test_sse_done.py

@@ -0,0 +1,112 @@
+"""Unit tests for SSE [DONE] sentinel tracking (issue #90848)."""
+
+from __future__ import annotations
+
+import json
+
+from hermes_cli.proxy.sse_done import DONE_SSE_FRAME, SseDoneTracker, content_type_is_sse
+
+
+def _data_line(obj) -> bytes:
+    if isinstance(obj, (bytes, bytearray)):
+        return b"data: " + bytes(obj) + b"\n\n"
+    if obj == "[DONE]":
+        return b"data: [DONE]\n\n"
+    return f"data: {json.dumps(obj)}\n\n".encode("utf-8")
+
+
+def test_complete_stream_without_done_appends():
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {"content": "LONGCAT_OK"}}]}))
+    tracker.feed(_data_line({"choices": [{"delta": {}, "finish_reason": "stop"}]}))
+    tracker.feed(
+        _data_line(
+            {
+                "choices": [],
+                "lastOne": True,
+                "usage": {"prompt_tokens": 1, "completion_tokens": 1},
+            }
+        )
+    )
+    assert tracker.should_append_done() is True
+    assert tracker.saw_done is False
+    assert DONE_SSE_FRAME.startswith(b"data: [DONE]")
+
+
+def test_finish_reason_alone_is_enough_without_last_one():
+    """Solar-shaped stream: finish_reason present, no lastOne, no [DONE]."""
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {"content": "SOLAR_OK"}}]}))
+    tracker.feed(_data_line({"choices": [{"delta": {}, "finish_reason": "stop"}]}))
+    assert tracker.should_append_done() is True
+
+
+def test_last_one_alone_is_enough_without_finish_reason():
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {"content": "x"}}]}))
+    tracker.feed(_data_line({"choices": [], "lastOne": True}))
+    assert tracker.should_append_done() is True
+
+
+def test_existing_done_is_not_duplicated():
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {}, "finish_reason": "stop"}]}))
+    tracker.feed(_data_line("[DONE]"))
+    assert tracker.should_append_done() is False
+    assert tracker.saw_done is True
+
+
+def test_error_event_blocks_done_synthesis():
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {"content": "partial"}}]}))
+    tracker.feed(_data_line({"error": {"message": "upstream failed", "type": "api_error"}}))
+    assert tracker.should_append_done() is False
+
+
+def test_error_finish_reason_blocks_done_synthesis():
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {}, "finish_reason": "error"}]}))
+    assert tracker.should_append_done() is False
+
+
+def test_malformed_event_blocks_done_synthesis():
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {}, "finish_reason": "stop"}]}))
+    tracker.feed(b'data: {"choices": [MALFORMED]}\n\n')
+    assert tracker.should_append_done() is False
+
+
+def test_truncated_trailing_event_blocks_done_synthesis():
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {}, "finish_reason": "stop"}]}))
+    tracker.feed(b'data: {"choices": [{"delta": {"content": "tail"}}]')
+    assert tracker.should_append_done() is False
+
+
+def test_interrupted_stream_never_appends_done():
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {}, "finish_reason": "stop"}]}))
+    tracker.mark_interrupted()
+    assert tracker.should_append_done() is False
+
+
+def test_incomplete_stream_without_terminal_marker_does_not_append():
+    tracker = SseDoneTracker()
+    tracker.feed(_data_line({"choices": [{"delta": {"content": "mid"}}]}))
+    assert tracker.should_append_done() is False
+
+
+def test_feed_preserves_chunk_boundaries_across_split_lines():
+    """A finish_reason frame split across TCP chunks must still be detected."""
+    tracker = SseDoneTracker()
+    payload = _data_line({"choices": [{"delta": {}, "finish_reason": "stop"}]})
+    mid = len(payload) // 2
+    tracker.feed(payload[:mid])
+    tracker.feed(payload[mid:])
+    assert tracker.should_append_done() is True
+
+
+def test_content_type_is_sse():
+    assert content_type_is_sse({"Content-Type": "text/event-stream"}) is True
+    assert content_type_is_sse({"content-type": "text/event-stream; charset=utf-8"}) is True
+    assert content_type_is_sse({"Content-Type": "application/json"}) is False

tests/run_agent/test_partial_stream_finish_reason.py

@@ -789,3 +789,34 @@ class TestSendTimePadMultimodalSafety:
         assert out[2]["content"] == ""
         # input list untouched (repair is copy-on-write)
         assert api_messages[1]["content"] == ""
+
+
+class TestPortalLastOneWithoutDone:
+    """#90848: complete Portal streams can end with lastOne=true and no
+    finish_reason / [DONE]. That is a clean terminal, not a drop."""
+
+    @patch("run_agent.AIAgent._create_request_openai_client")
+    @patch("run_agent.AIAgent._close_request_openai_client")
+    def test_last_one_usage_frame_is_a_clean_stop(self, _mock_close, mock_create):
+        def _portal_stream():
+            yield _make_stream_chunk(content="LONGCAT_OK")
+            yield SimpleNamespace(
+                choices=[],
+                model="meituan/longcat-2.0:free",
+                usage=SimpleNamespace(prompt_tokens=1, completion_tokens=1),
+                lastOne=True,
+            )
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create.side_effect = (
+            lambda *a, **kw: _portal_stream()
+        )
+        mock_create.return_value = mock_client
+
+        agent = _make_agent()
+        agent._fire_stream_delta = lambda text: None
+        response = agent._interruptible_streaming_api_call({})
+
+        assert getattr(response, "id", None) != PARTIAL_STREAM_STUB_ID
+        assert response.choices[0].finish_reason == "stop"
+        assert response.choices[0].message.content == "LONGCAT_OK"