fix(scheduler): 定时任务立即执行未生成执行记录 + 取消即停中断后台线程

5.44 前端「定时任务」页点击执行后，执行中心始终不生成新执行记录。
根因：自动调度触发后 task_id 卡住 _running_tasks，所有手动点击被静默吞掉
（return None），返回 200 但无新 execution。取消执行也不中断 Playwright 线程，
run_execution 阻塞在 worker_thread.join() 上，全局锁数小时不释放。

1. **scheduler_service.py** — run_task_once(task_id, manual=False) 新增
   manual 参数，手动触发绕过任务级 _running_tasks 守卫（仍受全局锁约束）

2. **scheduled_tasks.py** — POST /{task_id}/run 改为同步预检：
   全局锁占用 → 返回 status:skipped + running_execution_id；
   空闲 → manual=True 触发，返回 status:triggered

3. **execution_service.py** — 取消信号机制：
   request_cancel / is_cancel_requested / clear_cancel_requested 线程安全函数；
   cancel_execution 置 cancelled 后调用 request_cancel；
   run_all_cases_sync 用例循环开头检查并中断剩余用例；
   实时进度回调 _persist_case_result 按用例 commit

4. **playwright_executor.py** — 步骤边界检查取消请求：
   execute_step 重试循环开头查 is_cancel_requested → 置 _cancel_requested
   → break 退出重试；最终状态设为 skipped

5.44+5.202 已部署（deploy_scheduled_fix.py），容器 healthy
Co-Authored-By: Claude <noreply@anthropic.com>

HANDOFF_UI自动化.md

@@ -2,8 +2,8 @@
 
 > **生成时间**: 2026-08-19
 > **当前分支**: `platform-auto-test`
-> **最近提交**: `ac823858` feat(executor): 系统配置切换被测系统后用例执行自动跟随新地址
-> **状态**: 🟢 **会话29：修复「执行中无法取消」（MySQL 1205 行锁）+ SUT URL 校验加固，已实现并部署 5.60 生产——`run_execution` 置 running 后立即 commit 释放 InnoDB 行锁（取消接口由 ~50s 1205 超时降至 91ms 返回）；实机发现收尾状态判定命中 SQLAlchemy 身份映射（expire_on_commit=False 实例不刷新）导致 cancelled 被覆盖为 completed，改为列查询 + FOR UPDATE 修复；新增 `_validate_sut_url` 拒绝非法地址（如 192.168.5.4202）。后端 186 测试全过、实机取消演练通过、恢复用例跑通 100%**
+> **最近提交**: `1a3d52ad` fix(executor): 修复执行中无法取消（MySQL 1205 行锁）并加固 SUT URL 校验
+> **状态**: 🟢 **会话31：修复定时任务「立即执行」点击不生成执行记录 + 取消即停实时中断后台线程 + 实时进度按用例回调。** 已修复 4 个后端文件，已部署到 5.44+5.202 生产服务器，容器 healthy 且调度引擎正常；待用户在 5.44 前端点击执行验证。
 
 ---
 
@@ -248,11 +248,88 @@
 
 ---
 
+### ✅ 数据统计模块新增用例探索 + 本地验收 + `/sut-explore` Skill 沉淀（会话 30，2026-08-19）
+
+**背景与目标**：用户要求"访问被测系统，快速生成当前未包含的可执行 UI 自动化用例"。经 DB 盘点，数据统计模块（`module_parent_shujutongji`）顶层 0 条直接用例，且「会议概览/会议室概览/释放/满意度/参会人次/MCU/云视频」等关键词在用例库中均为 0 覆盖——正是本次新增用例的测试项。
+
+**发现过程**（Chrome DevTools MCP 真机探测）：
+- 登录流程有坑：登录后弹「同意《统一登录服务协议》」对话框，需点「确定」，否则停留登录页
+- 首页「常用功能」区域一次性列出所有顶层功能入口（新建会议/预定数据/使用数据/管理看板/会议概览/会议室概览等），是发现未覆盖功能的最佳入口
+- 数据统计页 9 个 Element UI 统计小部件（预定会议数据/使用会议室数据/用户签到数据/会议释放数据/满意度评价统计/用户参会人次统计/MCU视频会议使用时间统计/会议通知发送统计/云视频会议场次统计）
+- 直达 URL：`https://192.168.5.44/#/meetingV3?meetingV3=%2FmeetingV3%2F%23%2FStatisticsModule`（双重 hash 编码，从地址栏复制，勿手拼）
+
+**产出**：
+
+| 项目 | 说明 |
+|------|------|
+| 10 个新用例 | `generate_statistics_cases.py`（幂等生成，module_id=`module_parent_shujutongji`，case_type=ui，P0/P1） |
+| config | `update_statistics_cases_config.py`：全部 `{"auto_login": true, "headless": true}` |
+| 首步直达 | `fix_statistics_cases_steps.py`：首步 `click .block` → `navigate` 直达 StatisticsModule URL（headless 下微前端 `.block` 菜单不稳定） |
+| 页面注册 | `app/data/page_url_mapping.json` 新增 `statistics` 条目（url_patterns="StatisticsModule"、fingerprint="预定会议数据"、match_rules/skip_steps） |
+
+**4 轮迭代修复（本地执行引擎，每轮全量 10 用例 ~400s）**：
+
+| 轮次 | 结果 | 失败根因 |
+|------|------|----------|
+| V1 `exec_af7cebf3` | 1 fail | 首步 `.block:has-text("预定数据")` 点击 5s 超时（headless 微前端菜单不稳定）→ 改 navigate 直达 |
+| V2 `exec_ade9d259` | **4 pass / 6 fail** | 6 个断言 `input[value="日"]` 等失败 → Element UI `el-select` 渲染为 `<input readonly>`，query_selector 取不到 value |
+| V3 `exec_60011036` | **6 pass / 4 fail** | step5 查询按钮是纯图标按钮（无文本）→ 改断言 `.el-icon-download`；2 个时间维度下拉框选择器漏改 |
+| V4 `exec_e2a2a972` | **✅ 10 pass / 0 fail** | — |
+
+**关键踩坑（已沉淀为 Skill + 记忆，勿重复踩）**：
+1. **Element UI 下拉框** → 断言用 `input[readonly][value="日"]` 优先，兜底 `input[readonly][placeholder*="选择"]`
+2. **图标按钮无文本** → 查询/导出按钮是 `<i class="el-icon-download">`，用 `.el-icon-download` 或 `button.el-button--primary`
+3. **首步一律 navigate 直达**，不 click 菜单（headless 下 `.block` 不稳定）
+4. **双重 hash URL**（`%2F`=`/`、`%23`=`#`）→ 浏览器地址栏复制，勿手拼（拼错导航到 `/Invalid`）
+5. **Git Bash 下 curl 发中文 JSON body → 400 parse error** → 改用 `python requests.post(json=body)`
+6. **config 必配 `auto_login: true`**，否则 navigate 后跳登录页
+7. **page_url_mapping.json 单例不热加载** → 新增条目需重启 uvicorn（当前用例直接 navigate，此项非硬依赖）
+
+**Skill 沉淀**：新建 `.claude/skills/sut-explore/SKILL.md` —— `/sut-explore [功能关键词]` 一键执行「盘点覆盖率 → 浏览器探测 → 生成用例 → 创建执行 → 迭代修复」全流程；同时新增记忆 `sut-explore-pitfalls.md` 记录完整踩坑清单（已入 MEMORY.md 索引）。
+
+**遗留/待办**：
+- [ ] **同步 5.60 生产**：10 个新用例（MySQL test_cases）+ `page_url_mapping.json` 文件 → 用户确认的工作流是「本地验证可执行后统一补充 5.60」
+- [ ] 可继续用 `/sut-explore` 补充其他未覆盖功能（会议概览、会议室概览、Welink/SMC 统计等）
+
+---
+
+### ✅ 定时任务「立即执行」不生成执行记录修复 + 取消即停 + 实时进度回调（会话 31，2026-08-19）
+
+**用户反馈**：5.44 前端「UI自动化定时任务」页点击「执行」后，执行中心始终不生成新的执行任务。
+
+**根因（5.44 实机日志确认，双守卫问题）**：
+1. 容器 `2026-08-19T21:05:30Z` 部署重启后，调度引擎 `21:05:51` 自动触发了任务 `scheduled_e530e266`，创建 `exec_1371af03`（304 用例）进入 `run_execution`；
+2. 用户 `21:06:51`~`21:07:20` 连续点击「执行」，每次都命中 `scheduler_service.py` **任务级守卫**：`if task_id in _running_tasks: return None`——task_id 在 `run_execution` 整体返回前一直占用集合（后台 Playwright 线程跑完 304 用例才 `finally` discard）；
+3. 即使 `21:06:44` 用户取消了那次执行（status=cancelled），**取消不中断 Playwright 线程**，`run_execution` 仍阻塞在 `worker_thread.join()`，`_running_tasks` 与全局锁一直不释放 → 所有手动点击 200 OK 但被静默吞掉，执行中心永远无新记录。
+
+**修复（4 个后端文件）**：
+
+| 文件 | 改动 |
+|------|------|
+| `services/scheduler_service.py` | `run_task_once(task_id, manual=False)` 新增 `manual` 参数——**仅自动调度走任务级去重**；手动触发绕过 `_running_tasks` 守卫（用户意图就是"再跑一次"），仍受全局执行锁约束 |
+| `routers/scheduled_tasks.py` | `POST /{task_id}/run` 改为**同步预检**：全局锁占用 → 返回 `status:"skipped"` + `running_execution_id`；空闲 → `asyncio.create_task(run_task_once(task_id, manual=True))` 返回 `status:"triggered"`（前端可区分真触发/被跳过，不再无脑 200） |
+| `services/execution_service.py` | **取消信号机制**：模块级 `_cancel_requested: set` + 线程锁，`request_cancel/is_cancel_requested/clear_cancel_requested`；`cancel_execution` 置 cancelled 后调用 `request_cancel`；`run_all_cases_sync` 用例循环开头检查 `is_cancel_requested` → 中断剩余用例；`run_execution` 取锁后 `clear_cancel_requested`（防重跑残留）；执行器绑定 `_execution_id` |
+| `executors/playwright_executor.py` | 工作线程**步骤边界检查**：`execute_step` 重试循环开头查 `is_cancel_requested(self._execution_id)` → 置 `_cancel_requested` → `break` 退出重试；最终状态设置处若已取消 → `status=skipped`、"⏹ 检测到取消请求，中断执行" |
+
+**核心语义变化**：区分「自动调度触发」与「手动立即执行」——手动路径绕过任务级去重、给出可识别反馈；自动路径去重保护不变（避免同任务并发重复触发）。全局执行锁始终保证 UI 执行串行。
+
+**部署（5.44 + 5.202，脚本 `backend/scripts/deploy_scheduled_fix.py`）**：
+- SFTP 上传 4 个后端文件 → `docker restart plat-auto-test-app` → 容器 healthy；容器内 `ast.parse` 验证 4 文件均 OK
+- ⚠️ **容器内路径踩坑**：远程容器后端路径是 `/app/app/{rel}`（volume 挂载 `backend:/app`，`backend/app` 是 Python 包），不是 `/app/backend/app/{rel}`（首版验证脚本全部 MISSING，修正后 OK）
+
+**待验证**：
+- [ ] 用户在 5.44 前端点击「立即执行」→ 应生成新执行记录（此前只会假 200）
+- [ ] 执行中取消 → 工作线程应在用例/步骤边界尽快退出，`run_execution` 返回、锁释放，可立即再点执行
+- [ ] 前端暂未消费 `status` 字段（`ScheduledTasks.vue` 仍展示 `res.message`）；后端语义已正确，前端提示为可选增强
+
+---
+
 ## 📊 会话历史时间线
 
 | 会话 | 日期 | 主题 | 结果 |
 |------|------|------|------|
-| 29 | 08-19 | **修复「执行中无法取消」（MySQL 1205 行锁）+ SUT URL 校验加固**：置 running 后立即 commit 释放行锁；实机发现身份映射过期实例覆盖 cancelled 为 completed（改列查询 + FOR UPDATE）；`_validate_sut_url` 拒绝非法地址（192.168.5.4202）；结果保持 skipped | ✅ 后端 186 测试全过、已部署 5.60、实机取消演练通过（91ms + cancelled 保持）、恢复用例跑通 100% |
+| 31 | 08-19 | **定时任务「立即执行」不生成执行记录修复 + 取消即停**：双守卫根因（任务级 `_running_tasks` 卡住 + 取消不中断线程）→ `manual` 参数区分触发来源、`run` 接口预检返回 status、`request_cancel` 信号让工作线程用例/步骤边界退出 | ✅ 4 文件已修复并部署 5.44+5.202，容器 healthy；待前端点击执行验证 |
+| 30 | 08-19 | **数据统计模块新增用例探索 + 本地验收 + `/sut-explore` Skill 沉淀**：10 个新用例，V4 全量 10/10 通过；沉淀踩坑记录到记忆 + 可复用的探测 Skill | ✅ 本地验收通过，待同步 5.60 生产 | **修复「执行中无法取消」（MySQL 1205 行锁）+ SUT URL 校验加固**：置 running 后立即 commit 释放行锁；实机发现身份映射过期实例覆盖 cancelled 为 completed（改列查询 + FOR UPDATE）；`_validate_sut_url` 拒绝非法地址（192.168.5.4202）；结果保持 skipped | ✅ 后端 186 测试全过、已部署 5.60、实机取消演练通过（91ms + cancelled 保持）、恢复用例跑通 100% |
 | 28 | 08-19 | **系统配置切换被测系统后用例自动跟随（方案A·执行期地址重写）**：`_rebase_sut_url` 导航/接口主机重写（已知 SUT 主机→当前配置地址），130 条存量硬编码用例自动跟随、不动数据 | ✅ 后端 182 测试全过、已部署 5.60 生产 |
 | 27 | 08-19 | **UI自动化定时任务模块 + 系统配置被测系统URL**：调度引擎（30s 轮询 + 到期异步触发 + 冲突跳过）+ 定时任务 CRUD/启停/立即执行/运行记录 + 报告按模块分组并展示被测系统 URL + SUT URL 保存/重启持久化 + 前端定时任务页与系统配置卡片 | ✅ 按 PRD 工作流实现，后端 179 测试全过、前端 build 通过、真实端到端验收通过（已部署生产） |
 | 26 | 08-19 | **新建会议用例 step 11/17 执行失败修复**：保留页面直达 + 执行器策略5 增强（轮询等待/force勾选/校验防假通过）+ 用例 step 12 改「保持立即开始」（选择器修正零匹配坑）；部署 5.60 | ✅ 本地 + 生产各连跑 3 次全绿、零残留，提交 `c0dedbdf` |
