feat(service-monitor): 定时任务执行状态展示 + 立即执行按钮

- schedule.html: 表格新增「上次执行」列（时间+成功/失败/运行中徽章+报告链接）
- schedule.html: 操作列新增「执行」按钮，点击后后台线程执行、3秒轮询自动刷新
- schedule_service.py: 抽 _run_job_body 共享执行体、新增 run_now() 手动触发
- schedule_service.py: create_schedule 初始化 current_status:idle
- schedule_service.py: 异常路径 last_run_status 与 current_status 统一为 error
- routes.py: page_schedule 注入 run_badge/run_label；新增 POST .../schedules/<id>/run
- 响应式适配：平板端隐藏套件/上次执行/下次执行列
Co-Authored-By: Claude <noreply@anthropic.com>

skill/code/web/service_monitor/routes.py

@@ -113,9 +113,21 @@ def page_schedule():
         return redirect(url_for('auth.login'))
     user = _current_user()
     schedules = schedule_service.list_schedules()
-    # 为每个 schedule 生成自然语言描述
+    # 为每个 schedule 生成自然语言描述 + 执行状态徽章
+    _STATUS_MAP = {
+        "running": ("badge-running", "🔄 执行中"),
+        "success": ("badge-success", "✅ 成功"),
+        "failed":  ("badge-failed",  "❌ 失败"),
+        "error":   ("badge-error",   "⚠️ 异常"),
+        "idle":    ("badge-idle",    "⏸ 待执行"),
+    }
     for s in schedules:
         s['description'] = schedule_service._describe_schedule(s)
+        # current_status 优先，其次 last_run_status，默认 idle
+        status_key = s.get("current_status") or s.get("last_run_status") or "idle"
+        badge_cls, label = _STATUS_MAP.get(status_key, ("badge-idle", "⏸ 待执行"))
+        s['run_badge'] = badge_cls
+        s['run_label'] = label
     targets = target_service.list_targets(role=user.get('role', ''))
     return render_template(
         'service_monitor/schedule.html',
@@ -437,6 +449,19 @@ def api_toggle_schedule(schedule_id):
         return jsonify({"success": False, "error": {"code": 400, "message": str(e)}}), 400
 
 
+@bp.route('/api/service-monitor/schedules/<schedule_id>/run', methods=['POST'])
+def api_run_schedule(schedule_id):
+    """手动立即执行一次定时任务（后台线程，立即返回）。"""
+    guard = _require_admin_json()
+    if guard:
+        return guard
+    try:
+        result = schedule_service.run_now(schedule_id)
+        return jsonify({"success": True, **result})
+    except ValueError as e:
+        return jsonify({"success": False, "error": {"code": 400, "message": str(e)}}), 400
+
+
 # ============================================================
 # API：通知配置（管理员）
 # ============================================================

skill/code/web/service_monitor/services/schedule_service.py

@@ -239,6 +239,7 @@ def create_schedule(data: dict, created_by: str) -> dict:
         "start_date": start_date,
         "end_date": end_date,
         "enabled": True,
+        "current_status": "idle",  # 初始化执行状态
         "last_run_at": None,
         "last_run_status": None,
         "last_report_id": None,
@@ -458,13 +459,13 @@ def _remove_job(schedule_id: str) -> None:
         pass  # job 可能不存在
 
 
-def _execute_scheduled_job(schedule_id: str) -> None:
-    """定时任务执行入口（在 APScheduler 线程中调用）。"""
-    logger.info("定时任务触发: %s", schedule_id)
+def _run_job_body(schedule_id: str) -> None:
+    """定时任务执行体（不含 enabled 检查，供 APScheduler 和 run_now 共用）。"""
+    logger.info("定时任务执行开始: %s", schedule_id)
 
     sched = get_schedule(schedule_id)
-    if not sched or not sched.get("enabled"):
-        logger.warning("定时任务不存在或已禁用: %s", schedule_id)
+    if not sched:
+        logger.warning("定时任务不存在: %s", schedule_id)
         return
 
     # 标记执行中
@@ -491,7 +492,40 @@ def _execute_scheduled_job(schedule_id: str) -> None:
             report_id="",
             run_at=datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
         )
-        _update_current_status(schedule_id, "failed")
+        _update_current_status(schedule_id, "error")  # 与 last_run_status 一致
+
+
+def _execute_scheduled_job(schedule_id: str) -> None:
+    """定时任务执行入口（APScheduler 线程中调用），含 enabled 检查。"""
+    logger.info("定时任务触发: %s", schedule_id)
+
+    sched = get_schedule(schedule_id)
+    if not sched or not sched.get("enabled"):
+        logger.warning("定时任务不存在或已禁用: %s", schedule_id)
+        return
+
+    _run_job_body(schedule_id)
+
+
+def run_now(schedule_id: str) -> dict:
+    """手动立即执行一次定时任务（后台线程，立即返回，不阻塞 HTTP 请求）。
+
+    巡检可能耗时数分钟（单模块 90s 超时），故用后台线程执行。
+    不检查 enabled —— 手动执行是显式动作，禁用状态也应能跑。
+    页面通过 current_status 轮询感知 running → 完成。
+    """
+    sched = get_schedule(schedule_id)
+    if not sched:
+        raise ValueError("定时任务不存在")
+
+    # 避免重复触发：已在执行中则跳过
+    if sched.get("current_status") == "running":
+        raise ValueError("该任务正在执行中，请稍后刷新查看")
+
+    t = threading.Thread(target=_run_job_body, args=(schedule_id,), daemon=True)
+    t.start()
+    logger.info("手动触发定时任务: %s", schedule_id)
+    return {"schedule_id": schedule_id, "triggered": True}
 
 
 def reload_job(schedule_id: str) -> None: