[Engine] an error after a stream has started ends with the API's own error frame, not HTTP text in the body or silence

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
 

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)

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