fix: read dashboard history from one snapshot

docs/dashboard-overview-api.md

@@ -18,7 +18,7 @@
 - `market.review_count` 使用 History API repository 返回的 `total`，不使用当前已加载页的数组长度。
 - `personal.watchlist_count` 每次请求读取当前 runtime config；`active_monitor_count` 使用 Alert repository 的 `total`，不受 `page_size=100` 限制。
 - `personal.cached_position_count` 只读取非零缓存仓位 identity，不触发实时估值或写 snapshot，因此块包含 `cached_positions_only`。
-- `activity.recent_reports` 排除 market review，并按 history 分页继续读取直到获得最近 5 条非 market 报告、确认历史已耗尽或扫描达到 100 条；命中上限时返回 `recent_reports_history_scan_incomplete`。`task_stats` 只读当前 task queue 统计。
+- `activity.recent_reports` 排除 market review，并在一次数据库语句中读取最近 100 条历史及同一快照下的总数，再从该固定窗口选择最近 5 条非 market 报告；窗口外仍有历史且窗口内不足 5 条时返回 `recent_reports_history_scan_incomplete`。`task_stats` 只读当前 task queue 统计。
 - `system.refresh_starts_analysis` 固定为 `false`。该 service 不依赖 analyzer、LLM client、market-review generation 或 task submission 方法。
 
 ## What Changed

src/storage.py

@@ -3042,21 +3042,29 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
             # 构建 where 子句
             where_clause = and_(*conditions) if conditions else True
             
-            # 查询总数
-            total_query = select(func.count(AnalysisHistory.id)).where(where_clause)
-            total = session.execute(total_query).scalar() or 0
-            
-            # 查询分页数据
+            # Keep the page and its total in one database statement.  Dashboard
+            # callers use the total as the bounded-window completeness marker;
+            # a separate COUNT could observe a different snapshot from the rows.
+            total_count = func.count(AnalysisHistory.id).over().label("history_total")
             data_query = (
-                select(AnalysisHistory)
+                select(AnalysisHistory, total_count)
                 .where(where_clause)
                 .order_by(desc(AnalysisHistory.created_at), desc(AnalysisHistory.id))
                 .offset(offset)
                 .limit(limit)
             )
-            results = session.execute(data_query).scalars().all()
-            
-            return list(results), total
+            rows = session.execute(data_query).all()
+            if rows:
+                return [row[0] for row in rows], int(rows[0][1] or 0)
+
+            # An empty page after a non-zero offset still needs the full count
+            # for the existing paginated API contract.  Page one (including an
+            # empty table) is already complete from the single statement above.
+            if offset > 0 or limit <= 0:
+                total_query = select(func.count(AnalysisHistory.id)).where(where_clause)
+                total = session.execute(total_query).scalar() or 0
+                return [], int(total)
+            return [], 0
     
     def get_analysis_history_by_id(self, record_id: int) -> Optional[AnalysisHistory]:
         """

tests/test_analysis_history.py

@@ -19,6 +19,8 @@ from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+from sqlalchemy import event
+
 # Keep this test runnable when optional LLM runtime deps are not installed.
 try:
     import litellm  # noqa: F401
@@ -357,6 +359,26 @@ class AnalysisHistoryTestCase(unittest.TestCase):
         self.assertEqual(total, 2)
         self.assertEqual([record.id for record in records], [newer_id, older_id])
 
+    def test_first_history_page_reads_rows_and_total_in_one_statement(self) -> None:
+        self._save_history("stable_window_1")
+        self._save_history("stable_window_2")
+        select_statements = []
+
+        def capture_select(_conn, _cursor, statement, _parameters, _context, _executemany):
+            if statement.lstrip().upper().startswith("SELECT"):
+                select_statements.append(statement)
+
+        event.listen(self.db._engine, "before_cursor_execute", capture_select)
+        try:
+            records, total = self.db.get_analysis_history_paginated(limit=1)
+        finally:
+            event.remove(self.db._engine, "before_cursor_execute", capture_select)
+
+        self.assertEqual(len(records), 1)
+        self.assertEqual(total, 2)
+        self.assertEqual(len(select_statements), 1)
+        self.assertIn("OVER", select_statements[0].upper())
+
     def test_history_display_resolves_bare_jp_kr_code_from_stock_pool(self) -> None:
         result = self._build_result()
         result.code = "005930"