@@ -403,6 +480,12 @@ cd frontend && npm run build
 - [ ] 维护 jieba 自定义词典（专业术语分词优化）
 - [ ] Docker 容器安装中文字体（修复截图乱码，不影响定位）
 
+### 待验证（会话 31 产出）
+
+- [ ] **5.44 前端点击「立即执行」验证**：此前点击不会生成新执行记录；修复后应生成新 execution，`status` 应为 `triggered`
+- [ ] **5.44 取消即停验证**：执行中点击取消 → 工作线程在用例/步骤边界退出，`run_execution` 快速返回，锁释放，可立即再点执行
+- [ ] **前端 `status` 字段消费**（可选增强）：后端已返回 `status: triggered/skipped` + `running_execution_id`，前端可据此弹提示"已有执行正在运行"
+
 ---
 
-*本文档由 Claude Code 于 2026-08-19 更新（会话 29：修复「执行中无法取消」（MySQL 1205 行锁，置 running 后立即 commit 释放行锁）+ 实机发现身份映射过期实例覆盖 cancelled 为 completed（改列查询 + FOR UPDATE）+ SUT URL 校验加固，后端 186 测试全过、实机取消演练通过、已部署 5.60 生产）。*
+*本文档由 Claude Code 于 2026-08-19 更新（会话 31：定时任务「立即执行」不生成执行记录修复 + 取消即停 + 实时进度回调，4 文件已修复并部署 5.44+5.202；待前端点击执行验证）。*

backend/app/data/page_url_mapping.json

@@ -70,6 +70,26 @@
       "fingerprint": [
         {"selector": ".meeting_list", "min_count": 1}
       ]
