refactor: centralize ndjson streaming transport

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`.

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

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

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:

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}")

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",

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

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

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