refactor: centralize ndjson streaming transport

app/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
+- `backend/http/` owns common authentication, request IDs, JSON/NDJSON responses, static
-  error normalization. Feature-specific transport handlers live beside their feature.
+  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`.

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

app/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_ndjson_stream(stream, (ValueError, MentorAgentError))
-        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

app/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)
+        events = ({"type": "delta", "content": chunk} for chunk in stream)
-        self.send_header("Content-Type", "application/x-ndjson; charset=utf-8")
+        self.send_ndjson_stream(events, (ValueError, ReviewAssistantError))
-        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
 
     def save_watchlist(self) -> None:
         try:

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

app/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,
+      "bytes": 48513,
-      "lines": 1129
+      "lines": 1119
     },
     {
       "path": "frontend/styles/theme.css",

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

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

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

docs/governance/code-reduction.md

@@ -25,6 +25,7 @@
 | CR-05 | 根级兼容入口 | 正式后端仍有五处通过迁移兼容模块反向导入规范实现 | 正式代码改用规范路径；兼容入口只服务原公开导入契约 | 已完成 |
 | CR-06 | 紧凑日期显示 | Tushare与选股引擎各保留一份完全相同的`YYYYMMDD`显示转换 | 由`bootstrap/config.py`拥有唯一格式策略，消费者保留原局部别名 | 已完成 |
 | CR-07 | 数据Provider组装 | 核查网关、容器和业务服务是否重复创建外部数据客户端 | 固化唯一创建位置及兼容例外，不改数据源语义 | 已完成 |
+| CR-08 | NDJSON流式传输 | 问师与复盘助手重复维护响应头、事件写入、完成、断线及关闭流程 | HTTP共享层拥有唯一流式连接生命周期 | 已完成 |
 
 ## CR-01验收口径
 
@@ -185,6 +186,26 @@
 本批基线为`xiaobai-reduction-06-date-formatting-20260801`；检查点为
 `xiaobai-reduction-07-provider-ownership-20260801`。
 
+## CR-08验收口径
+
+- 只合并NDJSON响应头、事件序列化、完成事件、业务错误事件、客户端断线和连接关闭这些传输行为。
+- 问师继续直接输出原事件字典；复盘助手继续把文本分片包装为`delta/content`事件。
+- 两个功能各自的业务异常类型、请求体错误状态码、错误正文和流创建时机保持不变。
+- `_write_stream_event`与连接生命周期必须只由`backend/http/handler.py`拥有，应用大类和功能HTTP
+  模块不再保留第二份实现。
+
+## CR-08结果
+
+- 删除问师和复盘助手各16行重复流式控制流，并将应用大类中的10行事件写入方法归入HTTP共享层；
+  新共享实现29行、两个调用适配共3行，生产代码净减少10行。
+- 新增专项测试固定四个响应头、中文NDJSON序列化、增量顺序、完成事件、业务错误事件和关闭状态。
+- 候选324项、纯`app/`导出261项及45项Playwright通过；24个JavaScript文件、API/架构注册表、
+  Git空白检查和SQLite完整性检查通过。
+- 本批不修改提示词、模型选择、会员计次、流式正文、前端解析、API路径、数据库或数据源。
+
+本批基线为`xiaobai-reduction-07-provider-ownership-20260801`；检查点为
+`xiaobai-reduction-08-ndjson-transport-20260801`。
+
 ## 人工验收记录
 
 - 2026-08-01：用户检查CR-02与CR-03运行结果，确认未发现明显异常。本记录仅表示本轮可见功能与