+    },
+    {
+      "id": "statistics",
+      "name": "数据统计",
+      "url": "https://192.168.5.44/#/meetingV3?meetingV3=%2FmeetingV3%2F%23%2FStatisticsModule",
+      "url_patterns": [
+        "StatisticsModule"
+      ],
+      "fingerprint": [
+        {"selector": "text=预定会议数据", "min_count": 1}
+      ],
+      "match_rules": [
+        {"type": "step_name_contains", "value": "预定数据"},
+        {"type": "selector_contains", "value": "预定数据"},
+        {"type": "step_name_contains", "value": "数据统计"}
+      ],
+      "skip_steps": {
+        "step_name_contains": ["预定数据", "数据统计"],
+        "action_is": ["click"]
+      }
     }
   ]
 }

backend/app/executors/playwright_executor.py

@@ -202,6 +202,11 @@ class PlaywrightExecutor:
         # 语义化执行：当前页面 scope（页面注册表 id，登录/导航后同步，仅记录）
         self._current_scope: Optional[str] = None
 
+        # 取消请求标记：由 run_all_cases_sync 在创建执行器后设置执行 ID，
+        # execute_step 在每次重试前检查 is_cancel_requested 并提前退出
+        self._execution_id: Optional[str] = None
+        self._cancel_requested: bool = False
+
         # 确保截图目录存在
         os.makedirs(self.screenshot_dir, exist_ok=True)
 
@@ -1265,12 +1270,29 @@ class PlaywrightExecutor:
         attempt = 0
         max_attempts = self.step_retry_count + 1  # 至少执行一次
         last_error: Optional[Exception] = None
+        # 取消请求标志（由 cancel_execution → request_cancel 在 execute_step 之间置位）
+        self._cancel_requested = False
 
         while attempt < max_attempts:
             attempt += 1
             step_result.status = "running"
             step_result.error = None
 
+            # ★ 取消请求检查（步骤边界）：置位后立即中断，避免重试放大器
+            try:
+                from app.services.execution_service import is_cancel_requested
+
+                if is_cancel_requested(self._execution_id or ""):
+                    self._cancel_requested = True
+            except Exception:
+                pass
+            if self._cancel_requested:
+                logger.info(
+                    f"步骤 {step_result.order} 检测到取消请求，中断执行 "
+                    f"(case={self._ctx.case_name if self._ctx else ''})"
+                )
+                break
+
             try:
                 logger.debug(f"⏳ 执行步骤 {step_result.order}: {step_result.name} (action={action}, attempt={attempt}/{max_attempts})")
 
@@ -1402,6 +1424,12 @@ class PlaywrightExecutor:
         elif step_result.status != "passed":
             step_result.status = "failed"
 
+        # 取消请求中断：标记步骤为 skipped（不走失败重试/截图），
+        # 由 execute_case 在回填主结果时识别并保持 cancelled 语境
+        if self._cancel_requested:
+            step_result.status = "skipped"
+            step_result.log = "⏹ 检测到取消请求，中断执行"
+
         # 结束计时
         step_result.end_time = datetime.now()
         step_result.duration = (step_result.end_time - start_time).total_seconds()

backend/app/routers/scheduled_tasks.py

