From c9f240f46d1b346fc20f6f2b53676b0938b8f2e7 Mon Sep 17 00:00:00 2001 From: leefer Date: Sun, 2 Aug 2026 01:53:47 +0800 Subject: [PATCH] refactor: centralize ndjson streaming transport --- ARCHITECTURE.md | 5 +- backend/application.py | 10 --- backend/features/mentor/http.py | 17 +---- backend/features/review/http.py | 18 +---- backend/http/handler.py | 29 ++++++++ config/architecture-inventory.json | 14 +++- tests/test_http_streaming.py | 85 +++++++++++++++++++++++ tests/test_preservation_slice_accounts.py | 2 + tools/build_architecture_inventory.py | 4 ++ 9 files changed, 138 insertions(+), 46 deletions(-) create mode 100644 tests/test_http_streaming.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 8c6cb78..542357d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -30,8 +30,9 @@ background scheduler `backend/application.py` and `backend/bootstrap/`. - `backend/bootstrap/` owns process configuration, dependency construction, startup, and shared input/display-format contracts. It does not own feature behavior. -- `backend/http/` owns common authentication, request IDs, responses, static delivery, and - error normalization. Feature-specific transport handlers live beside their feature. +- `backend/http/` owns common authentication, request IDs, JSON/NDJSON responses, static + delivery, streaming connection lifecycle, and error normalization. Feature-specific + transport handlers live beside their feature. Exact POST endpoints that only delegate to one of those handlers use the explicit maps in `backend/application.py`; endpoints with path parameters, body handling, or special error semantics remain visible control flow in `RequestHandler`. diff --git a/backend/application.py b/backend/application.py index efea17b..f0abec4 100644 --- a/backend/application.py +++ b/backend/application.py @@ -1020,16 +1020,6 @@ class RequestHandler( return self.send_json({"error": "Not found"}, HTTPStatus.NOT_FOUND) - - - def _write_stream_event(self, payload: dict[str, Any]) -> None: - self.wfile.write( - (json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8") - ) - self.wfile.flush() - - - def save_reason(self) -> None: try: body = self.read_json_body() diff --git a/backend/features/mentor/http.py b/backend/features/mentor/http.py index 241b28f..cee3f0e 100644 --- a/backend/features/mentor/http.py +++ b/backend/features/mentor/http.py @@ -14,19 +14,4 @@ class MentorHttpMixin: except (ValueError, json.JSONDecodeError) as exc: self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) return - self.send_response(HTTPStatus.OK) - self.send_header("Content-Type", "application/x-ndjson; charset=utf-8") - self.send_header("Cache-Control", "no-cache, no-transform") - self.send_header("X-Accel-Buffering", "no") - self.send_header("Connection", "close") - self.end_headers() - try: - for event in stream: - self._write_stream_event(event) - self._write_stream_event({"type": "done"}) - except (ValueError, MentorAgentError) as exc: - self._write_stream_event({"type": "error", "error": str(exc)}) - except (BrokenPipeError, ConnectionResetError): - pass - finally: - self.close_connection = True + self.send_ndjson_stream(stream, (ValueError, MentorAgentError)) diff --git a/backend/features/review/http.py b/backend/features/review/http.py index ad66e52..251bfeb 100644 --- a/backend/features/review/http.py +++ b/backend/features/review/http.py @@ -26,22 +26,8 @@ class ReviewHttpMixin: except (ValueError, json.JSONDecodeError) as exc: self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) return - self.send_response(HTTPStatus.OK) - self.send_header("Content-Type", "application/x-ndjson; charset=utf-8") - self.send_header("Cache-Control", "no-cache, no-transform") - self.send_header("X-Accel-Buffering", "no") - self.send_header("Connection", "close") - self.end_headers() - try: - for chunk in stream: - self._write_stream_event({"type": "delta", "content": chunk}) - self._write_stream_event({"type": "done"}) - except (ValueError, ReviewAssistantError) as exc: - self._write_stream_event({"type": "error", "error": str(exc)}) - except (BrokenPipeError, ConnectionResetError): - pass - finally: - self.close_connection = True + events = ({"type": "delta", "content": chunk} for chunk in stream) + self.send_ndjson_stream(events, (ValueError, ReviewAssistantError)) def save_watchlist(self) -> None: try: diff --git a/backend/http/handler.py b/backend/http/handler.py index 186fb7e..59b3643 100644 --- a/backend/http/handler.py +++ b/backend/http/handler.py @@ -3,6 +3,7 @@ from __future__ import annotations import json import mimetypes import secrets +from collections.abc import Iterable from http import HTTPStatus from http.cookies import SimpleCookie from typing import Any @@ -142,5 +143,33 @@ class HttpTransportMixin: self.end_headers() self.wfile.write(content) + def _write_stream_event(self, payload: dict[str, Any]) -> None: + self.wfile.write( + (json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8") + ) + self.wfile.flush() + + def send_ndjson_stream( + self, + events: Iterable[dict[str, Any]], + error_types: tuple[type[Exception], ...], + ) -> None: + self.send_response(HTTPStatus.OK) + self.send_header("Content-Type", "application/x-ndjson; charset=utf-8") + self.send_header("Cache-Control", "no-cache, no-transform") + self.send_header("X-Accel-Buffering", "no") + self.send_header("Connection", "close") + self.end_headers() + try: + for event in events: + self._write_stream_event(event) + self._write_stream_event({"type": "done"}) + except error_types as exc: + self._write_stream_event({"type": "error", "error": str(exc)}) + except (BrokenPipeError, ConnectionResetError): + pass + finally: + self.close_connection = True + def log_message(self, format_string: str, *args: Any) -> None: print(f"[{self.log_date_time_string()}] {format_string % args}") diff --git a/config/architecture-inventory.json b/config/architecture-inventory.json index 625fdeb..413b671 100644 --- a/config/architecture-inventory.json +++ b/config/architecture-inventory.json @@ -287,6 +287,16 @@ "path": "backend/llm/transport.py" } ], + "http_transport": [ + { + "function": "send_json", + "path": "backend/http/handler.py" + }, + { + "function": "send_ndjson_stream", + "path": "backend/http/handler.py" + } + ], "css_layers": [ "/shared/tokens.css?v=20260729-1", "/styles/styles.css", @@ -364,8 +374,8 @@ }, { "path": "backend/application.py", - "bytes": 48749, - "lines": 1129 + "bytes": 48513, + "lines": 1119 }, { "path": "frontend/styles/theme.css", diff --git a/tests/test_http_streaming.py b/tests/test_http_streaming.py new file mode 100644 index 0000000..1d24823 --- /dev/null +++ b/tests/test_http_streaming.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import io +import json +import unittest + +from backend.http.handler import HttpTransportMixin + + +class StreamError(RuntimeError): + pass + + +class TransportStub(HttpTransportMixin): + def __init__(self) -> None: + self.statuses: list[int] = [] + self.response_headers: list[tuple[str, str]] = [] + self.wfile = io.BytesIO() + self.close_connection = False + + def send_response(self, status: int) -> None: + self.statuses.append(int(status)) + + def send_header(self, name: str, value: str) -> None: + self.response_headers.append((name, value)) + + def end_headers(self) -> None: + pass + + +class HttpStreamingTests(unittest.TestCase): + def test_stream_transport_preserves_headers_events_and_completion(self) -> None: + handler = TransportStub() + handler.send_ndjson_stream( + ({"type": "delta", "content": value} for value in ("甲", "乙")), + (StreamError,), + ) + + self.assertEqual(handler.statuses, [200]) + self.assertEqual( + dict(handler.response_headers), + { + "Content-Type": "application/x-ndjson; charset=utf-8", + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + "Connection": "close", + }, + ) + events = [ + json.loads(line) + for line in handler.wfile.getvalue().decode("utf-8").splitlines() + ] + self.assertEqual( + events, + [ + {"type": "delta", "content": "甲"}, + {"type": "delta", "content": "乙"}, + {"type": "done"}, + ], + ) + self.assertTrue(handler.close_connection) + + def test_stream_transport_preserves_feature_error_event(self) -> None: + def events(): + yield {"type": "delta", "content": "partial"} + raise StreamError("stream failed") + + handler = TransportStub() + handler.send_ndjson_stream(events(), (StreamError,)) + payloads = [ + json.loads(line) + for line in handler.wfile.getvalue().decode("utf-8").splitlines() + ] + self.assertEqual( + payloads, + [ + {"type": "delta", "content": "partial"}, + {"type": "error", "error": "stream failed"}, + ], + ) + self.assertTrue(handler.close_connection) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_preservation_slice_accounts.py b/tests/test_preservation_slice_accounts.py index f2c0eed..2c4170f 100644 --- a/tests/test_preservation_slice_accounts.py +++ b/tests/test_preservation_slice_accounts.py @@ -40,6 +40,8 @@ class AccountSliceStructureTests(unittest.TestCase): "require_access", "serve_static", "send_json", + "_write_stream_event", + "send_ndjson_stream", ): self.assertNotIn(method, RequestHandler.__dict__) self.assertIn(method, HttpTransportMixin.__dict__) diff --git a/tools/build_architecture_inventory.py b/tools/build_architecture_inventory.py index f5b84c6..948e651 100644 --- a/tools/build_architecture_inventory.py +++ b/tools/build_architecture_inventory.py @@ -179,6 +179,10 @@ def build() -> dict[str, Any]: {"function": "chat_completion", "path": "backend/llm/transport.py"}, {"function": "stream_chat_completion", "path": "backend/llm/transport.py"}, ], + "http_transport": [ + {"function": "send_json", "path": "backend/http/handler.py"}, + {"function": "send_ndjson_stream", "path": "backend/http/handler.py"}, + ], "css_layers": css_layers(html), "code_hotspots": code_hotspots(), }