fix(local): fail a reply the endpoint cuts off with an in-stream error

docs/architecture/chat.md

@@ -134,6 +134,7 @@ A chat thread's assistant turn stores how it ended in its content, as `ending`,
 | Interrupted | `{"type": "interrupted"}` | a quit, or the API going away, cut it off |
 
 - **A failed turn is kept,** with or without text, so the question stays with the reason it failed. A turn stopped before any text is deleted, both halves, with any image file no other turn of the thread points at: the person chose to stop and there is nothing to show.
+- **An error the endpoint sends inside the stream,** as llama.cpp and OpenRouter do once the first token has gone out, fails the turn the same way rather than ending it as a finished reply.
 - **History skips** a failed or interrupted reply with no text, together with its question, so the model never receives an empty answer or two questions in a row ([`history.py`](../../surfsense_local/backend/modules/chat/history.py)). One with text stays, as a partial reply always did.
 - **A thread is renamed** only when its first turn ends with answer text.
 

docs/architecture/connections.md

@@ -153,7 +153,7 @@ Loading a connection runs the egress check for its host. The bundled sd-server s
 
 ## Trying a model before assigning it
 
-- `chat-test` streams one answer from the chosen model, by default to a prompt asking for one short sentence, capped at 1,024 tokens and 600 characters, so a model whose capability is `unknown` can be seen answering before it becomes the chat model. An empty reply is a `502`, which is what a reasoning model returns when it spends the whole budget thinking.
+- `chat-test` streams one answer from the chosen model, by default to a prompt asking for one short sentence, capped at 1,024 tokens and 600 characters, so a model whose capability is `unknown` can be seen answering before it becomes the chat model. An empty reply is a `502`, which is what a reasoning model returns when it spends the whole budget thinking. An error the model sends inside its stream is a `502` as well, with its message, like a request the endpoint refuses, rather than the text before it passed off as the answer.
 - `image-test` generates one image, by default a blue circle on white, through the real image client and returns the bytes with `Cache-Control: no-store`. It creates no artifact.
 - `speech-test` voices the `prompt` sent, the Try dialog's line in the interface's language, through the podcast's speech client, in the `voice` sent, and returns the audio with `Cache-Control: no-store`. Without one the request carries no `voice`, so the server speaks in its own default where it has one, and says so where it has none; no voice is assumed. The Try dialog has an optional voice field for audio models and plays the clip.
 - `GET`, `POST` and `DELETE /llm/selection/audio_gen/voices` read the audio selection's voices (the server's list, else the user's), add one once the server has voiced the `text` sent (`502` with the server's words when it refuses), and remove one. A model on this computer is a `409`.

surfsense_local/backend/modules/llm/connections/router.py

@@ -26,6 +26,7 @@
     NonRetryableImageError,
     OpenAICompatibleImageProvider,
 )
