From 6f201abf06a729d3fb2f8fa70b70081c4539d156 Mon Sep 17 00:00:00 2001 From: "Gabriel A. Devenyi" Date: Sat, 31 Oct 2026 08:18:37 -0400 Subject: [PATCH] Streams: an error after the 200 is the API's own last frame (chat: error chunk then [DONE]; messages: error event; responses: response.failed), not a second HTTP response in the body or a silent end GpuBusy and EngineStuck no longer reach _post's handlers once a stream has started; EngineStarting and ModelBusy get the same branch, though no raise of them after the start was found. Any other exception gets the same frame with a plain sentence, and its traceback goes to the log. ClientGone and OSError keep their paths, and a stream that raises nothing is unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01W53iURzvDQMjnYemag8uvv --- serve/server.py | 40 +++++++++++ serve/test_parallel.py | 28 ++++++++ serve/test_stream_errors.py | 135 ++++++++++++++++++++++++++++++++++++ 3 files changed, 203 insertions(+) create mode 100644 serve/test_stream_errors.py diff --git a/serve/server.py b/serve/server.py index 768f6bcc374..ddb33dce29e 100644 --- a/serve/server.py +++ b/serve/server.py @@ -7916,6 +7916,26 @@ def _openai(self, req): err = {"error": {"type": "server_error", "message": str(e)}} # headers are sent, so no 400 now self._note(error=err["error"]) self.wfile.write(b"data: " + json.dumps(err).encode() + b"\n\ndata: [DONE]\n\n") + except (GpuBusy, EngineStarting, EngineStuck, ModelBusy) as e: + self._chat_failed(str(e)) # _post's handler would write HTTP into the stream's body + except ClientGone: + raise + except Exception: # no silent end: the client would take it for a whole answer + self._chat_failed(self._stream_crashed()) + + def _stream_crashed(self): + """An unexpected error in a started stream: the traceback for the log, a plain sentence for the client.""" + import traceback + print("[strata] error in a started stream:\n" + traceback.format_exc(), flush=True) + return "the server failed while answering; the server's log has the details" + + def _chat_failed(self, message): + err = {"error": {"type": "server_error", "message": message}} + self._note(error=err["error"]) + try: + self.wfile.write(b"data: " + json.dumps(err).encode() + b"\n\ndata: [DONE]\n\n") + except OSError: + self._note(outcome="disconnected") def _responses_ready(self, req, events): """Send a finished Responses result that did not run on the engine.""" @@ -8020,6 +8040,12 @@ def send(e): self._responses_failed(asm, str(e), "structured_output_failed", send) except ValueError as e: # the engine's ERR after the stream started self._responses_failed(asm, str(e), "server_error", send) + except (GpuBusy, EngineStarting, EngineStuck, ModelBusy) as e: # _post would write HTTP into the stream's body + self._responses_failed(asm, str(e), "server_error", send) + except ClientGone: + raise + except Exception: # no silent end: the client would take it for a whole answer + self._responses_failed(asm, self._stream_crashed(), "server_error", send) def _responses_failed(self, asm, message, code, send): e = asm.failed(message, code) @@ -8146,6 +8172,20 @@ def _anthropic(self, req): err = {"type": "error", "error": {"type": "api_error", "message": str(e)}} self._note(error=err["error"]) self.wfile.write(b"event: error\ndata: " + json.dumps(err).encode() + b"\n\n") + except (GpuBusy, EngineStarting, EngineStuck, ModelBusy) as e: # _post would write HTTP into the stream's body + self._anthropic_failed(str(e), "overloaded_error" if isinstance(e, GpuBusy) else "api_error") + except ClientGone: + raise + except Exception: # no silent end: the client would take it for a whole answer + self._anthropic_failed(self._stream_crashed(), "api_error") + + def _anthropic_failed(self, message, kind): + err = {"type": "error", "error": {"type": kind, "message": message}} + self._note(error=err["error"]) + try: + self.wfile.write(b"event: error\ndata: " + json.dumps(err).encode() + b"\n\n") + except OSError: + self._note(outcome="disconnected") return Handler diff --git a/serve/test_parallel.py b/serve/test_parallel.py index 1956ee908cc..a264c222239 100644 --- a/serve/test_parallel.py +++ b/serve/test_parallel.py @@ -433,6 +433,34 @@ def test_bgen_refused_for_good_is_a_retryable_503_and_frees_the_slots(self): self.assertEqual(self.engine.slot_busy, [False, False]) self.assertEqual(self.chat("fine again")["choices"][0]["message"]["content"], "ok, done.") + def test_bgen_refused_for_good_in_a_stream_is_each_apis_error_frame(self): + """The same refusal once the 200 is sent: the API's own last frame, not a second HTTP response in the body.""" + self.start(2) + self.engine.SLOT_PARK_S = 0.05 + text = [{"role": "user", "content": "BUSYALL hello"}] + cases = ( + ("/v1/chat/completions", {"messages": text, "max_tokens": 8}, b'data: {"error": ', b"data: [DONE]\n\n"), + ("/v1/messages", {"model": "m", "messages": text, "max_tokens": 8}, b"event: error\n", b"\n\n"), + ("/v1/responses", {"input": "BUSYALL hello", "max_output_tokens": 8}, b"event: response.failed\n", b"\n\n"), + ) + for path, body, frame, end in cases: + with self.subTest(path): + long = threading.Thread(target=self.chat, args=("LONGREPLY please",)) + long.start() + time.sleep(0.15) + req = urllib.request.Request(self.base + path, data=json.dumps({**body, "stream": True}).encode(), + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=60) as r: + self.assertEqual(r.status, 200) + got = r.read() + long.join(30) + self.assertNotIn(b"HTTP/1.", got) + self.assertIn(frame, got) + self.assertIn(b"slots are not free", got) + self.assertTrue(got.endswith(end), got[-80:]) + time.sleep(0.5) + self.assertEqual(self.engine.slot_busy, [False, False]) + def test_engine_reports_fewer_slots(self): self.start(4, fit=2) self.assertEqual(self.engine.batch, 2) diff --git a/serve/test_stream_errors.py b/serve/test_stream_errors.py new file mode 100644 index 00000000000..dd60e093d80 --- /dev/null +++ b/serve/test_stream_errors.py @@ -0,0 +1,135 @@ +"""An error after a stream has started: each API's own last frame, never a second HTTP response, never silence. + + python -m unittest serve.test_stream_errors -v +""" +import contextlib +import http.client +import io +import json +import types +import unittest +from pathlib import Path + +from serve import server +from serve.frontend import ChatTemplate +from serve.server import ByteTokenizer, MockEngine, Service, serve + +ROOT = Path(__file__).resolve().parents[1] +MESSAGES = [{"role": "user", "content": "hi"}] +CHAT = ("/v1/chat/completions", {"messages": MESSAGES, "stream": True}) +ANTHROPIC = ("/v1/messages", {"model": "m", "messages": MESSAGES, "max_tokens": 9, "stream": True}) +RESPONSES = ("/v1/responses", {"input": "hi", "stream": True}) + + +class FailsAfterOneToken(MockEngine): + failure = None + + def generate(self, *args, **kwargs): + tokens = list(super().generate(*args, **kwargs)) + yield tokens[0] + if self.failure is not None: + raise self.failure + yield from tokens[1:] + + +def frames(body): + return [f for f in body.decode().split("\n\n") if f] + + +def data_of(frame): + return json.loads(frame.partition("data: ")[2]) + + +class StreamErrors(unittest.TestCase): + @classmethod + def setUpClass(cls): + tok = ByteTokenizer() + cls.engine = FailsAfterOneToken(tok, "partial answer", max_context=131072) + cls.svc = Service(cls.engine, tok, ChatTemplate(ROOT / "serve/chat_template.jinja")) + cls.httpd = serve(cls.svc, port=0) + + @classmethod + def tearDownClass(cls): + cls.httpd.shutdown() + cls.httpd.server_close() + + def stream(self, api, failure): + self.engine.failure = failure + c = http.client.HTTPConnection("127.0.0.1", self.httpd.server_address[1], timeout=15) + try: + c.request("POST", api[0], json.dumps(api[1]), {"Content-Type": "application/json"}) + r = c.getresponse() + body = r.read() + finally: + c.close() + self.assertEqual(r.status, 200, body) + self.assertNotIn(b"HTTP/1.", body) + return frames(body) + + def chat_error(self, failure): + got = self.stream(CHAT, failure) + self.assertEqual(got[-1], "data: [DONE]") + return data_of(got[-2])["error"] + + def anthropic_error(self, failure): + got = self.stream(ANTHROPIC, failure) + self.assertTrue(got[-1].startswith("event: error\n"), got[-1]) + event = data_of(got[-1]) + self.assertEqual(event["type"], "error") + return event["error"] + + def responses_error(self, failure): + got = self.stream(RESPONSES, failure) + self.assertTrue(got[-1].startswith("event: response.failed\n"), got[-1]) + response = data_of(got[-1])["response"] + self.assertEqual(response["status"], "failed") + return response["error"] + + def test_a_refusal_after_the_start_is_the_apis_own_frame(self): + for failure in (server.GpuBusy("busy now", retry_after=5), server.EngineStuck("stuck now"), + server.EngineStarting("starting now"), server.ModelBusy("switching now")): + with self.subTest(type(failure).__name__): + self.assertEqual(self.chat_error(failure), {"type": "server_error", "message": str(failure)}) + self.assertEqual(self.responses_error(failure), {"code": "server_error", "message": str(failure)}) + self.assertEqual(self.anthropic_error(failure), + {"type": "overloaded_error" if isinstance(failure, server.GpuBusy) else "api_error", + "message": str(failure)}) + + def test_any_other_failure_is_a_frame_and_keeps_its_details_in_the_log(self): + failure = RuntimeError("a secret detail") + for read_error in (self.chat_error, self.anthropic_error, self.responses_error): + with self.subTest(read_error.__name__): + log = io.StringIO() + with contextlib.redirect_stdout(log): # the handler's thread prints to this sys.stdout too + error = read_error(failure) + self.assertIn("server's log", error["message"]) + self.assertNotIn("secret", error["message"]) + self.assertIn("[strata] error in a started stream", log.getvalue()) + self.assertIn("RuntimeError: a secret detail", log.getvalue()) + + def test_a_client_that_left_gets_no_frame(self): + for api in (CHAT, ANTHROPIC, RESPONSES): + with self.subTest(api[0]): + body = "\n\n".join(self.stream(api, server.ClientGone())) + for frame in ("[DONE]", "event: error", "response.failed"): + self.assertNotIn(frame, body) + + def test_a_stream_that_does_not_fail_ends_as_before(self): + self.assertEqual(self.stream(CHAT, None)[-1], "data: [DONE]") + self.assertTrue(self.stream(ANTHROPIC, None)[-1].startswith("event: message_stop\n")) + self.assertTrue(self.stream(RESPONSES, None)[-1].startswith("event: response.completed\n")) + + def test_a_frame_the_client_cannot_take_is_a_disconnect(self): + def refuses(_data): + raise BrokenPipeError() + for send in (lambda h, s: h._chat_failed(s, "x"), lambda h, s: h._anthropic_failed(s, "x", "api_error")): + noted = {} + fake = types.SimpleNamespace(wfile=types.SimpleNamespace(write=refuses), + _note=lambda **values: noted.update(values)) + send(self.httpd.RequestHandlerClass, fake) + self.assertEqual(noted["outcome"], "disconnected") + self.assertEqual(noted["error"]["message"], "x") # still a server error: the monitor's state is "error" + + +if __name__ == "__main__": + unittest.main()