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

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

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

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

@@ -220,6 +220,25 @@
       "runtime_role": "index observation fallback"
     }
   ],
+  "provider_construction": [
+    {
+      "client": "TushareClient",
+      "owner": "backend/data/providers/tushare.py",
+      "compatibility_fallback": "backend/features/market/service.py"
+    },
+    {
+      "client": "IfindHttpClient",
+      "owner": "backend/data/gateway.py"
+    },
+    {
+      "client": "MarketChartClient",
+      "owner": "backend/data/gateway.py"
+    },
+    {
+      "client": "WebRealtimeAggregator",
+      "owner": "backend/data/gateway.py"
+    }
+  ],
   "numeric_normalization": [
     {
       "function": "finite_number",
@@ -268,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",
@@ -345,8 +374,8 @@
     },
     {
       "path": "backend/application.py",
-      "bytes": 48749,
-      "lines": 1129
+      "bytes": 48513,
+      "lines": 1119
     },
     {
       "path": "frontend/styles/theme.css",

app/tests/test_data_gateway.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+import ast
 import unittest
 from datetime import datetime, timedelta
+from pathlib import Path
 
 from backend.data import (
     DataPolicyError,
@@ -45,8 +47,6 @@ class DataGatewayTests(unittest.TestCase):
         self.assertIs(gateway.chart_data.ifind, gateway.ifind)
 
     def test_server_has_no_direct_runtime_tushare_construction(self) -> None:
-        from pathlib import Path
-
         source = (
             Path(__file__).resolve().parents[1]
             / "backend"
@@ -57,6 +57,33 @@ class DataGatewayTests(unittest.TestCase):
         self.assertEqual(source.count("TushareClient(self.token)"), 1)
         self.assertIn("return gateway.tushare()", source)
 
+    def test_provider_construction_has_unique_declared_owners(self) -> None:
+        root = Path(__file__).resolve().parents[1]
+        owners = {
+            "EastmoneyChartClient": {"backend/data/gateway.py"},
+            "IfindHttpClient": {"backend/data/gateway.py"},
+            "IfindProvider": {"backend/data/gateway.py"},
+            "MarketChartClient": {"backend/data/gateway.py"},
+            "TushareClient": {"backend/features/market/service.py"},
+            "TushareProvider": {"backend/data/gateway.py"},
+            "WebRealtimeAggregator": {"backend/data/gateway.py"},
+        }
+        found = {name: set() for name in owners}
+        for path in (root / "backend").rglob("*.py"):
+            relative = path.relative_to(root).as_posix()
+            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+            for node in ast.walk(tree):
+                if not isinstance(node, ast.Call):
+                    continue
+                name = getattr(node.func, "id", None) or getattr(node.func, "attr", None)
+                if name in found:
+                    found[name].add(relative)
+        self.assertEqual(found, owners)
+        provider_source = (root / "backend/data/providers/tushare.py").read_text(
+            encoding="utf-8"
+        )
+        self.assertIn("client_factory: Callable[[str], TushareClient] = TushareClient", provider_source)
+
     def test_quality_gate_accepts_matching_daily_evidence(self) -> None:
         timezone = market_timezone()
         now = datetime(2026, 7, 29, 16, 0, tzinfo=timezone)

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

@@ -155,6 +155,12 @@ def build() -> dict[str, Any]:
             {"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation"},
             {"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation fallback"},
         ],
+        "provider_construction": [
+            {"client": "TushareClient", "owner": "backend/data/providers/tushare.py", "compatibility_fallback": "backend/features/market/service.py"},
+            {"client": "IfindHttpClient", "owner": "backend/data/gateway.py"},
+            {"client": "MarketChartClient", "owner": "backend/data/gateway.py"},
+            {"client": "WebRealtimeAggregator", "owner": "backend/data/gateway.py"},
+        ],
         "numeric_normalization": [
             {"function": "finite_number", "path": "backend/data/numbers.py"},
             {"function": "non_nan_number", "path": "backend/data/numbers.py"},
@@ -173,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

@@ -24,6 +24,8 @@
 | CR-04 | 数值归一化策略 | 四个业务模块分别保留两组完全相同的数值转换函数体 | 由`backend/data/numbers.py`集中拥有两种既有语义，消费者保留原局部别名 | 已完成 |
 | CR-05 | 根级兼容入口 | 正式后端仍有五处通过迁移兼容模块反向导入规范实现 | 正式代码改用规范路径；兼容入口只服务原公开导入契约 | 已完成 |
 | CR-06 | 紧凑日期显示 | Tushare与选股引擎各保留一份完全相同的`YYYYMMDD`显示转换 | 由`bootstrap/config.py`拥有唯一格式策略，消费者保留原局部别名 | 已完成 |
+| CR-07 | 数据Provider组装 | 核查网关、容器和业务服务是否重复创建外部数据客户端 | 固化唯一创建位置及兼容例外，不改数据源语义 | 已完成 |
+| CR-08 | NDJSON流式传输 | 问师与复盘助手重复维护响应头、事件写入、完成、断线及关闭流程 | HTTP共享层拥有唯一流式连接生命周期 | 已完成 |
 
 ## CR-01验收口径
 
@@ -162,6 +164,48 @@
 本批基线为`xiaobai-reduction-05-compatibility-boundaries-20260801`；检查点为
 `xiaobai-reduction-06-date-formatting-20260801`。
 
+## CR-07验收口径
+
+- iFinD、图表、实时观察器和Provider适配器必须只在`build_data_gateway`创建，并由容器共享。
+- Tushare必须继续通过实时Token供应器按需创建，不能为了减少对象数量缓存过期Token。
+- 市场服务中为原版隔离测试桩保留的一处`TushareClient(self.token)`是明确兼容例外，不得被误判为
+  第二条正式数据链路。
+- 不得合并Tushare、iFinD、东方财富和腾讯的传输、缓存、重试或降级逻辑。
+
+## CR-07结果
+
+- 全后端构造点扫描确认iFinD、MarketChart、东方财富图表和实时观察器均只有网关一个创建位置；
+  `ApplicationContainer`暴露的是同一对象引用，没有第二份客户端。
+- Tushare Provider使用动态Token供应器，市场服务只有一处已登记测试兼容回退；本批没有发现可安全
+  删除的生产实现，因此不为追求行数强行修改运行代码。
+- 架构清单新增Provider创建所有权，自动测试会在未来出现第二个未登记构造点时失败。
+- 候选322项、纯`app/`导出259项、24个JavaScript文件、API/架构注册表、Git空白检查和SQLite
+  完整性检查通过；本批不涉及页面、CSS或浏览器行为。
+- 本批不修改请求频率、缓存、重试、Token更新、数据源选择、计算口径、API、数据库或前端。
+
+本批基线为`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运行结果，确认未发现明显异常。本记录仅表示本轮可见功能与