feat(observability): worker 心跳告警闭环 + topic last_error 面板

apps/api/routers/system.py

@@ -2,6 +2,9 @@
 @author Color2333
 """
 
+import time
+from pathlib import Path
+
 from fastapi import APIRouter, Query
 
 from apps.api.deps import iso_dt, settings
@@ -15,6 +18,28 @@
 
 router = APIRouter()
 
+# worker 心跳文件（pm_data 共享卷，worker 写、backend 读）
+_WORKER_HEARTBEAT_FILE = Path("/app/data/worker_heartbeat.json")
+_HEARTBEAT_STALE_SECONDS = 1200  # 与 worker healthcheck 一致
+
+
+def _read_worker_heartbeat() -> dict:
+    """读共享卷 worker 心跳，返回 {ts, error, age_seconds, is_stale}；文件缺失返回 None。"""
+    import json
+
+    try:
+        data = json.loads(_WORKER_HEARTBEAT_FILE.read_text())
+        ts = float(data.get("ts", 0))
+        age = time.time() - ts
+        return {
+            "ts": ts,
+            "error": data.get("error"),
+            "age_seconds": int(age),
+            "is_stale": age > _HEARTBEAT_STALE_SECONDS,
+        }
+    except (OSError, ValueError, TypeError):
+        return None
+
 
 @router.get("/health")
 def health() -> dict:
@@ -28,6 +53,16 @@ def health() -> dict:
     }
 
 
+@router.get("/system/worker")
+def worker_status() -> dict:
+    """Worker 心跳状态（可观测性：供 Operations 面板 + 告警自检查询）"""
+    hb = _read_worker_heartbeat()
+    return {
+        "heartbeat": hb,
+        "stale_threshold": _HEARTBEAT_STALE_SECONDS,
+    }
+
+
 @router.get("/system/status")
 def system_status() -> dict:
     with session_scope() as session:
@@ -36,6 +71,8 @@ def system_status() -> dict:
         papers_total = PaperRepository(session).count_all()
         runs = PipelineRunRepository(session).list_latest(limit=50)
         failed = [r for r in runs if r.status.value == "failed"]
+        # 可观测性：附带 worker 心跳摘要 + 主题抓取错误计数
+        errored_topics = [t for t in topics if t.last_error]
         return {
             "health": health(),
             "counts": {
@@ -45,6 +82,8 @@ def system_status() -> dict:
                 "runs_latest_50": len(runs),
                 "failed_runs_latest_50": len(failed),
             },
+            "worker_heartbeat": _read_worker_heartbeat(),
+            "topic_errors": len(errored_topics),
             "latest_run": (
                 {
                     "pipeline_name": runs[0].pipeline_name,

apps/api/routers/topics.py

@@ -36,8 +36,12 @@ def _topic_dict(t, session=None) -> dict:
         "enable_date_filter": getattr(t, "enable_date_filter", False),
         "date_filter_days": getattr(t, "date_filter_days", 7),
         "paper_count": 0,
-        "last_run_at": None,
+        # last_run_at/last_error 读 TopicSubscription 真实抓取状态（PR1 存的），
+        # 此前被 CollectionAction.created_at 覆盖掩盖了抓取失败
+        "last_run_at": t.last_run_at.isoformat() if t.last_run_at else None,
+        "last_error": t.last_error,
         "last_run_count": None,
+        "last_action_at": None,  # 最近一次收集行动（与 last_run_at 区分）
     }
     if session is not None:
         from sqlalchemy import func, select
@@ -49,29 +53,32 @@ def _topic_dict(t, session=None) -> dict:
             select(func.count()).select_from(PaperTopic).where(PaperTopic.topic_id == t.id)
         )
         d["paper_count"] = cnt or 0
-        # 最近一次行动
+        # 最近一次收集行动（单独字段，不再覆盖 last_run_at）
         last_action = session.execute(
             select(CollectionAction)
             .where(CollectionAction.topic_id == t.id)
             .order_by(CollectionAction.created_at.desc())
             .limit(1)
         ).scalar_one_or_none()
         if last_action:
-            d["last_run_at"] = (
+            d["last_action_at"] = (
                 last_action.created_at.isoformat() if last_action.created_at else None
             )
             d["last_run_count"] = last_action.paper_count
     return d
 
 
 @router.get("/topics")
-def list_topics(enabled_only: bool = False) -> dict:
+def list_topics(enabled_only: bool = False, failed: bool = False) -> dict:
     from sqlalchemy import func, select
 
     from packages.storage.models import CollectionAction, PaperTopic
 
     with session_scope() as session:
         topics = TopicRepository(session).list_topics(enabled_only=enabled_only)
+        # failed=true：只返回最近抓取出错的 topic（供可观测性面板）
+        if failed:
+            topics = [t for t in topics if t.last_error]
         if not topics:
             return {"items": []}
         topic_ids = [t.id for t in topics]
@@ -110,12 +117,16 @@ def list_topics(enabled_only: bool = False) -> dict:
                 "enabled": t.enabled,
                 "created_at": t.created_at.isoformat() if t.created_at else None,
                 "paper_count": paper_counts.get(t.id, 0),
-                "last_run_at": None,
+                # last_run_at/last_error 读 TopicSubscription 真实抓取状态（PR1），
+                # 此前被 CollectionAction.created_at 覆盖，抓取失败被掩盖
+                "last_run_at": t.last_run_at.isoformat() if t.last_run_at else None,
+                "last_error": t.last_error,
+                "last_action_at": None,
                 "last_run_count": None,
             }
             last_action = latest_actions.get(t.id)
             if last_action:
-                d["last_run_at"] = (
+                d["last_action_at"] = (
                     last_action.created_at.isoformat() if last_action.created_at else None
                 )
                 d["last_run_count"] = last_action.paper_count

apps/worker/main.py

@@ -36,9 +36,13 @@
 setup_logging()
 logger = logging.getLogger(__name__)
 
-_HEALTH_FILE = Path("/tmp/worker_heartbeat")
+# 心跳改写共享卷 pm_data（/app/data），backend 也能读同一文件暴露 worker 状态。
+# 此前写 /tmp（容器内，后端读不到），可观测性端点无法查询 worker 健康。
+_HEALTH_FILE = Path("/app/data/worker_heartbeat.json")
 # 心跳健康判定：最近一次心跳距现在超过此秒数视为不健康（捕获 worker 卡死/全部任务失败）
 _HEARTBEAT_STALE_SECONDS = 1200  # 20 分钟（cron job 最小间隔 30min，留足缓冲）
+# 告警去重：同一告警类型在此秒数内不重复发邮件（防刷屏）
+_ALERT_DEDUP_SECONDS = 3600  # 1 小时
 
 
 def _write_heartbeat(error: str | None = None) -> None:
@@ -56,6 +60,92 @@ def _write_heartbeat(error: str | None = None) -> None:
         )
 
 
+def _read_heartbeat() -> dict | None:
+    """读共享卷心跳文件，供自检告警 + 端点查询。文件缺失/损坏返回 None。"""
+    import json
+
+    try:
+        return json.loads(_HEALTH_FILE.read_text())
+    except (OSError, ValueError, TypeError):
+        return None
+
+
+# 告警去重状态：{alert_key: last_alert_ts}（进程内，重启后重置——可接受）
+_last_alerts: dict[str, float] = {}
+
+
+def _send_alert(subject: str, html: str, alert_key: str) -> None:
+    """发告警邮件给 notify_default_to，带 1h 去重。SMTP 未配置时静默跳过。"""
+    now = time.time()
+    last = _last_alerts.get(alert_key, 0)
+    if now - last < _ALERT_DEDUP_SECONDS:
+        logger.debug(
+            "告警 %s 去重中（距上次 %.0fs < %ds），跳过",
+            alert_key,
+            now - last,
+            _ALERT_DEDUP_SECONDS,
+        )
+        return
+    from packages.config import get_settings
+    from packages.integrations.notifier import NotificationService
+
+    recipient = get_settings().notify_default_to
+    if not recipient:
+        logger.debug("notify_default_to 未配置，跳过告警 %s", alert_key)
+        return
+    ok = NotificationService().send_email_html(recipient, subject, html)
+    if ok:
+        _last_alerts[alert_key] = now
+        logger.info("告警邮件已发送: %s -> %s", alert_key, recipient)
+    else:
+        logger.warning("告警邮件发送失败（SMTP 未配置或出错）: %s", alert_key)
+
+
+def heartbeat_alert_job() -> None:
+    """每 10min 自检：心跳过期 + 主题抓取错误 → 发告警邮件（可观测性闭环）。
+
+    心跳过期说明 worker 卡死或全部 job 失败；主题 last_error 说明某主题抓取失败。
+    两者此前只记日志无人知，现在发邮件给 notify_default_to（带 1h 去重防刷屏）。
+    """
+    import html as html_lib
+
+    # 1. 心跳过期检查
+    hb = _read_heartbeat()
+    if hb is None or (time.time() - float(hb.get("ts", 0))) > _HEARTBEAT_STALE_SECONDS:
+        age = (
+            "未知（文件缺失/损坏）"
+            if hb is None
+            else f"{int(time.time() - float(hb.get('ts', 0)))}s"
+        )
+        body = (
+            f"<h2>⚠️ Worker 心跳过期</h2>"
+            f"<p>心跳距今 <b>{age}</b>（阈值 {_HEARTBEAT_STALE_SECONDS}s）。</p>"
+            f"<p>可能原因：worker 卡死、全部 job 失败、或容器异常。</p>"
+            f"<p>最近错误：{(hb or {}).get('error') or 'N/A'}</p>"
+            f"<p>请检查 worker 容器状态与日志。</p>"
+        )
+        _send_alert("[PaperMind] Worker 心跳过期告警", body, "heartbeat_stale")
+    # 2. 主题抓取错误检查
+    try:
+        with session_scope() as session:
+            topics = TopicRepository(session).list_topics(enabled_only=True)
+            errored = [t for t in topics if t.last_error]
+            if errored:
+                rows = "".join(
+                    f"<tr><td>{html_lib.escape(t.name)}</td>"
+                    f"<td>{html_lib.escape((t.last_error or '')[:200])}</td></tr>"
+                    for t in errored
+                )
+                body = (
+                    f"<h2>⚠️ 主题抓取错误（{len(errored)} 个）</h2>"
+                    f"<table border='1' cellpadding='6' style='border-collapse:collapse'>"
+                    f"<tr><th>主题</th><th>最近错误</th></tr>{rows}</table>"
+                )
+                _send_alert(f"[PaperMind] {len(errored)} 个主题抓取失败", body, "topic_errors")
+    except Exception:
+        logger.exception("heartbeat_alert_job 检查主题错误失败")
+
+
 def _update_topic_run_status(topic_id: str, *, error: str | None) -> None:
     """记录主题抓取的最近运行时间与错误（Critical #4：失败可查可补抓）。
 
@@ -304,6 +394,15 @@ def run_worker() -> None:
     )
     logger.info("✅ 已添加：每周图谱维护任务（UTC 周日 22:00）")
 
+    # 可观测性：心跳过期 + 主题抓取错误告警（每 10min 自检发邮件）
+    scheduler.add_job(
+        heartbeat_alert_job,
+        trigger=CronTrigger(minute="*/10"),
+        id="heartbeat_alert",
+        **_job_kwargs,
+    )
+    logger.info("✅ 已添加：心跳告警自检任务（每 10min）")
+
     # 优雅关闭（High 3f：等待进行中任务跑完，避免已下载 PDF 未 set_pdf_path
     # 的中间态丢失；wait=True + 60s 超时兜底）
     def _graceful_stop(*_: object) -> None: