feat(compare-alert): 报警末帧列补「停留Xs」(dwell-only,不显总帧数) (#226)

描述

## Summary
- 飞书报警卡片「末帧」列:给 failed(T1/T2/T6) 与 cancelled 兜底两条末帧路径补「末帧所在屏停留 Xs」,
  区分「一到结算页就崩(停留<1s)」vs「在结算页干转 40s 才放弃」;T5 卡死路径不变。
- trace_stuck.py:抽 _last_segment(末段扫描+时长,卡死判据与末帧停留共用);StuckPoint 加 dwell_ms;
  last_step / read_stuck_points.last 附 dwell;删重构后无人调用的 _platform_stuck。
- compare_alert_worker.py:新增 dwell-only 的 _fmt_last(只显环节·页面+停留,**不显总帧数**,
  规避「总帧数配末段时长」误导);failed / cancelled 兜底两处接上。
- 无新增列、无落库、无迁移。

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #226

app/core/compare_alert_worker.py

@@ -89,6 +89,16 @@ def _fmt_stuck(sp: trace_stuck.StuckPoint) -> str:
     return s
 
 
+def _fmt_last(sp: trace_stuck.StuckPoint) -> str:
+    """末帧路径(failed/兜底):环节·页面 + 末段停留时长,不显总帧数(总帧数配末段时长会误导)。
+    dwell_ms 为 None(缺 ts/时钟回退) → 只显环节。"""
+    s = sp.label()
+    if sp.dwell_ms is not None:
+        sec = round(sp.dwell_ms / 1000)
+        s += f" 停留{sec}s" if sec >= 1 else " 停留<1s"
+    return s
+
+
 def build_hits(
     records: list,
     *,
@@ -134,9 +144,9 @@ def build_hits(
                     cancelled_ms_threshold=cancelled_ms_threshold,
                     cancelled_step_threshold=cancelled_step_threshold,
                 )
-                # 没卡点但命中(超阈值放弃)→ 用末帧标「退出前在哪屏」(平台·环节·页面)
+                # 没卡点但命中(超阈值放弃)→ 末帧标「退出前在哪屏 + 在那屏停多久」(dwell-only,不显总帧数)
                 if hit is not None and res is not None and res.last is not None:
-                    hit = _dc_replace(hit, stuck_point=res.last.label())
+                    hit = _dc_replace(hit, stuck_point=_fmt_last(res.last))
             if hit is not None:
                 hits.append(hit)
         else:
@@ -156,12 +166,12 @@ def build_hits(
             ):
                 td = _trace_dir(base, rec.trace_url)
                 if td is not None:
-                    sp = trace_stuck.last_step(td)
+                    sp = trace_stuck.last_step(td, max_tail=max_tail)
                     reads += 1
                     if sp is not None:
-                        # failed 是「末帧停在哪」:last_step 返回的 frames 是该平台总帧数(非"卡住"帧数)、
-                        # stuck_ms=None,带上帧数/时长会误导,故只用 label() 显示环节
-                        hit = _dc_replace(hit, stuck_point=sp.label())
+                        # failed 是「末帧停在哪 + 在那屏停多久」:只显环节·页面 + 末段停留 dwell,
+                        # 不显总帧数(总帧数配末段时长会误导);缺 ts 时 _fmt_last 自动退化为只显环节
+                        hit = _dc_replace(hit, stuck_point=_fmt_last(sp))
             if hit is not None:
                 hits.append(hit)
     return hits

app/services/trace_stuck.py

@@ -47,6 +47,7 @@ class StuckPoint:
     frames: int               # 末段连续困住的帧数(上限 max_tail)
     stuck_ms: int | None = None  # 末段连续卡住的时长(ms);无 timestamp 时 None
     detected_page: str | None = None  # 末帧所在页面(pricebot detected_page 原值);无则 None
+    dwell_ms: int | None = None  # 末帧所在屏停留时长(ms);末帧路径(failed/兜底)用,无 ts → None
 
     def label(self) -> str:
         p = PLATFORM_LABELS.get(self.platform, self.platform)
@@ -90,10 +91,15 @@ def _read_head(path: Path, nbytes: int = 4096) -> tuple[str | None, str | None,
     return (ps.group(1) if ps else None, pg.group(1) if pg else None, ts.group(1) if ts else None)
 
 
-def _platform_stuck(
-    platform: str, step_files: list[Path], threshold: int, max_tail: int
-) -> StuckPoint | None:
-    """末帧往前数连续同 (pipeline_step, detected_page) 的帧数 ≥threshold → 卡死。"""
+def _last_segment(
+    step_files: list[Path], max_tail: int
+) -> tuple[str, str | None, int, int | None] | None:
+    """末帧往前数连续同 (pipeline_step, detected_page) 的一段。
+
+    返回 (pipeline_step, detected_page, count, dwell_ms);末帧 pipeline_step 抠不出 → None。
+    dwell_ms = 段末帧ts − 段首帧ts(ms);两端 ts 不全可解析、或负(帧钟回退) → None。
+    这是「末段停留」的唯一算法,卡死判据(≥threshold)与末帧停留(无门槛)都复用它。
+    """
     tail = step_files[-max_tail:]
     heads = [_read_head(p) for p in tail]  # [(ps, pg, ts), ...]
     last_ps, last_pg, _ = heads[-1]
@@ -105,15 +111,11 @@ def _platform_stuck(
             seg_ts.append(ts)
         else:
             break
-    count = len(seg_ts)
-    if count < threshold:
-        return None
-    # seg_ts[0]=末帧, seg_ts[-1]=段首帧;两端都能解析才算时长
     t_last, t_first = _parse_ts(seg_ts[0]), _parse_ts(seg_ts[-1])
-    stuck_ms = round((t_last - t_first).total_seconds() * 1000) if t_last and t_first else None
-    if stuck_ms is not None and stuck_ms < 0:
-        stuck_ms = None  # 时钟不单调(帧 timestamp 回退)→ 降级为不显示时长
-    return StuckPoint(platform, last_ps, count, stuck_ms, detected_page=last_pg)
+    dwell_ms = round((t_last - t_first).total_seconds() * 1000) if t_last and t_first else None
+    if dwell_ms is not None and dwell_ms < 0:
+        dwell_ms = None  # 时钟不单调(帧 timestamp 回退)→ 降级为不显示时长
+    return last_ps, last_pg, len(seg_ts), dwell_ms
 
 
 def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> StuckResult:
@@ -132,15 +134,20 @@ def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> Stuc
             if not step_files:
                 continue
             any_frames = True
-            sp = _platform_stuck(pdir.name, step_files, threshold, max_tail)
-            if sp is not None:
-                points.append(sp)
-            # 末帧:取帧数最多平台的末帧(与 last_step 同口径),带 detected_page
+            seg = _last_segment(step_files, max_tail)  # 每平台只扫一次末段
+            if seg is None:
+                # 末帧抠不出:整段跳过(不判卡死、也不当末帧候选)——与旧版 _platform_stuck→None
+                # + 独立 _read_head(末帧)→ps None 两处一并跳过等价(旧版两者都 key off 末帧)
+                continue
+            ps, pg, count, dwell_ms = seg
+            if count >= threshold:
+                points.append(StuckPoint(pdir.name, ps, count, dwell_ms, detected_page=pg))
+            # 末帧:取帧数最多平台的末帧(与 last_step 同口径),带 detected_page + 末段停留
             if len(step_files) > best_n:
-                ps, pg, _ = _read_head(step_files[-1])
-                if ps is not None:
-                    best_n = len(step_files)
-                    last = StuckPoint(pdir.name, ps, len(step_files), detected_page=pg)
+                best_n = len(step_files)
+                last = StuckPoint(
+                    pdir.name, ps, len(step_files), detected_page=pg, dwell_ms=dwell_ms
+                )
         if not any_frames:
             return StuckResult(readable=False, points=[])
         return StuckResult(readable=True, points=points, last=last)
@@ -148,8 +155,9 @@ def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> Stuc
         return StuckResult(readable=False, points=[])
 
 
-def last_step(trace_dir: Path) -> StuckPoint | None:
-    """failed 用:取帧数最多平台的末帧环节(不要求原地打转)。读不到 → None。"""
+def last_step(trace_dir: Path, *, max_tail: int) -> StuckPoint | None:
+    """failed 用:取帧数最多平台的末帧环节(不要求原地打转)。读不到 → None。
+    附末段停留 dwell_ms(末帧所在屏停留时长);无 ts/时钟回退 → None。"""
     try:
         if not trace_dir.is_dir():
             return None
@@ -164,9 +172,11 @@ def last_step(trace_dir: Path) -> StuckPoint | None:
         if best is None:
             return None
         _, platform, step_files = best
-        ps, pg, _ts = _read_head(step_files[-1])
-        if ps is None:
+        seg = _last_segment(step_files, max_tail)
+        if seg is None:
             return None
-        return StuckPoint(platform, ps, len(step_files), detected_page=pg)  # stuck_ms=None(failed 不算时长)
+        ps, pg, _count, dwell_ms = seg
+        # frames 仍=总帧数(选平台口径不变);dwell_ms=末帧所在屏停留
+        return StuckPoint(platform, ps, len(step_files), detected_page=pg, dwell_ms=dwell_ms)
     except OSError:
         return None

docs/superpowers/plans/2026-08-07-compare-alert-last-frame-dwell.md

@@ -0,0 +1,538 @@
+# 比价报警「末帧停留时长」实现计划
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 飞书报警卡片「末帧」列给 failed(T1/T2/T6) 与 cancelled 兜底两种末帧路径补上「末帧所在屏停留 Xs」，让运营一眼区分「一到结算页就崩(停留<1s)」vs「在结算页干转 40s 才放弃」。
+
+**Architecture:** 在 `trace_stuck.py` 抽一个末段扫描 helper `_last_segment`(复用现有 `_platform_stuck` 段扫描+时长逻辑，去掉 threshold 门槛)，供 `_platform_stuck`/`last_step`/`read_stuck_points` 三处复用；`StuckPoint` 新增 `dwell_ms` 字段承载「末帧所在屏停留」，与 `stuck_ms`(T5 卡死段)语义分离；worker 新增 dwell-only 格式化 `_fmt_last`(只显环节·页面+停留，**不显总帧数**，规避原注释担心的误导)。
+
+**Tech Stack:** Python 3.11 / FastAPI / pytest。纯 CPU+本地文件读，无 DB、无迁移、无外部调用。
+
+关联 spec：`docs/superpowers/specs/2026-08-07-compare-alert-last-frame-dwell-design.md`
+
+---
+
+## File Structure
+
+| 文件 | 职责 | 本计划改动 |
+|---|---|---|
+| `app/services/trace_stuck.py` | trace 末段读取+卡死判定(薄 IO+纯逻辑) | `StuckPoint` 加 `dwell_ms`；抽 `_last_segment`；`_platform_stuck`/`last_step`/`read_stuck_points` 复用它 |
+| `app/core/compare_alert_worker.py` | 报警编排 | 新增 `_fmt_last`；failed/cancelled 兜底两处末帧格式化换成它；`last_step` 调用传 `max_tail` |
+| `tests/test_trace_stuck.py` | trace_stuck 单测 | 新增 dwell 用例；`last_step` 调用加 `max_tail=40` |
+| `tests/test_compare_alert_stuck_worker.py` | worker 集成测 | 新增 `_frame_ts` helper + failed/兜底带 dwell 用例 + `_fmt_last` 单测 |
+
+`app/services/compare_alert_format.py` **不改**(末帧列是自由字符串)。
+
+**基线命令**(每个 Task 前后跑，避免全量 pytest 的先前债干扰)：
+```bash
+pytest tests/test_trace_stuck.py tests/test_compare_alert_stuck_worker.py -q
+```
+
+---
+
+## Task 1: `StuckPoint` 加 `dwell_ms` + 抽 `_last_segment`(重构，行为不变)
+
+**Files:**
+- Modify: `app/services/trace_stuck.py`(`StuckPoint` 定义 :43-57；`_platform_stuck` :93-116)
+- Test: `tests/test_trace_stuck.py`(现有测试作回归网，本 Task 不新增)
+
+> 纯重构 + 加一个默认 `None` 的新字段。无新外部行为，靠现有测试保绿。
+
+- [ ] **Step 1: 跑基线，确认现有测试全绿**
+
+Run: `pytest tests/test_trace_stuck.py -q`
+Expected: PASS(全绿)
+
+- [ ] **Step 2: `StuckPoint` 加 `dwell_ms` 字段**
+
+把 `app/services/trace_stuck.py` 的 `StuckPoint`(:43-57) 的字段区改为(仅加最后一行，`label()` 不动)：
+
+```python
+@dataclass(frozen=True)
+class StuckPoint:
+    platform: str
+    pipeline_step: str
+    frames: int               # 末段连续困住的帧数(上限 max_tail)
+    stuck_ms: int | None = None  # 末段连续卡住的时长(ms);无 timestamp 时 None
+    detected_page: str | None = None  # 末帧所在页面(pricebot detected_page 原值);无则 None
+    dwell_ms: int | None = None  # 末帧所在屏停留时长(ms);末帧路径(failed/兜底)用,无 ts → None
+```
+
+- [ ] **Step 3: 新增 `_last_segment` helper**
+
+在 `app/services/trace_stuck.py` 的 `_platform_stuck` **之前**插入(紧跟 `_read_head` 之后)：
+
+```python
+def _last_segment(
+    step_files: list[Path], max_tail: int
+) -> tuple[str, str | None, int, int | None] | None:
+    """末帧往前数连续同 (pipeline_step, detected_page) 的一段。
+
+    返回 (pipeline_step, detected_page, count, dwell_ms);末帧 pipeline_step 抠不出 → None。
+    dwell_ms = 段末帧ts − 段首帧ts(ms);两端 ts 不全可解析、或负(帧钟回退) → None。
+    这是「末段停留」的唯一算法,卡死判据(≥threshold)与末帧停留(无门槛)都复用它。
+    """
+    tail = step_files[-max_tail:]
+    heads = [_read_head(p) for p in tail]  # [(ps, pg, ts), ...]
+    last_ps, last_pg, _ = heads[-1]
+    if last_ps is None:
+        return None
+    seg_ts: list[str | None] = []  # 连续段的 timestamp(逆序:末帧在前)
+    for ps, pg, ts in reversed(heads):
+        if ps == last_ps and pg == last_pg:
+            seg_ts.append(ts)
+        else:
+            break
+    t_last, t_first = _parse_ts(seg_ts[0]), _parse_ts(seg_ts[-1])
+    dwell_ms = round((t_last - t_first).total_seconds() * 1000) if t_last and t_first else None
+    if dwell_ms is not None and dwell_ms < 0:
+        dwell_ms = None  # 时钟不单调(帧 timestamp 回退)→ 降级为不显示时长
+    return last_ps, last_pg, len(seg_ts), dwell_ms
+```
+
+- [ ] **Step 4: `_platform_stuck` 改为复用 `_last_segment`**
+
+把 `app/services/trace_stuck.py` 的整个 `_platform_stuck`(:93-116) 替换为：
+
+```python
+def _platform_stuck(
+    platform: str, step_files: list[Path], threshold: int, max_tail: int
+) -> StuckPoint | None:
+    """末帧往前数连续同 (pipeline_step, detected_page) 的帧数 ≥threshold → 卡死。"""
+    seg = _last_segment(step_files, max_tail)
+    if seg is None:
+        return None
+    ps, pg, count, dwell_ms = seg
+    if count < threshold:
+        return None
+    # 卡死:frames=末段帧数、stuck_ms=末段时长(同段自洽);dwell_ms 字段留默认 None(T5 用 stuck_ms)
+    return StuckPoint(platform, ps, count, dwell_ms, detected_page=pg)
+```
+
+> `stuck_ms` 收的就是 `_last_segment` 的 `dwell_ms` 值——T5 场景「末段=卡死段」，二者本是同一个量，故行为与改前完全一致。
+
+- [ ] **Step 5: 跑测试，确认行为不变(仍全绿)**
+
+Run: `pytest tests/test_trace_stuck.py -q`
+Expected: PASS(全绿；`test_stuck_ms_computed_from_timestamps` 等对 `stuck_ms`/`frames` 的断言不变)
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add app/services/trace_stuck.py
+git commit -m "refactor(compare-alert): 抽 _last_segment、StuckPoint 加 dwell_ms 字段
+
+末段扫描+时长算法抽成 _last_segment 供三处复用;StuckPoint 新增
+dwell_ms(默认 None、承载末帧所在屏停留),T5 行为不变。
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 2: `last_step` 附 `dwell_ms`(failed 路径数据)
+
+**Files:**
+- Modify: `app/services/trace_stuck.py`(`last_step` :151-172)
+- Modify: `app/core/compare_alert_worker.py`(`last_step` 调用 :159，本 Task 只传 `max_tail`、仍用 `label()`)
+- Test: `tests/test_trace_stuck.py`
+
+- [ ] **Step 1: 写失败测试(带 timestamp 的 dwell)**
+
+在 `tests/test_trace_stuck.py` 末尾追加：
+
+```python
+def test_last_step_computes_dwell_ms(tmp_path):
+    # 帧数最多平台末段 5 帧都在 checkout,ts :00→:08(每帧+2s) → 停留 8s
+    m = tmp_path / "meituan"
+    _frame_ts(m, 0, "enter_store", "store", "2026-08-07T12:00:00.000000")
+    _frame_ts(m, 1, "enter_store", "store", "2026-08-07T12:00:02.000000")
+    _frame_ts(m, 2, "enter_store", "store", "2026-08-07T12:00:04.000000")
+    for i in range(3, 8):  # step3..7 checkout,末段 5 帧
+        _frame_ts(m, i, "checkout", "checkout_page",
+                  f"2026-08-07T12:00:{(i - 3) * 2:02d}.000000")
+    sp = last_step(tmp_path, max_tail=40)
+    assert sp.platform == "meituan"
+    assert sp.pipeline_step == "checkout"
+    assert sp.frames == 8          # 总帧数(非末段)
+    assert sp.dwell_ms == 8000     # 末段 checkout :00→:08 = 8s
+
+
+def test_last_step_dwell_none_without_ts(tmp_path):
+    m = tmp_path / "meituan"
+    for i in range(5):
+        _frame(m, i, "checkout", "checkout_page")  # 无 timestamp
+    sp = last_step(tmp_path, max_tail=40)
+    assert sp.dwell_ms is None
+
+
+def test_last_step_dwell_zero_single_frame_segment(tmp_path):
+    # 末帧与前一帧不同屏 → 末段只有末帧 1 帧 → 停留 0(一到就是末屏)
+    m = tmp_path / "meituan"
+    for i in range(4):
+        _frame_ts(m, i, "enter_store", "store", f"2026-08-07T12:00:{i:02d}.000000")
+    _frame_ts(m, 4, "checkout", "checkout_page", "2026-08-07T12:00:10.000000")
+    sp = last_step(tmp_path, max_tail=40)
+    assert sp.pipeline_step == "checkout"
+    assert sp.dwell_ms == 0
+```
+
+同时把现有 `test_last_step_returns_busiest_platform_last_env`(约 :87-95) 里的调用改为传 `max_tail`：
+
+```python
+    sp = last_step(tmp_path, max_tail=40)
+```
+(断言不变——该 fixture 无 timestamp，`dwell_ms=None`=字段默认，精确 `StuckPoint(...)` 相等仍成立。)
+
+- [ ] **Step 2: 跑测试，确认新用例失败**
+
+Run: `pytest tests/test_trace_stuck.py -q`
+Expected: FAIL — `last_step() got an unexpected keyword argument 'max_tail'`(签名还没加 `max_tail`)
+
+- [ ] **Step 3: 改 `last_step`**
+
+把 `app/services/trace_stuck.py` 的整个 `last_step`(:151-172) 替换为：
+
+```python
+def last_step(trace_dir: Path, *, max_tail: int) -> StuckPoint | None:
+    """failed 用:取帧数最多平台的末帧环节(不要求原地打转)。读不到 → None。
+    附末段停留 dwell_ms(末帧所在屏停留时长);无 ts/时钟回退 → None。"""
+    try:
+        if not trace_dir.is_dir():
+            return None
+        best: tuple[int, str, list[Path]] | None = None
+        for pdir in sorted(trace_dir.iterdir()):
+            if not pdir.is_dir():
+                continue
+            step_files = sorted(pdir.glob("step_*.json"), key=_step_num)
+            # 平局(同帧数)时取字典序第一个平台(sorted 保证稳定)
+            if step_files and (best is None or len(step_files) > best[0]):
+                best = (len(step_files), pdir.name, step_files)
+        if best is None:
+            return None
+        _, platform, step_files = best
+        seg = _last_segment(step_files, max_tail)
+        if seg is None:
+            return None
+        ps, pg, _count, dwell_ms = seg
+        # frames 仍=总帧数(选平台口径不变);dwell_ms=末帧所在屏停留
+        return StuckPoint(platform, ps, len(step_files), detected_page=pg, dwell_ms=dwell_ms)
+    except OSError:
+        return None
+```
+
+- [ ] **Step 4: 同步 worker 的 `last_step` 调用(保持 `label()` 不变，避免签名破坏 worker 测试)**
+
+在 `app/core/compare_alert_worker.py` 把 :159 一行：
+
+```python
+                sp = trace_stuck.last_step(td)
+```
+改为：
+```python
+                sp = trace_stuck.last_step(td, max_tail=max_tail)
+```
+(本 Task 只改调用签名；末帧格式化换成 `_fmt_last` 留到 Task 4。)
+
+- [ ] **Step 5: 跑测试，确认全绿**
+
+Run: `pytest tests/test_trace_stuck.py tests/test_compare_alert_stuck_worker.py -q`
+Expected: PASS(新 dwell 用例过；worker 集成测因 `last_step` 仍用 `label()`、行为不变，全绿)
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add app/services/trace_stuck.py app/core/compare_alert_worker.py tests/test_trace_stuck.py
+git commit -m "feat(compare-alert): last_step 附末段停留 dwell_ms
+
+failed 末帧路径拿到「末帧所在屏停留」;frames 仍为总帧数、口径不变。
+worker 调用同步传 max_tail(格式化留待接线)。
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 3: `read_stuck_points` 的 `last` 附 `dwell_ms`(cancelled 兜底数据)
+
+**Files:**
+- Modify: `app/services/trace_stuck.py`(`read_stuck_points` :119-148)
+- Test: `tests/test_trace_stuck.py`
+
+- [ ] **Step 1: 写失败测试**
+
+在 `tests/test_trace_stuck.py` 末尾追加：
+
+```python
+def test_read_stuck_points_last_has_dwell(tmp_path):
+    # 没卡死(末段<threshold),但末帧末段带 ts → last.dwell_ms 有值
+    m = tmp_path / "meituan"
+    _frame_ts(m, 0, "enter_store", "store", "2026-08-07T12:00:00.000000")
+    _frame_ts(m, 1, "enter_store", "store", "2026-08-07T12:00:02.000000")
+    for i in range(2, 5):  # 末段 checkout 3 帧 :00→:04
+        _frame_ts(m, i, "checkout", "checkout_page",
+                  f"2026-08-07T12:00:{(i - 2) * 2:02d}.000000")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.points == []               # 末段 3<15,没卡死
+    assert res.last.pipeline_step == "checkout"
+    assert res.last.frames == 5           # 总帧数
+    assert res.last.dwell_ms == 4000      # 末段 :00→:04 = 4s
+```
+
+- [ ] **Step 2: 跑测试，确认失败**
+
+Run: `pytest tests/test_trace_stuck.py::test_read_stuck_points_last_has_dwell -q`
+Expected: FAIL — `assert None == 4000`(`last.dwell_ms` 还没填)
+
+- [ ] **Step 3: 改 `read_stuck_points`(每平台一次 `_last_segment`，零增量 IO)**
+
+把 `app/services/trace_stuck.py` 的整个 `read_stuck_points`(:119-148) 替换为：
+
+```python
+def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> StuckResult:
+    """逐平台判卡死。目录不存在/异常/无平台帧 → StuckResult(readable=False, points=[])。"""
+    try:
+        if not trace_dir.is_dir():
+            return StuckResult(readable=False, points=[])
+        points: list[StuckPoint] = []
+        any_frames = False
+        last: StuckPoint | None = None
+        best_n = -1
+        for pdir in sorted(trace_dir.iterdir()):
+            if not pdir.is_dir():
+                continue
+            step_files = sorted(pdir.glob("step_*.json"), key=_step_num)
+            if not step_files:
+                continue
+            any_frames = True
+            seg = _last_segment(step_files, max_tail)  # 每平台只扫一次末段
+            if seg is None:
+                continue  # 末帧抠不出:不判卡死、也不当末帧候选
+            ps, pg, count, dwell_ms = seg
+            if count >= threshold:
+                points.append(StuckPoint(pdir.name, ps, count, dwell_ms, detected_page=pg))
+            # 末帧:取帧数最多平台的末帧(与 last_step 同口径),带 detected_page + 末段停留
+            if len(step_files) > best_n:
+                best_n = len(step_files)
+                last = StuckPoint(
+                    pdir.name, ps, len(step_files), detected_page=pg, dwell_ms=dwell_ms
+                )
+        if not any_frames:
+            return StuckResult(readable=False, points=[])
+        return StuckResult(readable=True, points=points, last=last)
+    except OSError:
+        return StuckResult(readable=False, points=[])
+```
+
+> 行为等价校验：`points` 的 `StuckPoint` 仍是 `frames=末段count / stuck_ms=末段时长`(T5 卡死，同改前)；`last` 仍是 `frames=总帧数`，只是多带 `dwell_ms`。末帧 `ps is None` 的平台整段跳过(不判卡死、不更新 `last`)，与改前 `_platform_stuck→None` + `if ps is not None` 一致。
+
+- [ ] **Step 4: 跑测试，确认全绿**
+
+Run: `pytest tests/test_trace_stuck.py -q`
+Expected: PASS(新用例过；`test_read_stuck_points_returns_last_frame`、`test_per_platform_one_stuck_one_normal` 等精确断言仍相等)
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add app/services/trace_stuck.py tests/test_trace_stuck.py
+git commit -m "feat(compare-alert): read_stuck_points.last 附 dwell_ms
+
+cancelled 兜底末帧拿到末段停留;循环改为每平台一次 _last_segment,
+判卡死与末帧候选共用同一次末段扫描,零增量 IO。
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 4: worker `_fmt_last` + 末帧列接线(dwell-only 显示)
+
+**Files:**
+- Modify: `app/core/compare_alert_worker.py`(新增 `_fmt_last`；cancelled 兜底 :138-139、failed :161-164)
+- Test: `tests/test_compare_alert_stuck_worker.py`
+
+- [ ] **Step 1: 写失败测试(`_fmt_last` 三态 + failed/兜底集成)**
+
+在 `tests/test_compare_alert_stuck_worker.py` 顶部把 import 改为(加 `_fmt_last`)：
+
+```python
+from app.core.compare_alert_worker import _fmt_last, _fmt_stuck, build_hits
+```
+
+在 `_frame` helper(约 :23-28) 之后新增带 timestamp 的 fixture helper：
+
+```python
+def _frame_ts(pdir: Path, idx: int, step: str, page: str, ts: str) -> None:
+    pdir.mkdir(parents=True, exist_ok=True)
+    body = {"pipeline_step": step, "detected_page": page, "timestamp": ts,
+            "windows": [{"n": ["x" * 200]}]}
+    (pdir / f"step_{idx:03d}.json").write_text(
+        json.dumps(body, ensure_ascii=False), encoding="utf-8"
+    )
+```
+
+在文件末尾追加：
+
+```python
+# ---- _fmt_last 单测(末帧路径:环节·页面 + 停留,不显总帧数)----
+
+def test_fmt_last_with_dwell():
+    sp = StuckPoint("meituan", "checkout", frames=480,
+                    detected_page="checkout_page", dwell_ms=8000)
+    assert _fmt_last(sp) == "美团·结算·checkout_page 停留8s"
+
+
+def test_fmt_last_sub_second():
+    sp = StuckPoint("meituan", "checkout", frames=480,
+                    detected_page="checkout_page", dwell_ms=300)
+    assert _fmt_last(sp) == "美团·结算·checkout_page 停留<1s"
+
+
+def test_fmt_last_without_dwell():
+    sp = StuckPoint("meituan", "checkout", frames=480,
+                    detected_page="checkout_page", dwell_ms=None)
+    assert _fmt_last(sp) == "美团·结算·checkout_page"
+
+
+# ---- 末帧路径带 dwell 集成 ----
+
+def test_failed_stuck_point_has_dwell(tmp_path):
+    # failed 末帧 5 帧都在 checkout,ts :00→:08 → stuck_point 附「停留8s」
+    p = tmp_path / "20260807_f" / "meituan"
+    for i in range(5):
+        _frame_ts(p, i, "checkout", "checkout_page",
+                  f"2026-08-07T12:00:{i * 2:02d}.000000")
+    rec = _Rec(status="failed", fail_reason="启动超时",
+               trace_url="https://x/traces/20260807_f/")
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
+    assert len(hits) == 1
+    assert hits[0].alert_type == "T2"
+    assert hits[0].stuck_point == "美团·结算·checkout_page 停留8s"
+
+
+def test_cancelled_fallback_stuck_point_has_dwell(tmp_path):
+    # cancelled 超阈值(95s)但末段 5<15 不卡死 → 兜底,末帧带 ts → 附「停留8s」
+    p = tmp_path / "20260807_c" / "eleme"
+    _frame_ts(p, 0, "set_address", "home", "2026-08-07T12:00:00.000000")
+    for i in range(1, 6):  # enter_store 5 帧 :02→:10
+        _frame_ts(p, i, "enter_store", "store",
+                  f"2026-08-07T12:00:{i * 2:02d}.000000")
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/20260807_c/",
+               total_ms=95000, step_count=40)
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
+    assert len(hits) == 1
+    assert hits[0].alert_type == "T5"
+    assert hits[0].stuck_point == "饿了么·进店·store 停留8s"
+```
+
+- [ ] **Step 2: 跑测试，确认失败**
+
+Run: `pytest tests/test_compare_alert_stuck_worker.py -q`
+Expected: FAIL — `ImportError: cannot import name '_fmt_last'`
+
+- [ ] **Step 3: 新增 `_fmt_last`**
+
+在 `app/core/compare_alert_worker.py` 的 `_fmt_stuck`(:84-89) **之后**插入：
+
+```python
+def _fmt_last(sp: trace_stuck.StuckPoint) -> str:
+    """末帧路径(failed/兜底):环节·页面 + 末段停留时长,不显总帧数(总帧数配末段时长会误导)。
+    dwell_ms 为 None(缺 ts/时钟回退) → 只显环节。"""
+    s = sp.label()
+    if sp.dwell_ms is not None:
+        sec = round(sp.dwell_ms / 1000)
+        s += f" 停留{sec}s" if sec >= 1 else " 停留<1s"
+    return s
+```
+
+- [ ] **Step 4: cancelled 兜底接线**
+
+在 `app/core/compare_alert_worker.py` 把 :137-139：
+
+```python
+                # 没卡点但命中(超阈值放弃)→ 用末帧标「退出前在哪屏」(平台·环节·页面)
+                if hit is not None and res is not None and res.last is not None:
+                    hit = _dc_replace(hit, stuck_point=res.last.label())
+```
+替换为：
+```python
+                # 没卡点但命中(超阈值放弃)→ 末帧标「退出前在哪屏 + 在那屏停多久」(dwell-only,不显总帧数)
+                if hit is not None and res is not None and res.last is not None:
+                    hit = _dc_replace(hit, stuck_point=_fmt_last(res.last))
+```
+
+- [ ] **Step 5: failed 接线 + 改注释**
+
+在 `app/core/compare_alert_worker.py` 把 :161-164：
+
+```python
+                    if sp is not None:
+                        # failed 是「末帧停在哪」:last_step 返回的 frames 是该平台总帧数(非"卡住"帧数)、
+                        # stuck_ms=None,带上帧数/时长会误导,故只用 label() 显示环节
+                        hit = _dc_replace(hit, stuck_point=sp.label())
+```
+替换为：
+```python
+                    if sp is not None:
+                        # failed 是「末帧停在哪 + 在那屏停多久」:只显环节·页面 + 末段停留 dwell,
+                        # 不显总帧数(总帧数配末段时长会误导);缺 ts 时 _fmt_last 自动退化为只显环节
+                        hit = _dc_replace(hit, stuck_point=_fmt_last(sp))
+```
+
+- [ ] **Step 6: 跑测试，确认全绿**
+
+Run: `pytest tests/test_compare_alert_stuck_worker.py -q`
+Expected: PASS(新 dwell 用例过；`test_failed_gets_stuck_point_appended`、`test_cancelled_readable_not_stuck_long_duration_reports` 等无 ts 用例因 `dwell_ms=None`→只显环节，断言仍成立)
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add app/core/compare_alert_worker.py tests/test_compare_alert_stuck_worker.py
+git commit -m "feat(compare-alert): 末帧列显「停留Xs」(dwell-only,不显总帧数)
+
+failed 与 cancelled 兜底两条末帧路径用 _fmt_last 显示环节·页面+末段停留;
+只带 dwell、不带总帧数,规避原注释担心的「总帧数配末段时长」误导;
+缺 ts/时钟回退降级只显环节。
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 5: 全量回归 + lint
+
+**Files:** 无(仅校验)
+
+- [ ] **Step 1: 跑两测试文件全绿**
+
+Run: `pytest tests/test_trace_stuck.py tests/test_compare_alert_stuck_worker.py -q`
+Expected: PASS(全绿)
+
+- [ ] **Step 2: 跑其余 compare_alert 相关测试(确认没连带破坏)**
+
+Run: `pytest tests/test_compare_alert_format.py tests/test_compare_alert_rules.py tests/test_compare_alert_fallback.py -q`
+Expected: PASS(本计划未碰这些路径，应全绿)
+
+- [ ] **Step 3: ruff 检查改动文件**
+
+Run: `ruff check app/services/trace_stuck.py app/core/compare_alert_worker.py tests/test_trace_stuck.py tests/test_compare_alert_stuck_worker.py`
+Expected: `All checks passed!`(如有可自动修的用 `ruff check --fix` 同名文件；有则改后重跑 Step 1)
+
+- [ ] **Step 4: 如 Step 3 有 `--fix` 改动则 commit**
+
+```bash
+git add -A
+git commit -m "style(compare-alert): ruff 清理
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## 完成标准(Definition of Done)
+
+- 飞书卡片「末帧」列：failed 与 cancelled 兜底两种路径显示 `平台·环节·页面 停留Xs`(有 ts 时)或 `平台·环节·页面`(缺 ts)。
+- T5 卡死路径显示不变(`平台·环节 N帧/Xs`)。
+- `pytest tests/test_trace_stuck.py tests/test_compare_alert_stuck_worker.py -q` 全绿。
+- `ruff check` 改动文件通过。
+- 无新增列、无落库、无迁移。

docs/superpowers/specs/2026-08-07-compare-alert-last-frame-dwell-design.md

@@ -0,0 +1,121 @@
+# 比价报警「末帧停留时长」增强设计
+
+- 日期：2026-08-07
+- 分支：feat-compare-alert-last-frame-dwell
+- 关联：
+  - `docs/superpowers/specs/2026-08-04-compare-fail-alert-design.md`（一期报警）
+  - `docs/superpowers/specs/2026-08-05-compare-stuck-detection-design.md`（卡死定位，本期母 spec）
+
+## 1. 背景与问题
+
+飞书报警卡片已有独立的「末帧」列（`app/services/compare_alert_format.py` `_TABLE_COLUMNS`，值 = `AlertHit.stuck_point`）。当前三种末帧口径里，**只有 T5 卡死带时长**，另两种只显「停在哪屏」、不显「在那屏停了多久」：
+
+| 场景 | 「末帧」列现状 | 有时长？ |
+|---|---|---|
+| T5 cancelled·判出原地卡死 | `平台·环节 N帧/Xs`（`_fmt_stuck`） | ✅ `stuck_ms` |
+| cancelled·兜底（超阈值放弃、没判出卡点） | `平台·环节·页面`（`res.last.label()`） | ❌ |
+| failed（T1/T2/T6） | `平台·环节·页面`（`last_step().label()`） | ❌ |
+
+问题：后两种看不出「碰一下结算页就崩」和「在结算页干转 40s 才放弃」的区别——而这个区别对定位 failed / 深度放弃很关键。卡片已有的「用时」列量的是**整场**耗时，不是**末屏**停留，二者互补不重复。
+
+### 1.1 为什么现在没有
+
+不是缺数据，是当初**故意**没算：`last_step()` / `read_stuck_points().last` 返回的 `frames` 是该平台**总帧数**（非末段停留帧），`stuck_ms=None`。`compare_alert_worker.py` 里有注释明确「带上帧数/时长会误导」，所以只用了 `label()`。数据其实现成——每帧 `timestamp` 已被 `_read_head` 抠出，末段时长算法已在 `_platform_stuck` 里（`stuck_ms`）。
+
+## 2. 目标
+
+- 给 **failed** 和 **cancelled 兜底** 两种末帧路径补上「末帧所在屏停留 Xs」。
+- 口径与 T5 的 `stuck_ms` 一致（同一种「末段连续同屏时长」），一个卡片里不出现两种「时长含义」。
+- **规避原注释担心的误导**：末帧路径只显停留时长、**不显总帧数**。
+- 稳：缺时间戳 / 时钟回退 / 读不到 trace → 降级只显环节，**绝不阻断报警**（延续母 spec 铁律）。
+
+## 3. 口径定义
+
+**末段停留 `dwell_ms`** = 从末帧往前、连续 `(pipeline_step, detected_page)` 都与末帧相同的那一段的时长（= 段末帧 `timestamp` − 段首帧 `timestamp`，round 到 ms）。
+
+- 与卡死判据 `stuck_ms` **同一算法**，唯一区别：**去掉 `count ≥ threshold` 门槛**（末帧路径不要求原地打转，只问「末屏停了多久」）。
+- 两端 `timestamp` 都能解析才有值；负时长（帧钟非单调/回退）→ `None`（沿用 `_platform_stuck` 现有降级）。
+- failed 与 cancelled 兜底都取「帧数最多平台」的末段停留（与既有 `last` / `last_step` 选平台口径一致）。
+
+## 4. 数据模型
+
+`app/services/trace_stuck.py` 的 `StuckPoint` 新增一个字段：
+
+```python
+@dataclass(frozen=True)
+class StuckPoint:
+    platform: str
+    pipeline_step: str
+    frames: int                        # 末段连续困住的帧数（上限 max_tail）
+    stuck_ms: int | None = None        # 判为卡死那段的时长；T5 用
+    detected_page: str | None = None
+    dwell_ms: int | None = None        # 新增：末帧所在屏停留时长；末帧路径用
+```
+
+`stuck_ms` 与 `dwell_ms` **并存、语义分离**：
+
+| 字段 | 含义 | 谁填/谁用 | 与 `frames` 关系 |
+|---|---|---|---|
+| `stuck_ms` | 判为卡死那段的时长 | `_platform_stuck` 填、T5 显 | `frames`=末段卡住帧数，**同段自洽** |
+| `dwell_ms` | 末帧所在屏停留 | 末帧路径填、末帧列显 | `frames`=平台总帧数，**不参与显示** |
+
+> 为什么不复用 `stuck_ms`：末帧路径返回的 `StuckPoint` 里 `frames` 是**总帧数**，若把末段停留塞进 `stuck_ms`，对象内部「帧数（总）」与「时长（末段）」不同段、不自洽，且会诱使误用 `_fmt_stuck` 打印出「总帧数 / 末段时长」——正是第 1.1 节要规避的误导。新加独立字段 + 专用 dwell-only 格式化，语义干净。
+
+## 5. 改动
+
+### 5.1 `services/trace_stuck.py`
+
+- 抽末段扫描逻辑（复用现有 `_platform_stuck` 的段扫描 + 时长计算，去掉 `count ≥ threshold` 门槛），产出末段 `(count, dwell_ms, detected_page)`。
+- `last_step()`：除末帧 head 外，读该平台末段几帧 head，算 `dwell_ms` 填入返回的 `StuckPoint`（`frames` 仍 = 总帧数，语义不变）。
+- `read_stuck_points()` 的 `last`：补 `dwell_ms`。该平台 tail 在逐平台判卡死时已读过，几乎零增量 IO。
+
+### 5.2 `core/compare_alert_worker.py`
+
+新增 dwell-only 格式化（**不显总帧数**是规避误导的关键）：
+
+```python
+def _fmt_last(sp: trace_stuck.StuckPoint) -> str:
+    """末帧路径：环节·页面 + 停留时长，不显总帧数。dwell_ms 为 None → 只显环节。"""
+    s = sp.label()
+    if sp.dwell_ms is not None:
+        sec = round(sp.dwell_ms / 1000)
+        s += f" 停留{sec}s" if sec >= 1 else " 停留<1s"
+    return s
+```
+
+- failed 路径：`sp.label()` → `_fmt_last(sp)`。
+- cancelled 兜底：`res.last.label()` → `_fmt_last(res.last)`。
+- 改掉原「带上帧数/时长会误导」注释（现在只带 dwell、不带总帧数，不再误导）。
+
+### 5.3 `services/compare_alert_format.py`
+
+**不改**。「末帧」列本就是自由字符串。
+
+## 6. 显示效果
+
+| 场景 | 改前 | 改后 |
+|---|---|---|
+| failed 结算页秒崩 | 美团·结算·checkout | 美团·结算·checkout **停留<1s** |
+| failed 结算页干转 | 美团·结算·checkout | 美团·结算·checkout **停留40s** |
+| cancelled 兜底 | 饿了么·进店·store | 饿了么·进店·store **停留8s** |
+| 缺 ts / 时钟回退 | 只环节 | 只环节（无停留） |
+| T5 卡死 | 美团·加菜 110帧/32s | 不变 |
+
+## 7. 降级与物理边界
+
+- **降级（不阻断报警）**：缺 `timestamp` / 帧钟非单调 / 读不到 trace → `dwell_ms=None` → 只显环节。任何 trace 异常仍在 `trace_stuck` 内降级。
+- **物理边界**：帧 `timestamp` 只到末帧。若比价在**写完末帧之后**才彻底冻死（不再落帧），这段测不到 → `dwell≈0`。要覆盖它得用 `abort时间 − 末帧ts`，但那是 **DB `updated_at`（SQLite UTC）vs pricebot 帧钟**、跨源跨时区——worker 对「别混钟」很谨慎（见冷启动水位注释），**不引入混钟**。所以报的是「帧级末段停留」：`dwell≈0` = 一到这屏就死，`dwell=30s` = 在这屏干转——低估本身也是信号。
+
+## 8. 测试
+
+- **`trace_stuck` 单测**：
+  - 新增：带 `timestamp` 的末帧 `dwell_ms` 用例（仿 `test_stuck_ms_computed_from_timestamps`）——覆盖末段多帧算出停留、末段单帧 → 0、无 ts → None、时钟回退 → None。
+  - 更新：`test_last_step_returns_busiest_platform_last_env`、`test_read_stuck_points_returns_last_frame` 的精确 `StuckPoint` 断言（多 `dwell_ms` 字段）。
+- **worker 集成**：新增「failed / cancelled 兜底带 dwell」用例（fixture 帧带 `timestamp`）；现有不带 ts 的用例不受影响（`dwell=None` → 只显环节）。
+- **`_fmt_last` 单测**：有 dwell（≥1s）/ `<1s`（sec 四舍五入为 0）/ 无 dwell 三态。
+
+## 9. 不做（YAGNI）
+
+- 不混 DB 钟补「纯末尾冻死」。
+- 不动 T5 卡死路径 / `timing.json` / 逐帧 profile（母 spec 第 12 节「不做每帧耗时」指 `timing.json` 逐帧 profile；本期用帧 `timestamp` 算的段时长是两回事，数据现成、报警用得上）。
+- 不加列、不落库、不迁移。

tests/test_compare_alert_stuck_worker.py

@@ -2,7 +2,7 @@
 import json
 from pathlib import Path
 
-from app.core.compare_alert_worker import _fmt_stuck, build_hits
+from app.core.compare_alert_worker import _fmt_last, _fmt_stuck, build_hits
 from app.services.trace_stuck import StuckPoint
 
 
@@ -28,6 +28,15 @@ def _frame(pdir: Path, idx: int, step: str, page: str) -> None:
     )
 
 
+def _frame_ts(pdir: Path, idx: int, step: str, page: str, ts: str) -> None:
+    pdir.mkdir(parents=True, exist_ok=True)
+    body = {"pipeline_step": step, "detected_page": page, "timestamp": ts,
+            "windows": [{"n": ["x" * 200]}]}
+    (pdir / f"step_{idx:03d}.json").write_text(
+        json.dumps(body, ensure_ascii=False), encoding="utf-8"
+    )
+
+
 _KW = dict(
     stuck_threshold=15, max_tail=40, max_trace_reads=30,
     cancelled_ms_threshold=90000, cancelled_step_threshold=30,
@@ -144,3 +153,61 @@ def test_fmt_stuck_with_ms():
 def test_fmt_stuck_without_ms():
     sp = StuckPoint(platform="meituan", pipeline_step="add_one_dish", frames=110, stuck_ms=None)
     assert _fmt_stuck(sp) == "美团·加菜 110帧"
+
+
+# ---- _fmt_last 单测(末帧路径:环节·页面 + 停留,不显总帧数)----
+
+def test_fmt_last_with_dwell():
+    sp = StuckPoint("meituan", "checkout", frames=480,
+                    detected_page="checkout_page", dwell_ms=8000)
+    assert _fmt_last(sp) == "美团·结算·checkout_page 停留8s"
+
+
+def test_fmt_last_sub_second():
+    sp = StuckPoint("meituan", "checkout", frames=480,
+                    detected_page="checkout_page", dwell_ms=300)
+    assert _fmt_last(sp) == "美团·结算·checkout_page 停留<1s"
+    # <1s 阈值边界:round(500/1000)=0 → <1s;round(999/1000)=1 → 停留1s
+    sp500 = StuckPoint("meituan", "checkout", frames=480,
+                       detected_page="checkout_page", dwell_ms=500)
+    assert _fmt_last(sp500) == "美团·结算·checkout_page 停留<1s"
+    sp999 = StuckPoint("meituan", "checkout", frames=480,
+                       detected_page="checkout_page", dwell_ms=999)
+    assert _fmt_last(sp999) == "美团·结算·checkout_page 停留1s"
+
+
+def test_fmt_last_without_dwell():
+    sp = StuckPoint("meituan", "checkout", frames=480,
+                    detected_page="checkout_page", dwell_ms=None)
+    assert _fmt_last(sp) == "美团·结算·checkout_page"
+
+
+# ---- 末帧路径带 dwell 集成 ----
+
+def test_failed_stuck_point_has_dwell(tmp_path):
+    # failed 末帧 5 帧都在 checkout,ts :00→:08 → stuck_point 附「停留8s」
+    p = tmp_path / "20260807_f" / "meituan"
+    for i in range(5):
+        _frame_ts(p, i, "checkout", "checkout_page",
+                  f"2026-08-07T12:00:{i * 2:02d}.000000")
+    rec = _Rec(status="failed", fail_reason="启动超时",
+               trace_url="https://x/traces/20260807_f/")
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
+    assert len(hits) == 1
+    assert hits[0].alert_type == "T2"
+    assert hits[0].stuck_point == "美团·结算·checkout_page 停留8s"
+
+
+def test_cancelled_fallback_stuck_point_has_dwell(tmp_path):
+    # cancelled 超阈值(95s)但末段 5<15 不卡死 → 兜底,末帧带 ts → 附「停留8s」
+    p = tmp_path / "20260807_c" / "eleme"
+    _frame_ts(p, 0, "set_address", "home", "2026-08-07T12:00:00.000000")
+    for i in range(1, 6):  # enter_store 5 帧 :02→:10
+        _frame_ts(p, i, "enter_store", "store",
+                  f"2026-08-07T12:00:{i * 2:02d}.000000")
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/20260807_c/",
+               total_ms=95000, step_count=40)
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
+    assert len(hits) == 1
+    assert hits[0].alert_type == "T5"
+    assert hits[0].stuck_point == "饿了么·进店·store 停留8s"

tests/test_trace_stuck.py

@@ -91,7 +91,7 @@ def test_last_step_returns_busiest_platform_last_env(tmp_path):
     e = tmp_path / "eleme"
     for i in range(3):
         _frame(e, i, "enter_store", "store")
-    sp = last_step(tmp_path)
+    sp = last_step(tmp_path, max_tail=40)
     assert sp == StuckPoint("meituan", "add_one_dish", 20, detected_page="meal_detail_popup")
 
 
@@ -172,3 +172,71 @@ def test_stuck_ms_none_when_clock_goes_backwards(tmp_path):
     res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
     assert len(res.points) == 1
     assert res.points[0].stuck_ms is None  # 负时长降级为 None
+
+
+def test_last_step_computes_dwell_ms(tmp_path):
+    # 帧数最多平台末段 5 帧都在 checkout,ts :00→:08(每帧+2s) → 停留 8s
+    m = tmp_path / "meituan"
+    _frame_ts(m, 0, "enter_store", "store", "2026-08-07T12:00:00.000000")
+    _frame_ts(m, 1, "enter_store", "store", "2026-08-07T12:00:02.000000")
+    _frame_ts(m, 2, "enter_store", "store", "2026-08-07T12:00:04.000000")
+    for i in range(3, 8):  # step3..7 checkout,末段 5 帧
+        _frame_ts(m, i, "checkout", "checkout_page",
+                  f"2026-08-07T12:00:{(i - 3) * 2:02d}.000000")
+    sp = last_step(tmp_path, max_tail=40)
+    assert sp.platform == "meituan"
+    assert sp.pipeline_step == "checkout"
+    assert sp.frames == 8          # 总帧数(非末段)
+    assert sp.dwell_ms == 8000     # 末段 checkout :00→:08 = 8s
+
+
+def test_last_step_dwell_none_without_ts(tmp_path):
+    m = tmp_path / "meituan"
+    for i in range(5):
+        _frame(m, i, "checkout", "checkout_page")  # 无 timestamp
+    sp = last_step(tmp_path, max_tail=40)
+    assert sp.dwell_ms is None
+
+
+def test_last_step_dwell_zero_single_frame_segment(tmp_path):
+    # 末帧与前一帧不同屏 → 末段只有末帧 1 帧 → 停留 0(一到就是末屏)
+    m = tmp_path / "meituan"
+    for i in range(4):
+        _frame_ts(m, i, "enter_store", "store", f"2026-08-07T12:00:{i:02d}.000000")
+    _frame_ts(m, 4, "checkout", "checkout_page", "2026-08-07T12:00:10.000000")
+    sp = last_step(tmp_path, max_tail=40)
+    assert sp.pipeline_step == "checkout"
+    assert sp.dwell_ms == 0
+
+
+def test_read_stuck_points_last_has_dwell(tmp_path):
+    # 没卡死(末段<threshold),但末帧末段带 ts → last.dwell_ms 有值
+    m = tmp_path / "meituan"
+    _frame_ts(m, 0, "enter_store", "store", "2026-08-07T12:00:00.000000")
+    _frame_ts(m, 1, "enter_store", "store", "2026-08-07T12:00:02.000000")
+    for i in range(2, 5):  # 末段 checkout 3 帧 :00→:04
+        _frame_ts(m, i, "checkout", "checkout_page",
+                  f"2026-08-07T12:00:{(i - 2) * 2:02d}.000000")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.points == []               # 末段 3<15,没卡死
+    assert res.last.pipeline_step == "checkout"
+    assert res.last.frames == 5           # 总帧数
+    assert res.last.dwell_ms == 4000      # 末段 :00→:04 = 4s
+
+
+def test_read_stuck_points_last_falls_through_when_busiest_last_frame_corrupt(tmp_path):
+    # 最忙平台末帧损坏(抠不出环节)→ _last_segment=None → 整段跳过(重构 continue 分支)
+    # → last 落到次忙的干净平台。锁定 read_stuck_points 重构的最险等价分支。
+    m = tmp_path / "meituan"
+    m.mkdir()
+    for i in range(7):
+        _frame(m, i, "add_one_dish", "menu")
+    (m / "step_007.json").write_bytes(b'{"pipeline_step": "add\xff')  # 末帧截断 UTF-8
+    e = tmp_path / "eleme"
+    for i in range(3):
+        _frame(e, i, "enter_store", "store")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.points == []                    # 谁都没卡死
+    assert res.last is not None
+    assert res.last.platform == "eleme"        # 最忙的 meituan(8帧)末帧损坏被跳过,last 落到 eleme
+    assert res.last.pipeline_step == "enter_store"