fix: restore heaven interpretation persistence dependency

app/backend/application.py

@@ -128,14 +128,19 @@ class DashboardService(
             profile_supplier=self._resolved_llm_profile,
         )
         self.screener.ensure_builtin_strategies()
-        self._background_stop = threading.Event()
+
-        self._background_thread = self.jobs.start_scheduler(
+    def start_background_jobs(self) -> threading.Thread:
+        return self.jobs.start_scheduler(
             self._background_refresh_tick,
-            self._background_stop,
             interval_seconds=5,
             initial_delay_seconds=3,
         )
 
+    def stop_background_jobs(self, timeout_seconds: float = 5) -> bool:
+        scheduler_stopped = self.jobs.stop_scheduler(timeout_seconds)
+        workers_stopped = self.jobs.wait_for_idle(timeout_seconds)
+        return scheduler_stopped and workers_stopped
+
 
     def _load_system_credentials(self, environment: dict[str, str]) -> dict[str, Any]:
         encrypted = self.database.get_system_setting("credentials")

app/backend/bootstrap/runtime.py

@@ -16,12 +16,13 @@ def main(handler_class: type[Any] | None = None, service: Any | None = None) ->
     parser.add_argument("--port", type=int, default=8765)
     args = parser.parse_args()
     server = ThreadingHTTPServer((args.host, args.port), handler_class)
+    try:
+        service.start_background_jobs()
         print(f"Xiaobai Review Web is running at http://{args.host}:{args.port}")
         print("Press Ctrl+C to stop.")
-    try:
         server.serve_forever()
     except KeyboardInterrupt:
         pass
     finally:
-        service._background_stop.set()
+        service.stop_background_jobs()
         server.server_close()

app/backend/features/heaven/service.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import copy
 import json
 import re
+import secrets
 from datetime import date, datetime, timedelta
 from typing import Any
 

app/backend/jobs/runner.py

@@ -18,6 +18,9 @@ class InProcessJobRunner:
         self.repository = repository
         self._locks: dict[str, threading.Lock] = {}
         self._locks_guard = threading.Lock()
+        self._scheduler_guard = threading.Lock()
+        self._scheduler_stop = threading.Event()
+        self._scheduler_thread: threading.Thread | None = None
 
     def submit(
         self, job_id: str, idempotency_key: str, action: JobAction,
@@ -52,28 +55,43 @@ class InProcessJobRunner:
         return True
 
     def start_scheduler(
-        self, callback: Callable[[], None], stop_event: threading.Event,
+        self, callback: Callable[[], None], interval_seconds: float,
-        interval_seconds: float, initial_delay_seconds: float = 0,
+        initial_delay_seconds: float = 0,
     ) -> threading.Thread:
+        with self._scheduler_guard:
+            current = self._scheduler_thread
+            if current is not None and current.is_alive():
+                return current
+            self._scheduler_stop.clear()
+
             def schedule_loop() -> None:
-            if stop_event.wait(initial_delay_seconds):
+                if self._scheduler_stop.wait(initial_delay_seconds):
                     return
-            while not stop_event.is_set():
+                while not self._scheduler_stop.is_set():
                     try:
                         callback()
                     except Exception:
-                    # Submitted jobs persist their own failures; the scheduler must stay alive.
+                        # Submitted jobs persist failures; the scheduler must stay alive.
                         pass
-                stop_event.wait(interval_seconds)
+                    self._scheduler_stop.wait(interval_seconds)
 
             thread = threading.Thread(
                 target=schedule_loop,
                 name="background-job-scheduler",
                 daemon=True,
             )
+            self._scheduler_thread = thread
             thread.start()
             return thread
 
+    def stop_scheduler(self, timeout_seconds: float = 5) -> bool:
+        with self._scheduler_guard:
+            thread = self._scheduler_thread
+            self._scheduler_stop.set()
+            if thread is not None and thread is not threading.current_thread():
+                thread.join(max(0, timeout_seconds))
+            return thread is None or not thread.is_alive()
+
     def wait_for_idle(self, timeout_seconds: float = 5) -> bool:
         deadline = time.monotonic() + max(0, timeout_seconds)
         while time.monotonic() <= deadline:

app/backend/llm/service.py

@@ -4,6 +4,7 @@ from datetime import datetime, timezone
 from typing import Any
 from urllib.parse import urlparse
 
+from backend.bootstrap.config import validate_text
 from backend.features.screener.compiler import LLMCompilerError, test_llm_connection
 
 

app/config/architecture-inventory.json

@@ -354,8 +354,8 @@
     },
     {
       "path": "backend/features/heaven/service.py",
-      "bytes": 63123,
+      "bytes": 63138,
-      "lines": 1303
+      "lines": 1304
     },
     {
       "path": "backend/features/market/insights.py",
@@ -374,8 +374,8 @@
     },
     {
       "path": "backend/application.py",
-      "bytes": 47552,
+      "bytes": 47769,
-      "lines": 1087
+      "lines": 1092
     },
     {
       "path": "frontend/styles/theme.css",

app/tests/test_background_lifecycle.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import ast
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from backend.bootstrap import runtime
+
+
+class BackgroundLifecycleTests(unittest.TestCase):
+    def test_service_construction_does_not_start_the_scheduler(self) -> None:
+        source = (Path(__file__).parents[1] / "backend" / "application.py").read_text(
+            encoding="utf-8"
+        )
+        tree = ast.parse(source)
+        service_class = next(
+            node for node in tree.body
+            if isinstance(node, ast.ClassDef) and node.name == "DashboardService"
+        )
+        constructor = next(
+            node for node in service_class.body
+            if isinstance(node, ast.FunctionDef) and node.name == "__init__"
+        )
+        called_methods = {
+            node.func.attr for node in ast.walk(constructor)
+            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
+        }
+        self.assertNotIn("start_scheduler", called_methods)
+
+    def test_runtime_starts_jobs_after_bind_and_stops_before_close(self) -> None:
+        events: list[str] = []
+
+        class Service:
+            def start_background_jobs(self) -> None:
+                events.append("start-jobs")
+
+            def stop_background_jobs(self) -> None:
+                events.append("stop-jobs")
+
+        class Server:
+            def __init__(self, address: tuple[str, int], handler: object) -> None:
+                events.append("bind")
+
+            def serve_forever(self) -> None:
+                events.append("serve")
+                raise KeyboardInterrupt
+
+            def server_close(self) -> None:
+                events.append("close")
+
+        with (
+            patch.object(runtime, "ThreadingHTTPServer", Server),
+            patch.object(sys, "argv", ["server.py", "--port", "8797"]),
+        ):
+            runtime.main(object, Service())
+
+        self.assertEqual(
+            events, ["bind", "start-jobs", "serve", "stop-jobs", "close"]
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

app/tests/test_heaven_readings.py

@@ -124,6 +124,50 @@ class HeavenReadingTests(unittest.TestCase):
             len(self.database.list_heaven_readings(self.owner["id"], "fortune")), 1
         )
 
+    def test_trend_interpretation_saves_successful_agent_result(self):
+        service = DashboardService.__new__(DashboardService)
+        service.database = self.database
+        service._request_context = threading.local()
+        service._request_context.user_id = self.owner["id"]
+        setup = {
+            "trade_date": "20260723",
+            "chart": {
+                "available": True,
+                "sector": "银行",
+                "stock": {"code": "000001", "name": "平安银行"},
+                "hexagram": {
+                    "name": "中孚",
+                    "lines": [],
+                    "transformed": {"name": "小畜"},
+                },
+                "movement": {},
+            },
+        }
+
+        with patch.object(service, "heaven_setup", return_value=setup), patch.object(
+            service,
+            "_call_heaven_agent",
+            return_value=(
+                {"answer": "完整的解势结果", "model": "test", "latency_ms": 1},
+                "primary",
+            ),
+        ):
+            result = service.heaven_interpret(
+                {
+                    "mode": "trend",
+                    "trade_date": "2026-07-23",
+                    "stock_code": "000001",
+                }
+            )
+
+        self.assertFalse(result["reused"])
+        self.assertEqual(result["answer"], "完整的解势结果")
+        self.assertEqual(result["reading"]["subject"], "000001 平安银行")
+        self.assertEqual(
+            self.database.list_heaven_readings(self.owner["id"], "trend")[0]["answer"],
+            "完整的解势结果",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

app/tests/test_job_runner.py

@@ -68,6 +68,29 @@ class JobRunnerTests(unittest.TestCase):
         )
         self.assertEqual(self.repository.recent(1)[0]["status"], "failed")
 
+    def test_scheduler_start_is_idempotent_and_stop_waits_for_exit(self) -> None:
+        first = self.runner.start_scheduler(
+            lambda: None, interval_seconds=60, initial_delay_seconds=60
+        )
+        second = self.runner.start_scheduler(
+            lambda: None, interval_seconds=60, initial_delay_seconds=60
+        )
+        self.assertIs(first, second)
+        self.assertTrue(first.is_alive())
+        self.assertTrue(self.runner.stop_scheduler())
+        self.assertFalse(first.is_alive())
+
+    def test_scheduler_can_restart_after_an_orderly_stop(self) -> None:
+        first = self.runner.start_scheduler(
+            lambda: None, interval_seconds=60, initial_delay_seconds=60
+        )
+        self.assertTrue(self.runner.stop_scheduler())
+        second = self.runner.start_scheduler(
+            lambda: None, interval_seconds=60, initial_delay_seconds=60
+        )
+        self.assertIsNot(first, second)
+        self.assertTrue(self.runner.stop_scheduler())
+
 
 if __name__ == "__main__":
     unittest.main()

app/tests/test_llm_gateway.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import unittest
+from unittest.mock import patch
 
 from backend.llm import LLMGateway, LLMGatewayError
+from backend.llm.service import LLMServiceMixin
 
 
 class ProviderFailure(RuntimeError):
@@ -113,6 +115,31 @@ class LLMGatewayTests(unittest.TestCase):
         with self.assertRaisesRegex(LLMGatewayError, "额度已用完"):
             gateway.call("mentor", "mentor-v1", lambda model: "unused", (ProviderFailure,))
 
+    def test_saved_system_model_can_reach_the_connection_probe(self) -> None:
+        service = LLMServiceMixin()
+        service._system_credentials = {
+            "llm_models": [
+                {
+                    "id": "primary-model",
+                    "name": "主模型",
+                    "api_key": "secret",
+                    "base_url": "https://model.example/v1",
+                    "model": "model-name",
+                }
+            ]
+        }
+        service.llm_gateway = self.gateway()
+        expected = {"ok": True, "reply": "OK"}
+        with patch(
+            "backend.llm.service.test_llm_connection", return_value=expected
+        ) as connection_probe:
+            result = service.test_system_llm_profile("primary-model", {})
+
+        self.assertEqual(result, expected)
+        connection_probe.assert_called_once_with(
+            "secret", "https://model.example/v1", "model-name"
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

app/tools/run_preservation_runtime.py

@@ -32,16 +32,21 @@ def main() -> None:
     from server import RequestHandler, SERVICE
 
     server = ThreadingHTTPServer(("127.0.0.1", args.port), RequestHandler)
+    try:
+        if hasattr(SERVICE, "start_background_jobs"):
+            SERVICE.start_background_jobs()
         print(
             f"Preservation runtime is running at http://127.0.0.1:{args.port} "
             f"with database {SERVICE.database.path}",
             flush=True,
         )
-    try:
         server.serve_forever()
     except KeyboardInterrupt:
         pass
     finally:
+        if hasattr(SERVICE, "stop_background_jobs"):
+            SERVICE.stop_background_jobs()
+        else:
             SERVICE._background_stop.set()
         server.server_close()
 

docs/governance/code-reduction.md

@@ -28,6 +28,7 @@
 | CR-08 | NDJSON流式传输 | 问师与复盘助手重复维护响应头、事件写入、完成、断线及关闭流程 | HTTP共享层拥有唯一流式连接生命周期 | 已完成 |
 | CR-09 | 应用服务门面 | 两个仅服务股池iFinD补全的方法仍错位在全局`DashboardService` | 原函数体机械归位到`PoolServiceMixin` | 已完成 |
 | CR-10 | Repository所有权 | 五行行业阶段覆盖的三项持久化方法仍错位在根级数据库门面 | 原函数体机械归位到问天Repository，兼容数据继续保留 | 已完成 |
+| CR-11 | 后台任务生命周期 | 导入应用即启动调度线程，启动早于端口绑定，停止只置位但不等待 | 运行时显式启停，每个Runner只拥有一个可等待的调度线程 | 已完成 |
 
 ## CR-01验收口径
 
@@ -247,6 +248,52 @@
 本批基线为`xiaobai-reduction-09-service-facade-20260801`；检查点为
 `xiaobai-reduction-10-repository-ownership-20260801`。
 
+## CR-11验收口径
+
+- 导入`backend.application`或构造`DashboardService`不得启动后台调度；必须先成功绑定HTTP端口，
+  再由运行时显式启动。
+- 同一`InProcessJobRunner`重复启动调度器必须返回同一活动线程，停止必须置位并在限定时间内等待退出，
+  有序停止后允许重新启动。
+- 运行时关闭顺序固定为：停止调度、等待已提交任务、关闭HTTP服务器；迁移对比工具继续兼容原版入口。
+- 三个任务的注册定义、5秒刷新频率、3秒初始延迟、幂等键、锁、重试、业务函数和结果不得改变。
+- 声明的超时继续是目标与审计字段；Python线程不能安全强杀，本批不伪造硬取消能力。
+
+## CR-11结果
+
+- 删除`DashboardService`构造阶段的调度副作用，端口占用、模块导入和单元测试不再提前创建后台写线程；
+  `backend/bootstrap/runtime.py`成为正式启动与停止所有者。
+- `InProcessJobRunner`集中持有调度停止事件和线程引用；重复启动幂等，停止可等待，原任务锁、持久化运行
+  状态、成功幂等、失败记录和后续重试逻辑保持不变。
+- 新增语法树与运行顺序门禁，固定“构造不启动”“绑定后启动”“停止后关服”，并补齐重复启动与重启测试。
+- 候选328项、纯`app/`导出265项和45项Playwright通过；24个JavaScript文件、API/架构注册表、
+  Git空白检查和SQLite完整性检查通过。
+- 本批没有可安全删除的重复任务实现；为补齐原先缺失的生命周期，生产代码净增加24行。增加内容仅为
+  调度状态、幂等启停和运行时委托，不新增业务层、任务或兼容包装。
+- 本批不修改页面、CSS、API、数据源、刷新计算、自动选股条件、数据库结构、权限或LLM。
+
+本批基线为`xiaobai-reduction-10-repository-ownership-20260801`；检查点为
+`xiaobai-reduction-11-background-jobs-20260801`。
+
+## CR-11人工验收修正
+
+- 2026-08-02人工验收发现本地页面可访问，但行情与LLM同时无法连接。第一原因是验收服务由受限
+  自动化会话启动，子进程继承了禁止外部网络访问的权限；重新在主机正常网络权限下启动后恢复。
+- 随后的主模型连接测试暴露`LLMServiceMixin`机械迁移时遗漏`validate_text`导入，导致请求在真正
+  访问模型前抛出`NameError`并关闭HTTP连接；恢复原依赖并增加保存模型连接探测的运行契约测试。
+- 网站自身实测`000001`返回Tushare日K 60根、分时242点；主模型
+  `MiniMax-M2.7-highspeed`在2236毫秒内回复`OK`，证明服务进程的数据与LLM出网链路均已恢复。
+- 修正后候选329项、纯`app/`导出266项通过；本次只恢复缺失导入和测试，不修改模型配置、额度、
+  提示词、回退策略、行情来源或计算逻辑。
+- 继续验收观势时发现模型已经成功返回，但问天服务在保存解势记录前关闭了HTTP连接。原因是原版
+  `server.py`已有的`secrets`导入在机械拆分到问天服务时遗漏，生成非观气记录去重键时触发
+  `NameError`。迁移版恢复该标准库依赖，并增加“模型成功返回后保存观势结果”的完整服务回归测试。
+- 使用`000001 平安银行`完成真实页面复测：六爻安全门6/6通过，解势结果正常返回并写入历史，
+  `8797`错误日志为空。该修正不改变提示词、模型选择、额度、卦象计算、记录结构或前端行为。
+
+本修正基线为`xiaobai-reduction-11-background-jobs-20260801`；检查点为
+`xiaobai-reduction-11-runtime-connectivity-fix-20260802`；后续解势修正检查点为
+`xiaobai-reduction-11-heaven-interpret-fix-20260802`。
+
 ## 人工验收记录
 
 - 2026-08-01：用户检查CR-02与CR-03运行结果，确认未发现明显异常。本记录仅表示本轮可见功能与