feat(runtime): partial notify and diagnostics after scheduler timeout (#2338)

* feat(runtime): partial notify and diagnostics after scheduler timeout

After a hard timeout, scan already-saved analyses and enrich last_error
with completed/pending counts; optional push via DSA_TIMEOUT_PARTIAL_NOTIFY.

Refs #2328

* test(runtime): cover timeout partial delivery helpers

Refs #2328

* docs: document DSA_TIMEOUT_PARTIAL_NOTIFY

Refs #2328

* fix(config): use switch ui_control for timeout partial notify

DSA_TIMEOUT_PARTIAL_NOTIFY used ui_control=toggle, which SystemConfigResponse rejects and broke GET /config in backend-tests 1/3.

* docs(runtime): document timeout partial fail-open for operators

Channel exceptions are swallowed after the analysis lock is released, so they cannot keep status.running true. Collect/import failures stay in warning logs because last_error cannot distinguish them from zero completions.

.env.example

@@ -826,6 +826,8 @@ SCHEDULE_TIME=18:00
 SCHEDULE_TIMES=
 # Web/API runtime scheduler 单次分析硬超时（秒，最小 60 秒）
 DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS=2700
+# 硬超时后是否推送已落库个股的部分完成通知（true/false；默认 true，未设置即开启，此前超时不推送）
+DSA_TIMEOUT_PARTIAL_NOTIFY=true
 # 定时模式启动时是否立即执行一次分析（true/false）
 # 若未显式设置，定时模式会沿用 RUN_IMMEDIATELY 的运行时覆盖语义以兼容旧配置
 SCHEDULE_RUN_IMMEDIATELY=true

docs/CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+- [新功能] Web/API runtime scheduler 硬超时后扫描已落库分析历史，**默认发送**部分完成通知（`DSA_TIMEOUT_PARTIAL_NOTIFY` 未设置或为 true；此前超时不推送已落库个股），并在 `last_error` 中记录 `completed/pending` 摘要；可用 `DSA_TIMEOUT_PARTIAL_NOTIFY=false` 关闭推送（Refs #2328）。
 - [测试] 修复股票名称解析冷启动超时并发测试的同步竞态：在放行后台抓取前确认两个等待者均已结束并返回空结果，避免 Docker 发布门禁偶发失败。
 - [文档] 将仓库内所有 SerpApi 链接统一更新为新的赞助转化追踪地址。
 - [修复] 智能导入兼容带 UTF-8 BOM 的 CSV 与剪贴板文本，避免 `code` 表头被误当成数据并丢失有效股票代码。

docs/full-guide.md

@@ -496,6 +496,7 @@ daily_stock_analysis/
 | `SCHEDULE_TIME` | 定时执行时间 | `18:00` |
 | `SCHEDULE_TIMES` | 多个定时执行时间，逗号分隔；为空时使用 `SCHEDULE_TIME` | 空 |
 | `DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS` | Web/API runtime scheduler 单次分析硬超时（秒，最小 60 秒）；超时后终止独立分析进程，不阻塞后续任务 | `2700` |
+| `DSA_TIMEOUT_PARTIAL_NOTIFY` | **默认开启（行为变化）**：硬超时后对已落库个股发送「部分完成」通知；此前超时不推送。`false` 时仍写入结构化 `last_error`，但不推送 | `true` |
 | `LOG_DIR` | 日志目录 | `./logs` |
 | `SAVE_CONTEXT_SNAPSHOT` | 保存分析历史 `context_snapshot`；设为 `false` 时新历史不保存 enhanced_context、market_phase_summary、AnalysisContextPack overview 或诊断快照，但不关闭当次 Prompt 低敏摘要 | `true` |
 
@@ -890,10 +891,13 @@ python main.py --schedule --no-run-immediately
 | `SCHEDULE_TIME` | 每日执行时间 (HH:MM) | `18:00` | `09:30` |
 | `SCHEDULE_TIMES` | 多个每日执行时间，逗号分隔；为空时使用 `SCHEDULE_TIME` | 空 | `09:20,12:30,15:10,18:00` |
 | `DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS` | Web/API runtime scheduler 单次分析硬超时（秒，最小 60 秒） | `2700` | `3600` |
+| `DSA_TIMEOUT_PARTIAL_NOTIFY` | **默认开启（行为变化）**：硬超时后是否推送已落库个股的部分完成通知；此前超时不推送 | `true` | `false` |
 | `SCHEDULE_RUN_IMMEDIATELY` | 定时模式启动时是否立即运行一次；未显式设置时沿用 `RUN_IMMEDIATELY` 的运行时覆盖语义 | `true` | `false` |
 | `RUN_IMMEDIATELY` | 非定时模式启动时是否立即运行一次；同时作为未显式设置 `SCHEDULE_RUN_IMMEDIATELY` 时的 legacy 回退 | `true` | `false` |
 | `TRADING_DAY_CHECK_ENABLED` | 交易日检查：非交易日跳过执行；设为 `false` 可强制执行 | `true` | `false` |
 
+> 超时部分完成排障：通知渠道异常只记 warning（`Partial timeout notification failed`）并跳过推送，**不抛出、不占用** `status().running`（notify 在分析锁释放后的后台线程）。扫库 / `src.storage` 导入失败同样 fail-open：API 上可能仍是 baseline `timed out after Ns`，或被 enrich 成 `completed=0`（与「确实没有落库」无法区分）。运维请搜 `Failed to collect completed analyses after timeout`、`Timeout partial delivery failed open`、`Partial timeout notification failed`。合入后首次发布请关注通知量，避免默认开启造成骚扰；可设 `DSA_TIMEOUT_PARTIAL_NOTIFY=false` 关闭推送。
+
 例如在 Docker 中配置：
 
 ```bash

docs/full-guide_EN.md

@@ -417,11 +417,14 @@ For the notification baseline, diagnostics, and deployment notes, see [Notificat
 | `SCHEDULE_TIME` | Scheduled execution time | `18:00` |
 | `SCHEDULE_TIMES` | Multiple scheduled execution times, comma-separated; falls back to `SCHEDULE_TIME` when empty | empty |
 | `DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS` | Hard timeout in seconds for each Web/API runtime scheduler analysis (minimum 60); the isolated analysis process is terminated on timeout so later runs can continue | `2700` |
+| `DSA_TIMEOUT_PARTIAL_NOTIFY` | **Default on (behavior change):** after a hard timeout, send a partial notification for analyses already saved to history; previously timeout sent no per-stock push. When `false`, structured `last_error` is still recorded but no push is sent | `true` |
 | `SCHEDULE_RUN_IMMEDIATELY` | Run once immediately when scheduler mode starts; when unset it keeps following the legacy `RUN_IMMEDIATELY` runtime override | `true` |
 | `RUN_IMMEDIATELY` | Run once immediately for non-scheduler startup; also acts as the legacy fallback when `SCHEDULE_RUN_IMMEDIATELY` is unset | `true` |
 | `LOG_DIR` | Log directory | `./logs` |
 | `SAVE_CONTEXT_SNAPSHOT` | Persist analysis-history `context_snapshot`. When false, new history records do not save enhanced_context, market_phase_summary, AnalysisContextPack overview, or diagnostic snapshots, but current-run prompt summaries remain enabled | `true` |
 
+> Timeout partial-delivery ops: a notification-channel exception is logged (`Partial timeout notification failed`) and skipped; it is not re-raised and cannot keep `status().running` true (notify runs on a daemon thread after the analysis lock is released). Storage import / collect failures also fail open: `status().last_error` may stay at the baseline `timed out after Ns` string, or be enriched to `completed=0`, which is indistinguishable from “nothing was saved”. Grep `Failed to collect completed analyses after timeout`, `Timeout partial delivery failed open`, and `Partial timeout notification failed`. After the first release, watch notification volume; set `DSA_TIMEOUT_PARTIAL_NOTIFY=false` to disable pushes.
+
 > Behavior notes:
 > - When `TICKFLOW_API_KEY` is configured, TickFlow is instantiated as an optional A-share daily K-line data source and CN market-review enhancer. `TICKFLOW_PRIORITY` only affects the generic A-share daily K-line/provider fallback chain. Realtime quote priority is controlled separately by `REALTIME_SOURCE_PRIORITY`; TickFlow realtime quotes are used only when that list explicitly includes `tickflow`, and any source listed before `tickflow` is tried first.
 > - All SH/SZ indices currently registered in `IndexRegistry` (the `sh`/`sz`/`csi` prefixes; see the [Index watchlist configuration](#index-watchlist-configuration) registry notes and `scripts/stock_index_seeds/index_registry.csv` for the full list) bypass generic priority sorting when given as explicit-market inputs (exchange-suffix forms such as `000016.SH` are also accepted) and use the fixed Tencent → AkShare → TickFlow → YFinance fallback chain; unconfigured or unavailable providers are skipped. Bare `000016`-style inputs remain stocks and do not enter the index chain. This fixed chain ignores `EFINANCE_PRIORITY`, `AKSHARE_PRIORITY`, `TUSHARE_PRIORITY`, `TICKFLOW_PRIORITY`, `PYTDX_PRIORITY`, `BAOSTOCK_PRIORITY`, `YFINANCE_PRIORITY`, and `TENCENT_PRIORITY`. Existing stock and realtime-quote ordering is unchanged.

src/core/config_registry.py

@@ -3119,6 +3119,35 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
         ],
         "warning_codes": [],
     },
+    "DSA_TIMEOUT_PARTIAL_NOTIFY": {
+        "title": "Timeout Partial Notification",
+        "description": (
+            "After a Web/API runtime scheduler hard timeout, send a partial "
+            "notification for analyses already saved to history."
+        ),
+        "category": "system",
+        "data_type": "boolean",
+        "ui_control": "switch",
+        "is_sensitive": False,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": "true",
+        "options": [],
+        "validation": {},
+        "display_order": 13,
+        "help_key": "settings.system.schedule",
+        "examples": [
+            "DSA_TIMEOUT_PARTIAL_NOTIFY=true",
+            "DSA_TIMEOUT_PARTIAL_NOTIFY=false",
+        ],
+        "docs": [
+            {
+                "label": "Full guide: configuration",
+                "href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#其他配置",
+            },
+        ],
+        "warning_codes": [],
+    },
     "HTTP_PROXY": {
         "title": "HTTP Proxy",
         "description": "Optional HTTP proxy endpoint.",

src/services/analysis_timeout_partial.py

@@ -0,0 +1,361 @@
+# -*- coding: utf-8 -*-
+"""Timeout partial delivery helpers for RuntimeSchedulerService.
+
+When the Web/API runtime scheduler hard-timeout kills a worker, successful
+per-stock analyses may already be persisted. This module collects those rows
+and optionally sends a partial notification with richer diagnostics.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Sequence
+
+from sqlalchemy import and_, desc, select
+
+logger = logging.getLogger(__name__)
+
+TIMEOUT_PARTIAL_NOTIFY_ENV = "DSA_TIMEOUT_PARTIAL_NOTIFY"
+_FALSEY = {"0", "false", "no", "off"}
+_TRUTHY = {"1", "true", "yes", "on"}
+
+
+@dataclass(frozen=True)
+class CompletedAnalysisSummary:
+    """One stock analysis already saved before the timeout kill."""
+
+    code: str
+    name: Optional[str] = None
+    operation_advice: Optional[str] = None
+    sentiment_score: Optional[int] = None
+    created_at: Optional[datetime] = None
+    history_id: Optional[int] = None
+
+
+@dataclass(frozen=True)
+class TimeoutPartialOutcome:
+    """Result of timeout partial collection + optional notify."""
+
+    completed: List[CompletedAnalysisSummary]
+    pending_codes: List[str]
+    notified: bool
+    notify_skipped_reason: Optional[str]
+    error_message: str
+
+
+def is_timeout_partial_notify_enabled() -> bool:
+    """Return whether ``DSA_TIMEOUT_PARTIAL_NOTIFY`` allows timeout notify.
+
+    Default: enabled (true) when unset. Accepts common truthy/falsey strings.
+    """
+    raw = os.getenv(TIMEOUT_PARTIAL_NOTIFY_ENV)
+    if raw is None or not str(raw).strip():
+        return True
+    normalized = str(raw).strip().lower()
+    if normalized in _FALSEY:
+        return False
+    if normalized in _TRUTHY:
+        return True
+    logger.warning(
+        "Invalid %s=%r; treating as enabled",
+        TIMEOUT_PARTIAL_NOTIFY_ENV,
+        raw,
+    )
+    return True
+
+
+def resolve_expected_stock_codes(
+    stock_codes: Optional[Sequence[str]],
+    *,
+    config: Optional[Any] = None,
+) -> List[str]:
+    """Resolve the stock universe for this run.
+
+    When ``stock_codes`` is provided, normalize and return it. Otherwise fall
+    back to ``config.stock_list`` (caller may pass a loaded Config).
+    """
+    if stock_codes is not None:
+        return _normalize_codes(stock_codes)
+
+    if config is not None:
+        stock_list = getattr(config, "stock_list", None) or []
+        return _normalize_codes(stock_list)
+
+    return []
+
+
+def collect_completed_analyses_since(
+    *,
+    run_started_at: datetime,
+    expected_codes: Sequence[str],
+    db: Optional[Any] = None,
+) -> List[CompletedAnalysisSummary]:
+    """Load successful analysis_history rows created at/after run start.
+
+    Only codes in ``expected_codes`` are considered when that list is non-empty.
+    When the same code has multiple rows in the window, keep the latest.
+    Fail-open: DB/import errors return an empty list after logging a warning.
+    Callers cannot tell "no rows yet" from "collect failed"; operators must
+    grep ``Failed to collect completed analyses after timeout``. ``status()``
+    may then show ``completed=0`` or keep the baseline timeout ``last_error``.
+    """
+    try:
+        rows = _query_history_rows_since(
+            run_started_at=run_started_at,
+            expected_codes=expected_codes,
+            db=db,
+        )
+        return _summaries_from_history_rows(rows, expected_codes)
+    except Exception as exc:  # noqa: BLE001 - timeout path must stay fail-open
+        logger.warning("Failed to collect completed analyses after timeout: %s", exc)
+        return []
+
+
+def _query_history_rows_since(
+    *,
+    run_started_at: datetime,
+    expected_codes: Sequence[str],
+    db: Optional[Any] = None,
+) -> List[Any]:
+    """Fetch raw analysis_history ORM rows for the timeout window."""
+    storage_mod = _resolve_storage_module()
+    database = db if db is not None else storage_mod.get_db()
+    AnalysisHistory = storage_mod.AnalysisHistory
+    codes = _normalize_codes(expected_codes)
+    conditions = [
+        AnalysisHistory.created_at >= run_started_at,
+        AnalysisHistory.report_type != "market_review",
+    ]
+    if codes:
+        conditions.append(AnalysisHistory.code.in_(codes))
+
+    with database.get_session() as session:
+        return list(
+            session.execute(
+                select(AnalysisHistory)
+                .where(and_(*conditions))
+                .order_by(desc(AnalysisHistory.created_at), desc(AnalysisHistory.id))
+                .limit(500)
+            ).scalars().all()
+        )
+
+
+def _resolve_storage_module() -> Any:
+    """Load ``src.storage``, cleaning up broken partial imports on failure.
+
+    On import failure, drop ``src.storage`` / ``src.storage.*`` from
+    ``sys.modules`` and raise. The timeout collect path fail-opens to an
+    empty list; this is not surfaced on ``status().last_error`` beyond a
+    possible ``completed=0`` enrich. Watch the warning log, not the API.
+    """
+    import importlib
+    import sys
+
+    existing = sys.modules.get("src.storage")
+    if existing is not None:
+        return existing
+    try:
+        return importlib.import_module("src.storage")
+    except Exception as exc:
+        for name in list(sys.modules):
+            if name == "src.storage" or name.startswith("src.storage."):
+                sys.modules.pop(name, None)
+        raise RuntimeError("storage unavailable for timeout partial collect") from exc
+
+
+def _summaries_from_history_rows(
+    rows: Sequence[Any],
+    expected_codes: Sequence[str],
+) -> List[CompletedAnalysisSummary]:
+    """Convert ORM/history-like rows into latest-per-code summaries."""
+    codes = _normalize_codes(expected_codes)
+    latest_by_code: Dict[str, CompletedAnalysisSummary] = {}
+    for row in rows:
+        code = str(getattr(row, "code", "") or "").strip()
+        if not code or code in latest_by_code:
+            continue
+        if codes and code not in set(codes):
+            continue
+        latest_by_code[code] = CompletedAnalysisSummary(
+            code=code,
+            name=getattr(row, "name", None),
+            operation_advice=getattr(row, "operation_advice", None),
+            sentiment_score=getattr(row, "sentiment_score", None),
+            created_at=getattr(row, "created_at", None),
+            history_id=getattr(row, "id", None),
+        )
+
+    if codes:
+        return [latest_by_code[code] for code in codes if code in latest_by_code]
+    return list(latest_by_code.values())
+
+
+def format_timeout_error_message(
+    *,
+    timeout_seconds: int,
+    completed: Sequence[CompletedAnalysisSummary],
+    pending_codes: Sequence[str],
+) -> str:
+    """Build structured ``last_error`` text for RuntimeSchedulerService.status()."""
+    completed_codes = [item.code for item in completed]
+    pending = [str(code).strip() for code in pending_codes if str(code).strip()]
+    parts = [
+        f"runtime scheduled analysis timed out after {timeout_seconds}s",
+        f"completed={len(completed_codes)}",
+        f"pending={len(pending)}",
+    ]
+    if completed_codes:
+        parts.append(f"completed_codes={','.join(completed_codes)}")
+    if pending:
+        parts.append(f"pending_codes={','.join(pending)}")
+    return "; ".join(parts)
+
+
+def build_partial_timeout_report(
+    *,
+    timeout_seconds: int,
+    completed: Sequence[CompletedAnalysisSummary],
+    pending_codes: Sequence[str],
+) -> str:
+    """Build Markdown body for the partial-success timeout notification."""
+    lines = [
+        "## 定时分析超时（部分完成）",
+        "",
+        f"- 硬超时：`{timeout_seconds}` 秒",
+        f"- 已完成：`{len(completed)}` 只",
+        f"- 未完成：`{len(list(pending_codes))}` 只",
+        "",
+    ]
+    if completed:
+        lines.append("### 已落库结果")
+        for item in completed:
+            name = item.name or "-"
+            advice = item.operation_advice or "-"
+            score = item.sentiment_score if item.sentiment_score is not None else "-"
+            lines.append(f"- `{item.code}` {name}｜建议：{advice}｜评分：{score}")
+        lines.append("")
+    if pending_codes:
+        lines.append("### 未完成")
+        lines.append(", ".join(f"`{code}`" for code in pending_codes))
+        lines.append("")
+    lines.append(
+        "_本轮因 runtime scheduler 硬超时被终止；已完成个股结果已写入分析历史，"
+        "未完成部分可稍后手动重跑。_"
+    )
+    return "\n".join(lines)
+
+
+def send_partial_timeout_notification(
+    report: str,
+    *,
+    completed_codes: Sequence[str],
+    no_notify: bool = False,
+) -> bool:
+    """Send ``report`` via NotificationService unless ``no_notify`` is set.
+
+    Returns True when a send was attempted and succeeded.
+
+    Channel/import exceptions are swallowed: this function logs a warning
+    (``Partial timeout notification failed``) and returns False. It does not
+    re-raise, so it cannot keep ``RuntimeSchedulerService`` ``_run_lock``
+    held or flip ``status().running`` — notify runs on a daemon thread after
+    that lock is already released. Structured ``last_error`` is still returned
+    by ``handle_runtime_analysis_timeout`` with ``notify_skipped_reason=send_failed``.
+    """
+    if no_notify:
+        return False
+    if not is_timeout_partial_notify_enabled():
+        return False
+    if not report or not str(report).strip():
+        return False
+
+    try:
+        from src.notification import NotificationService
+
+        service = NotificationService()
+        return bool(
+            service.send(
+                report,
+                email_stock_codes=list(completed_codes) or None,
+                route_type="system_error",
+                severity="warning",
+                dedup_key=f"runtime_timeout_partial:{','.join(completed_codes)}",
+            )
+        )
+    except Exception as exc:  # noqa: BLE001 - never break scheduler on notify failure
+        logger.warning("Partial timeout notification failed: %s", exc)
+        return False
+
+
+def handle_runtime_analysis_timeout(
+    *,
+    timeout_seconds: int,
+    run_started_at: datetime,
+    stock_codes: Optional[Sequence[str]] = None,
+    no_notify: bool = False,
+    config: Optional[Any] = None,
+    db: Optional[Any] = None,
+) -> TimeoutPartialOutcome:
+    """Orchestrate collect → format error → optional partial notify.
+
+    Intended call site: ``RuntimeSchedulerService._run_analysis_with_watchdog``
+    timeout branch, after the worker process tree is terminated.
+    """
+    expected = resolve_expected_stock_codes(stock_codes, config=config)
+    completed = collect_completed_analyses_since(
+        run_started_at=run_started_at,
+        expected_codes=expected,
+        db=db,
+    )
+    completed_code_set = {item.code for item in completed}
+    pending_codes = [code for code in expected if code not in completed_code_set]
+    error_message = format_timeout_error_message(
+        timeout_seconds=timeout_seconds,
+        completed=completed,
+        pending_codes=pending_codes,
+    )
+
+    notified = False
+    notify_skipped_reason: Optional[str] = None
+    if no_notify:
+        notify_skipped_reason = "no_notify"
+    elif not is_timeout_partial_notify_enabled():
+        notify_skipped_reason = "env_disabled"
+    elif not completed:
+        notify_skipped_reason = "no_completed_results"
+    else:
+        report = build_partial_timeout_report(
+            timeout_seconds=timeout_seconds,
+            completed=completed,
+            pending_codes=pending_codes,
+        )
+        notified = send_partial_timeout_notification(
+            report,
+            completed_codes=[item.code for item in completed],
+            no_notify=False,
+        )
+        if not notified:
+            notify_skipped_reason = "send_failed"
+
+    return TimeoutPartialOutcome(
+        completed=list(completed),
+        pending_codes=pending_codes,
+        notified=notified,
+        notify_skipped_reason=notify_skipped_reason,
+        error_message=error_message,
+    )
+
+
+def _normalize_codes(codes: Sequence[Any]) -> List[str]:
+    seen = set()
+    ordered: List[str] = []
+    for raw in codes:
+        code = str(raw or "").strip()
+        if not code or code in seen:
+            continue
+        seen.add(code)
+        ordered.append(code)
+    return ordered

src/services/runtime_scheduler.py

@@ -359,6 +359,107 @@ class RuntimeSchedulerService:
             )
             return DEFAULT_RUNTIME_SCHEDULER_TIMEOUT_SECONDS
 
+    def _build_timeout_last_error(
+        self,
+        *,
+        timeout_seconds: int,
+        run_started_at: datetime,
+        stock_codes: Optional[List[str]],
+    ) -> str:
+        """Collect partial DB results and build a structured timeout error.
+
+        Notify-channel failures are fail-open and do not change this string.
+        Collect/import failures are also fail-open: this may return a
+        ``completed=0`` structured message (indistinguishable from no saved
+        rows) or, if the helper itself raises, the baseline timeout fallback.
+        ``status().last_error`` therefore cannot be used as a collect-failure
+        signal; grep ``Failed to collect completed analyses after timeout``
+        or ``Timeout partial delivery failed open``.
+        """
+        fallback = f"runtime scheduled analysis timed out after {timeout_seconds}s"
+        try:
+            from src.services.analysis_timeout_partial import handle_runtime_analysis_timeout
+
+            config = None
+            try:
+                config = self._config_provider()
+            except Exception as exc:  # noqa: BLE001 - config is optional for diagnostics
+                logger.warning("Timeout partial notify could not load config: %s", exc)
+
+            no_notify = bool(getattr(self._make_schedule_args(), "no_notify", False))
+            outcome = handle_runtime_analysis_timeout(
+                timeout_seconds=timeout_seconds,
+                run_started_at=run_started_at,
+                stock_codes=stock_codes,
+                no_notify=no_notify,
+                config=config,
+            )
+            if outcome.notified:
+                logger.info(
+                    "Timeout partial notification sent: completed=%s pending=%s",
+                    len(outcome.completed),
+                    len(outcome.pending_codes),
+                )
+            elif outcome.notify_skipped_reason:
+                logger.info(
+                    "Timeout partial notification skipped (%s): completed=%s pending=%s",
+                    outcome.notify_skipped_reason,
+                    len(outcome.completed),
+                    len(outcome.pending_codes),
+                )
+            return outcome.error_message or fallback
+        except Exception as exc:  # noqa: BLE001 - never lose the original timeout signal
+            logger.warning("Timeout partial delivery failed open: %s", exc)
+            return fallback
+
+    def _apply_timeout_partial_outcome(self, context: Dict[str, Any]) -> None:
+        """Enrich timeout diagnostics after the run lock is released.
+
+        Runs in a daemon thread so DB/import work cannot delay the next run or
+        keep the watchdog finally block occupied.
+
+        Consistency with ``status()``:
+        - The watchdog first writes a baseline ``timed out after Ns`` string
+          under ``_analysis_process_lock``, then releases ``_run_lock``.
+        - This thread later replaces ``_last_error`` with the structured
+          completed/pending message, still under ``_analysis_process_lock``.
+        - The replace is skipped if ``generation`` no longer matches (a newer
+          run started) or if ``_last_error`` no longer contains
+          ``timed out after`` (another outcome already replaced it).
+        - ``status()`` reads ``_last_error`` without that lock. CPython
+          pointer assignment is atomic, so callers observe either the baseline
+          or the fully replaced string—never a torn mix. They may briefly see
+          the baseline until this thread finishes.
+        - Notify-channel exceptions stay inside this thread and cannot
+          re-acquire ``_run_lock``. Collect/import fail-open is only visible
+          in warning logs, not as a distinct ``last_error`` code.
+        """
+        generation = context.get("generation")
+
+        def _enrich() -> None:
+            try:
+                enriched = self._build_timeout_last_error(
+                    timeout_seconds=int(context["timeout_seconds"]),
+                    run_started_at=context["run_started_at"],
+                    stock_codes=context.get("stock_codes"),
+                )
+            except Exception as exc:  # noqa: BLE001 - baseline timeout error remains
+                logger.warning("Timeout partial enrichment failed open: %s", exc)
+                return
+            with self._analysis_process_lock:
+                if generation is not None and generation != self._analysis_generation:
+                    return
+                current = self._last_error or ""
+                if "timed out after" not in current:
+                    return
+                self._last_error = enriched
+
+        threading.Thread(
+            target=_enrich,
+            daemon=True,
+            name="runtime-timeout-partial",
+        ).start()
+
     def _run_analysis_with_watchdog(
         self,
         stock_codes: Optional[List[str]] = None,
@@ -374,6 +475,7 @@ class RuntimeSchedulerService:
                 generation = self._analysis_generation
 
         result_queue = None
+        timeout_partial_context: Optional[Dict[str, Any]] = None
         try:
             context = multiprocessing.get_context("spawn")
             result_queue = context.Queue()
@@ -383,12 +485,13 @@ class RuntimeSchedulerService:
                 name="runtime-scheduled-analysis",
             )
             timeout = self._analysis_timeout_seconds()
+            run_started_at = datetime.now()
             with self._analysis_process_lock:
                 if generation != self._analysis_generation:
                     return
                 process.start()
                 self._analysis_process = process
-                self._last_run_at = datetime.now().isoformat()
+                self._last_run_at = run_started_at.isoformat()
 
             result = None
             deadline = time.monotonic() + timeout
@@ -411,7 +514,16 @@ class RuntimeSchedulerService:
                 with self._analysis_process_lock:
                     if generation != self._analysis_generation:
                         return
-                    self._last_error = f"runtime scheduled analysis timed out after {timeout}s"
+                    # Baseline error first so status.running can clear quickly.
+                    self._last_error = (
+                        f"runtime scheduled analysis timed out after {timeout}s"
+                    )
+                    timeout_partial_context = {
+                        "timeout_seconds": timeout,
+                        "run_started_at": run_started_at,
+                        "stock_codes": stock_codes,
+                        "generation": generation,
+                    }
                 return
 
             if result is None:
@@ -457,6 +569,8 @@ class RuntimeSchedulerService:
             if result_queue is not None:
                 result_queue.cancel_join_thread()
                 result_queue.close()
+            if timeout_partial_context is not None:
+                self._apply_timeout_partial_outcome(timeout_partial_context)
 
     def _start_analysis_watchdog(
         self,
@@ -682,6 +796,8 @@ class RuntimeSchedulerService:
             "next_run_at": next_run,
             "last_run_at": self._last_run_at,
             "last_success_at": self._last_success_at,
+            # Unlocked read: may briefly show the baseline timeout string
+            # until _apply_timeout_partial_outcome finishes. See that method.
             "last_error": self._last_error,
             "last_skipped_at": self._last_skipped_at,
             "last_skip_reason": self._last_skip_reason,

tests/test_analysis_timeout_partial_notify.py

@@ -0,0 +1,387 @@
+# -*- coding: utf-8 -*-
+"""Unit tests for timeout partial delivery helpers."""
+
+from __future__ import annotations
+
+import os
+import time
+import unittest
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from src.services.analysis_timeout_partial import (
+    CompletedAnalysisSummary,
+    TIMEOUT_PARTIAL_NOTIFY_ENV,
+    build_partial_timeout_report,
+    collect_completed_analyses_since,
+    format_timeout_error_message,
+    handle_runtime_analysis_timeout,
+    is_timeout_partial_notify_enabled,
+    resolve_expected_stock_codes,
+    send_partial_timeout_notification,
+)
+
+
+class TimeoutPartialHelpersTests(unittest.TestCase):
+    def tearDown(self) -> None:
+        os.environ.pop(TIMEOUT_PARTIAL_NOTIFY_ENV, None)
+
+    def test_env_defaults_enabled_and_parses_false(self) -> None:
+        os.environ.pop(TIMEOUT_PARTIAL_NOTIFY_ENV, None)
+        self.assertTrue(is_timeout_partial_notify_enabled())
+        os.environ[TIMEOUT_PARTIAL_NOTIFY_ENV] = "false"
+        self.assertFalse(is_timeout_partial_notify_enabled())
+
+    def test_resolve_expected_stock_codes_prefers_explicit_list(self) -> None:
+        config = SimpleNamespace(stock_list=["000001", "000002"])
+        self.assertEqual(
+            resolve_expected_stock_codes(["600519", "600519", ""], config=config),
+            ["600519"],
+        )
+        self.assertEqual(
+            resolve_expected_stock_codes(None, config=config),
+            ["000001", "000002"],
+        )
+
+    def test_format_timeout_error_message_includes_counts(self) -> None:
+        completed = [
+            CompletedAnalysisSummary(code="600519", name="贵州茅台"),
+            CompletedAnalysisSummary(code="000001", name="平安银行"),
+        ]
+        message = format_timeout_error_message(
+            timeout_seconds=90,
+            completed=completed,
+            pending_codes=["300750"],
+        )
+        self.assertIn("timed out after 90s", message)
+        self.assertIn("completed=2", message)
+        self.assertIn("pending=1", message)
+        self.assertIn("completed_codes=600519,000001", message)
+        self.assertIn("pending_codes=300750", message)
+
+    def test_build_partial_timeout_report_lists_completed_and_pending(self) -> None:
+        report = build_partial_timeout_report(
+            timeout_seconds=120,
+            completed=[
+                CompletedAnalysisSummary(
+                    code="600519",
+                    name="贵州茅台",
+                    operation_advice="买入",
+                    sentiment_score=80,
+                )
+            ],
+            pending_codes=["300750"],
+        )
+        self.assertIn("部分完成", report)
+        self.assertIn("600519", report)
+        self.assertIn("300750", report)
+        self.assertIn("120", report)
+
+    def test_summaries_from_history_rows_keeps_latest_per_code(self) -> None:
+        from src.services.analysis_timeout_partial import _summaries_from_history_rows
+
+        older = SimpleNamespace(
+            id=1,
+            code="600519",
+            name="old",
+            operation_advice="持有",
+            sentiment_score=50,
+            created_at=datetime(2026, 1, 1, 10, 0, 0),
+        )
+        newer = SimpleNamespace(
+            id=2,
+            code="600519",
+            name="new",
+            operation_advice="买入",
+            sentiment_score=70,
+            created_at=datetime(2026, 1, 1, 11, 0, 0),
+        )
+        rows = _summaries_from_history_rows(
+            [newer, older],
+            ["600519", "300750"],
+        )
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0].code, "600519")
+        self.assertEqual(rows[0].name, "new")
+        self.assertEqual(rows[0].history_id, 2)
+
+    def test_timeout_partial_notify_registry_uses_switch_ui_control(self) -> None:
+        from api.v1.schemas.system_config import SystemConfigFieldSchema
+        from src.core.config_registry import get_field_definition
+
+        field = get_field_definition("DSA_TIMEOUT_PARTIAL_NOTIFY")
+        self.assertEqual(field["data_type"], "boolean")
+        self.assertEqual(field["ui_control"], "switch")
+        self.assertEqual(field["default_value"], "true")
+        SystemConfigFieldSchema.model_validate(field)
+
+    def test_collect_completed_analyses_since_fail_open(self) -> None:
+        with patch(
+            "src.services.analysis_timeout_partial._query_history_rows_since",
+            side_effect=RuntimeError("db down"),
+        ):
+            rows = collect_completed_analyses_since(
+                run_started_at=datetime.now(),
+                expected_codes=["600519"],
+            )
+        self.assertEqual(rows, [])
+
+    def test_resolve_storage_module_cleans_partial_import_on_failure(self) -> None:
+        import sys
+
+        from src.services.analysis_timeout_partial import _resolve_storage_module
+
+        original = {
+            name: module
+            for name, module in sys.modules.items()
+            if name == "src.storage" or name.startswith("src.storage.")
+        }
+        for name in list(original):
+            sys.modules.pop(name, None)
+
+        def _boom(name: str, *args, **kwargs):
+            if name == "src.storage":
+                sys.modules["src.storage"] = object()
+                sys.modules["src.storage.models"] = object()
+                raise ImportError("broken storage")
+            raise AssertionError(f"unexpected import: {name}")
+
+        try:
+            with patch("importlib.import_module", side_effect=_boom):
+                with self.assertRaises(RuntimeError) as ctx:
+                    _resolve_storage_module()
+            self.assertIn("storage unavailable", str(ctx.exception))
+            self.assertNotIn("src.storage", sys.modules)
+            self.assertFalse(
+                any(name.startswith("src.storage.") for name in sys.modules)
+            )
+        finally:
+            for name in list(sys.modules):
+                if name == "src.storage" or name.startswith("src.storage."):
+                    sys.modules.pop(name, None)
+            sys.modules.update(original)
+
+    def test_collect_fail_open_when_storage_import_fails(self) -> None:
+        with patch(
+            "src.services.analysis_timeout_partial._resolve_storage_module",
+            side_effect=RuntimeError("storage unavailable for timeout partial collect"),
+        ):
+            rows = collect_completed_analyses_since(
+                run_started_at=datetime.now(),
+                expected_codes=["600519"],
+            )
+        self.assertEqual(rows, [])
+
+    def test_send_respects_no_notify_and_env(self) -> None:
+        self.assertFalse(
+            send_partial_timeout_notification(
+                "hello",
+                completed_codes=["600519"],
+                no_notify=True,
+            )
+        )
+        os.environ[TIMEOUT_PARTIAL_NOTIFY_ENV] = "0"
+        self.assertFalse(
+            send_partial_timeout_notification(
+                "hello",
+                completed_codes=["600519"],
+                no_notify=False,
+            )
+        )
+
+    def test_send_swallows_channel_exception_and_returns_false(self) -> None:
+        notification_mod = MagicMock()
+        notification_mod.NotificationService.return_value.send.side_effect = RuntimeError(
+            "webhook 500"
+        )
+        with patch.dict("sys.modules", {"src.notification": notification_mod}):
+            self.assertFalse(
+                send_partial_timeout_notification(
+                    "hello",
+                    completed_codes=["600519"],
+                    no_notify=False,
+                )
+            )
+        notification_mod.NotificationService.return_value.send.assert_called_once()
+
+    @patch("src.services.analysis_timeout_partial.collect_completed_analyses_since")
+    def test_handle_keeps_error_message_when_channel_raises(
+        self,
+        collect_mock: MagicMock,
+    ) -> None:
+        collect_mock.return_value = [
+            CompletedAnalysisSummary(code="600519", name="贵州茅台"),
+        ]
+        notification_mod = MagicMock()
+        notification_mod.NotificationService.return_value.send.side_effect = RuntimeError(
+            "channel down"
+        )
+
+        with patch.dict("sys.modules", {"src.notification": notification_mod}):
+            outcome = handle_runtime_analysis_timeout(
+                timeout_seconds=60,
+                run_started_at=datetime(2026, 1, 1, 12, 0, 0),
+                stock_codes=["600519", "300750"],
+                no_notify=False,
+            )
+
+        self.assertFalse(outcome.notified)
+        self.assertEqual(outcome.notify_skipped_reason, "send_failed")
+        self.assertIn("completed=1", outcome.error_message)
+        self.assertIn("pending=1", outcome.error_message)
+
+    @patch("src.services.analysis_timeout_partial.send_partial_timeout_notification")
+    @patch("src.services.analysis_timeout_partial.collect_completed_analyses_since")
+    def test_handle_runtime_analysis_timeout_notifies_when_completed(
+        self,
+        collect_mock: MagicMock,
+        send_mock: MagicMock,
+    ) -> None:
+        collect_mock.return_value = [
+            CompletedAnalysisSummary(code="600519", name="贵州茅台"),
+        ]
+        send_mock.return_value = True
+
+        outcome = handle_runtime_analysis_timeout(
+            timeout_seconds=60,
+            run_started_at=datetime(2026, 1, 1, 12, 0, 0),
+            stock_codes=["600519", "300750"],
+            no_notify=False,
+            config=SimpleNamespace(stock_list=[]),
+            db=MagicMock(),
+        )
+
+        self.assertTrue(outcome.notified)
+        self.assertEqual(outcome.pending_codes, ["300750"])
+        self.assertIn("completed=1", outcome.error_message)
+        self.assertIn("pending=1", outcome.error_message)
+        send_mock.assert_called_once()
+
+    @patch("src.services.analysis_timeout_partial.send_partial_timeout_notification")
+    @patch("src.services.analysis_timeout_partial.collect_completed_analyses_since")
+    def test_handle_skips_notify_without_completed(
+        self,
+        collect_mock: MagicMock,
+        send_mock: MagicMock,
+    ) -> None:
+        collect_mock.return_value = []
+        outcome = handle_runtime_analysis_timeout(
+            timeout_seconds=60,
+            run_started_at=datetime.now(),
+            stock_codes=["600519"],
+            no_notify=False,
+        )
+        self.assertFalse(outcome.notified)
+        self.assertEqual(outcome.notify_skipped_reason, "no_completed_results")
+        send_mock.assert_not_called()
+
+
+class RuntimeSchedulerTimeoutPartialIntegrationTests(unittest.TestCase):
+    def test_build_timeout_last_error_uses_partial_helper(self) -> None:
+        from src.services.runtime_scheduler import RuntimeSchedulerService
+
+        config = SimpleNamespace(stock_list=["600519", "300750"])
+        service = RuntimeSchedulerService(config_provider=lambda: config)
+        completed = [CompletedAnalysisSummary(code="600519", name="贵州茅台")]
+
+        with patch(
+            "src.services.analysis_timeout_partial.collect_completed_analyses_since",
+            return_value=completed,
+        ), patch(
+            "src.services.analysis_timeout_partial.send_partial_timeout_notification",
+            return_value=True,
+        ):
+            message = service._build_timeout_last_error(
+                timeout_seconds=90,
+                run_started_at=datetime(2026, 1, 1, 12, 0, 0),
+                stock_codes=["600519", "300750"],
+            )
+
+        self.assertIn("timed out after 90s", message)
+        self.assertIn("completed=1", message)
+        self.assertIn("pending=1", message)
+        self.assertIn("completed_codes=600519", message)
+        self.assertIn("pending_codes=300750", message)
+
+    def test_timeout_branch_sets_structured_last_error(self) -> None:
+        from src.services.runtime_scheduler import RuntimeSchedulerService
+
+        config = SimpleNamespace(stock_list=["600519", "300750"])
+        service = RuntimeSchedulerService(config_provider=lambda: config)
+        service._analysis_timeout_seconds = lambda: 1
+
+        class _AliveProcess:
+            def __init__(self, *args, **kwargs):
+                self.pid = 424242
+                self.exitcode = None
+                self._alive = True
+
+            def start(self) -> None:
+                return None
+
+            def is_alive(self) -> bool:
+                return self._alive
+
+            def join(self, timeout=None) -> None:
+                return None
+
+            def terminate(self) -> None:
+                self._alive = False
+
+            def kill(self) -> None:
+                self._alive = False
+
+        class _EmptyQueue:
+            def get(self, timeout=None):
+                from queue import Empty
+
+                raise Empty
+
+            def cancel_join_thread(self) -> None:
+                return None
+
+            def close(self) -> None:
+                return None
+
+        fake_context = SimpleNamespace(
+            Queue=lambda: _EmptyQueue(),
+            Process=_AliveProcess,
+        )
+
+        with patch(
+            "src.services.runtime_scheduler.multiprocessing.get_context",
+            return_value=fake_context,
+        ), patch(
+            "src.services.runtime_scheduler._terminate_analysis_process_tree",
+        ), patch.object(
+            service,
+            "_build_timeout_last_error",
+            return_value=(
+                "runtime scheduled analysis timed out after 1s; "
+                "completed=1; pending=1; completed_codes=600519; pending_codes=300750"
+            ),
+        ):
+            # Watchdog finally always releases the shared lock.
+            self.assertTrue(service._run_lock.acquire(blocking=False))
+            service._run_analysis_with_watchdog(
+                ["600519", "300750"],
+                lock_held=True,
+            )
+
+        deadline = time.time() + 2
+        last_error = service.status()["last_error"]
+        while (
+            last_error is None or "completed=1" not in last_error
+        ) and time.time() < deadline:
+            time.sleep(0.02)
+            last_error = service.status()["last_error"]
+
+        self.assertIsNotNone(last_error)
+        self.assertIn("timed out after 1s", last_error)
+        self.assertIn("completed=1", last_error)
+        self.assertIn("pending_codes=300750", last_error)
+
+
+if __name__ == "__main__":
+    unittest.main()