diff --git a/docs/architecture/chat.md b/docs/architecture/chat.md index 55784a9748..cce37593cb 100644 --- a/docs/architecture/chat.md +++ b/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. diff --git a/docs/architecture/connections.md b/docs/architecture/connections.md index f0fec7da1f..b34a650b3c 100644 --- a/docs/architecture/connections.md +++ b/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`. diff --git a/surfsense_local/backend/modules/llm/connections/router.py b/surfsense_local/backend/modules/llm/connections/router.py index 35b09462b5..ec208beacb 100644 --- a/surfsense_local/backend/modules/llm/connections/router.py +++ b/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}" diff --git a/surfsense_local/backend/modules/llm/providers/openai_compatible/chat.py b/surfsense_local/backend/modules/llm/providers/openai_compatible/chat.py index 0c2cb6413a..42f6c2f340 100644 --- a/surfsense_local/backend/modules/llm/providers/openai_compatible/chat.py +++ b/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): diff --git a/surfsense_local/backend/tests/integration/llm/conftest.py b/surfsense_local/backend/tests/integration/llm/conftest.py index e7848effd6..b2499968db 100644 --- a/surfsense_local/backend/tests/integration/llm/conftest.py +++ b/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": diff --git a/surfsense_local/backend/tests/integration/llm/test_connections.py b/surfsense_local/backend/tests/integration/llm/test_connections.py index 6c190714e7..7f7945fb9d 100644 --- a/surfsense_local/backend/tests/integration/llm/test_connections.py +++ b/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" diff --git a/surfsense_local/backend/tests/unit/llm/providers/openai_compatible/test_streamed_error.py b/surfsense_local/backend/tests/unit/llm/providers/openai_compatible/test_streamed_error.py new file mode 100644 index 0000000000..ff0dfcfe17 --- /dev/null +++ b/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