feat(compare-alert): 卡点独立成列(AlertHit.stuck_point + 卡片第5列),reason 去重简化

- AlertHit 加 stuck_point: str | None = None 字段(格式化好的「平台·环节 帧/s」)
- classify_cancelled_fallback reason 简化为「深度放弃」(耗时/步数已在「用时」列,不重复)
- build_hits 加 _fmt_stuck helper;cancelled 卡死 stuck_point=「美团·加菜 110帧/32s」;
  failed stuck_point=环节标签、reason 不再附「卡在 X」
- format_alert_card 列序改为 时间/手机号/用时/失败原因/卡点/版本/trace (7列)
- 同步更新 test_compare_alert_stuck_worker / _fallback / _format / _rules 断言

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/core/compare_alert_worker.py

@@ -81,6 +81,14 @@ def _trace_dir(base: Path, trace_url: str | None) -> Path | None:
     return base / name
 
 
+def _fmt_stuck(sp) -> str:
+    """StuckPoint → 「平台·环节 110帧/32s」;stuck_ms 为 None 时省略时长。"""
+    s = f"{sp.label()} {sp.frames}帧"
+    if sp.stuck_ms is not None:
+        s += f"/{round(sp.stuck_ms / 1000)}s"
+    return s
+
+
 def build_hits(
     records: list,
     *,
@@ -114,8 +122,8 @@ def build_hits(
                     reads += 1
             if res is not None and res.readable:
                 if res.points:
-                    reason = "卡在 " + "、".join(sp.label() for sp in res.points)
-                    hit = make_hit(rec, "T5", reason)
+                    stuck = "、".join(_fmt_stuck(sp) for sp in res.points)
+                    hit = _dc_replace(make_hit(rec, "T5", "深度放弃"), stuck_point=stuck)
                 else:
                     hit = None  # 读到且确认没卡 → 不报
             else:
@@ -146,7 +154,7 @@ def build_hits(
                     sp = trace_stuck.last_step(td)
                     reads += 1
                     if sp is not None:
-                        hit = _dc_replace(hit, reason=f"{hit.reason}｜卡在 {sp.label()}")
+                        hit = _dc_replace(hit, stuck_point=sp.label())
             if hit is not None:
                 hits.append(hit)
     return hits

app/services/compare_alert.py

@@ -30,6 +30,7 @@ class AlertHit:
     user_id: int | None
     total_ms: int | None = None
     step_count: int | None = None
+    stuck_point: str | None = None
 
 
 def make_hit(rec: Any, alert_type: str, reason: str) -> AlertHit:
@@ -59,10 +60,7 @@ def classify_cancelled_fallback(
         step is not None and step > cancelled_step_threshold
     )
     if deep:
-        return make_hit(
-            rec, "T5",
-            f"深度放弃·等待 {round((ms or 0) / 1000)}s / {step or 0} 步后退出",
-        )
+        return make_hit(rec, "T5", "深度放弃")
     return None
 
 

app/services/compare_alert_format.py

@@ -168,6 +168,7 @@ def _build_table_rows(
                 "phone": phone,
                 "cost": _cost_cell(h.total_ms, h.step_count),
                 "reason": h.reason,
+                "stuck": h.stuck_point or "-",
                 "ver": h.app_version or "-",
                 "trace": trace,
             })
@@ -179,6 +180,7 @@ _TABLE_COLUMNS = [
     {"name": "phone",  "display_name": "手机号",  "data_type": "text"},
     {"name": "cost",   "display_name": "用时",    "data_type": "text"},
     {"name": "reason", "display_name": "失败原因", "data_type": "text"},
+    {"name": "stuck",  "display_name": "卡点",    "data_type": "text"},
     {"name": "ver",    "display_name": "版本",    "data_type": "text"},
     {"name": "trace",  "display_name": "trace",  "data_type": "lark_md"},
 ]

tests/test_compare_alert_format.py

@@ -265,7 +265,7 @@ def test_card_body_markdown_summary():
     assert "合计 3 条" in md["content"]
 
 
-def test_card_has_table_six_columns():
+def test_card_has_table_seven_columns():
     card = format_alert_card(
         _card_hits(2),
         window_label="w",
@@ -280,9 +280,15 @@ def test_card_has_table_six_columns():
     table = elements[1]
     assert table["tag"] == "table"
     cols = table["columns"]
-    assert len(cols) == 6
+    assert len(cols) == 7
     display_names = [c["display_name"] for c in cols]
-    assert display_names == ["时间", "手机号", "用时", "失败原因", "版本", "trace"]
+    assert display_names == ["时间", "手机号", "用时", "失败原因", "卡点", "版本", "trace"]
+    # 「卡点」列在「失败原因」后、「版本」前
+    names = [c["name"] for c in cols]
+    reason_idx = names.index("reason")
+    stuck_idx = names.index("stuck")
+    ver_idx = names.index("ver")
+    assert reason_idx < stuck_idx < ver_idx
     # trace 列用 lark_md
     trace_col = next(c for c in cols if c["name"] == "trace")
     assert trace_col["data_type"] == "lark_md"
@@ -523,3 +529,44 @@ def test_card_table_has_page_size_and_header_style():
     assert "header_style" in table
     assert table["header_style"].get("background_style") == "grey"
     assert table["header_style"].get("bold") is True
+
+
+def test_card_stuck_point_shown_in_row():
+    """stuck_point 有值时行中 stuck 列正确显示；无值时显示「-」。"""
+    import dataclasses
+
+    from app.services.compare_alert import make_hit
+
+    class _Rec:
+        trace_id = "sp1"
+        status = "cancelled"
+        total_ms = 90000
+        step_count = 20
+        fail_reason = None
+        information = None
+        app_version = "v1.0"
+        created_at = datetime(2026, 8, 5, 10, 0)
+        trace_url = None
+        user_id = None
+
+    hit_with = dataclasses.replace(
+        make_hit(_Rec(), "T5", "深度放弃"),
+        stuck_point="美团·加菜 110帧/32s",
+    )
+    hit_without = make_hit(_Rec(), "T5", "深度放弃")  # stuck_point=None
+
+    card_with = format_alert_card(
+        [hit_with],
+        window_label="w", phone_map={}, interval_min=5,
+        max_detail_per_type=20, max_total=50,
+    )
+    row_with = card_with["body"]["elements"][1]["rows"][0]
+    assert row_with["stuck"] == "美团·加菜 110帧/32s"
+
+    card_without = format_alert_card(
+        [hit_without],
+        window_label="w", phone_map={}, interval_min=5,
+        max_detail_per_type=20, max_total=50,
+    )
+    row_without = card_without["body"]["elements"][1]["rows"][0]
+    assert row_without["stuck"] == "-"

tests/test_compare_alert_rules.py

@@ -58,4 +58,4 @@ def test_reason_texts():
     t1_empty = classify_record(_rec(status="failed", fail_reason=None, information=None), **KW)
     assert t1_empty.reason == "技术失败·比价过程出错"
     t5 = classify_record(_rec(status="cancelled", total_ms=98000, step_count=26), **KW)
-    assert t5.reason == "深度放弃·等待 98s / 26 步后退出"
+    assert t5.reason == "深度放弃"

tests/test_compare_alert_stuck_worker.py

@@ -2,7 +2,8 @@
 import json
 from pathlib import Path
 
-from app.core.compare_alert_worker import build_hits
+from app.core.compare_alert_worker import _fmt_stuck, build_hits
+from app.services.trace_stuck import StuckPoint
 
 
 class _Rec:
@@ -42,7 +43,9 @@ def test_cancelled_stuck_reports_via_trace(tmp_path):
     hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
     assert len(hits) == 1
     assert hits[0].alert_type == "T5"
-    assert "卡在" in hits[0].reason and "美团·加菜" in hits[0].reason
+    assert hits[0].reason == "深度放弃"
+    assert "美团·加菜" in hits[0].stuck_point
+    assert "帧" in hits[0].stuck_point
 
 
 def test_cancelled_readable_not_stuck_no_report(tmp_path):
@@ -80,7 +83,8 @@ def test_failed_gets_stuck_point_appended(tmp_path):
     hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
     assert len(hits) == 1
     assert hits[0].alert_type == "T2"
-    assert "卡在 美团·加菜" in hits[0].reason
+    assert "美团·加菜" in hits[0].stuck_point
+    assert "卡在" not in hits[0].reason
 
 
 def test_max_trace_reads_zero_skips_trace(tmp_path):
@@ -107,7 +111,19 @@ def test_shared_reads_budget_across_branches(tmp_path):
     kw = {**_KW, "max_trace_reads": 1}
     hits = build_hits([cancelled, failed], work_log_dir=str(tmp_path), **kw)
     assert len(hits) == 2
-    # cancelled 消耗了唯一预算 → 报卡死
-    assert hits[0].alert_type == "T5" and "卡在" in hits[0].reason
-    # failed 超预算 → 仍报 T2,但不附卡点
-    assert hits[1].alert_type == "T2" and "卡在" not in hits[1].reason
+    # cancelled 消耗了唯一预算 → 报卡死，卡点在 stuck_point
+    assert hits[0].alert_type == "T5" and hits[0].stuck_point is not None
+    # failed 超预算 → 仍报 T2,但 stuck_point 为 None
+    assert hits[1].alert_type == "T2" and hits[1].stuck_point is None
+
+
+# ---- _fmt_stuck 单测 ----
+
+def test_fmt_stuck_with_ms():
+    sp = StuckPoint(platform="meituan", pipeline_step="add_one_dish", frames=110, stuck_ms=32000)
+    assert _fmt_stuck(sp) == "美团·加菜 110帧/32s"
+
+
+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帧"