feat: cache agent data tool results (#1117)

docs/CHANGELOG.md

@@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [修复] 修复 Windows 桌面端转抄后端 stdout/stderr 时中文日志可能乱码的问题，统一优先使用 UTF-8 并兼容本地代码页回退
 - [改进] Docker 发布工作流收敛为更清晰的正式发布与手动补发链路，并统一官方 Docker Hub 镜像名为 `zhulinsen/daily_stock_analysis`
 - [文档] 补充官方镜像拉取、`docker run` 用法与 `.env` / 数据目录映射说明，不再仅覆盖 Compose 部署路径
+- [改进] Agent 日线工具优先复用本地缓存，并持久化新获取的日线与新闻情报
 - [修复] GitHub Actions 每日分析工作流补齐 `LLM_CHANNELS`、多 Key 与常用 `LLM_<NAME>_*` 渠道变量透传，避免本地可用的多模型配置在云端定时任务中失效（Fixes #1063, #872）
 - [文档] 修正 `feishu_sender.py` 中飞书自定义机器人 Webhook 消息格式示例为 interactive card JSON，并补充飞书自动化 Webhook 触发器配置教程（参数 JSON 与 `card.elements[0].text.content` 字段映射）。
 

docs/full-guide.md

@@ -1192,6 +1192,14 @@ A: 检查是否启用了 Actions，以及 cron 表达式是否正确（注意是
 - On online failure, fallback to latest cached rate and mark `is_stale=true`.
 - Main snapshot/risk pipeline stays available even when online FX fetch is unavailable.
 
+## Agent 工具数据缓存与持久化
+
+- `get_daily_history` 会先尝试复用本地 `stock_daily` 日线缓存；缓存新鲜且至少覆盖首页默认的 30 条记录时，不再重复请求外部数据源。
+- 当 Agent 请求的天数多于本地缓存记录数时，工具会返回实际可用记录，并通过 `partial_cache=true`、`requested_days`、`actual_records` 标明这是部分缓存命中。
+- 缓存缺失或过期时，工具仍会按原逻辑从数据源获取日线数据；获取成功后会 best-effort 写回 `stock_daily`，保存失败不会阻断 Agent 回复。
+- `search_stock_news` 与 `search_comprehensive_intel` 成功返回后会 best-effort 写入 `news_intel`，复用现有 URL / fallback key 去重逻辑。
+- `get_realtime_quote` 不复用 `stock_daily` 作为实时行情缓存，也不会把盘中实时行情写入日线表；如需实时行情缓存，应单独设计实时行情存储。
+
 ## Portfolio P0 PR3 (Web + Agent Consumption)
 
 ### Web consumption page

docs/full-guide_EN.md

@@ -976,6 +976,14 @@ A: Check if Actions is enabled, and if cron expression is correct (note it's UTC
 - If upstream FX fetch fails, the page may still remain stale after refresh and will explain the fallback result inline.
 - When `PORTFOLIO_FX_UPDATE_ENABLED=false`, the refresh API returns an explicit disabled status and the page shows that online FX refresh is disabled instead of implying that no refreshable pairs exist.
 
+## Agent Tool Data Cache And Persistence
+
+- `get_daily_history` first tries to reuse local `stock_daily` daily-bar cache; when the cache is fresh and contains at least the dashboard default of 30 records, it avoids another external data-source request.
+- If Agent asks for more days than the local cache contains, the tool returns the available records and marks the response with `partial_cache=true`, `requested_days`, and `actual_records`.
+- When the cache is missing or stale, the tool keeps the original data-source fetch path; successful fetches are written back to `stock_daily` on a best-effort basis, and write failures do not block the Agent response.
+- `search_stock_news` and `search_comprehensive_intel` persist successful results to `news_intel` on a best-effort basis, reusing the existing URL / fallback-key deduplication logic.
+- `get_realtime_quote` does not use `stock_daily` as a realtime-quote cache and does not write intraday quotes into the daily-bar table; realtime quote caching should use a dedicated realtime store if needed.
+
 ---
 
 For more questions, please [submit an Issue](https://github.com/ZhuLinsen/daily_stock_analysis/issues)

src/agent/tools/data_tools.py

@@ -12,7 +12,7 @@ Tools:
 import logging
 from datetime import date
 from threading import Lock
-from typing import Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 from src.agent.tools.registry import ToolParameter, ToolDefinition
 
@@ -20,6 +20,8 @@ logger = logging.getLogger(__name__)
 
 _fetcher_manager_singleton = None
 _fetcher_manager_lock = Lock()
+_DAILY_HISTORY_DEFAULT_DAYS = 60
+_DAILY_HISTORY_MAX_DAYS = 365
 
 
 def _get_fetcher_manager():
@@ -51,6 +53,59 @@ def _get_db():
     return get_db()
 
 
+def _normalize_history_days(days: Any) -> Tuple[int, Dict[str, Any]]:
+    """Normalize LLM-provided history window and return response metadata."""
+    requested_days = days
+    warning = None
+    try:
+        if isinstance(days, bool):
+            raise ValueError("bool is not a valid days value")
+        effective_days = int(days)
+    except (TypeError, ValueError):
+        effective_days = _DAILY_HISTORY_DEFAULT_DAYS
+        warning = (
+            f"Invalid days value {requested_days!r}; "
+            f"using default {_DAILY_HISTORY_DEFAULT_DAYS}."
+        )
+
+    if effective_days < 1:
+        effective_days = 1
+        warning = f"days must be >= 1; using {effective_days}."
+    elif effective_days > _DAILY_HISTORY_MAX_DAYS:
+        effective_days = _DAILY_HISTORY_MAX_DAYS
+        warning = f"days exceeds max {_DAILY_HISTORY_MAX_DAYS}; truncated."
+
+    metadata: Dict[str, Any] = {}
+    if warning is not None:
+        metadata.update(
+            {
+                "warning": warning,
+                "requested_days": requested_days,
+                "effective_days": effective_days,
+            }
+        )
+    return effective_days, metadata
+
+
+def _history_code_candidates(stock_code: str) -> Tuple[List[str], str]:
+    """Return cache lookup candidates plus canonical write code."""
+    from data_provider.base import canonical_stock_code, normalize_stock_code
+
+    raw_code = str(stock_code or "").strip()
+    normalized_code = canonical_stock_code(normalize_stock_code(raw_code))
+    candidates: List[str] = []
+    for candidate in (canonical_stock_code(raw_code), normalized_code):
+        if candidate and candidate not in candidates:
+            candidates.append(candidate)
+    return candidates, normalized_code
+
+
+def _append_history_metadata(response: dict, metadata: Dict[str, Any]) -> dict:
+    if metadata:
+        response.update(metadata)
+    return response
+
+
 def _compact_fundamental_context(fundamental_context: dict) -> dict:
     """Reduce token footprint for tool responses while keeping key semantics."""
     if not isinstance(fundamental_context, dict):
@@ -234,25 +289,56 @@ get_realtime_quote_tool = ToolDefinition(
 
 def _handle_get_daily_history(stock_code: str, days: int = 60) -> dict:
     """Get daily OHLCV history data."""
+    effective_days, metadata = _normalize_history_days(days)
+
     from src.services.history_loader import load_history_df
-    df, source = load_history_df(stock_code, days=days)
+    df, source = load_history_df(stock_code, days=effective_days)
 
     if df is None or df.empty:
-        return {"error": f"No historical data available for {stock_code}"}
+        return _append_history_metadata(
+            {"error": f"No historical data available for {stock_code}"},
+            metadata,
+        )
+
+    if source != "db_cache":
+        _, normalized_code = _history_code_candidates(stock_code)
+        try:
+            saved_count = _get_db().save_daily_data(df, normalized_code, source)
+            logger.info(
+                "Agent daily history persisted for %s (source=%s, new_records=%s)",
+                normalized_code,
+                source,
+                saved_count,
+            )
+        except Exception as exc:
+            logger.warning(
+                "Agent daily history persistence failed for %s: %s",
+                normalized_code,
+                exc,
+            )
 
     # Convert DataFrame to list of dicts (last N records)
-    records = df.tail(min(days, len(df))).to_dict(orient="records")
+    records = df.tail(min(effective_days, len(df))).to_dict(orient="records")
     # Ensure date is string
     for r in records:
         if "date" in r:
             r["date"] = str(r["date"])
 
-    return {
-        "code": stock_code,
+    response_code = stock_code
+    if source == "db_cache" and records:
+        response_code = records[-1].get("code") or response_code
+
+    return _append_history_metadata({
+        "code": response_code,
         "source": source,
+        "cache_hit": source == "db_cache",
+        "requested_days": effective_days,
+        "effective_days": effective_days,
+        "actual_records": len(records),
+        "partial_cache": source == "db_cache" and len(records) < effective_days,
         "total_records": len(records),
         "data": records,
-    }
+    }, metadata)
 
 
 get_daily_history_tool = ToolDefinition(

src/agent/tools/search_tools.py

@@ -8,19 +8,66 @@ Tools:
 """
 
 import logging
-from typing import Optional
 
 from src.agent.tools.registry import ToolParameter, ToolDefinition
 
 logger = logging.getLogger(__name__)
 
 
+def _get_db():
+    """Lazy import for DatabaseManager."""
+    from src.storage import get_db
+    return get_db()
+
+
 def _get_search_service():
     """Return shared SearchService singleton."""
     from src.search_service import get_search_service
     return get_search_service()
 
 
+def _canonical_search_code(stock_code: str) -> str:
+    from data_provider.base import canonical_stock_code, normalize_stock_code
+
+    return canonical_stock_code(normalize_stock_code(str(stock_code or "").strip()))
+
+
+def _persist_news_response(
+    *,
+    stock_code: str,
+    stock_name: str,
+    dimension: str,
+    response,
+) -> None:
+    """Best-effort news persistence for Agent search tools."""
+    if not response or not getattr(response, "success", False) or not getattr(response, "results", None):
+        return
+
+    code = _canonical_search_code(stock_code)
+    try:
+        saved_count = _get_db().save_news_intel(
+            code=code,
+            name=stock_name,
+            dimension=dimension,
+            query=response.query,
+            response=response,
+            query_context=None,
+        )
+        logger.info(
+            "Agent news intel persisted for %s (dimension=%s, new_records=%s)",
+            code,
+            dimension,
+            saved_count,
+        )
+    except Exception as exc:
+        logger.warning(
+            "Agent news intel persistence failed for %s (dimension=%s): %s",
+            code,
+            dimension,
+            exc,
+        )
+
+
 def _handle_search_stock_news(stock_code: str, stock_name: str) -> dict:
     """Search latest news for a stock."""
     service = _get_search_service()
@@ -37,6 +84,13 @@ def _handle_search_stock_news(stock_code: str, stock_name: str) -> dict:
             "error": response.error_message,
         }
 
+    _persist_news_response(
+        stock_code=stock_code,
+        stock_name=stock_name,
+        dimension="latest_news",
+        response=response,
+    )
+
     return {
         "query": response.query,
         "provider": response.provider,
@@ -104,6 +158,12 @@ def _handle_search_comprehensive_intel(stock_code: str, stock_name: str) -> dict
     dimensions = {}
     for dim_name, response in intel_results.items():
         if response and response.success:
+            _persist_news_response(
+                stock_code=stock_code,
+                stock_name=stock_name,
+                dimension=dim_name,
+                response=response,
+            )
             dimensions[dim_name] = {
                 "query": response.query,
                 "results_count": len(response.results),

src/services/history_loader.py

@@ -10,13 +10,14 @@ from __future__ import annotations
 
 import contextvars
 import logging
-from datetime import date, timedelta
+from datetime import date, datetime, timedelta
 from threading import Lock
-from typing import Optional, Tuple
+from typing import Any, List, Optional, Tuple
 
 import pandas as pd
 
 logger = logging.getLogger(__name__)
+_CACHE_MIN_RECORDS = 30
 
 # ---------------------------------------------------------------------------
 # Frozen target date (ContextVar) – set once per stock in pipeline, read by
@@ -59,6 +60,69 @@ def _get_fetcher_manager():
 # ---------------------------------------------------------------------------
 # DB-first history loader
 # ---------------------------------------------------------------------------
+def _history_code_candidates(stock_code: str) -> Tuple[List[str], str]:
+    from data_provider.base import canonical_stock_code, normalize_stock_code
+
+    raw_code = str(stock_code or "").strip()
+    normalized_code = canonical_stock_code(normalize_stock_code(raw_code))
+    candidates: List[str] = []
+    for candidate in (canonical_stock_code(raw_code), normalized_code):
+        if candidate and candidate not in candidates:
+            candidates.append(candidate)
+    return candidates, normalized_code
+
+
+def _coerce_bar_date(value: Any) -> date:
+    if isinstance(value, datetime):
+        return value.date()
+    if isinstance(value, date):
+        return value
+    if isinstance(value, str):
+        try:
+            return datetime.strptime(value[:10], "%Y-%m-%d").date()
+        except ValueError:
+            return date.min
+    if hasattr(value, "date"):
+        try:
+            coerced = value.date()
+            return coerced if isinstance(coerced, date) else date.min
+        except Exception:
+            return date.min
+    return date.min
+
+
+def _bar_date(bar: Any) -> date:
+    row_date = _coerce_bar_date(getattr(bar, "date", None))
+    if row_date != date.min:
+        return row_date
+    if hasattr(bar, "to_dict"):
+        try:
+            return _coerce_bar_date((bar.to_dict() or {}).get("date"))
+        except Exception:
+            return date.min
+    return date.min
+
+
+def _select_best_bars(db, stock_code: str, start: date, end: date) -> Tuple[Optional[str], list]:
+    candidates, normalized_code = _history_code_candidates(stock_code)
+    best_code = None
+    best_bars = []
+    best_key = None
+
+    for candidate in candidates:
+        bars = list(db.get_data_range(candidate, start, end) or [])
+        if not bars:
+            continue
+        latest_date = max(_bar_date(bar) for bar in bars)
+        key = (latest_date, len(bars), candidate == normalized_code)
+        if best_key is None or key > best_key:
+            best_key = key
+            best_code = candidate
+            best_bars = bars
+
+    return best_code, best_bars
+
+
 def load_history_df(
     stock_code: str,
     days: int = 60,
@@ -70,7 +134,6 @@ def load_history_df(
     actual provider name on network fallback.  Returns ``(None, "none")`` when
     both paths fail.
     """
-    from data_provider.base import canonical_stock_code, normalize_stock_code
     from src.storage import get_db
 
     # Resolve effective end date
@@ -84,15 +147,12 @@ def load_history_df(
     start = end - timedelta(days=int(days * 1.8) + 10)
 
     # --- 1. DB lookup (canonical code, then prefix-stripped fallback) ------
-    code = canonical_stock_code(stock_code)
     try:
         db = get_db()
-        bars = db.get_data_range(code, start, end)
-        if not bars:
-            alt = normalize_stock_code(stock_code)
-            if alt != code:
-                bars = db.get_data_range(alt, start, end)
-        if bars and len(bars) >= max(int(days * 0.3), 5):
+        _code, bars = _select_best_bars(db, stock_code, start, end)
+        required_records = max(min(days, _CACHE_MIN_RECORDS), 1)
+        latest_date = max((_bar_date(bar) for bar in bars), default=date.min)
+        if bars and latest_date >= end and len(bars) >= required_records:
             df = pd.DataFrame([b.to_dict() for b in bars])
             logger.debug(
                 "load_history_df(%s): %d bars from DB (requested %d)",

tests/test_data_tools_daily_history_cache.py

@@ -0,0 +1,230 @@
+# -*- coding: utf-8 -*-
+"""Tests for Agent get_daily_history DB cache reuse."""
+
+from datetime import date, timedelta
+from types import SimpleNamespace
+from typing import Optional
+import unittest
+from unittest.mock import MagicMock, patch
+
+import pandas as pd
+
+from src.agent.tools.data_tools import _handle_get_daily_history
+from src.services.history_loader import reset_frozen_target_date, set_frozen_target_date
+
+
+class _DailyRow:
+    def __init__(self, code: str, row_date: date, close: float) -> None:
+        self.code = code
+        self.date = row_date
+        self.open = close - 1
+        self.high = close + 1
+        self.low = close - 2
+        self.close = close
+        self.volume = 1000
+        self.amount = 10000
+        self.pct_chg = 1.2
+        self.ma5 = close - 0.5
+        self.ma10 = close - 1
+        self.ma20 = close - 2
+        self.volume_ratio = 1.1
+        self.data_source = "unit-test"
+
+    def to_dict(self):
+        return {
+            "code": self.code,
+            "date": self.date,
+            "open": self.open,
+            "high": self.high,
+            "low": self.low,
+            "close": self.close,
+            "volume": self.volume,
+            "amount": self.amount,
+            "pct_chg": self.pct_chg,
+            "ma5": self.ma5,
+            "ma10": self.ma10,
+            "ma20": self.ma20,
+            "volume_ratio": self.volume_ratio,
+            "data_source": self.data_source,
+        }
+
+
+def _rows(code: str, latest: date, count: int):
+    return [
+        _DailyRow(code, latest - timedelta(days=offset), close=100 + offset)
+        for offset in range(count)
+    ]
+
+
+class _FakeDb:
+    def __init__(self, rows_by_code=None, save_error: Optional[Exception] = None) -> None:
+        self.rows_by_code = rows_by_code or {}
+        self.save_error = save_error
+        self.save_daily_data = MagicMock(side_effect=self._save_daily_data)
+
+    def get_data_range(self, code: str, start_date: date, end_date: date):
+        rows = [
+            row
+            for row in self.rows_by_code.get(code, [])
+            if start_date <= row.date <= end_date
+        ]
+        return sorted(rows, key=lambda row: row.date)
+
+    def _save_daily_data(self, df, code: str, source: str):
+        if self.save_error:
+            raise self.save_error
+        return len(df)
+
+
+class DailyHistoryCacheToolTest(unittest.TestCase):
+    def _run_with_frozen_date(self, target: date, stock_code: str, days: int):
+        token = set_frozen_target_date(target)
+        try:
+            return _handle_get_daily_history(stock_code, days=days)
+        finally:
+            reset_frozen_target_date(token)
+
+    def test_uses_fresh_partial_db_cache_without_fetching(self) -> None:
+        target = date(2026, 4, 24)
+        db = _FakeDb({"600519": _rows("600519", target, 30)})
+        manager = SimpleNamespace(get_daily_data=MagicMock())
+
+        with patch("src.storage.get_db", return_value=db), \
+             patch("src.services.history_loader._get_fetcher_manager", return_value=manager):
+            result = self._run_with_frozen_date(target, "600519", days=60)
+
+        self.assertEqual(result["source"], "db_cache")
+        self.assertTrue(result["cache_hit"])
+        self.assertTrue(result["partial_cache"])
+        self.assertEqual(result["actual_records"], 30)
+        self.assertEqual(result["requested_days"], 60)
+        self.assertEqual(result["data"][0]["date"], str(target - timedelta(days=29)))
+        self.assertEqual(result["data"][-1]["date"], str(target))
+        manager.get_daily_data.assert_not_called()
+
+    def test_prefers_fuller_candidate_when_dates_tie(self) -> None:
+        target = date(2026, 4, 24)
+        db = _FakeDb(
+            {
+                "1810.HK": _rows("1810.HK", target, 40),
+                "HK01810": _rows("HK01810", target, 30),
+            }
+        )
+        manager = SimpleNamespace(get_daily_data=MagicMock())
+
+        with patch("src.storage.get_db", return_value=db), \
+             patch("src.services.history_loader._get_fetcher_manager", return_value=manager):
+            result = self._run_with_frozen_date(target, "1810.HK", days=60)
+
+        self.assertEqual(result["code"], "1810.HK")
+        self.assertEqual(result["actual_records"], 40)
+        manager.get_daily_data.assert_not_called()
+
+    def test_prefers_normalized_candidate_when_dates_and_counts_tie(self) -> None:
+        target = date(2026, 4, 24)
+        db = _FakeDb(
+            {
+                "1810.HK": _rows("1810.HK", target, 30),
+                "HK01810": _rows("HK01810", target, 30),
+            }
+        )
+        manager = SimpleNamespace(get_daily_data=MagicMock())
+
+        with patch("src.storage.get_db", return_value=db), \
+             patch("src.services.history_loader._get_fetcher_manager", return_value=manager):
+            result = self._run_with_frozen_date(target, "1810.HK", days=60)
+
+        self.assertEqual(result["code"], "HK01810")
+        self.assertEqual(result["actual_records"], 30)
+        manager.get_daily_data.assert_not_called()
+
+    def test_fetches_and_persists_when_cache_is_stale(self) -> None:
+        target = date(2026, 4, 24)
+        db = _FakeDb({"600519": _rows("600519", target - timedelta(days=1), 30)})
+        df = pd.DataFrame(
+            [
+                {"date": target, "open": 1, "high": 2, "low": 0.5, "close": 1.5},
+            ]
+        )
+        manager = SimpleNamespace(get_daily_data=MagicMock(return_value=(df, "Fetcher")))
+
+        with patch("src.storage.get_db", return_value=db), \
+             patch("src.agent.tools.data_tools._get_db", return_value=db), \
+             patch("src.services.history_loader._get_fetcher_manager", return_value=manager):
+            result = self._run_with_frozen_date(target, "600519", days=60)
+
+        manager.get_daily_data.assert_called_once_with("600519", days=60)
+        db.save_daily_data.assert_called_once_with(df, "600519", "Fetcher")
+        self.assertFalse(result["cache_hit"])
+        self.assertEqual(result["source"], "Fetcher")
+
+    def test_save_failure_does_not_hide_fetched_data(self) -> None:
+        target = date(2026, 4, 24)
+        db = _FakeDb(save_error=RuntimeError("db locked"))
+        df = pd.DataFrame(
+            [
+                {"date": target, "open": 1, "high": 2, "low": 0.5, "close": 1.5},
+            ]
+        )
+        manager = SimpleNamespace(get_daily_data=MagicMock(return_value=(df, "Fetcher")))
+
+        with patch("src.storage.get_db", return_value=db), \
+             patch("src.agent.tools.data_tools._get_db", return_value=db), \
+             patch("src.services.history_loader._get_fetcher_manager", return_value=manager):
+            result = self._run_with_frozen_date(target, "600519", days=60)
+
+        self.assertEqual(result["total_records"], 1)
+        self.assertEqual(result["data"][0]["date"], str(target))
+
+    def test_db_read_exception_falls_back_to_fetch(self) -> None:
+        target = date(2026, 4, 24)
+        df = pd.DataFrame(
+            [{"date": target, "open": 1, "high": 2, "low": 0.5, "close": 1.5}]
+        )
+        manager = SimpleNamespace(get_daily_data=MagicMock(return_value=(df, "Fetcher")))
+        broken_db = MagicMock()
+        broken_db.get_data_range.side_effect = RuntimeError("db corrupted")
+        broken_db.save_daily_data.return_value = 1
+
+        with patch("src.storage.get_db", return_value=broken_db), \
+             patch("src.agent.tools.data_tools._get_db", return_value=broken_db), \
+             patch("src.services.history_loader._get_fetcher_manager", return_value=manager):
+            result = self._run_with_frozen_date(target, "600519", days=60)
+
+        manager.get_daily_data.assert_called_once_with("600519", days=60)
+        self.assertFalse(result["cache_hit"])
+        self.assertEqual(result["source"], "Fetcher")
+
+    def test_days_one_cache_hit_with_single_fresh_record(self) -> None:
+        target = date(2026, 4, 24)
+        db = _FakeDb({"600519": _rows("600519", target, 1)})
+        manager = SimpleNamespace(get_daily_data=MagicMock())
+
+        with patch("src.storage.get_db", return_value=db), \
+             patch("src.services.history_loader._get_fetcher_manager", return_value=manager):
+            result = self._run_with_frozen_date(target, "600519", days=1)
+
+        self.assertTrue(result["cache_hit"])
+        self.assertEqual(result["actual_records"], 1)
+        self.assertFalse(result["partial_cache"])
+        manager.get_daily_data.assert_not_called()
+
+    def test_days_are_normalized_with_warning(self) -> None:
+        target = date(2026, 4, 24)
+        db = _FakeDb()
+        df = pd.DataFrame([{"date": target, "close": 1.5}])
+        manager = SimpleNamespace(get_daily_data=MagicMock(return_value=(df, "Fetcher")))
+
+        with patch("src.storage.get_db", return_value=db), \
+             patch("src.agent.tools.data_tools._get_db", return_value=db), \
+             patch("src.services.history_loader._get_fetcher_manager", return_value=manager):
+            result = self._run_with_frozen_date(target, "600519", days=999)
+
+        manager.get_daily_data.assert_called_once_with("600519", days=365)
+        self.assertEqual(result["requested_days"], 999)
+        self.assertEqual(result["effective_days"], 365)
+        self.assertIn("warning", result)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_search_tools_persistence.py

@@ -0,0 +1,121 @@
+# -*- coding: utf-8 -*-
+"""Tests for Agent search tool news persistence."""
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from src.agent.tools.search_tools import (
+    _handle_search_comprehensive_intel,
+    _handle_search_stock_news,
+)
+from src.search_service import SearchResponse, SearchResult
+
+
+def _response(query: str, *, success: bool = True) -> SearchResponse:
+    return SearchResponse(
+        query=query,
+        provider="UnitSearch",
+        success=success,
+        error_message=None if success else "search failed",
+        results=[
+            SearchResult(
+                title="新闻标题",
+                snippet="新闻摘要",
+                url="https://example.com/news",
+                source="example.com",
+                published_date="2026-04-24",
+            )
+        ] if success else [],
+    )
+
+
+class SearchToolsPersistenceTest(unittest.TestCase):
+    def test_search_stock_news_persists_successful_response(self) -> None:
+        response = _response("贵州茅台 600519 latest news")
+        service = SimpleNamespace(
+            is_available=True,
+            search_stock_news=MagicMock(return_value=response),
+        )
+        db = SimpleNamespace(save_news_intel=MagicMock(return_value=1))
+
+        with patch("src.agent.tools.search_tools._get_search_service", return_value=service), \
+             patch("src.agent.tools.search_tools._get_db", return_value=db):
+            result = _handle_search_stock_news("600519", "贵州茅台")
+
+        self.assertTrue(result["success"])
+        db.save_news_intel.assert_called_once_with(
+            code="600519",
+            name="贵州茅台",
+            dimension="latest_news",
+            query=response.query,
+            response=response,
+            query_context=None,
+        )
+
+    def test_search_comprehensive_intel_persists_successful_dimensions_only(self) -> None:
+        latest = _response("latest")
+        failed = _response("risk", success=False)
+        service = SimpleNamespace(
+            is_available=True,
+            search_comprehensive_intel=MagicMock(
+                return_value={"latest_news": latest, "risk_check": failed}
+            ),
+            format_intel_report=MagicMock(return_value="report"),
+        )
+        db = SimpleNamespace(save_news_intel=MagicMock(return_value=1))
+
+        with patch("src.agent.tools.search_tools._get_search_service", return_value=service), \
+             patch("src.agent.tools.search_tools._get_db", return_value=db):
+            result = _handle_search_comprehensive_intel("600519", "贵州茅台")
+
+        self.assertEqual(result["report"], "report")
+        self.assertEqual(list(result["dimensions"].keys()), ["latest_news"])
+        db.save_news_intel.assert_called_once_with(
+            code="600519",
+            name="贵州茅台",
+            dimension="latest_news",
+            query=latest.query,
+            response=latest,
+            query_context=None,
+        )
+
+    def test_persistence_failure_keeps_search_result(self) -> None:
+        response = _response("贵州茅台 600519 latest news")
+        service = SimpleNamespace(
+            is_available=True,
+            search_stock_news=MagicMock(return_value=response),
+        )
+        db = SimpleNamespace(save_news_intel=MagicMock(side_effect=RuntimeError("db locked")))
+
+        with patch("src.agent.tools.search_tools._get_search_service", return_value=service), \
+             patch("src.agent.tools.search_tools._get_db", return_value=db):
+            result = _handle_search_stock_news("600519", "贵州茅台")
+
+        self.assertTrue(result["success"])
+        self.assertEqual(result["results_count"], 1)
+
+    def test_unavailable_or_failed_search_does_not_persist(self) -> None:
+        unavailable = SimpleNamespace(is_available=False)
+        db = SimpleNamespace(save_news_intel=MagicMock())
+        with patch("src.agent.tools.search_tools._get_search_service", return_value=unavailable), \
+             patch("src.agent.tools.search_tools._get_db", return_value=db):
+            result = _handle_search_stock_news("600519", "贵州茅台")
+
+        self.assertIn("error", result)
+        db.save_news_intel.assert_not_called()
+
+        failed = SimpleNamespace(
+            is_available=True,
+            search_stock_news=MagicMock(return_value=_response("latest", success=False)),
+        )
+        with patch("src.agent.tools.search_tools._get_search_service", return_value=failed), \
+             patch("src.agent.tools.search_tools._get_db", return_value=db):
+            result = _handle_search_stock_news("600519", "贵州茅台")
+
+        self.assertFalse(result["success"])
+        db.save_news_intel.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()