@@ -249,12 +249,15 @@ async def run_task_now(
 
     后台异步创建执行记录并运行，执行完成后自动生成报告。
 
+    若当前已有 UI 执行任务在运行，返回跳过状态（不创建执行记录），
+    前端可据此提示用户"请等待当前执行完成"。
+
     Args:
         task_id (str): 任务ID
         db (AsyncSession): 数据库会话
 
     Returns:
-        dict: 触发结果
+        dict: 触发结果，含 status 字段（triggered / skipped）
     """
     try:
         result = await db.execute(
@@ -264,11 +267,27 @@ async def run_task_now(
         if not task:
             raise HTTPException(status_code=404, detail=f"定时任务不存在: {task_id}")
 
-        # 后台异步执行（不阻塞请求）
-        asyncio.create_task(run_task_once(task_id))
+        # 预检全局执行锁：已有 UI 执行在运行 → 返回明确跳过状态
+        from app.services.execution_service import is_execution_running
+
+        is_running, running_id = is_execution_running()
+        if is_running:
+            logger.info(
+                f"[定时任务] 手动触发跳过：已有执行 {running_id} 正在运行"
+            )
+            return {
+                "message": f"已有执行 {running_id} 正在运行，请等待完成后再执行",
+                "task_id": task_id,
+                "status": "skipped",
+                "running_execution_id": running_id,
+            }
+
+        # 后台异步执行（manual=True 绕过任务级去重守卫）
+        asyncio.create_task(run_task_once(task_id, manual=True))
         return {
             "message": f"任务「{task.name}」已触发执行",
             "task_id": task_id,
+            "status": "triggered",
         }
     except HTTPException:
         raise

backend/app/services/execution_service.py

@@ -43,6 +43,7 @@ logger = logging.getLogger(__name__)
 # 确保 UI 自动化用例串行执行，避免并发执行导致浏览器页面状态互相干扰
 # 根因：Playwright 浏览器实例共享页面状态，并发执行会导致状态竞争
 _execution_lock = threading.Lock()
+# 正在运行的执行任务ID（全局 UI 执行锁）
 _running_execution_id: Optional[str] = None
 _running_execution_lock = threading.Lock()
 
@@ -62,6 +63,8 @@ def set_running_execution(execution_id: Optional[str]) -> None:
     """
     设置/清除当前运行中的执行ID
 
+    同时维护取消请求集合：设置新执行时清空旧的取消标记。
+
     Args:
         execution_id: 执行ID，None 表示清除
     """
@@ -74,6 +77,52 @@ def set_running_execution(execution_id: Optional[str]) -> None:
             logger.info("[执行锁] 清除运行状态")
 
 
+# 被请求取消的执行ID集合（供 Playwright 工作线程快速退出循环）
+_cancel_requested: set = set()
+_cancel_lock = threading.Lock()
+
+
+def request_cancel(execution_id: str) -> None:
+    """
+    请求取消指定执行（线程安全）
+
+    仅标记取消意图，不阻塞调用方；Playwright 工作线程在用例/步骤边界检查
+    该标记并提前退出，从而使 run_execution 尽快返回、释放全局锁。
+
+    Args:
+        execution_id (str): 执行记录ID
+    """
+    with _cancel_lock:
+        _cancel_requested.add(execution_id)
+
+
+def is_cancel_requested(execution_id: str) -> bool:
+    """
+    检查指定执行是否被请求取消（线程安全）
+
+    Args:
+        execution_id (str): 执行记录ID
+
+    Returns:
+        bool: 是否已请求取消
+    """
+    with _cancel_lock:
+        return execution_id in _cancel_requested
+
+
+def clear_cancel_requested(execution_id: str) -> None:
+    """
+    清除指定执行的取消请求标记
+
+    在创建新执行或执行重跑时调用，避免旧取消标记残留导致新执行一启动就被中断。
+
+    Args:
+        execution_id (str): 执行记录ID
+    """
+    with _cancel_lock:
+        _cancel_requested.discard(execution_id)
+
+
 class ExecutionService:
     """
     执行服务类
@@ -276,6 +325,9 @@ class ExecutionService:
         # 获取执行锁
         set_running_execution(execution_id)
 
+        # 清除旧取消标记（防止重跑时残留）
+        clear_cancel_requested(execution_id)
+
         # 更新状态为运行中并立即提交：
         # 若不提交，长执行期间事务会一直持有 executions 行锁（MySQL InnoDB），
         # 取消接口的 UPDATE 将被阻塞（Lock wait timeout exceeded 1205），
@@ -315,16 +367,52 @@ class ExecutionService:
         # 在线程池中为每个用例使用独立执行器
         loop = asyncio.get_event_loop()
 
+        async def _persist_case_result(case_result_obj, result):
+            """单个用例执行完毕：立即写库 + 更新统计 + 广播（跨线程提交到事件循环）"""
+            try:
+                # 已取消 → 不覆盖 skipped 状态，直接跳过写入
+                status_check = await self.db.execute(
+                    select(Execution.status).where(Execution.id == execution_id)
+                )
+                if status_check.scalar_one_or_none() == "cancelled":
+                    return
+
+                case_result_obj.status = result.status
+                case_result_obj.duration = result.duration
+                case_result_obj.end_time = datetime.now()
+                case_result_obj.retry_count = result.retry_count
+                case_result_obj.error_message = result.error_message
+                case_result_obj.steps_result = [
+                    {"order": s.order, "name": s.name, "action": s.action,
+                     "status": s.status, "duration": s.duration,
+                     "screenshot": s.screenshot, "log": s.log, "error": s.error}
+                    for s in result.steps_result
+                ]
+                execution.passed = await self._count_results(execution_id, "passed")
+                execution.failed = await self._count_results(execution_id, "failed")
+                execution.skipped = await self._count_results(execution_id, "skipped")
+                execution.pass_rate = execution.calculate_pass_rate()
+                await self.db.commit()  # 立即提交，前端轮询可见
+
+                await manager.broadcast(execution_id, {
+                    "type": "case_complete",
+                    "data": case_result_obj.to_dict(),
+                })
+            except Exception as e:
+                logger.error(f"实时写入用例结果失败: {e}")
+                await self.db.rollback()
+
         # 定义一个共享的执行函数，复用同一个执行器
-        def run_all_cases_sync(cases_list, exec_config, ev_loop, broadcast_fn):
+        def run_all_cases_sync(cases_list, exec_config, ev_loop, broadcast_fn, result_callback):
             """
             在一个执行器中执行所有用例，保持登录态
 
             Args:
-                cases_list: 用例字典列表
+                cases_list: (case_result_obj, case_dict) 元组列表
                 exec_config: 执行配置
                 ev_loop: 主事件循环（用于线程安全广播）
                 broadcast_fn: 异步广播函数（async broadcast）
+                result_callback: 用例完成后的实时写库回调（async）
             """
             logger.info(f"[run_all_cases_sync] 开始执行, 用例数: {len(cases_list)}")
 
@@ -336,6 +424,8 @@ class ExecutionService:
             executor_config.setdefault("base_url", settings.TARGET_URL)
 
             executor = PlaywrightExecutor(config=executor_config)
+            # 传入 execution_id，供 execute_step 检查取消请求
+            executor._execution_id = execution_id
             results = []
 
             try:
@@ -343,7 +433,16 @@ class ExecutionService:
                 logger.info(f"[run_all_cases_sync] 执行器启动成功")
 
                 # 第一个用例会自动登录
-                for i, case_dict in enumerate(cases_list):
+                for i, (case_result_obj, case_dict) in enumerate(cases_list):
+                    # ★ 取消请求检查（用例边界）：取消后立即中断剩余用例，
+                    # 使 run_execution 尽快返回并释放全局锁，不再后台空跑数小时。
+                    if is_cancel_requested(execution_id):
+                        logger.info(
+                            f"[run_all_cases_sync] 检测到取消请求，中断剩余用例 "
+                            f"({i}/{len(cases_list)} 已执行)"
+                        )
+                        break
+
                     case_id = case_dict.get("id", "")
                     case_name = case_dict.get("name", "")
                     steps_count = len(case_dict.get("steps", []))
@@ -378,6 +477,15 @@ class ExecutionService:
                     logger.info(f"[run_all_cases_sync] 用例 {case_name} 执行完成, status={result.status}")
                     results.append(result)
 
+                    # ★ 实时写库：跨线程提交到事件循环（不阻塞 Playwright 线程）
+                    try:
+                        asyncio.run_coroutine_threadsafe(
+                            result_callback(case_result_obj, result),
+                            ev_loop,
+                        )
+                    except Exception as e:
+                        logger.error(f"✗ 用例结果回调提交失败: {str(e)}")
+
                 logger.info(f"[run_all_cases_sync] 所有用例执行完成, 结果数: {len(results)}")
                 return results
             except Exception as e:
@@ -408,10 +516,11 @@ class ExecutionService:
             def _playwright_worker():
                 try:
                     exec_holder["data"] = run_all_cases_sync(
-                        [c["case_dict"] for c in cases_to_run],
+                        [(c["case_result"], c["case_dict"]) for c in cases_to_run],
                         config,
                         loop,
                         manager.broadcast,
+                        _persist_case_result,
                     )
                 except Exception as e:  # noqa: BLE001 - 线程内异常需透传到主协程
                     exec_holder["err"] = e
@@ -429,64 +538,10 @@ class ExecutionService:
                 raise exec_holder["err"]
             all_results = exec_holder.get("data", [])
 
-            # 若执行期间已被用户取消，结果保持取消时的 skipped 状态，不再写入真实结果
-            # with_for_update 为当前读：MySQL REPEATABLE READ 下普通 SELECT 走快照读，
-            # 看不到本事务之外已提交的取消（会读到 running 旧快照），必须加锁读最新已提交状态
-            cancel_check = await self.db.execute(
-                select(Execution.status).where(Execution.id == execution_id).with_for_update()
-            )
-            exec_was_cancelled = cancel_check.scalar_one_or_none() == "cancelled"
-
-            # 更新结果
-            for i, case_info in enumerate(cases_to_run):
-                case_result = case_info["case_result"]
-                if exec_was_cancelled:
-                    # 执行已在运行中被取消：保留 skipped，不覆盖真实结果
-                    case_result.status = "skipped"
-                    case_result.error_message = "执行任务已取消"
-                    await self.db.flush()
-                    continue
-                result = all_results[i]
-
-                # 更新状态为运行中（广播用例开始）
-                case_result.status = "running"
-                case_result.start_time = datetime.now()
-                await self.db.flush()
-
-                await manager.broadcast(execution_id, {
-                    "type": "case_start",
-                    "data": {
-                        "execution_id": execution_id,
-                        "case_id": case_result.case_id,
-                        "case_name": case_result.case_name,
-                    },
-                })
-
-                # 更新结果
-                case_result.status = result.status
-                case_result.duration = result.duration
-                case_result.end_time = datetime.now()
-                case_result.retry_count = result.retry_count
-                case_result.error_message = result.error_message
-                case_result.steps_result = [
-                    {"order": s.order, "name": s.name, "action": s.action,
-                     "status": s.status, "duration": s.duration,
-                     "screenshot": s.screenshot, "log": s.log, "error": s.error}
-                    for s in result.steps_result
-                ]
-
-                # 更新执行统计
-                execution.passed = await self._count_results(execution_id, "passed")
-                execution.failed = await self._count_results(execution_id, "failed")
-                execution.skipped = await self._count_results(execution_id, "skipped")
-                execution.pass_rate = execution.calculate_pass_rate()
-                await self.db.flush()
-
-                # 广播用例完成
-                await manager.broadcast(execution_id, {
-                    "type": "case_complete",
-                    "data": case_result.to_dict(),
-                })
+            # ★ 实时写入模式：每个用例的结果已在 _persist_case_result 中逐条 commit，
+            # 此处不再批量写库（原来 join 后再批量写导致执行期间 DB 无任何更新）。
+            # 取消场景：cancel_execution 已把未执行用例标记 skipped；
+            # _persist_case_result 写入前也会检查 cancelled 状态，不会覆盖 skipped。
 
         except Exception as e:
             logger.error(f"执行过程异常: {str(e)}")
@@ -562,6 +617,10 @@ class ExecutionService:
         execution.end_time = datetime.now()
         await self.db.flush()
 
+        # ★ 请求取消：置位取消标记，Playwright 工作线程在用例/步骤边界检查并提前退出，
+        # 使 run_execution 尽快返回、释放全局锁，不再后台空跑剩余用例。
+        request_cancel(execution_id)
+
         # 标记剩余待执行的用例为跳过
         remaining = await self._get_pending_results(execution_id)
         for r in remaining:

backend/app/services/scheduler_service.py

@@ -235,7 +235,7 @@ async def _run_security_task_once(
     return execution.id
 
 
-async def run_task_once(task_id: str) -> Optional[str]:
+async def run_task_once(task_id: str, manual: bool = False) -> Optional[str]:
     """
     执行一次定时任务（创建执行记录 → 运行 → 自动生成报告 → 推进下次调度）
 
@@ -244,11 +244,15 @@ async def run_task_once(task_id: str) -> Optional[str]:
 
     Args:
         task_id (str): 定时任务 ID
+        manual (bool): 是否为手动触发。手动触发时绕过任务级去重守卫
+                       （`_running_tasks` 检查），但仍受全局执行锁约束。
+                       同一任务自动调度与手动触发互不阻塞。
 
     Returns:
         Optional[str]: 本次创建的执行记录 ID；跳过/失败时返回 None
     """
-    if task_id in _running_tasks:
+    # 仅自动调度走任务级去重；手动触发允许跳过（用户意图就是"再跑一次"）
+    if not manual and task_id in _running_tasks:
         logger.info(f"[定时任务] 任务 {task_id} 正在执行中，忽略重复触发")
         return None