+from modules.llm.providers.openai_compatible.chat import StreamedError
 from modules.llm.providers.openai_compatible.speech import (
     NonRetryableSpeechError,
     RemoteSpeech,
@@ -320,6 +321,10 @@ async def test_connection_chat(
             status.HTTP_429_TOO_MANY_REQUESTS,
             detail={"code": "plan_limit", "message": str(error)},
         ) from error
+    except StreamedError as error:
+        raise HTTPException(
+            status.HTTP_502_BAD_GATEWAY, f"chat request failed: {error}"
+        ) from error
     except httpx.HTTPError as error:
         raise HTTPException(
             status.HTTP_502_BAD_GATEWAY, f"chat request failed: {error}"

surfsense_local/backend/modules/llm/providers/openai_compatible/chat.py

@@ -34,6 +34,10 @@
 MAX_ERROR_CHARS = 400
 
 
+class StreamedError(RuntimeError):
+    """A failure the endpoint reported inside a stream it had already begun."""
+
+
 class OpenAICompatibleChatProvider:
     name = "openai_compatible"
 
@@ -264,6 +268,11 @@ def _delta(line: str) -> Delta | None:
     if not payload or payload == "[DONE]":
         return None
     chunk = json.loads(payload)
+    # Past the first token the status is already 200, so llama.cpp and
+    # OpenRouter send a failure as a chunk. Skipped, it ended a cut-off reply
+    # as though the model had finished.
+    if error := chunk.get("error"):
+        raise StreamedError(_streamed_message(error))
     if progress := _prompt_progress(chunk.get("prompt_progress")):
         return Delta("", progress=progress)
     choices = chunk.get("choices")
@@ -289,6 +298,13 @@ def _reuse(line: str) -> tuple[int, int] | None:
     return chunk_reuse(chunk) if isinstance(chunk, dict) else None
 
 
+def _streamed_message(error: object) -> str:
+    message = error.get("message") if isinstance(error, dict) else error
+    if isinstance(message, str) and message.strip():
+        return message.strip()[:MAX_ERROR_CHARS]
+    return "the endpoint reported an error mid-stream"
+
+
 def _prompt_progress(reported: object) -> PromptProgress | None:
     """llama.cpp counts the cached prefix as processed; the work is what lies past it."""
     if not isinstance(reported, dict):

surfsense_local/backend/tests/integration/llm/conftest.py

@@ -211,6 +211,9 @@ async def download(url, destination, *, sha256=None, transport=None):
 # finish thinking inside.
 REASONING_DELTAS = ["Okay, "] * 200
 REMOTE_THINKS = False
+# Set, /chat/completions sends this chunk after CHAT_DELTAS, as OpenRouter
+# reports a failure once the status is already 200.
+STREAM_ERROR: dict | None = None
 REMOTE_REQUESTS: list[tuple[str, str]] = []
 # Set, the image routes answer with this URL instead of inline bytes, as a
 # provider serving its images from a CDN does.
@@ -263,6 +266,8 @@ def do_POST(self) -> None:
                 f"data: {json.dumps({'choices': [{'delta': {field: text}}]})}\n\n"
                 for field, text in deltas
             ]
+            if STREAM_ERROR is not None:
+                frames.append(f"data: {json.dumps(STREAM_ERROR)}\n\n")
             frames.append("data: [DONE]\n\n")
             self._send("".join(frames).encode())
         elif self.path == "/responses":

surfsense_local/backend/tests/integration/llm/test_connections.py

@@ -465,6 +465,34 @@ async def test_a_thinking_model_on_a_connection_is_not_called_broken(
     assert tested.json() == {"reply": "Hello"}
 
 
+async def test_an_error_the_model_sends_mid_reply_fails_the_chat_test(
+    client: AsyncClient, openai_server: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A 502 with the provider's reason, as a refused request gets, not a 500."""
+    monkeypatch.setattr(conftest, "CHAT_DELTAS", ["Hel"])
+    monkeypatch.setattr(
+        conftest,
+        "STREAM_ERROR",
+        {
+            "id": "gen-1",
+            "object": "chat.completion.chunk",
+            "error": {"code": "server_error", "message": "Provider disconnected"},
+            "choices": [
+                {"index": 0, "delta": {"content": ""}, "finish_reason": "error"}
+            ],
+        },
+    )
+    connection = await _connect(client, openai_server)
+
+    tested = await client.post(
+        f"/llm/connections/{connection['id']}/chat-test",
+        json={"model": "anthropic/claude-3.5-sonnet"},
+    )
+
+    assert tested.status_code == 502
+    assert tested.json() == {"detail": "chat request failed: Provider disconnected"}
+
+
 SD15 = "v1-5-pruned_Q4_0"
 SDXL = "sd_xl_base_1.0_0_Q4_0"
 

surfsense_local/backend/tests/unit/llm/providers/openai_compatible/test_streamed_error.py

@@ -0,0 +1,58 @@
+"""An error an endpoint sends inside its stream fails the reply.
+
+Once tokens have gone out the status is already 200, so llama.cpp and OpenRouter
+report a later failure as a `data:` chunk carrying `error`. Read as an empty
+delta, a reply cut off by it was stored and shown as if it had finished.
+"""
+
+import httpx
+import pytest
+
+from modules.llm.providers.openai_compatible.chat import (
+    OpenAICompatibleChatProvider,
+    StreamedError,
+)
+from modules.llm.providers.types import Delta, Message
+
+pytestmark = pytest.mark.unit
+
+# The shapes each endpoint documents or sends, after some answer text.
+LLAMACPP = '{"error":{"code":500,"message":"decode failed","type":"server_error"}}'
+OPENROUTER = (
+    '{"id":"gen-1","object":"chat.completion.chunk",'
+    '"error":{"code":"server_error","message":"decode failed"},'
+    '"choices":[{"index":0,"delta":{"content":""},"finish_reason":"error"}]}'
+)
+
+
+def endpoint(*chunks: str) -> OpenAICompatibleChatProvider:
+    """A 200 stream of these chunks, as an endpoint past its first token sends."""
+    body = "".join(f"data: {chunk}\n\n" for chunk in chunks)
+    return OpenAICompatibleChatProvider(
+        "http://endpoint/v1",
+        transport=httpx.MockTransport(lambda _: httpx.Response(200, text=body)),
+    )
+
+
+@pytest.mark.parametrize(
+    "error", [LLAMACPP, OPENROUTER], ids=["llamacpp", "openrouter"]
+)
+async def test_an_error_after_text_fails_the_reply_with_its_message(error: str) -> None:
+    """The text before the error still arrives; the stream then fails, not ends."""
+    provider = endpoint('{"choices":[{"delta":{"content":"Revenue"}}]}', error)
+    received: list[Delta] = []
+
+    with pytest.raises(StreamedError, match="decode failed"):
+        async for delta in provider.chat_deltas("m", [Message("user", "hi")]):
+            received.append(delta)
+
+    assert received == [Delta("Revenue")]
+
+
+async def test_an_error_sent_as_a_string_still_fails_the_reply() -> None:
+    """Not every endpoint nests a message, and a bare error is still a failure."""
+    provider = endpoint('{"error":"overloaded"}')
+
+    with pytest.raises(StreamedError, match="overloaded"):
+        async for _ in provider.chat("m", [Message("user", "hi")]):
+            pass