feat(compare-alert): failed 混合单某平台真技术失败被业务 headline 盖住时补判 T1 (#222)

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

alembic/versions/comparison_updated_at.py

@@ -0,0 +1,50 @@
+"""comparison_record 加 updated_at 列(报警水位)+ 回填现有行 + 索引
+
+新增 updated_at:server_default + onupdate = func.now()(DB 时钟)。比价失败报警 worker 用它做
+单调水位(WHERE updated_at > watermark)。加列后回填现有行 = created_at,避免冷启动 max(updated_at)
+为 NULL;再置 NOT NULL + 建索引 ix_comparison_updated(水位查询按它)。batch 模式兼容 SQLite。
+
+Revision ID: comparison_updated_at
+Revises: deepseek_v4_flash_price
+Create Date: 2026-08-04 00:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "comparison_updated_at"
+down_revision: str | Sequence[str] | None = "deepseek_v4_flash_price"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # 1) 先加可空列(不带 default,避免各库对 add-column-with-default 的差异)
+    with op.batch_alter_table("comparison_record", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True))
+
+    # 2) 回填现有行 = created_at(全新环境表为空,回填 no-op)
+    op.get_bind().execute(
+        sa.text(
+            "UPDATE comparison_record SET updated_at = created_at WHERE updated_at IS NULL"
+        )
+    )
+
+    # 3) 置 NOT NULL + server_default + 建索引
+    with op.batch_alter_table("comparison_record", schema=None) as batch_op:
+        batch_op.alter_column(
+            "updated_at",
+            existing_type=sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        )
+        batch_op.create_index("ix_comparison_updated", ["updated_at"], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("comparison_record", schema=None) as batch_op:
+        batch_op.drop_index("ix_comparison_updated")
+        batch_op.drop_column("updated_at")

app/admin/repositories/comparison_outcome.py

@@ -0,0 +1,61 @@
+"""admin 比价记录展示口径：把「流程跑完、外部原因致结果缺失」的记录判为成功。
+
+#209 / C 端落库把 store_not_found / items_not_found / store_closed / no_delivery /
+unsupported 归一化成 status='failed'；admin 排查视角改按记录级原始业务结局
+（raw_payload.record_status）重判——这些「外部缺失」算成功（附缺失提示），
+只有纯技术故障才是失败。仅 admin 用，不碰 C 端 / #209 落库。
+"""
+from __future__ import annotations
+
+from sqlalchemy import func
+
+from app.models.comparison import ComparisonRecord
+
+# 记录级原始业务结局里算「成功（流程跑完）」的集合；其余（failed / 未知）才是技术故障。
+ADMIN_SUCCESS_OUTCOMES = frozenset({
+    "success", "below_minimum", "store_closed",
+    "store_not_found", "items_not_found", "no_delivery", "unsupported",
+})
+# 有缺失的成功 → 感叹号 hover 提示；success 本身无提示。
+OUTCOME_HINTS = {
+    "below_minimum": "未满起送",
+    "store_closed": "门店打烊",
+    "store_not_found": "未找到店",
+    "items_not_found": "未找到菜",
+    "no_delivery": "单点不配送",
+    "unsupported": "平台·场景不支持",
+}
+
+
+def derive_admin_outcome(raw_payload: dict | None, status: str) -> tuple[str, str | None]:
+    """(admin_status, outcome_hint)。列表 Python 层派生（raw_payload 已随 ORM 加载）。"""
+    if status in ("cancelled", "running"):
+        return status, None
+    raw = raw_payload or {}
+    # 原始结局：优先 raw.record_status，其次 raw.status，兜底 status 列
+    # （兼容迁移未覆盖、细分值残留在 status 列的老记录；与下方 SQL 口径一致）。
+    original = raw.get("record_status") or raw.get("status") or status
+    if original in ADMIN_SUCCESS_OUTCOMES:
+        return "success", OUTCOME_HINTS.get(original)
+    return "failed", None
+
+
+def _original_expr():
+    """SQL：原始结局 = coalesce(nullif(raw.record_status,''), nullif(raw.status,''), status 列)。
+    nullif('') 让空串与 Python `or` 口径一致地跳过（as_string 跨方言，#209 迁移已验证）。"""
+    return func.coalesce(
+        func.nullif(ComparisonRecord.raw_payload["record_status"].as_string(), ""),
+        func.nullif(ComparisonRecord.raw_payload["status"].as_string(), ""),
+        ComparisonRecord.status,
+    )
+
+
+def admin_success_sql():
+    """SQL 层 admin 成功判定（概览 / 大盘的 case / where 共用）。
+
+    排除 cancelled / running（生命周期态，不看结局）；其余按原始结局 ∈ S。
+    coalesce 兜底 status 列 → original 永非 NULL、且兼容 status 列残留的细分值。
+    """
+    return ComparisonRecord.status.notin_(("cancelled", "running")) & _original_expr().in_(
+        tuple(ADMIN_SUCCESS_OUTCOMES)
+    )

app/admin/repositories/queries.py

@@ -5,13 +5,14 @@
 """
 from __future__ import annotations
 
-from datetime import date, datetime, time, timedelta, timezone
+from datetime import UTC, date, datetime, time, timedelta
 from decimal import ROUND_HALF_UP, Decimal
 from zoneinfo import ZoneInfo
 
 from sqlalchemy import Select, and_, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session
 
+from app.admin.repositories.comparison_outcome import admin_success_sql, derive_admin_outcome
 from app.core import rewards
 from app.core.config import settings
 from app.models.ad_ecpm import AdEcpmRecord
@@ -39,33 +40,15 @@ from app.repositories import activity, ad_ecpm
 # 折算成可提现现金时,非广告金币来源的排除集(广告单独统计、人工调整不算"赚取")
 _NON_TASK_BIZ_TYPES = ("reward_video", "feed_ad_reward", "admin_grant", "admin_deduct")
 
-# comparison_record historically persisted a few granular business outcomes as
-# top-level statuses. Admin filters and metrics expose lifecycle buckets while
-# retaining the raw values until the data migration has run everywhere.
-_COMPARISON_STATUS_ALIASES = {
-    "success": ("success", "below_minimum"),
-    "failed": (
-        "failed",
-        "store_closed",
-        "store_not_found",
-        "items_not_found",
-        "no_delivery",
-        "unsupported",
-    ),
-    "cancelled": ("cancelled",),
-    "running": ("running",),
-}
-_COMPARISON_SUCCESS_STATUSES = _COMPARISON_STATUS_ALIASES["success"]
-_COMPARISON_FAILED_STATUSES = _COMPARISON_STATUS_ALIASES["failed"]
-_COMPARISON_COMPLETED_STATUSES = (
-    *_COMPARISON_SUCCESS_STATUSES,
-    *_COMPARISON_FAILED_STATUSES,
-)
-
 
 def _comparison_status_condition(status: str):
-    values = _COMPARISON_STATUS_ALIASES.get(status, (status,))
-    return ComparisonRecord.status.in_(values)
+    """列表/概览「状态」筛选：success/failed 按 admin 口径（与显示/统计一致）；
+    cancelled/running 按 status 列生命周期。"""
+    if status == "success":
+        return admin_success_sql()
+    if status == "failed":
+        return ~admin_success_sql() & ComparisonRecord.status.notin_(("cancelled", "running"))
+    return ComparisonRecord.status == status
 
 # 信息流点位场景 → 金币记录「赚取途径」展示名;NULL/未知 = 历史未分类。
 _FEED_SCENE_LABEL = {
@@ -371,10 +354,10 @@ def _comparison_conditions(
         conditions.append(ComparisonRecord.product_names.like(f"%{product}%"))
     beijing = ZoneInfo("Asia/Shanghai")
     if date_from is not None:
-        start_utc = datetime.combine(date_from, time.min, tzinfo=beijing).astimezone(timezone.utc)
+        start_utc = datetime.combine(date_from, time.min, tzinfo=beijing).astimezone(UTC)
         conditions.append(ComparisonRecord.created_at >= start_utc)
     if date_to is not None:
-        end_utc = datetime.combine(date_to + timedelta(days=1), time.min, tzinfo=beijing).astimezone(timezone.utc)
+        end_utc = datetime.combine(date_to + timedelta(days=1), time.min, tzinfo=beijing).astimezone(UTC)
         conditions.append(ComparisonRecord.created_at < end_utc)
     return conditions
 
@@ -414,6 +397,7 @@ def list_comparison_records(
     rev = ad_ecpm.revenue_yuan_by_trace(db, [it.trace_id for it in items])
     for it in items:
         it.ad_revenue_yuan = rev.get(it.trace_id, 0.0)
+        it.admin_status, it.outcome_hint = derive_admin_outcome(it.raw_payload, it.status)
     return items, next_cursor, total
 
 
@@ -437,8 +421,8 @@ def _round_duration_ms(value) -> int | None:
     return int(Decimal(str(value)).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
 
 
-def _comparison_duration_aggregate_stmt(conditions: list, status: str, quantiles: tuple[float, ...]):
-    """PostgreSQL 耗时聚合语句；每种状态只返回一行。"""
+def _comparison_duration_aggregate_stmt(conditions: list, status_filter, quantiles: tuple[float, ...]):
+    """PostgreSQL 耗时聚合语句；status_filter 为已构造的口径条件表达式，每口径只返回一行。"""
     return select(
         func.avg(ComparisonRecord.total_ms),
         *(
@@ -447,7 +431,7 @@ def _comparison_duration_aggregate_stmt(conditions: list, status: str, quantiles
         ),
     ).where(
         *conditions,
-        _comparison_status_condition(status),
+        status_filter,
         ComparisonRecord.total_ms.is_not(None),
     )
 
@@ -456,13 +440,13 @@ def _comparison_duration_aggregates(
     db: Session,
     *,
     conditions: list,
-    status: str,
+    status_filter,
     quantiles: tuple[float, ...],
 ) -> list[int | None]:
     """返回平均值和各分位数；生产 PG 在数据库内聚合，SQLite 仅作测试回退。"""
     if db.bind is not None and db.bind.dialect.name == "postgresql":
         row = db.execute(
-            _comparison_duration_aggregate_stmt(conditions, status, quantiles)
+            _comparison_duration_aggregate_stmt(conditions, status_filter, quantiles)
         ).one()
         return [_round_duration_ms(value) for value in row]
 
@@ -472,7 +456,7 @@ def _comparison_duration_aggregates(
             select(ComparisonRecord.total_ms)
             .where(
                 *conditions,
-                _comparison_status_condition(status),
+                status_filter,
                 ComparisonRecord.total_ms.is_not(None),
             )
             .order_by(ComparisonRecord.total_ms)
@@ -499,16 +483,14 @@ def comparison_records_summary(
         user_id=user_id, phone=phone, status=status, business_type=business_type,
         store=store, product=product, date_from=date_from, date_to=date_to,
     )
+    _success = admin_success_sql()
     row = db.execute(
         select(
             func.count(ComparisonRecord.id),
-            func.sum(case((ComparisonRecord.status.in_(_COMPARISON_COMPLETED_STATUSES), 1), else_=0)),
-            func.sum(case((ComparisonRecord.status.in_(_COMPARISON_SUCCESS_STATUSES), 1), else_=0)),
+            func.sum(case((_success | (ComparisonRecord.status == "failed"), 1), else_=0)),
+            func.sum(case((_success, 1), else_=0)),
             func.avg(ComparisonRecord.llm_cost_yuan),
-            func.sum(case((
-                ComparisonRecord.status.in_(_COMPARISON_SUCCESS_STATUSES)
-                & (ComparisonRecord.saved_amount_cents > 0), 1
-            ), else_=0)),
+            func.sum(case((_success & (ComparisonRecord.saved_amount_cents > 0), 1), else_=0)),
             func.sum(case((ComparisonRecord.status == "cancelled", 1), else_=0)),
         ).where(*conditions)
     ).one()
@@ -520,13 +502,13 @@ def comparison_records_summary(
     success_duration_stats = _comparison_duration_aggregates(
         db,
         conditions=conditions,
-        status="success",
+        status_filter=admin_success_sql(),
         quantiles=(0.05, 0.5, 0.95, 0.99),
     )
     cancelled_duration_stats = _comparison_duration_aggregates(
         db,
         conditions=conditions,
-        status="cancelled",
+        status_filter=(ComparisonRecord.status == "cancelled"),
         quantiles=(0.05, 0.5, 0.95),
     )
     success_rate_denominator = started - cancelled
@@ -551,12 +533,13 @@ def comparison_records_summary(
 
 
 def get_comparison_record(db: Session, record_id: int) -> ComparisonRecord | None:
-    """admin 取单条比价记录(任意用户,不限本人;附 phone/nickname 瞬态)。"""
+    """admin 取单条比价记录(任意用户,不限本人;附 phone/nickname + admin 口径瞬态)。"""
     rec = db.get(ComparisonRecord, record_id)
     if rec is not None:
         _attach_user_info(db, [rec])
         _attach_comparison_order_status(db, [rec])
         _attach_comparison_device_details([rec])
+        rec.admin_status, rec.outcome_hint = derive_admin_outcome(rec.raw_payload, rec.status)
     return rec
 
 
@@ -585,8 +568,8 @@ def _heartbeat_seconds_ago(last: datetime | None) -> int | None:
     if last is None:
         return None
     if last.tzinfo is None:
-        last = last.replace(tzinfo=timezone.utc)
-    return int((datetime.now(timezone.utc) - last).total_seconds())
+        last = last.replace(tzinfo=UTC)
+    return int((datetime.now(UTC) - last).total_seconds())
 
 
 def _device_model_from_id(device_id: str) -> str:
@@ -639,7 +622,7 @@ def _attach_device_user_info(db: Session, devices: list[DeviceLiveness]) -> None
 def _liveness_cutoff() -> datetime:
     """掉线判定分界:此刻 - HEARTBEAT_TIMEOUT_MINUTES。心跳早于它 = 掉线(同 list_overdue 口径)。"""
     timeout_min = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES))
-    return datetime.now(timezone.utc) - timedelta(minutes=timeout_min)
+    return datetime.now(UTC) - timedelta(minutes=timeout_min)
 
 
 def list_device_liveness(
@@ -815,11 +798,11 @@ def list_all_withdraw_orders(
         stmt = stmt.where(date_col <= _as_utc(date_to))
 
     # tz-aware:列为 timestamptz,比较绝对时刻、与 DB 会话时区无关(同 _as_utc / stats.py)
-    now = datetime.now(timezone.utc)
+    now = datetime.now(UTC)
     today_start = (
         datetime.now(ZoneInfo("Asia/Shanghai"))
         .replace(hour=0, minute=0, second=0, microsecond=0)
-        .astimezone(timezone.utc)
+        .astimezone(UTC)
     )
     if quick_filter == "abnormal":
         stmt = stmt.where(
@@ -866,8 +849,8 @@ def _as_utc(value: datetime) -> datetime:
     生产会话非 UTC 时筛选边界会整体偏移——故统一 tz-aware(与 stats.py / withdraw_summary 一致)。
     无时区入参按 UTC 解释。"""
     if value.tzinfo is None:
-        return value.replace(tzinfo=timezone.utc)
-    return value.astimezone(timezone.utc)
+        return value.replace(tzinfo=UTC)
+    return value.astimezone(UTC)
 
 
 def withdraw_list_enrichment(
@@ -1032,7 +1015,7 @@ def withdraw_summary(db: Session, *, source: str | None = None) -> dict:
     today_start = (
         datetime.now(ZoneInfo("Asia/Shanghai"))
         .replace(hour=0, minute=0, second=0, microsecond=0)
-        .astimezone(timezone.utc)
+        .astimezone(UTC)
     )
 
     def _today_count(status: str) -> int:
@@ -1207,8 +1190,8 @@ def withdraw_risk_flags(
     if user and user.status != "active":
         flags.append(f"账号状态:{user.status}")
     if user and user.created_at:
-        created_at = user.created_at.replace(tzinfo=timezone.utc) if user.created_at.tzinfo is None else user.created_at
-        if datetime.now(timezone.utc) - created_at < timedelta(hours=24):
+        created_at = user.created_at.replace(tzinfo=UTC) if user.created_at.tzinfo is None else user.created_at
+        if datetime.now(UTC) - created_at < timedelta(hours=24):
             flags.append("新注册用户")
     # 历史异常提现拆「拒绝」「失败」两类(口径不同:拒绝=人工驳回退款,失败=打款失败退款)
     rejected_n = sum(1 for item in recent_withdraws if item.status == "rejected")
@@ -1419,7 +1402,7 @@ def _cn_wall_to_utc(dt: datetime) -> datetime:
     """coin_transaction 存的是北京 wall-clock(naive,见 wallet.grant_coins「存北京 wall-clock」),转成 UTC naive,
     与广告表(func.now() UTC)统一 —— 让本函数按同一绝对时刻排序、且前端 apiTime(把无时区时间当 UTC 再 +8 展示)
     口径一致;否则签到会比实际多显示 8 小时(北京时间又被 +8)。"""
-    return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)
+    return dt.replace(tzinfo=rewards.CN_TZ).astimezone(UTC).replace(tzinfo=None)
 
 
 def _coin_record_sort_key(row: dict) -> datetime:

app/admin/repositories/stats.py

@@ -12,6 +12,7 @@ from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
 from sqlalchemy import case, func, or_, select
 from sqlalchemy.orm import Session
 
+from app.admin.repositories.comparison_outcome import admin_success_sql
 from app.admin.repositories.coupon_data import _percentile
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
@@ -257,13 +258,14 @@ def dashboard_overview(
         ComparisonRecord.created_at >= start_local,
         ComparisonRecord.created_at < end_local,
     )
+    _period_success = admin_success_sql()
     period_comparison_stats = db.execute(
         select(
             func.count(ComparisonRecord.id),
             func.coalesce(
                 func.sum(
                     case(
-                        (ComparisonRecord.status.in_(("success", "failed")), 1),
+                        (_period_success | (ComparisonRecord.status == "failed"), 1),
                         else_=0,
                     )
                 ),
@@ -276,7 +278,7 @@ def dashboard_overview(
                 0,
             ),
             func.coalesce(
-                func.sum(case((ComparisonRecord.status == "success", 1), else_=0)),
+                func.sum(case((_period_success, 1), else_=0)),
                 0,
             ),
             func.coalesce(func.sum(ComparisonRecord.llm_cost_yuan), 0.0),

app/admin/schemas/comparison.py

@@ -22,6 +22,10 @@ class AdminComparisonListItem(BaseModel):
     # admin 是 debug 工具,无条件下发 trace_url(不看 user.debug_trace_enabled)
     trace_url: str | None = None
     status: str  # success / failed / cancelled / running;旧细分值由前端兼容映射
+    # admin 展示口径(见 repositories/comparison_outcome):成功含「跑完但外部缺失」,
+    # 纯技术故障才 failed;outcome_hint 非空=有缺失,前端标感叹号。
+    admin_status: str = "success"
+    outcome_hint: str | None = None
     information: str | None = None
     store_name: str | None = None
     product_names: str | None = None   # 下单商品名派生串(顿号分隔;「商品」列展示 + 商品搜索)

app/api/v1/compare_record.py

@@ -37,6 +37,8 @@ logger = logging.getLogger("shagua.compare_record")
 
 router = APIRouter(prefix="/api/v1/compare", tags=["compare-record"])
 
+COMPARE_DAILY_LIMIT_MESSAGE = "今日比价额度用完啦，明天再来吧～"
+
 
 @router.post(
     "/start",
@@ -78,11 +80,7 @@ def reserve_compare_start(
     except crud_compare.DailyCompareStartLimitExceeded:
         raise HTTPException(
             status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-            detail=(
-                f"今日已比价超过{policy.limit}次，请明天再试"
-                if policy.limit is not None
-                else "今日比价次数已达上限，请明天再试"
-            ),
+            detail=COMPARE_DAILY_LIMIT_MESSAGE,
         ) from None
     except crud_compare.ComparisonTraceOwnershipError:
         raise HTTPException(

app/core/compare_alert_worker.py

@@ -0,0 +1,319 @@
+"""比价失败报警后台任务:周期扫 comparison_record 新落定记录 → 规则命中 → 飞书汇总。
+
+结构仿 heartbeat_monitor_worker(单实例文件锁 + asyncio 轮询 + 优雅退出);发送与 DB 全同步,
+放 asyncio.to_thread。水位存 app_config(key=compare_alert.last_watermark,值=上次处理的最大
+updated_at ISO 串),查询用 updated_at 自身比较、规避时区。见 spec 第 5/6 节。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import os
+import time
+from collections.abc import Iterator
+from dataclasses import replace as _dc_replace
+from datetime import datetime
+from pathlib import Path
+
+from sqlalchemy import func, select
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.core.config import settings
+from app.core.rewards import CN_TZ
+from app.db.session import SessionLocal
+from app.integrations import feishu_notifier
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.models.user import User
+from app.services import trace_stuck
+from app.services.compare_alert import (
+    AlertHit,
+    classify_cancelled_fallback,
+    classify_record,
+    make_hit,
+)
+from app.services.compare_alert_format import format_alert_card
+
+logger = logging.getLogger("shagua.compare_alert")
+
+WATERMARK_KEY = "compare_alert.last_watermark"
+_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "compare_alert.lock"
+
+
+def _read_watermark(db) -> datetime | None:
+    row = db.get(AppConfig, WATERMARK_KEY)
+    if row is None or not row.value:
+        return None
+    try:
+        return datetime.fromisoformat(row.value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _write_watermark(db, value: datetime) -> None:
+    iso = value.isoformat()
+    row = db.get(AppConfig, WATERMARK_KEY)
+    if row is None:
+        db.add(AppConfig(key=WATERMARK_KEY, value=iso, updated_by_admin_id=None))
+    else:
+        row.value = iso
+    db.commit()
+
+
+def _send_card(card: dict) -> None:
+    """发飞书交互卡片(webhook 为空则只打日志、不外发)。失败抛 FeishuNotifyError 由调用方处理。"""
+    webhook = settings.COMPARE_ALERT_FEISHU_WEBHOOK
+    if not webhook:
+        title = card.get("header", {}).get("title", {}).get("content", "")
+        logger.info("[compare-alert] webhook 未配置,仅打印: title=%s", title)
+        return
+    feishu_notifier.send_feishu_card(
+        webhook, card, timeout=settings.COMPARE_ALERT_FEISHU_TIMEOUT_SEC
+    )
+
+
+def _trace_dir(base: Path, trace_url: str | None) -> Path | None:
+    """URL → trace 目录 Path;trace_url 缺失/无法解析 → None。"""
+    name = trace_stuck.dir_name_from_trace_url(trace_url)
+    if not name:
+        return None
+    return base / name
+
+
+def _fmt_stuck(sp: trace_stuck.StuckPoint) -> 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,
+    *,
+    work_log_dir: str,
+    stuck_threshold: int,
+    max_tail: int,
+    max_trace_reads: int,
+    cancelled_ms_threshold: int,
+    cancelled_step_threshold: int,
+    timeout_keywords: tuple[str, ...],
+    unrecognized_keywords: tuple[str, ...],
+    biz_exclude_keywords: tuple[str, ...],
+) -> list[AlertHit]:
+    """编排:cancelled 先读 trace,判出原地卡点就带卡点报 T5,否则(可读没卡点/读不到)一律回退
+    耗时/步数兜底——超长放弃照报(卡点列留空);failed 命中后附末段卡点。
+
+    trace 读取限量 max_trace_reads 次/轮;任何 trace 异常都在 trace_stuck 内部降级为「读不到」,
+    cancelled 因而走耗时兜底、failed 不附卡点,绝不影响报警发送。trace 只做「锦上添花」标注卡点,
+    绝不因「可读但没判出卡点」把超长放弃吞掉(线上 trace 几乎总可读,否则 total_ms 阈值形同虚设)。
+    """
+    base = Path(work_log_dir) if work_log_dir else None
+    reads = 0
+    hits: list = []
+    for rec in records:
+        if rec.status == "cancelled":
+            res = None
+            if base is not None and reads < max_trace_reads:
+                td = _trace_dir(base, rec.trace_url)
+                if td is not None:
+                    res = trace_stuck.read_stuck_points(
+                        td, threshold=stuck_threshold, max_tail=max_tail
+                    )
+                    reads += 1
+            if res is not None and res.readable and res.points:
+                stuck = "、".join(_fmt_stuck(sp) for sp in res.points)
+                hit = _dc_replace(make_hit(rec, "T5", "深度放弃"), stuck_point=stuck)
+            else:
+                # trace 判出卡点 → 上面带卡点报。其余一律回退耗时/步数兜底:可读但没判出卡点、
+                # 读不到、无 trace,都过 total_ms/step 阈值——超长放弃照报,不再因「trace 可读但不
+                # 原地卡」把超长放弃整条吞掉(线上 trace 几乎总可读,否则耗时阈值形同虚设)。
+                hit = classify_cancelled_fallback(
+                    rec,
+                    cancelled_ms_threshold=cancelled_ms_threshold,
+                    cancelled_step_threshold=cancelled_step_threshold,
+                )
+                # 没卡点但命中(超阈值放弃)→ 用末帧标「退出前在哪屏」(平台·环节·页面)
+                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())
+            if hit is not None:
+                hits.append(hit)
+        else:
+            hit = classify_record(
+                rec,
+                cancelled_ms_threshold=cancelled_ms_threshold,
+                cancelled_step_threshold=cancelled_step_threshold,
+                timeout_keywords=timeout_keywords,
+                unrecognized_keywords=unrecognized_keywords,
+                biz_exclude_keywords=biz_exclude_keywords,
+            )
+            if (
+                hit is not None
+                and hit.alert_type in ("T1", "T2", "T6")
+                and base is not None
+                and reads < max_trace_reads
+            ):
+                td = _trace_dir(base, rec.trace_url)
+                if td is not None:
+                    sp = trace_stuck.last_step(td)
+                    reads += 1
+                    if sp is not None:
+                        # failed 是「末帧停在哪」:last_step 返回的 frames 是该平台总帧数(非"卡住"帧数)、
+                        # stuck_ms=None,带上帧数/时长会误导,故只用 label() 显示环节
+                        hit = _dc_replace(hit, stuck_point=sp.label())
+            if hit is not None:
+                hits.append(hit)
+    return hits
+
+
+def _scan_and_alert() -> None:
+    """一轮:读水位 → 查有更新记录 → 规则 → 有命中发飞书 → 成功推进水位。同步,放 to_thread 调。"""
+    with SessionLocal() as db:
+        watermark = _read_watermark(db)
+        if watermark is None:
+            # 冷启动:水位 = 当前 max(updated_at),不回溯历史失败。空表则本轮不建水位、下轮再说——
+            # 不用 datetime.now():那是本地时钟,与 SQLite 的 updated_at(UTC CURRENT_TIMESTAMP)
+            # 不同源、会差 8h,导致新记录永远追不上水位。只用 DB 产出的 updated_at 值。
+            max_updated = db.scalar(select(func.max(ComparisonRecord.updated_at)))
+            if max_updated is None:
+                logger.info("[compare-alert] 冷启动:表空,待有记录后再建水位")
+                return
+            _write_watermark(db, max_updated)
+            logger.info("[compare-alert] 冷启动,水位初始化=%s", max_updated)
+            return
+
+        records = list(
+            db.scalars(
+                select(ComparisonRecord)
+                .where(ComparisonRecord.updated_at > watermark)
+                .order_by(ComparisonRecord.updated_at.asc())
+            )
+        )
+        if not records:
+            return
+        batch_max = max(r.updated_at for r in records)
+
+        hits = build_hits(
+            records,
+            work_log_dir=settings.COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR,
+            stuck_threshold=settings.COMPARE_ALERT_STUCK_FRAME_THRESHOLD,
+            max_tail=settings.COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES,
+            max_trace_reads=settings.COMPARE_ALERT_TRACE_MAX_RECORDS,
+            cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
+            cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
+            timeout_keywords=settings.compare_alert_timeout_keywords,
+            unrecognized_keywords=settings.compare_alert_unrecognized_keywords,
+            biz_exclude_keywords=settings.compare_alert_biz_exclude_keywords,
+        )
+
+        if hits or settings.COMPARE_ALERT_SEND_EMPTY:
+            label = datetime.now(CN_TZ).strftime("%Y-%m-%d %H:%M")
+            interval_min = max(1, settings.COMPARE_ALERT_SCAN_INTERVAL_SEC // 60)
+
+            if hits:
+                # join User 取手机号
+                uids = {h.user_id for h in hits if h.user_id is not None}
+                if uids:
+                    users = db.scalars(select(User).where(User.id.in_(uids)))
+                    phone_map = {u.id: u.phone for u in users}
+                else:
+                    phone_map = {}
+            else:
+                phone_map = {}
+
+            card = format_alert_card(
+                hits,
+                window_label=label,
+                phone_map=phone_map,
+                interval_min=interval_min,
+                max_detail_per_type=settings.COMPARE_ALERT_MAX_DETAIL_PER_TYPE,
+                max_total=settings.COMPARE_ALERT_MAX_TOTAL,
+                cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
+                cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
+            )
+
+            try:
+                _send_card(card)
+            except feishu_notifier.FeishuNotifyError:
+                logger.warning("[compare-alert] 发送失败,水位不推进、下轮补发", exc_info=True)
+                return  # 不推进水位
+
+        _write_watermark(db, batch_max)
+        if hits:
+            logger.info("[compare-alert] 本轮命中 %d 条,水位推进到 %s", len(hits), batch_max)
+
+
+# ---- 单实例锁 + 轮询循环(结构同 heartbeat_monitor_worker)---------------------
+def _touch_lock() -> None:
+    with contextlib.suppress(FileNotFoundError):
+        os.utime(_LOCK_PATH, None)
+
+
+@contextlib.contextmanager
+def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
+    _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd: int | None = None
+    try:
+        try:
+            fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+        except FileExistsError:
+            try:
+                age = time.time() - _LOCK_PATH.stat().st_mtime
+            except FileNotFoundError:
+                age = stale_after_sec + 1
+            if age > stale_after_sec:
+                with contextlib.suppress(FileNotFoundError):
+                    _LOCK_PATH.unlink()
+                try:
+                    fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+                except FileExistsError:
+                    fd = None
+        if fd is None:
+            yield False
+            return
+        os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
+        yield True
+    finally:
+        if fd is not None:
+            os.close(fd)
+            with contextlib.suppress(FileNotFoundError):
+                _LOCK_PATH.unlink()
+
+
+async def _run_loop() -> None:
+    interval = max(60, int(settings.COMPARE_ALERT_SCAN_INTERVAL_SEC))
+    lock_stale_after = max(interval * 3, 600)
+    with _single_instance_lock(lock_stale_after) as lock_acquired:
+        if not lock_acquired:
+            logger.warning("compare-alert skipped: another worker owns lock")
+            return
+        logger.info("compare-alert worker started interval=%ss", interval)
+        try:
+            while True:
+                try:
+                    _touch_lock()
+                    await asyncio.to_thread(_scan_and_alert)
+                except SQLAlchemyError:
+                    logger.exception("compare-alert db error")
+                except Exception:  # noqa: BLE001 - 后台任务不因单次异常退出
+                    logger.exception("compare-alert unexpected error")
+                await asyncio.sleep(interval)
+        except asyncio.CancelledError:
+            logger.info("compare-alert worker stopped")
+            raise
+
+
+def start_compare_alert_worker() -> asyncio.Task | None:
+    if not settings.COMPARE_ALERT_ENABLED:
+        logger.info("compare-alert worker disabled")
+        return None
+    return asyncio.create_task(_run_loop(), name="compare-alert-worker")
+
+
+async def stop_compare_alert_worker(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task

app/core/config.py

@@ -132,6 +132,26 @@ class Settings(BaseSettings):
     HEARTBEAT_TIMEOUT_MINUTES: int = 60          # 多久没心跳算掉线(1 小时,避免短暂离线误判被杀)
     HEARTBEAT_SCAN_INTERVAL_SEC: int = 60        # 扫描周期
 
+    # ===== 比价失败报警(常驻 worker 周期扫 comparison_record → 飞书汇总)=====
+    COMPARE_ALERT_ENABLED: bool = False               # 总开关(默认关;启用用 .env COMPARE_ALERT_ENABLED=true 覆盖,别改这默认值);关时 worker 不启动
+    COMPARE_ALERT_SCAN_INTERVAL_SEC: int = 900        # 扫描间隔(默认 15min,可配 1800=30min)
+    COMPARE_ALERT_FEISHU_WEBHOOK: str = ""             # 群机器人 webhook(敏感,放 .env 别硬编码进代码);空则 worker 仅打日志不外发
+    COMPARE_ALERT_FEISHU_TIMEOUT_SEC: float = 10.0     # 飞书 POST 读/连超时
+    COMPARE_ALERT_CANCELLED_MS_THRESHOLD: int = 90000  # T5 耗时阈值(ms)
+    COMPARE_ALERT_CANCELLED_STEP_THRESHOLD: int = 30   # T5 步数阈值
+    COMPARE_ALERT_TIMEOUT_KEYWORDS: str = "超时,启动,加载"       # T2 关键词(逗号分隔)
+    COMPARE_ALERT_UNRECOGNIZED_KEYWORDS: str = "未识别"          # T6 关键词
+    COMPARE_ALERT_BIZ_EXCLUDE_KEYWORDS: str = "未找到,打烊,起送,门店,店内,不配送,这些菜,未入驻,休息"  # T1 information 业务词排除
+    COMPARE_ALERT_MAX_DETAIL_PER_TYPE: int = 20        # 单类型明细截断
+    COMPARE_ALERT_MAX_TOTAL: int = 50                  # 本期总命中截断(超则只给计数)
+    COMPARE_ALERT_SEND_EMPTY: bool = False             # 无命中是否发「本期无异常」简讯
+
+    # ===== 卡死定位(读 pricebot trace 末段判原地打转)=====
+    COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR: str = ""      # pricebot work_logs 绝对路径(敏感,放 .env);空=跳过 trace、cancelled 全走保底
+    COMPARE_ALERT_STUCK_FRAME_THRESHOLD: int = 15      # 末段连续同环节达此帧数判卡死
+    COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES: int = 40      # 每平台最多往前读多少帧
+    COMPARE_ALERT_TRACE_MAX_RECORDS: int = 30          # 每轮最多对多少条命中记录读 trace(限量)
+
     # ===== 短信 =====
     SMS_MOCK: bool = True
     SMS_CODE_TTL_SEC: int = 300
@@ -217,6 +237,22 @@ class Settings(BaseSettings):
             phones.add(self.test_account_phone)
         return frozenset(phones)
 
+    def _csv(self, raw: str) -> tuple[str, ...]:
+        """逗号分隔字符串 → 去空白非空元组(报警关键词解析共用)。"""
+        return tuple(w.strip() for w in raw.split(",") if w.strip())
+
+    @property
+    def compare_alert_timeout_keywords(self) -> tuple[str, ...]:
+        return self._csv(self.COMPARE_ALERT_TIMEOUT_KEYWORDS)
+
+    @property
+    def compare_alert_unrecognized_keywords(self) -> tuple[str, ...]:
+        return self._csv(self.COMPARE_ALERT_UNRECOGNIZED_KEYWORDS)
+
+    @property
+    def compare_alert_biz_exclude_keywords(self) -> tuple[str, ...]:
+        return self._csv(self.COMPARE_ALERT_BIZ_EXCLUDE_KEYWORDS)
+
     # ===== 美团联盟 CPS =====
     # 未配置时所有 /api/v1/meituan/* 接口 200 返空(优雅降级),不影响登录/领券等其他业务。
     MT_CPS_APP_KEY: str = ""

app/core/rewards.py

@@ -58,7 +58,7 @@ WITHDRAW_MAX_CENTS: int = 5_000_000  # 5 万元
 # 规则(2026-07-09 拍板):
 #   - 新人档(is_newbie):账号历史一次性,"发起就算用过"(任意状态含被拒),用过即不再下发;
 #     0.1 与 0.3 各自独立同天可各提一次,且不参与常规档"每日选一个额度"互斥。
-#   - 常规档:按北京日计次(0.5×3 / 10×1 / 20×1),三档每天只能选一个。
+#   - 常规档:按北京日计次(0.5×3 / 10×1 / 20×1 / 100×1),四档每天只能选一个。
 # invite_cash(邀请页)本轮无档位概念,不在此表。改档位=改这里发版。
 class WithdrawTier(NamedTuple):
     amount_cents: int
@@ -74,6 +74,7 @@ WITHDRAW_TIERS_COIN_CASH: tuple[WithdrawTier, ...] = (
     WithdrawTier(50, "0.5", None, 3, False),
     WithdrawTier(1000, "10", None, 1, False),
     WithdrawTier(2000, "20", None, 1, False),
+    WithdrawTier(10000, "100", None, 1, False),
 )
 
 

app/integrations/feishu_notifier.py

@@ -0,0 +1,48 @@
+"""飞书群自定义机器人发送(text / post 消息)。
+
+自定义机器人「关键词」验证:消息 content.text 必须含机器人配置的关键词,否则飞书返回 code!=0
+(如 19024 Key Words Not Found)——本项目消息由 compare_alert_format 生成,标题已含「比价失败报警」。
+不需要签名(sign)/IP 白名单。文档:https://open.feishu.cn/document/client-docs/bot-v3/add-custom-bot
+"""
+from __future__ import annotations
+
+import httpx
+
+
+class FeishuNotifyError(Exception):
+    """飞书发送失败(网络错误 / 非 2xx / 业务 code!=0,含关键词不匹配)。"""
+
+
+def _post_feishu(webhook_url: str, payload: dict, timeout: float) -> None:
+    """内部 helper:POST payload 到飞书 webhook 并校验响应。失败抛 FeishuNotifyError。"""
+    try:
+        resp = httpx.post(webhook_url, json=payload, timeout=timeout)
+    except httpx.HTTPError as e:
+        raise FeishuNotifyError(f"feishu request failed: {e}") from e
+    if resp.status_code >= 300:
+        raise FeishuNotifyError(f"feishu http {resp.status_code}: {resp.text[:200]}")
+    try:
+        data = resp.json()
+    except ValueError as e:
+        raise FeishuNotifyError(f"feishu bad json: {resp.text[:200]}") from e
+    code = data.get("code", data.get("StatusCode", 0))
+    if code not in (0, None):
+        raise FeishuNotifyError(f"feishu code={code} msg={data.get('msg') or data.get('StatusMessage')}")
+
+
+def send_feishu_text(webhook_url: str, text: str, *, timeout: float = 10.0) -> None:
+    """POST 一条 text 消息到飞书群机器人 webhook。失败(网络/HTTP/业务 code)抛 FeishuNotifyError。"""
+    payload = {"msg_type": "text", "content": {"text": text}}
+    _post_feishu(webhook_url, payload, timeout)
+
+
+def send_feishu_post(webhook_url: str, title: str, content: list, *, timeout: float = 10.0) -> None:
+    """发飞书富文本(post)。content 是段落数组,每段是元素数组[{tag:text/a,...}]。失败抛 FeishuNotifyError。"""
+    payload = {"msg_type": "post", "content": {"post": {"zh_cn": {"title": title, "content": content}}}}
+    _post_feishu(webhook_url, payload, timeout)
+
+
+def send_feishu_card(webhook_url: str, card: dict, *, timeout: float = 10.0) -> None:
+    """发飞书交互卡片(interactive)。card 为 schema 2.0 卡片 dict。失败抛 FeishuNotifyError。"""
+    payload = {"msg_type": "interactive", "card": card}
+    _post_feishu(webhook_url, payload, timeout)

app/main.py

@@ -45,6 +45,10 @@ from app.api.v1.user import router as user_router
 from app.api.v1.wallet import router as wallet_router
 from app.api.v1.wxpay import router as wxpay_router
 from app.core import media
+from app.core.compare_alert_worker import (
+    start_compare_alert_worker,
+    stop_compare_alert_worker,
+)
 from app.core.config import settings
 from app.core.cps_reconcile_worker import (
     start_cps_reconcile_worker,
@@ -123,6 +127,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     observe_task = start_observe_worker()
     inactivity_task = start_inactivity_reset_worker()
     llm_cost_backfill_task = start_llm_cost_backfill_worker()
+    compare_alert_task = start_compare_alert_worker()
     try:
         yield
     finally:
@@ -133,6 +138,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         await stop_observe_worker(observe_task)
         await stop_inactivity_reset_worker(inactivity_task)
         await stop_llm_cost_backfill_worker(llm_cost_backfill_task)
+        await stop_compare_alert_worker(compare_alert_task)
         await aclose_pricebot_client()
         mt_meituan.close_client()
         logger.info("shutting down")

app/models/comparison.py

@@ -49,6 +49,9 @@ class ComparisonRecord(Base):
         # 单列 user_id 索引只能过滤,排序仍要把该用户全部记录取出来排一遍;这条复合索引的**反向扫**
         # 恰好等于 (created_at DESC, id DESC),PG 直接取前 n 条、免排序。列序不能动。
         Index("ix_comparison_user_created", "user_id", "created_at", "id"),
+        # 比价报警 worker 的水位查询 WHERE updated_at > watermark 走它。显式命名(不用列上
+        # index=True 的自动名 ix_comparison_record_updated_at)以与迁移 create_index 同名、免 autogenerate 漂移。
+        Index("ix_comparison_updated", "updated_at"),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -160,6 +163,15 @@ class ComparisonRecord(Base):
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
     )
+    # 记录任一次更新(建 running 行 / done / abort 落终态)的 DB 时钟时间。比价报警 worker 的
+    # 水位列:按 updated_at 单调推进扫描,任何记录落定/更新都刷新它 > 水位、必被下轮扫到,
+    # 根治 created_at 水位漏掉「慢失败」(落定延迟 p99 达 7-10min)。onupdate 在 ORM UPDATE 时自动刷新。
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
 
     def __repr__(self) -> str:  # pragma: no cover
         return (

app/repositories/wallet.py

@@ -735,8 +735,8 @@ def withdraw_tier_states(db: Session, user_id: int, source: str = "coin_cash") -
       - 新人档(0.1/0.3):账号历史一次性——进行中(reviewing/pending)或成功打款(success)即视为
         已用,直接**从返回列表消失**;被拒/转账失败/解绑退回(均已退款、钱没到手)则恢复可提,不永久
         占用资格。两档各自独立互不影响,不参与"每日选一个额度"互斥。
-      - 常规档(0.5×3 / 10×1 / 20×1):按北京日计次,"发起就算占用"(当天创建的单不论最终状态
-        都计入,被拒/失败不退当天名额);三档每天只能选一个,选定后其余两档当天 other_tier_selected。
+      - 常规档(0.5×3 / 10×1 / 20×1 / 100×1):按北京日计次,"发起就算占用"(当天创建的单不论最终状态
+        都计入,被拒/失败不退当天名额);四档每天只能选一个,选定后其余三档当天 other_tier_selected。
       - invite_cash 本轮无档位概念 → 返回空列表(邀请页客户端仍用本地写死档位,行为不变)。
     余额是否足够由客户端本地判断(余额随兑换实时变化,不在此快照)。
     """
@@ -785,7 +785,7 @@ def withdraw_tier_states(db: Session, user_id: int, source: str = "coin_cash") -
             out.append({
                 "amount_cents": t.amount_cents, "label": t.label, "badge": t.badge,
                 "is_newbie": True, "available": True, "disabled_reason": None,
-                "remaining_today": 1,
+                "remaining_today": 1, "daily_limit": t.daily_limit,
             })
             continue
         used = today_counts.get(t.amount_cents, 0)
@@ -798,7 +798,9 @@ def withdraw_tier_states(db: Session, user_id: int, source: str = "coin_cash") -
         out.append({
             "amount_cents": t.amount_cents, "label": t.label, "badge": t.badge,
             "is_newbie": False, "available": available, "disabled_reason": reason,
-            "remaining_today": remaining,
+            # daily_limit 一起下发:客户端要靠它区分「本档能提 3 次」和「本档只能提 1 次」——
+            # 只看 remaining_today 分不出「0.5 已提两次剩 1」与「10 元一次没提剩 1」。
+            "remaining_today": remaining, "daily_limit": t.daily_limit,
         })
     return out
 

app/schemas/welfare.py

@@ -88,6 +88,14 @@ class WithdrawTierOut(BaseModel):
         description="不可提原因:quota_exhausted(今日次数满) / other_tier_selected(今日已选其他额度)",
     )
     remaining_today: int = Field(0, description="今日剩余可提次数")
+    daily_limit: int = Field(
+        1,
+        description=(
+            "该档每日可提次数上限(rewards.WITHDRAW_TIERS_COIN_CASH.daily_limit)。"
+            "客户端据此判「本档可提多次」——只有 daily_limit>1 且 remaining_today<daily_limit"
+            "(= 今天已提过至少一次)才展示「今日还可提N次」角标。"
+        ),
+    )
 
 
 class WithdrawInfoOut(BaseModel):

app/services/compare_alert.py

@@ -0,0 +1,127 @@
+"""比价失败报警规则:一条记录 → 命中的 AlertHit(或 None)。
+
+纯函数、不碰 DB(阈值/关键词由调用方从 config 传入),便于单测与调阈值。判定顺序保证四类互斥:
+  failed → fail_reason 空(且 information 非业务)=T1 / 含未识别=T6 / 含超时词=T2 / 其余业务不报;
+  cancelled 且耗时或步数超阈值=T5(深度放弃);success/running 不报。
+口径依据见 docs/superpowers/specs/2026-08-04-compare-fail-alert-design.md 第 3 节。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+ALERT_TYPE_LABELS: dict[str, str] = {
+    "T1": "系统技术失败",
+    "T6": "商品识别失败",
+    "T2": "启动/超时失败",
+    "T5": "深度放弃(cancelled)",
+}
+
+
+@dataclass(frozen=True)
+class AlertHit:
+    trace_id: str
+    alert_type: str
+    reason: str
+    app_version: str | None
+    created_at: datetime | None
+    trace_url: str | None
+    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:
+    return AlertHit(
+        trace_id=rec.trace_id,
+        alert_type=alert_type,
+        reason=reason,
+        app_version=getattr(rec, "app_version", None),
+        created_at=getattr(rec, "created_at", None),
+        trace_url=getattr(rec, "trace_url", None),
+        user_id=getattr(rec, "user_id", None),
+        total_ms=getattr(rec, "total_ms", None),
+        step_count=getattr(rec, "step_count", None),
+    )
+
+
+def classify_cancelled_fallback(
+    rec: Any,
+    *,
+    cancelled_ms_threshold: int,
+    cancelled_step_threshold: int,
+) -> AlertHit | None:
+    """cancelled 保底判定(读不到 trace 时用):超耗时或步数阈值 → T5 深度放弃,否则 None。纯函数。"""
+    ms = rec.total_ms
+    step = rec.step_count
+    deep = (ms is not None and ms > cancelled_ms_threshold) or (
+        step is not None and step > cancelled_step_threshold
+    )
+    if deep:
+        return make_hit(rec, "T5", "深度放弃")
+    return None
+
+
+def _target_technical_failure_reason(
+    rec: Any, biz_exclude_keywords: tuple[str, ...]
+) -> str | None:
+    """某目标平台「真技术失败」(pricebot 原始 status=='failed' 且 reason 非业务话术)的原因;无 → None。
+
+    记录级 fail_reason 是「展示口径」——一条比价里若某平台是干净业务结局(京东未找到菜),它会被派生
+    成 headline,盖住另一平台的真技术崩溃(淘宝'比价过程出错')。这里扫 raw_payload.platform_results
+    补判:任一目标平台 status='failed' 且 reason 不含业务词(pricebot 偶把打烊/不配送漏标成 failed,
+    用 biz_exclude 过滤掉这些业务误标)→ 真技术崩溃。缺 raw_payload/platform_results / 结构异常 → None。
+    """
+    raw = getattr(rec, "raw_payload", None)
+    pr = raw.get("platform_results") if isinstance(raw, dict) else None
+    if not isinstance(pr, dict):
+        return None
+    for v in pr.values():
+        if not isinstance(v, dict) or v.get("is_source"):
+            continue
+        if v.get("status") != "failed":
+            continue
+        reason = (v.get("reason") or "").strip()
+        if not any(w in reason for w in biz_exclude_keywords):
+            return reason or "比价过程出错"
+    return None
+
+
+def classify_record(
+    rec: Any,
+    *,
+    cancelled_ms_threshold: int,
+    cancelled_step_threshold: int,
+    timeout_keywords: tuple[str, ...],
+    unrecognized_keywords: tuple[str, ...],
+    biz_exclude_keywords: tuple[str, ...],
+) -> AlertHit | None:
+    """判定单条记录是否触发报警。rec 需有 status/fail_reason/information/total_ms/step_count/
+    trace_id/app_version 属性(ComparisonRecord 或等价对象)。"""
+    status = rec.status
+    if status == "failed":
+        fail_reason = rec.fail_reason
+        if fail_reason is None:
+            info = (rec.information or "").strip()
+            if info and any(w in info for w in biz_exclude_keywords):
+                return None
+            return make_hit(rec, "T1", f"技术失败·{info[:80] or '比价过程出错'}")
+        if any(w in fail_reason for w in unrecognized_keywords):
+            return make_hit(rec, "T6", f"识别失败·{fail_reason[:80]}")
+        if any(w in fail_reason for w in timeout_keywords):
+            return make_hit(rec, "T2", fail_reason[:80])
+        # fail_reason 是干净业务 headline,但可能盖住某目标平台的真技术失败(比价过程出错)→ 补判 T1。
+        tech = _target_technical_failure_reason(rec, biz_exclude_keywords)
+        if tech:
+            return make_hit(rec, "T1", f"技术失败·{tech[:80]}")
+        return None  # 纯业务失败,不报
+    if status == "cancelled":
+        return classify_cancelled_fallback(
+            rec,
+            cancelled_ms_threshold=cancelled_ms_threshold,
+            cancelled_step_threshold=cancelled_step_threshold,
+        )
+    return None
+

app/services/compare_alert_format.py

@@ -0,0 +1,268 @@
+"""AlertHit[] → 飞书群机器人消息。
+
+提供三个格式化函数:
+- format_alert_message: 纯文本(保留,已有集成测试依赖)。
+- format_alert_post: 富文本 post(行式明细:时间｜手机｜版本｜原因｜trace 超链接)。
+- format_alert_card: schema 2.0 卡片 + table 组件(正式发送格式)。
+
+按触发类型分组,每类给计数 + 明细(trace/版本/原因)。两级截断防报警风暴:单类型超
+max_detail_per_type 只列前 N + 「另有 M 条」;本期总量超 max_total 只给各类型计数、提示去分析库查。
+标题含关键词「比价失败报警」——飞书自定义机器人用关键词验证,消息必须含它,否则被拒收。
+"""
+from __future__ import annotations
+
+from app.services.compare_alert import ALERT_TYPE_LABELS, AlertHit
+
+ALERT_KEYWORD = "比价失败报警"
+
+_TYPE_ORDER = ("T1", "T6", "T2", "T5")
+
+
+def _detail_line(h: AlertHit) -> str:
+    ver = h.app_version or "?"
+    return f"   - trace {h.trace_id} | {ver} | {h.reason}"
+
+
+def format_alert_message(
+    hits: list[AlertHit],
+    *,
+    window_label: str,
+    max_detail_per_type: int,
+    max_total: int,
+) -> str:
+    total = len(hits)
+    header = f"🚨 {ALERT_KEYWORD} · {window_label} · 本期触发 {total} 条"
+
+    grouped: dict[str, list[AlertHit]] = {t: [] for t in _TYPE_ORDER}
+    for h in hits:
+        grouped.setdefault(h.alert_type, []).append(h)
+
+    lines = [header]
+    counts_only = total > max_total
+    for t in _TYPE_ORDER:
+        bucket = grouped.get(t) or []
+        if not bucket:
+            continue
+        lines.append(f"• {ALERT_TYPE_LABELS[t]} {len(bucket)} 条")
+        if counts_only:
+            continue
+        shown = bucket[:max_detail_per_type]
+        lines.extend(_detail_line(h) for h in shown)
+        if len(bucket) > max_detail_per_type:
+            lines.append(f"   …另有 {len(bucket) - max_detail_per_type} 条")
+    if counts_only:
+        lines.append(f"(本期命中超 {max_total} 条,仅列计数,明细见分析库 comparison_record)")
+    return "\n".join(lines)
+
+
+def format_alert_post(
+    hits: list[AlertHit],
+    *,
+    window_label: str,
+    phone_map: dict[int, str],
+    max_detail_per_type: int,
+    max_total: int,
+) -> tuple[str, list]:
+    """行式富文本:返回 (title, content)。title 含 ALERT_KEYWORD(飞书关键词验证)。
+
+    content: 摘要段(各类型计数) + 表头段 + 明细行(每条「时间｜手机｜版本｜原因｜」+ trace 超链接 a 元素)。
+    phone_map: {user_id: phone};明细手机号取 phone_map.get(hit.user_id) or "-"。
+    截断规则:总命中 > max_total → 只出摘要+各类型计数(不列明细);
+              否则明细最多列前 max_detail_per_type 条,超出加「…另有 N 条」。
+    """
+    total = len(hits)
+    title = f"🚨 {ALERT_KEYWORD} · {window_label}"
+
+    grouped: dict[str, list[AlertHit]] = {t: [] for t in _TYPE_ORDER}
+    for h in hits:
+        grouped.setdefault(h.alert_type, []).append(h)
+
+    # 摘要段:各类型计数
+    summary_parts = [f"{ALERT_TYPE_LABELS[t]} {len(grouped[t])}" for t in _TYPE_ORDER if grouped.get(t)]
+    summary_text = f"合计 {total} 条：" + " ｜ ".join(summary_parts)
+    content: list[list[dict]] = [
+        [{"tag": "text", "text": summary_text}],
+    ]
+
+    counts_only = total > max_total
+    if counts_only:
+        content.append([{"tag": "text", "text": f"(本期命中超 {max_total} 条,仅列计数,明细见分析库 comparison_record)"}])
+        return title, content
+
+    # 表头段
+    content.append([{"tag": "text", "text": "时间 ｜ 手机号 ｜ 版本 ｜ 失败原因 ｜ trace"}])
+
+    # 明细行(按类型顺序展开,每条一段)
+    shown_count = 0
+    for t in _TYPE_ORDER:
+        bucket = grouped.get(t) or []
+        if not bucket:
+            continue
+        for h in bucket[:max_detail_per_type]:
+            time_str = h.created_at.strftime("%m-%d %H:%M") if h.created_at else "-"
+            phone = phone_map.get(h.user_id) if h.user_id is not None else None
+            phone = phone or "-"
+            ver = h.app_version or "-"
+            row: list[dict] = [{"tag": "text", "text": f"{time_str} ｜ {phone} ｜ {ver} ｜ {h.reason} ｜ "}]
+            if h.trace_url:
+                row.append({"tag": "a", "text": "trace", "href": h.trace_url})
+            else:
+                row.append({"tag": "text", "text": h.trace_id[:16]})
+            content.append(row)
+            shown_count += 1
+        if len(bucket) > max_detail_per_type:
+            content.append([{"tag": "text", "text": f"…另有 {len(bucket) - max_detail_per_type} 条"}])
+
+    return title, content
+
+
+# ---- format_alert_card (schema 2.0 卡片 + table 组件) ----
+
+def _cost_cell(total_ms: int | None, step_count: int | None) -> str:
+    """组合「用时」列值。有 ms → '{N}s'，有 step_count → '{M}步'，两者用 ' / ' 连；都无 → '-'。"""
+    parts = []
+    if total_ms is not None:
+        parts.append(f"{round(total_ms / 1000)}s")
+    if step_count is not None:
+        parts.append(f"{step_count}步")
+    return " / ".join(parts) or "-"
+
+
+def _build_card(title_text: str, elements: list[dict]) -> dict:
+    """组装 schema 2.0 红色 header 卡片;三条路径只需决定 elements。"""
+    return {
+        "schema": "2.0",
+        "header": {
+            "title": {"tag": "plain_text", "content": title_text},
+            "template": "red",
+        },
+        "body": {"elements": elements},
+    }
+
+
+def _build_table_rows(
+    hits: list[AlertHit],
+    *,
+    phone_map: dict[int, str],
+    max_detail_per_type: int,
+) -> list[dict]:
+    """按 _TYPE_ORDER 顺序展开，每类型最多 max_detail_per_type 条。"""
+    grouped: dict[str, list[AlertHit]] = {t: [] for t in _TYPE_ORDER}
+    for h in hits:
+        if h.alert_type in grouped:
+            grouped[h.alert_type].append(h)
+        # 非 _TYPE_ORDER 类型静默跳过(与既有行为一致)
+
+    rows = []
+    for t in _TYPE_ORDER:
+        bucket = grouped.get(t) or []
+        for h in bucket[:max_detail_per_type]:
+            time_str = h.created_at.strftime("%m-%d %H:%M") if h.created_at else "-"
+            phone = (phone_map.get(h.user_id) if h.user_id is not None else None) or "-"
+            trace = (
+                f"[链接]({h.trace_url})" if h.trace_url
+                else (h.trace_id or "")[:12]
+            )
+            rows.append({
+                "time": time_str,
+                "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,
+            })
+    return rows
+
+
+_TABLE_COLUMNS = [
+    {"name": "time",   "display_name": "时间",    "data_type": "text"},
+    {"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"},
+]
+
+
+def _type_criteria(t: str, cancelled_ms_threshold: int, cancelled_step_threshold: int) -> str:
+    """各触发类型的「判据」文案,展示在卡片摘要里让收报警的人一眼知道为什么报。
+    T5 带当前配置阈值(耗时 ms→s、步数);T1 覆盖两条来源(整场系统错 + 混合单里任一平台 status=failed)。"""
+    if t == "T5":
+        return f"耗时>{round(cancelled_ms_threshold / 1000)}s 或 步数>{cancelled_step_threshold}"
+    if t == "T1":
+        return "无业务原因的系统错 或 任一平台 status=failed"
+    if t == "T2":
+        return "原因含 超时/启动/加载"
+    if t == "T6":
+        return "原因含 未识别"
+    return ""
+
+
+def format_alert_card(
+    hits: list[AlertHit],
+    *,
+    window_label: str,
+    phone_map: dict[int, str],
+    interval_min: int,
+    max_detail_per_type: int,
+    max_total: int,
+    cancelled_ms_threshold: int = 90000,
+    cancelled_step_threshold: int = 30,
+) -> dict:
+    """返回飞书 schema 2.0 卡片 dict（配合 send_feishu_card 发送）。
+
+    - header: template=red，title 含 ALERT_KEYWORD（飞书关键词验证必须）。
+    - body 第一个元素: markdown 摘要（数据范围 + 合计 + 各类型计数）。
+    - 空 hits: 只有摘要「本期无异常」。
+    - total > max_total: 只有摘要（提示去 comparison_record 查），不加 table。
+    - 否则: 第二个元素为 table（列序 time/phone/cost/reason/stuck/ver/trace）。
+    """
+    total = len(hits)
+    title_text = f"🚨 {ALERT_KEYWORD} · {window_label}"
+
+    # ---------- 空 hits ----------
+    if total == 0:
+        md_content = f"数据范围：近 {interval_min} 分钟\n本期无异常"
+        return _build_card(title_text, [{"tag": "markdown", "content": md_content}])
+
+    # ---------- 摘要 ----------
+    grouped_count: dict[str, int] = {}
+    for h in hits:
+        grouped_count[h.alert_type] = grouped_count.get(h.alert_type, 0) + 1
+    count_parts = [
+        f"{ALERT_TYPE_LABELS[t]} {grouped_count[t]}"
+        for t in _TYPE_ORDER
+        if grouped_count.get(t)
+    ]
+    md_content = (
+        f"数据范围：近 {interval_min} 分钟\n"
+        f"**合计 {total} 条**：" + " ｜ ".join(count_parts)
+    )
+
+    # ---------- 判据说明（本期出现的类型各给一行判据,T5 带当前配置阈值）----------
+    criteria_parts = [
+        f"{ALERT_TYPE_LABELS[t]}={_type_criteria(t, cancelled_ms_threshold, cancelled_step_threshold)}"
+        for t in _TYPE_ORDER
+        if grouped_count.get(t)
+    ]
+    if criteria_parts:
+        md_content += "\n判据：" + " ｜ ".join(criteria_parts)
+
+    # ---------- 截断：超 max_total 只出摘要 ----------
+    if total > max_total:
+        md_content += f"\n超 {max_total} 条仅列计数，明细见分析库 comparison_record"
+        return _build_card(title_text, [{"tag": "markdown", "content": md_content}])
+
+    # ---------- 常规：摘要 + table ----------
+    rows = _build_table_rows(hits, phone_map=phone_map, max_detail_per_type=max_detail_per_type)
+    table_element = {
+        "tag": "table",
+        "page_size": 10,
+        "row_height": "low",
+        "header_style": {"background_style": "grey", "bold": True},
+        "columns": _TABLE_COLUMNS,
+        "rows": rows,
+    }
+    return _build_card(title_text, [{"tag": "markdown", "content": md_content}, table_element])

app/services/trace_stuck.py

@@ -0,0 +1,172 @@
+"""比价卡死定位:读 pricebot trace 末段,判某平台是否原地打转(卡死)。
+
+同机直读 {WORK_LOG_DIR}/{dir_name}/{platform}/step_*.json,只取头部字段
+(pipeline_step/detected_page),不解析后面的无障碍树(windows,占单帧 99% 体积)。
+判据与降级见 docs/superpowers/specs/2026-08-05-compare-stuck-detection-design.md。
+"""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+
+# pipeline_step / 平台名 → 中文(映射不到原样显示英文,不阻断)。按 pricebot 实际枚举补全。
+PIPELINE_STEP_LABELS: dict[str, str] = {
+    "set_address": "定位",
+    "enter_store": "进店",
+    "add_one_dish": "加菜",
+    "match_dish": "找菜",
+    "checkout": "结算",
+}
+PLATFORM_LABELS: dict[str, str] = {
+    "meituan": "美团",
+    "eleme": "饿了么",
+    "jd_waimai": "京东外卖",
+}
+
+_PIPE_RE = re.compile(r'"pipeline_step":\s*"([^"]*)"')
+_PAGE_RE = re.compile(r'"detected_page":\s*"([^"]*)"')
+_TS_RE = re.compile(r'"timestamp":\s*"([^"]*)"')
+_STEP_NUM_RE = re.compile(r"step_(\d+)")
+
+
+def _parse_ts(s: str | None) -> datetime | None:
+    if not s:
+        return None
+    try:
+        return datetime.fromisoformat(s)
+    except (ValueError, TypeError):
+        return None
+
+
+@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
+
+    def label(self) -> str:
+        p = PLATFORM_LABELS.get(self.platform, self.platform)
+        s = PIPELINE_STEP_LABELS.get(self.pipeline_step, self.pipeline_step)
+        base = f"{p}·{s}"
+        if self.detected_page:
+            base += f"·{self.detected_page}"  # 页面暂用 pricebot 原值(英文),无中文映射
+        return base
+
+
+@dataclass(frozen=True)
+class StuckResult:
+    readable: bool               # trace 是否读到(区分「读不到」与「读到但没卡」)
+    points: list[StuckPoint]     # 卡死平台列表;readable=True 且空 = 确认没卡死
+    last: StuckPoint | None = None  # 末帧(帧数最多平台的末帧,带 detected_page);读不到 → None
+
+
+def dir_name_from_trace_url(trace_url: str | None) -> str | None:
+    """.../traces/{dir_name}/ → dir_name;空/异常 → None。"""
+    if not trace_url:
+        return None
+    name = trace_url.rstrip("/").rsplit("/", 1)[-1]
+    return name or None
+
+
+def _step_num(path: Path) -> int:
+    m = _STEP_NUM_RE.search(path.name)
+    return int(m.group(1)) if m else -1
+
+
+def _read_head(path: Path, nbytes: int = 4096) -> tuple[str | None, str | None, str | None]:
+    """只读文件头部,抠 (pipeline_step, detected_page, timestamp)。它们在 json 最前面。"""
+    try:
+        with open(path, encoding="utf-8", errors="replace") as f:
+            head = f.read(nbytes)
+    except OSError:
+        return None, None, None
+    ps = _PIPE_RE.search(head)
+    pg = _PAGE_RE.search(head)
+    ts = _TS_RE.search(head)
+    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 → 卡死。"""
+    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
+    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)
+
+
+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
+            sp = _platform_stuck(pdir.name, step_files, threshold, max_tail)
+            if sp is not None:
+                points.append(sp)
+            # 末帧:取帧数最多平台的末帧(与 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)
+        if not any_frames:
+            return StuckResult(readable=False, points=[])
+        return StuckResult(readable=True, points=points, last=last)
+    except OSError:
+        return StuckResult(readable=False, points=[])
+
+
+def last_step(trace_dir: Path) -> StuckPoint | None:
+    """failed 用:取帧数最多平台的末帧环节(不要求原地打转)。读不到 → 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
+        ps, pg, _ts = _read_head(step_files[-1])
+        if ps is None:
+            return None
+        return StuckPoint(platform, ps, len(step_files), detected_page=pg)  # stuck_ms=None(failed 不算时长)
+    except OSError:
+        return None

docs/guides/比价结果卡片-状态口径与交互参考.md

@@ -0,0 +1,156 @@
+# 比价结果卡片 · 状态口径与交互参考
+
+> **定位**：把「比价结果页每张平台卡片」的 11 种展示分类，逐一映射到后端状态与判定字段，供 **App 端做互动/文案调整时参考**。
+> **适用**：外卖比价结果页（`CompResultScreen`）。
+> **数据源**：基于三仓代码梳理（pricebot 判定 → app-server 透传/落库 → android 映射/渲染）。
+> **整理日期**：2026-07-31。代码行号会漂移，改动后以实际代码为准。
+
+---
+
+## 0. 快速须知（三个容易踩的坑）
+
+1. **`ok` 不是 `success`**：逐平台卡片的成功态字符串是 **`ok`**（端侧 `rowToUi` 判 `row.status == "ok"`）；`success/failed/cancelled/running` 是**记录级** `ComparisonRecord.status`（admin 后台用）。两层状态串不同，别混。
+2. **未知状态一律兜底成 `Failed`（#9）**：端侧映射用 `else -> CompareResultStatus.Failed`，所以后端 `unsupported`、`failed`、以及**任何端侧没显式处理的新状态**都会显示成 #9。pricebot 以后加状态，端侧不同步就会「消失」进 #9。
+3. **状态判定分三层来源**（见 §2）：不是所有卡都由后端 status 决定，#10/#11 完全是端侧本地判定，后端零感知。
+
+---
+
+## 1. 主对照表：11 类卡片 ↔ 后端状态
+
+映射函数：`CompareProgressRepository.kt` — 新路径 `rowToUi:339-375`、老路径 `summaryToUi:509-544`、端侧补齐 `onComparisonResultsReady:266-315`。
+
+| # | 分类 | Android 枚举 | 后端逐平台 `status` | `Found` 内细分依据 | 状态来源 | 当前交互 |
+|---|---|---|---|---|---|---|
+| 1 | 全网最低赢家 | `Found` | `ok`（有价） | `is_best && !has_dish_diff` → isLowest | pricebot | 去购买（黄） |
+| 2 | 没有可用优惠（原选择） | `Found` | `ok` | `role=source`/is_user_original + 无可用优惠 | pricebot | 查看（灰） |
+| 3 | 其他成功 | `Found` | `ok` | `role=target`，非 best | pricebot | 查看（灰） |
+| 4 | 少菜 / 部分缺 | `Found` + 提示条 | `ok` | `skipped_dish_names` 非空 / `skipped_dish_count>0` | pricebot | 随主卡 |
+| 5 | 未满起送 | `BelowMinimum` | `below_minimum` | — | pricebot | 查看（灰） |
+| 6 | 门店打烊 | `StoreClosed` | `store_closed` | — | pricebot | 查看（灰） |
+| 7 | 没有您点的商品 | `ItemsNotFound` | `items_not_found` | — | pricebot | 查看（灰） |
+| 8 | 无对应商家 | `StoreNotFound` | `store_not_found` | — | pricebot **或端侧补齐** | 查看（灰） |
+| 9 | 比价失败兜底 | `Failed` | `failed`/`unsupported`/**任何未知值**（`else`） | — | pricebot **或端侧**（整场没跑成） | 无 |
+| 10 | 未安装 | `NotInstalled` | **无**（后端零感知） | 端侧：该平台没装（`getPackageInfo`） | **端侧本地** | 去安装 |
+| 11 | 未选择 / 本次未比 | `NotComparedThisTime` | **无**（后端零感知） | 端侧：`selectedPlatformIds` 没勾这家 | **端侧本地** | 无（设计要「重试」，未接） |
+| 表外 | 单点不配送 | `NoDelivery` | `no_delivery` | — | pricebot | 查看（灰） |
+
+> **表外提醒**：`no_delivery`（单点不配送/需搭配主食）在代码里是独立状态（复用未起送灰框样式），但**不在原 11 类分类表内**——做分类时需给它安个位置或明确并入 #5/#7。
+
+---
+
+## 2. 状态来源三分层（做互动最该记住的）
+
+**A. pricebot 逐平台 `status` 直接决定（5/6/7/8/9 + no_delivery）**
+端侧只做「字符串 → 枚举」映射，互动依附后端判定。
+
+**B. `ok` 成功态再靠 `platforms[]` 字段细分（1/2/3/4）**
+同一个 `Found`，靠 `is_best`、`has_dish_diff`、`role`、`skipped_dish_names` 分出四种。这些字段都是 pricebot 在 `platforms[]` 下发的。
+
+**C. 纯端侧本地判定，后端完全无感知（10/11，以及 8 的补齐分支）**
+依据：**装机态**（`getPackageInfo`）+ **勾选态**（`selectedPlatformIds`）+ **整场是否跑成**（`failureReason`）。
+见 `onComparisonResultsReady:266-315`：走了选平台弹窗没勾 → #11；装了 App 没找到店 → #8；都没装 → #10（`marketPackage` 指向子 App 包名，引导下载子 App）。
+
+---
+
+## 3. 菜品差异口径（有相似菜品 / 少菜 / 规格近似）
+
+「有相似菜品」**不是独立卡**，而是 `status=Found`（有价）成功卡上、因「菜品对不齐」触发的黄色提醒模块 `MismatchBox`。
+
+### 3.1 开关：`hasDishDiff`
+定义在 `CompareMockResults.kt:101-109`。`status=Found` 且满足**任一**即为 true：
+- `items` 里有 `similar=true` 的菜（近似替代）
+- `approxDishNames` 非空（规格近似）
+- `skippedDishNames` 非空（完全缺失）
+- `skippedDishCount > 0`（只有数量）
+
+**权威值优先取后端** `platforms[].has_dish_diff`（端侧 `dishDiffOverride`，`CompareMockResults.kt:91-93`），老路径才端侧本地算。与后端 `is_best` 竞选同判据。
+
+### 3.2 `MismatchBox` 的三种行（`CompRowCard.kt:236-250` 触发 / `460-516` 渲染）
+
+| 展示文案 | 触发字段 | 语义 |
+|---|---|---|
+| **本平台没有「orig」 / 已换成近似「name」** | `items` 里 `similar=true && orig!=null`（similarPairs） | **已发生近似替换**（有 orig→name 替换对） |
+| 规格近似「X」，请下单前核对 | `approxDishNames` | 已近似，但提示核对 |
+| 本平台没有「X」，也无相似菜 | `skippedDishNames` | 完全缺失、无替代 |
+| 本平台缺少 N 个菜品，已按可购买商品计价 | `skippedDishCount>0` 且无菜名 | 只有数量时的兜底 |
+
+> **易混近亲**：`complexSpecDishNames`（`CompareMockResults.kt:86-88`）走的是**另一个浅黄条** `ComplexSpecNotice`，文案「规格较复杂，请核对」，语义是"需人工核对规格、**不改价格有效性**"，与"已换成近似"不是一回事。
+
+### 3.3 连锁后果（为什么降级成"查看"）
+`has_dish_diff=true` → `isLowest = is_best && !has_dish_diff` 必为 false → **踢出「全网最低」评定**，排到"仅供参考"分组，不给绶带/省¥红章/去购买黄按钮，只能灰"查看"。
+
+### 3.4 边界
+- **vs #4「完全缺失」**：同一个 `MismatchBox` 的不同行，**可并存**（一个平台既有近似替换又有完全缺失）。它们是同一张成功卡里分段列出，不是两张卡。
+- **vs #7「该店没有您点的商品（items_not_found）」**：分水岭是**有没有比出价**。能凑出单、有价 → `Found` + MismatchBox；整单找不到菜、无价 → `items_not_found` 灰框无价卡。
+- **vs #1 赢家**：互斥。
+
+---
+
+## 4. 交互调整参考要点
+
+1. **文案/提示权已在前端**：卡片副文案由端侧 `fullReasonForStatus(status)` 按状态本地生成（`CompareProgressRepository.kt:51-59`，注释明写"标题/文案决策收回前端"）。后端逐平台原因字段 `notFoundReason`（= `platform_results[].reason`）**线上常空**，端侧兜底。→ **改提示文案、加互动引导、加动效，纯端侧就能做，不用改 pricebot。**
+2. **可空字段决定互动形态**：`price` 可空（6/7/8/9/10/11 无价 → `¥??` 或隐藏）；`store_name` 可空（退化平台名）。`LOCATED_STATUSES`（`CompareProgressRepository.kt:46-47`：`ok/below_minimum/no_delivery/store_closed/items_not_found`）决定 header 用店名还是平台名。做点击/跳转互动前要判空。
+3. **#10/#11 端侧完全自主**：引导安装、引导勾选、重试这类互动无需后端配合，随便调。
+4. **区分"真没店"vs"没跑成"**：`failureReason` 非空（网络/上游 502）时，补齐卡统一兜底文案而非"未找到店"（`CompareProgressRepository.kt:104-107`，避免谎报）。"重试整场"的互动应挂这个场景，而不是 #8。
+5. **带动态数字的文案要后端补字段**：如"还差 ¥8 起送"——`platforms[]` 目前不下发起送差额，想要这种提示得让 pricebot 补字段。
+6. **#4 少菜 / 有相似菜品是成功卡上的模块，不是独立卡**：它们的互动依附主卡（1/2/3）。
+
+---
+
+## 5. 关键代码位置索引
+
+### Android（`shaguabijia-app-android`）
+| 关注点 | 文件:行 |
+|---|---|
+| 后端 status → 枚举映射（新/老路径） | `CompareProgressRepository.kt:339-375` / `509-544` |
+| 端侧文案本地生成 | `CompareProgressRepository.kt:51-59`（`fullReasonForStatus`） |
+| header 店名/平台名切换 | `CompareProgressRepository.kt:46-47`（`LOCATED_STATUSES`） |
+| 端侧补齐 #8/#10/#11（装机+勾选+failureReason） | `CompareProgressRepository.kt:266-315` |
+| 卡片 when 主分支 | `CompRowCard.kt:83-128` |
+| `MismatchBox`（菜品差异条） | `CompRowCard.kt:236-250`（触发）/ `460-516`（渲染） |
+| 未安装卡 / 本次未比卡 | `CompRowCard.kt:596-697` / `720-790` |
+| 按钮（去购买/查看/重试） | `CompRowCard.kt:1148-1180` / `1210` / `1214`；枚举 `GrayCardCta:132` |
+| 数据模型 `CompareResult` / `hasDishDiff` / `CompareDish` | `CompareMockResults.kt:34-110` / `101-109` / `116` |
+| 网络 DTO 解析 | `Protocol.kt:530-549` / `576-665` |
+| 生产结果页 / VM | `CompResultScreen.kt` / `CompResultViewModel.kt` |
+
+### 后端（`shaguabijia-app-server`）
+| 关注点 | 文件:行 |
+|---|---|
+| 逐平台 `status` 枚举 | `app/schemas/compare_record.py:81-85` |
+| 记录模型 / `status` / `skipped_dish_names` / `raw_payload` | `app/models/comparison.py:39-169` / `:101` / `:121` / `:123` |
+| 落库与状态派生（running/done/abort） | `app/repositories/comparison.py`（`harvest_*`、`_derive*`） |
+| `has_dish_diff` 兜底判 is_best | `app/repositories/comparison.py:379` |
+| 记录级失败原因派生 | `app/repositories/comparison.py:118-127`（`_derive_fail_display`） |
+| admin 读取接口 | `app/admin/routers/comparison.py` / `app/admin/repositories/queries.py` |
+
+---
+
+## 6. 当前实现现状与「目标设计」的差异（待办）
+
+11 类骨架**都已实现、生产页 `CompResultScreen` 真实可达**（非 mock）。但对照原 11 类目标设计，仍有以下差异：
+
+| 项 | 现状 | 差异 / 待办 |
+|---|---|---|
+| #11「重试」按钮 | `NotComparedCard` 无任何 CTA；`GrayCardCta.Retry`+`RetryCta` 定义了但**没挂到任何卡**（死代码） | 唯一**功能级缺口**：要么接上，要么把设计改成"无按钮" |
+| 各类文案措辞 | 偏功能陈述（如"该店当前已打烊，暂时无法比价"） | 与目标口语+emoji（"…打烊休息啦😴…"）不一致；改端侧 `fullReasonForStatus` 即可（后端 `reason` 常空、无需动 pricebot） |
+| #1 优惠区 | 已改**聚合**"已自动帮您应用优惠，共减 ¥X" | 与"优惠 tags（逐项标签）"不同（产品已决策聚合） |
+| #4 少菜 | 成功卡上的 `MismatchBox` 提示条 | 非独立卡（按"原生为准"这多为有意） |
+| `no_delivery` | 代码有独立状态 | 原 11 类表未含，需补位或并入 |
+
+---
+
+## 7. admin 记录页口径（与 C 端不同，勿混）
+
+admin 比价记录页 / 概览 / 大盘用**技术完成率**口径：`below_minimum / store_closed /
+store_not_found / items_not_found / no_delivery / unsupported`（流程跑完、只是外部原因致结果缺失）
+**记为成功**（前端标绿「成功」+ 感叹号，hover 显示缺失原因），只有纯技术故障 `failed` 才算失败。
+派生见 `app/admin/repositories/comparison_outcome.py`（原始结局取 `raw_payload.record_status`，
+`admin_success_sql()` 供概览/大盘/列表筛选共用，`derive_admin_outcome()` 供列表逐行下发 `outcome_hint`）。
+
+这**刻意宽于** C 端 / #209 落库口径（那边 not_found 类归 `failed`）——admin 关心「系统有没有跑成」，
+C 端关心「有没有省到钱」。故 **admin 成功率 ≠ C 端 / 首页轮播口径**，对不上是设计使然、非 bug。
+
+---
+
+*本文档为跨端口径参考，非契约。字段/行号以三仓实际代码为准。*

docs/superpowers/plans/2026-08-05-compare-stuck-detection.md

@@ -0,0 +1,735 @@
+# 比价卡死定位报警增强 Implementation Plan
+
+> **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:** 让 cancelled 报警用 trace 末段「原地打转」判卡死并定位卡在哪个环节，读不到 trace 回退耗时/帧数保底；failed 类附卡点。
+
+**Architecture:** 判定保持纯函数（`compare_alert.py`），trace 读取单独成层（`trace_stuck.py`，同机直读 pricebot work_logs、只读帧头部字段），worker 编排（cancelled trace 优先 + 保底、failed 附卡点）。卡点拼进 `AlertHit.reason`，复用现有 `format_alert_post` 展示，不依赖卡片 table 固化。
+
+**Tech Stack:** Python 3.11+、FastAPI、SQLAlchemy、pytest。无新依赖（仅标准库 `re`/`json`/`pathlib`/`dataclasses`）。
+
+参考 spec：`docs/superpowers/specs/2026-08-05-compare-stuck-detection-design.md`
+
+---
+
+## File Structure
+
+- **Create** `app/services/trace_stuck.py` — 卡死判据 + 薄 IO。`StuckPoint`/`StuckResult`、`read_stuck_points`、`last_step`、`dir_name_from_trace_url`、文案表。
+- **Create** `tests/test_trace_stuck.py` — trace_stuck 单测。
+- **Modify** `app/core/config.py` — 加 4 个配置项。
+- **Modify** `app/services/compare_alert.py` — `_hit` 改公开 `make_hit`；新增纯函数 `classify_cancelled_fallback`；`classify_record` 的 cancelled 分支改调它（行为不变）。
+- **Create** `tests/test_compare_alert_fallback.py` — `classify_cancelled_fallback`/`make_hit` 单测。
+- **Modify** `app/core/compare_alert_worker.py` — 新增 `build_hits`/`_trace_dir` 编排；`_scan_and_alert` 用 `build_hits` 替换 `classify_batch`。
+- **Create** `tests/test_compare_alert_stuck_worker.py` — `build_hits` 集成测。
+
+---
+
+## Task 1: 配置项
+
+**Files:**
+- Modify: `app/core/config.py:147`（在 `COMPARE_ALERT_SEND_EMPTY` 行后追加）
+
+- [ ] **Step 1: 加 4 个配置字段**
+
+在 `app/core/config.py` 第 147 行 `COMPARE_ALERT_SEND_EMPTY: bool = False ...` 之后，紧接着追加：
+
+```python
+    # ===== 卡死定位(读 pricebot trace 末段判原地打转)=====
+    COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR: str = ""      # pricebot work_logs 绝对路径(敏感,放 .env);空=跳过 trace、cancelled 全走保底
+    COMPARE_ALERT_STUCK_FRAME_THRESHOLD: int = 15      # 末段连续同环节达此帧数判卡死
+    COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES: int = 40      # 每平台最多往前读多少帧
+    COMPARE_ALERT_TRACE_MAX_RECORDS: int = 30          # 每轮最多对多少条命中记录读 trace(限量)
+```
+
+- [ ] **Step 2: 跑现有测试确认不破**
+
+Run: `pytest tests/ -q -k "config or defaults"`
+Expected: PASS（新增字段都有默认值，不影响 `test_defaults`）
+
+- [ ] **Step 3: Commit**
+
+```bash
+git add app/core/config.py
+git commit -m "feat(compare-alert): 卡死定位 4 个配置项
+
+Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
+```
+
+---
+
+## Task 2: trace_stuck 模块
+
+**Files:**
+- Create: `app/services/trace_stuck.py`
+- Test: `tests/test_trace_stuck.py`
+
+- [ ] **Step 1: 写失败测试**
+
+创建 `tests/test_trace_stuck.py`：
+
+```python
+"""trace_stuck 单测:用 tmp 造 step_*.json(只含头部字段)验证卡死判据。"""
+import json
+from pathlib import Path
+
+from app.services.trace_stuck import (
+    StuckPoint,
+    dir_name_from_trace_url,
+    last_step,
+    read_stuck_points,
+)
+
+
+def _frame(pdir: Path, idx: int, step: str, page: str) -> None:
+    """造一帧 step json:头部放 pipeline_step/detected_page,尾部塞大 windows 模拟真实。"""
+    pdir.mkdir(parents=True, exist_ok=True)
+    body = {
+        "trace_id": "t", "step": idx, "platform": pdir.name,
+        "pipeline_step": step, "detected_page": page,
+        "windows": [{"nodes": ["x" * 200]}],
+    }
+    (pdir / f"step_{idx:03d}.json").write_text(
+        json.dumps(body, ensure_ascii=False), encoding="utf-8"
+    )
+
+
+def test_stuck_when_tail_repeats_same_step(tmp_path):
+    pdir = tmp_path / "meituan"
+    for i in range(20):
+        _frame(pdir, i, "add_one_dish", "meal_detail_popup")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.readable is True
+    assert res.points == [StuckPoint("meituan", "add_one_dish", 20)]
+
+
+def test_not_stuck_when_progressing(tmp_path):
+    pdir = tmp_path / "eleme"
+    _frame(pdir, 0, "set_address", "home")
+    for i in range(1, 8):
+        _frame(pdir, i, "enter_store", "store")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.readable is True
+    assert res.points == []
+
+
+def test_adding_many_dishes_not_stuck_when_page_changes(tmp_path):
+    # add_one_dish 重复但 detected_page 在跳(换菜)=推进,不判卡死
+    pdir = tmp_path / "meituan"
+    for i in range(20):
+        _frame(pdir, i, "add_one_dish", "menu" if i % 2 == 0 else "dish_popup")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.points == []
+
+
+def test_below_threshold_not_stuck(tmp_path):
+    pdir = tmp_path / "meituan"
+    for i in range(10):  # < 15
+        _frame(pdir, i, "add_one_dish", "meal_detail_popup")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.points == []
+
+
+def test_missing_dir_not_readable(tmp_path):
+    res = read_stuck_points(tmp_path / "nope", threshold=15, max_tail=40)
+    assert res.readable is False
+    assert res.points == []
+
+
+def test_empty_dir_no_platform_frames_not_readable(tmp_path):
+    (tmp_path / "emptysub").mkdir()
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.readable is False
+
+
+def test_per_platform_one_stuck_one_normal(tmp_path):
+    m = tmp_path / "meituan"
+    for i in range(18):
+        _frame(m, i, "add_one_dish", "meal_detail_popup")
+    e = tmp_path / "eleme"
+    _frame(e, 0, "set_address", "home")
+    for i in range(1, 6):
+        _frame(e, i, "enter_store", "store")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.readable is True
+    assert res.points == [StuckPoint("meituan", "add_one_dish", 18)]
+
+
+def test_last_step_returns_busiest_platform_last_env(tmp_path):
+    m = tmp_path / "meituan"
+    for i in range(20):
+        _frame(m, i, "add_one_dish", "meal_detail_popup")
+    e = tmp_path / "eleme"
+    for i in range(3):
+        _frame(e, i, "enter_store", "store")
+    sp = last_step(tmp_path)
+    assert sp == StuckPoint("meituan", "add_one_dish", 20)
+
+
+def test_dir_name_from_trace_url():
+    assert dir_name_from_trace_url("https://x/traces/20260804_1_abc/") == "20260804_1_abc"
+    assert dir_name_from_trace_url("https://x/traces/20260804_1_abc") == "20260804_1_abc"
+    assert dir_name_from_trace_url("") is None
+    assert dir_name_from_trace_url(None) is None
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `pytest tests/test_trace_stuck.py -q`
+Expected: FAIL（`ModuleNotFoundError: app.services.trace_stuck`）
+
+- [ ] **Step 3: 实现 trace_stuck.py**
+
+创建 `app/services/trace_stuck.py`：
+
+```python
+"""比价卡死定位:读 pricebot trace 末段,判某平台是否原地打转(卡死)。
+
+同机直读 {WORK_LOG_DIR}/{dir_name}/{platform}/step_*.json,只取头部字段
+(pipeline_step/detected_page),不解析后面的无障碍树(windows,占单帧 99% 体积)。
+判据与降级见 docs/superpowers/specs/2026-08-05-compare-stuck-detection-design.md。
+"""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from pathlib import Path
+
+# pipeline_step / 平台名 → 中文(映射不到原样显示英文,不阻断)。按 pricebot 实际枚举补全。
+PIPELINE_STEP_LABELS: dict[str, str] = {
+    "set_address": "定位",
+    "enter_store": "进店",
+    "add_one_dish": "加菜",
+    "match_dish": "找菜",
+    "checkout": "结算",
+}
+PLATFORM_LABELS: dict[str, str] = {
+    "meituan": "美团",
+    "eleme": "饿了么",
+    "jd_waimai": "京东外卖",
+}
+
+_PIPE_RE = re.compile(r'"pipeline_step":\s*"([^"]*)"')
+_PAGE_RE = re.compile(r'"detected_page":\s*"([^"]*)"')
+_STEP_NUM_RE = re.compile(r"step_(\d+)")
+
+
+@dataclass(frozen=True)
+class StuckPoint:
+    platform: str
+    pipeline_step: str
+    frames: int          # 末段连续困住的帧数(上限 max_tail)
+
+    def label(self) -> str:
+        p = PLATFORM_LABELS.get(self.platform, self.platform)
+        s = PIPELINE_STEP_LABELS.get(self.pipeline_step, self.pipeline_step)
+        return f"{p}·{s}"
+
+
+@dataclass(frozen=True)
+class StuckResult:
+    readable: bool               # trace 是否读到(区分「读不到」与「读到但没卡」)
+    points: list[StuckPoint]     # 卡死平台列表;readable=True 且空 = 确认没卡死
+
+
+def dir_name_from_trace_url(trace_url: str | None) -> str | None:
+    """.../traces/{dir_name}/ → dir_name;空/异常 → None。"""
+    if not trace_url:
+        return None
+    name = trace_url.rstrip("/").rsplit("/", 1)[-1]
+    return name or None
+
+
+def _step_num(path: Path) -> int:
+    m = _STEP_NUM_RE.search(path.name)
+    return int(m.group(1)) if m else -1
+
+
+def _read_head(path: Path, nbytes: int = 4096) -> tuple[str | None, str | None]:
+    """只读文件头部,抠 (pipeline_step, detected_page)。它们在 json 最前面。"""
+    with open(path, "r", encoding="utf-8") as f:
+        head = f.read(nbytes)
+    ps = _PIPE_RE.search(head)
+    pg = _PAGE_RE.search(head)
+    return (ps.group(1) if ps else None, pg.group(1) if pg else None)
+
+
+def _platform_stuck(
+    platform: str, step_files: list[Path], threshold: int, max_tail: int
+) -> StuckPoint | None:
+    """末帧往前数连续同 (pipeline_step, detected_page) 的帧数 ≥threshold → 卡死。"""
+    tail = step_files[-max_tail:]
+    heads = [_read_head(p) for p in tail]
+    last_ps, last_pg = heads[-1]
+    if last_ps is None:
+        return None
+    count = 0
+    for ps, pg in reversed(heads):
+        if ps == last_ps and pg == last_pg:
+            count += 1
+        else:
+            break
+    if count >= threshold:
+        return StuckPoint(platform, last_ps, count)
+    return None
+
+
+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
+        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
+            sp = _platform_stuck(pdir.name, step_files, threshold, max_tail)
+            if sp is not None:
+                points.append(sp)
+        if not any_frames:
+            return StuckResult(readable=False, points=[])
+        return StuckResult(readable=True, points=points)
+    except OSError:
+        return StuckResult(readable=False, points=[])
+
+
+def last_step(trace_dir: Path) -> StuckPoint | None:
+    """failed 用:取帧数最多平台的末帧环节(不要求原地打转)。读不到 → 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)
+            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
+        ps, _pg = _read_head(step_files[-1])
+        if ps is None:
+            return None
+        return StuckPoint(platform, ps, len(step_files))
+    except OSError:
+        return None
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `pytest tests/test_trace_stuck.py -q`
+Expected: PASS（9 passed）
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add app/services/trace_stuck.py tests/test_trace_stuck.py
+git commit -m "feat(compare-alert): trace_stuck 卡死判据(末段原地打转)
+
+Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
+```
+
+---
+
+## Task 3: compare_alert 抽出 fallback + 公开 make_hit
+
+**Files:**
+- Modify: `app/services/compare_alert.py`（`_hit`→`make_hit`；新增 `classify_cancelled_fallback`；cancelled 分支改调它）
+- Test: `tests/test_compare_alert_fallback.py`
+
+- [ ] **Step 1: 写失败测试**
+
+创建 `tests/test_compare_alert_fallback.py`：
+
+```python
+"""classify_cancelled_fallback / make_hit 单测。"""
+from app.services.compare_alert import classify_cancelled_fallback, make_hit
+
+
+class _Rec:
+    def __init__(self, **kw):
+        self.trace_id = kw.get("trace_id", "t")
+        self.status = kw.get("status", "cancelled")
+        self.total_ms = kw.get("total_ms")
+        self.step_count = kw.get("step_count")
+        self.fail_reason = kw.get("fail_reason")
+        self.information = kw.get("information")
+        self.app_version = kw.get("app_version")
+        self.created_at = kw.get("created_at")
+        self.trace_url = kw.get("trace_url")
+        self.user_id = kw.get("user_id")
+
+
+def test_fallback_deep_by_ms():
+    hit = classify_cancelled_fallback(
+        _Rec(total_ms=95000, step_count=5),
+        cancelled_ms_threshold=90000, cancelled_step_threshold=30,
+    )
+    assert hit is not None and hit.alert_type == "T5" and "深度放弃" in hit.reason
+
+
+def test_fallback_deep_by_step():
+    hit = classify_cancelled_fallback(
+        _Rec(total_ms=1000, step_count=35),
+        cancelled_ms_threshold=90000, cancelled_step_threshold=30,
+    )
+    assert hit is not None and hit.alert_type == "T5"
+
+
+def test_fallback_shallow_none():
+    hit = classify_cancelled_fallback(
+        _Rec(total_ms=5000, step_count=3),
+        cancelled_ms_threshold=90000, cancelled_step_threshold=30,
+    )
+    assert hit is None
+
+
+def test_make_hit_carries_fields():
+    hit = make_hit(_Rec(trace_id="tx", app_version="0.6.0"), "T5", "卡在 美团·加菜")
+    assert hit.trace_id == "tx"
+    assert hit.reason == "卡在 美团·加菜"
+    assert hit.app_version == "0.6.0"
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `pytest tests/test_compare_alert_fallback.py -q`
+Expected: FAIL（`ImportError: cannot import name 'classify_cancelled_fallback'`）
+
+- [ ] **Step 3: 改 compare_alert.py**
+
+在 `app/services/compare_alert.py`：
+
+(a) 把 `def _hit(` 改名为 `def make_hit(`（第 33 行），并把 `classify_record` 内 4 处 `_hit(` 调用改成 `make_hit(`（原 T1/T6/T2/T5 分支）。
+
+(b) 在 `make_hit` 之后、`classify_record` 之前，新增：
+
+```python
+def classify_cancelled_fallback(
+    rec: Any,
+    *,
+    cancelled_ms_threshold: int,
+    cancelled_step_threshold: int,
+) -> AlertHit | None:
+    """cancelled 保底判定(读不到 trace 时用):超耗时或步数阈值 → T5 深度放弃,否则 None。纯函数。"""
+    ms = rec.total_ms
+    step = rec.step_count
+    deep = (ms is not None and ms > cancelled_ms_threshold) or (
+        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 None
+```
+
+(c) 把 `classify_record` 里的 cancelled 分支（原 `if status == "cancelled":` 那整段）替换为：
+
+```python
+    if status == "cancelled":
+        return classify_cancelled_fallback(
+            rec,
+            cancelled_ms_threshold=cancelled_ms_threshold,
+            cancelled_step_threshold=cancelled_step_threshold,
+        )
+    return None
+```
+
+（`classify_record` 行为不变，只是把 cancelled 逻辑抽到 `classify_cancelled_fallback`。）
+
+- [ ] **Step 4: 跑测试确认通过（含现有 rules 测试不回归）**
+
+Run: `pytest tests/test_compare_alert_fallback.py tests/test_compare_alert_rules.py -q`
+Expected: PASS（新测试 4 passed，现有 rules 测试仍全 PASS）
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add app/services/compare_alert.py tests/test_compare_alert_fallback.py
+git commit -m "feat(compare-alert): 抽出 classify_cancelled_fallback + 公开 make_hit
+
+Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
+```
+
+---
+
+## Task 4: worker 编排 build_hits
+
+**Files:**
+- Modify: `app/core/compare_alert_worker.py`（加 imports、`_trace_dir`、`build_hits`；`_scan_and_alert` 改用 `build_hits`）
+- Test: `tests/test_compare_alert_stuck_worker.py`
+
+- [ ] **Step 1: 写失败测试**
+
+创建 `tests/test_compare_alert_stuck_worker.py`：
+
+```python
+"""build_hits 集成测:cancelled trace 优先/保底切换、failed 附卡点、限量。"""
+import json
+from pathlib import Path
+
+from app.core.compare_alert_worker import build_hits
+
+
+class _Rec:
+    def __init__(self, **kw):
+        self.trace_id = kw.get("trace_id", "t")
+        self.status = kw.get("status", "cancelled")
+        self.total_ms = kw.get("total_ms")
+        self.step_count = kw.get("step_count")
+        self.fail_reason = kw.get("fail_reason")
+        self.information = kw.get("information")
+        self.app_version = kw.get("app_version")
+        self.created_at = kw.get("created_at")
+        self.trace_url = kw.get("trace_url")
+        self.user_id = kw.get("user_id")
+
+
+def _frame(pdir: Path, idx: int, step: str, page: str) -> None:
+    pdir.mkdir(parents=True, exist_ok=True)
+    body = {"pipeline_step": step, "detected_page": page, "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,
+    timeout_keywords=("超时",), unrecognized_keywords=("未识别",), biz_exclude_keywords=(),
+)
+
+
+def test_cancelled_stuck_reports_via_trace(tmp_path):
+    for i in range(18):
+        _frame(tmp_path / "20260804_x" / "meituan", i, "add_one_dish", "meal_detail_popup")
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_x/",
+               total_ms=5000, step_count=3)  # 保底不会中,靠 trace 判卡死
+    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
+
+
+def test_cancelled_readable_not_stuck_no_report(tmp_path):
+    # trace 确认没卡(在推进);即便 total_ms/step 超阈值也不报(信 trace,不回退保底)
+    p = tmp_path / "20260804_y" / "eleme"
+    _frame(p, 0, "set_address", "home")
+    for i in range(1, 6):
+        _frame(p, i, "enter_store", "store")
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_y/",
+               total_ms=95000, step_count=40)
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
+    assert hits == []
+
+
+def test_cancelled_unreadable_falls_back(tmp_path):
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/nope/",
+               total_ms=95000, step_count=3)
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
+    assert len(hits) == 1
+    assert hits[0].alert_type == "T5" and "深度放弃" in hits[0].reason
+
+
+def test_no_work_log_dir_uses_fallback(tmp_path):
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/y/",
+               total_ms=95000, step_count=3)
+    hits = build_hits([rec], work_log_dir="", **_KW)
+    assert len(hits) == 1 and "深度放弃" in hits[0].reason
+
+
+def test_failed_gets_stuck_point_appended(tmp_path):
+    for i in range(20):
+        _frame(tmp_path / "20260804_f" / "meituan", i, "add_one_dish", "meal_detail_popup")
+    rec = _Rec(status="failed", fail_reason="启动超时",
+               trace_url="https://x/traces/20260804_f/")
+    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
+
+
+def test_max_trace_reads_zero_skips_trace(tmp_path):
+    for i in range(18):
+        _frame(tmp_path / "20260804_z" / "meituan", i, "add_one_dish", "meal_detail_popup")
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_z/",
+               total_ms=95000, step_count=3)
+    kw = {**_KW, "max_trace_reads": 0}
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **kw)
+    # 没读 trace → 回退保底 → deep(95s) → 深度放弃
+    assert len(hits) == 1 and "深度放弃" in hits[0].reason
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `pytest tests/test_compare_alert_stuck_worker.py -q`
+Expected: FAIL（`ImportError: cannot import name 'build_hits'`）
+
+- [ ] **Step 3: 改 compare_alert_worker.py**
+
+(a) 顶部 imports 段，把
+```python
+from app.services.compare_alert import classify_batch
+from app.services.compare_alert_format import ALERT_KEYWORD, format_alert_post
+```
+改为
+```python
+from dataclasses import replace as _dc_replace
+
+from app.services import trace_stuck
+from app.services.compare_alert import (
+    classify_cancelled_fallback,
+    classify_record,
+    make_hit,
+)
+from app.services.compare_alert_format import ALERT_KEYWORD, format_alert_post
+```
+
+(b) 在 `_scan_and_alert` 之前新增两个函数：
+
+```python
+def _trace_dir(base: Path, trace_url: str | None) -> Path | None:
+    name = trace_stuck.dir_name_from_trace_url(trace_url)
+    if not name:
+        return None
+    return base / name
+
+
+def build_hits(
+    records: list,
+    *,
+    work_log_dir: str,
+    stuck_threshold: int,
+    max_tail: int,
+    max_trace_reads: int,
+    cancelled_ms_threshold: int,
+    cancelled_step_threshold: int,
+    timeout_keywords: tuple[str, ...],
+    unrecognized_keywords: tuple[str, ...],
+    biz_exclude_keywords: tuple[str, ...],
+) -> list:
+    """编排:cancelled 走 trace 优先(读到确认没卡则不报,读不到回退保底);failed 附卡点。
+
+    trace 读取限量 max_trace_reads 次/轮;任何 trace 异常都在 trace_stuck 内部降级为
+    「读不到」,cancelled 因而回退保底、failed 不附卡点,绝不影响报警发送。
+    """
+    base = Path(work_log_dir) if work_log_dir else None
+    reads = 0
+    hits: list = []
+    for rec in records:
+        if rec.status == "cancelled":
+            res = None
+            if base is not None and reads < max_trace_reads:
+                td = _trace_dir(base, rec.trace_url)
+                if td is not None:
+                    res = trace_stuck.read_stuck_points(
+                        td, threshold=stuck_threshold, max_tail=max_tail
+                    )
+                    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)
+                else:
+                    hit = None  # 读到且确认没卡 → 不报
+            else:
+                hit = classify_cancelled_fallback(
+                    rec,
+                    cancelled_ms_threshold=cancelled_ms_threshold,
+                    cancelled_step_threshold=cancelled_step_threshold,
+                )
+            if hit is not None:
+                hits.append(hit)
+        else:
+            hit = classify_record(
+                rec,
+                cancelled_ms_threshold=cancelled_ms_threshold,
+                cancelled_step_threshold=cancelled_step_threshold,
+                timeout_keywords=timeout_keywords,
+                unrecognized_keywords=unrecognized_keywords,
+                biz_exclude_keywords=biz_exclude_keywords,
+            )
+            if (
+                hit is not None
+                and hit.alert_type in ("T1", "T2", "T6")
+                and base is not None
+                and reads < max_trace_reads
+            ):
+                td = _trace_dir(base, rec.trace_url)
+                if td is not None:
+                    sp = trace_stuck.last_step(td)
+                    reads += 1
+                    if sp is not None:
+                        hit = _dc_replace(hit, reason=f"{hit.reason}｜卡在 {sp.label()}")
+            if hit is not None:
+                hits.append(hit)
+    return hits
+```
+
+(c) 在 `_scan_and_alert` 里，把
+```python
+        hits = classify_batch(
+            records,
+            cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
+            cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
+            timeout_keywords=settings.compare_alert_timeout_keywords,
+            unrecognized_keywords=settings.compare_alert_unrecognized_keywords,
+            biz_exclude_keywords=settings.compare_alert_biz_exclude_keywords,
+        )
+```
+替换为
+```python
+        hits = build_hits(
+            records,
+            work_log_dir=settings.COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR,
+            stuck_threshold=settings.COMPARE_ALERT_STUCK_FRAME_THRESHOLD,
+            max_tail=settings.COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES,
+            max_trace_reads=settings.COMPARE_ALERT_TRACE_MAX_RECORDS,
+            cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
+            cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
+            timeout_keywords=settings.compare_alert_timeout_keywords,
+            unrecognized_keywords=settings.compare_alert_unrecognized_keywords,
+            biz_exclude_keywords=settings.compare_alert_biz_exclude_keywords,
+        )
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `pytest tests/test_compare_alert_stuck_worker.py -q`
+Expected: PASS（6 passed）
+
+- [ ] **Step 5: 跑报警相关全量测试确认不回归**
+
+Run: `pytest tests/ -q -k "compare_alert or trace_stuck"`
+Expected: PASS（全绿）
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add app/core/compare_alert_worker.py tests/test_compare_alert_stuck_worker.py
+git commit -m "feat(compare-alert): worker 编排 build_hits(cancelled trace 优先+保底、failed 附卡点)
+
+Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
+```
+
+---
+
+## 收尾
+
+- [ ] **全量测试**：`pytest -q`（对齐 preexisting 失败基线，不新增失败）
+- [ ] **lint**：`ruff check app/services/trace_stuck.py app/core/compare_alert_worker.py app/services/compare_alert.py`
+- [ ] **本地联调（可选）**：把 `.env` 的 `COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR` 指向本地 `E:\project\pricebot-backend\data\work_logs`，用真实 cancelled trace 目录验证卡点文案。
+- [ ] **清理临时脚本**：`git rm --cached` 无关，直接删 `scripts/_probe_trace_timing.py`（若确认不再用，另行确认 `scripts/_test_alert_card.py`）。
+
+## 不在本 plan（后续单独排）
+
+- 卡片 schema 2.0 table 组件固化（`format_alert_card` + `send_feishu_card`，当前仍在 `scripts/_test_alert_card.py`）——卡点已随 `reason` 在现有 `format_alert_post` 展示，不阻塞本功能。
+- pricebot 侧改动（本方案零改 pricebot）。
+- `PIPELINE_STEP_LABELS` 全枚举补全（映射不到原样英文，可随线上观察增量补）。

docs/superpowers/specs/2026-08-04-compare-fail-alert-design.md

@@ -0,0 +1,197 @@
+# 比价失败报警机制 · 设计文档
+
+- **日期**：2026-08-04
+- **状态**：设计待评审（v2，含数据复审修订）
+- **范围**：app-server（`shaguabijia-app-server`）
+- **数据源**：`comparison_record` 单表（线上快照已导入本地 `cr_analysis` 分析库，3867 行，覆盖 2026-06-09 ~ 08-04）
+
+## 复审修订记录（2026-08-04, v2）
+
+结合 `cr_analysis` 实测数据复审后的改动：
+1. **[必修] 水位改用 `updated_at`**：原 `created_at` 水位会系统性漏报慢失败（落定延迟 p99 达 7.5–10min）。改为给 `comparison_record` 加 `updated_at` 列、水位按 `updated_at` 单调推进（§5/§6）。
+2. **新增规则 T6「商品识别失败」**：「未识别到商品」96 条纳入报警，作为识别能力信号（§3）。
+3. **T1 加业务词排除**：清掉 `fail_reason IS NULL` 里 7 条 `information` 实为业务的误报（§3）。
+4. 补充：NULL 语义、时区口径、`business_type` 复核、单窗口截断阈值（§7/§9）。
+5. 数据证伪、未采纳的改动：T2 关键词已完备（35 条技术词全被「超时/启动/加载」覆盖），不扩。
+
+---
+
+## 1. 背景与目标
+
+比价（外卖 `business_type=food`）由客户端无障碍自动化 + pricebot 多平台 LLM 驱动，链路长、失败形态多。目前**没有任何主动发现失败的手段**——只能人工翻库或等用户反馈。
+
+**目标**：新增一个**近实时、记录级**的比价失败报警机制。每隔 15–30 分钟扫描新落定的比价记录，逐条按预定规则判定「是否属于需要关注的失败」，把命中的记录**汇总成一条飞书消息**发到告警群，并**报出每条触发的原因**。
+
+**关键设计取向**（均由数据分析与评审确认）：
+- **粒度是「记录」不是「失败率」**：逐条判定，不算比率、不设样本量门槛、不做基线对比。日均比价量小（完成约 45 条/天），比率方案在小窗口会剧烈抖动；记录级方案规避了这个问题。
+- **只报「技术性失败」「识别失败」与「深度放弃」**，不报正常业务结局。
+- **有触发才发，无触发静默**：不刷屏。
+
+## 2. 数据分析依据（基线）
+
+全量 3867 条记录级 `status` 分布（详见附录 A）：
+
+| status | 数量 | 占比 | 说明 |
+|---|---|---|---|
+| success | 1501 | 38.8% | 成功（含 `below_minimum` 未满起送，被归一为 success） |
+| cancelled | 1394 | 36.0% | 用户中途退出 |
+| failed | 957 | 24.7% | 失败（T1 技术 386 + T2 超时 35 + T6 识别 96 + 业务 440） |
+| running | 15 | 0.4% | 悬挂未收尾 |
+
+支撑规则设计的关键事实：
+- **`failed` 是混合桶**：`fail_reason IS NULL` 的 393 条是纯系统技术失败（其中「比价过程出错，请稍后重试」占 294，是 `_GENERIC_INFO` 兜底黑话）；`fail_reason` 非空的多为业务结局，但夹杂「启动淘宝超时」等技术问题 35 条、「未识别到商品」96 条。
+- **业务失败不该报**：打烊、无此店、无此菜、未起送、单点不配送是正常结局。
+- **落定延迟很长**：`total_ms`（≈ 记录从建行到落定的时长）p99 = failed 449s、cancelled 602s，max 16min。**这是水位必须用 `updated_at` 而非 `created_at` 的直接依据**。
+- **cancelled 缺退出上下文**：99.3% 终止原因就一句「用户终止比价」，且 100% 没有 `platforms`/结果数据（都在 `running` 阶段被中止）。唯一可用信号是退出时机（`total_ms`/`step_count`）。cancelled 的 `step_count` 中位 5、p90 31；`total_ms` 中位 24s、p90 124s。参照系：一次成功比价中位 113s / 38 步。
+
+## 3. 报警规则（v1）
+
+worker 每轮查询「上次水位之后有更新」的记录，对每条按下表判定；命中任一即计入本期汇总。四个规则互斥（一条记录最多归一类）。
+
+| 类型 | 判定条件（SQL 语义） | 触发原因文案 | 历史量(2月) |
+|---|---|---|---|
+| **T1 系统技术失败** | `status='failed' AND fail_reason IS NULL AND (information IS NULL OR information !~ 业务词)` | `技术失败·{information 去空白截断; 空则"比价过程出错"}` | ≈386 |
+| **T6 商品识别失败** | `status='failed' AND fail_reason ~ '未识别'` | `识别失败·未识别到商品` | ≈96 |
+| **T2 超时/启动失败** | `status='failed' AND fail_reason IS NOT NULL AND fail_reason ~ 超时关键词` | `{fail_reason}`（如「启动淘宝超时」） | ≈35 |
+| **T5 cancelled 深度放弃** | `status='cancelled' AND (total_ms > 90000 OR step_count > 30)` | `深度放弃·等待 {total_ms/1000 取整}s / {step_count} 步后退出` | ≈250 |
+
+**判定顺序（保证互斥）**：
+1. `status='failed'`：
+   - `fail_reason IS NULL` → 若 `information` 命中**业务词**（`未找到|打烊|起送|门店|店内|不配送|这些菜|未入驻|休息`）则**不报**（业务失败漏派生 fail_reason，约 7 条）；否则 **T1**。
+   - `fail_reason` 含「未识别」→ **T6**。
+   - `fail_reason` 含超时关键词（`超时|启动|加载`）→ **T2**。
+   - 其余（干净业务原因）→ **不报**。
+2. `status='cancelled'` 且（`total_ms>90000` 或 `step_count>30`）→ **T5**；否则不报。
+3. `status IN ('success','running')` → 不报。
+
+**阈值/关键词（可配初值）**：T5 的 `90000ms`/`30步` 取自 cancelled 分布约 p90（评审选定「B 中档」）。超时关键词 `超时,启动,加载`、识别关键词 `未识别`、业务排除词均可配。
+
+每条命中记录在汇总里附带：`trace_id`、`app_version`、`business_type`、触发原因文案、`created_at`。
+
+## 4. 非目标与暂缓项
+
+| 项 | 处理 | 原因 |
+|---|---|---|
+| 业务失败（打烊/无店/无菜/未起送/单点不配送） | **不报** | 正常业务结局 |
+| success、早退 cancelled（≤90s 且 ≤30 步） | **不报** | 无报警价值 |
+| **T3 running 悬挂** | **本期暂缓** | 评审决定先不报；但见下方 🔴 |
+| **T4 单平台适配失效**（`platforms[].status='failed'`） | 暂不纳入（未来增强） | 量大、与整体失败重叠、噪音高 |
+| 失败率 / cancelled 率等**比率型**指标 | 不做 | 本设计是记录级 |
+
+> 🔴 **待独立排查的回归线索**（非本报警范围，留档）：`running` 悬挂 15 条**全部集中在 2026-07-28 之后**，此前两个月几乎为 0。强烈提示某次发版后 `harvest_done`/`harvest_abort` 收尾链路（`app/repositories/comparison.py`）回归，建议单独开 issue。排查确认后可在 v2 把 T3 作为独立高频告警加回。
+
+## 5. 架构与组件
+
+沿用项目现有**常驻 asyncio worker** 范式（与 `heartbeat_monitor_worker` 等一致）。各组件单一职责、可独立测试：
+
+| 组件 | 路径 | 职责 |
+|---|---|---|
+| **数据模型改动** | `app/models/comparison.py` + alembic 迁移 | `comparison_record` 新增 `updated_at`（`server_default=func.now()`, `onupdate=func.now()`）+ 索引 `ix_comparison_updated`。为水位提供单调递增的落定时间。 |
+| **规则模块** | `app/services/compare_alert.py` | 纯函数：输入一批 ORM 记录 → 输出 `[(记录, 触发类型, 原因文案)]`。判定逻辑与阈值全在此，无 I/O，易测易调。 |
+| **扫描 worker** | `app/core/compare_alert_worker.py` | 仿 `heartbeat_monitor_worker`：单实例文件锁 + `asyncio` 轮询 + 优雅退出。每轮：读水位 → 查有更新记录 → 调规则 → 有命中则格式化并发飞书 → 推进水位。 |
+| **飞书通知器** | `app/integrations/feishu_notifier.py` | 实现群机器人 webhook 发送。发送失败抛异常由 worker 处理。 |
+| **水位存储** | 复用 `app_config` 表 | key=`compare_alert.last_watermark`，value=上次处理的最大 `updated_at`。 |
+| **启停挂载** | `app/main.py` lifespan | `start_compare_alert_worker()` / `stop_compare_alert_worker()`，与现有 worker 同处注册。 |
+
+> **`onupdate` 生效前提**：现有 `harvest_done`/`harvest_abort`/`upsert_record` 均走 ORM `setattr`+`commit` 更新，`onupdate=func.now()` 会自动刷新 `updated_at`，无需改写路径。
+
+## 6. 数据流与水位管理（updated_at 方案）
+
+```
+每 interval 秒：
+  读 app_config['compare_alert.last_watermark'] → watermark
+    （空 → 冷启动：watermark = 当前 max(updated_at)，只报之后新落定的，不回溯历史）
+  查 comparison_record
+     WHERE updated_at > watermark
+     ORDER BY updated_at ASC
+  逐条套 T1/T6/T2/T5 规则 → 命中集合（按类型分组）
+  若命中集合非空：
+     格式化飞书消息 → feishu_notifier.send()
+       成功 → 水位 = 本批 max(updated_at)
+       失败 → 不更新水位（log），下一轮重扫补发
+  若命中集合为空：
+     水位 = 本批 max(updated_at)（无记录则不动；可选 SEND_EMPTY 发简讯）
+```
+
+- **零漏报**：任何记录落定/更新时 `updated_at` 刷新为当前 DB 时钟 > 水位，必被下一轮扫到——无论 `created_at` 多早、落定多慢（根治了 `created_at` 水位漏掉慢失败的问题）。
+- **规避时区**：水位存的是 DB 产出的 `updated_at` 值，查询用 `updated_at` 自身比较，**不依赖 worker 本地时钟与 DB 时钟对齐**（`created_at` 存 naive 北京、`func.now()` 为 DB 时钟，二者口径不同，但本方案只用 `updated_at` 自比较，不受影响）。
+- **发送失败不推进水位**：保证不漏；恢复后一次补发。
+- **一条记录可能被扫多次**（running 更新→落定更新，`updated_at` 变两次）：但只有落定后 `status` 才命中规则，running 阶段扫到不命中，无副作用；不会重复报。
+
+## 7. 飞书消息格式
+
+群机器人消息（文本或富文本 `post`），按类型分组：
+
+```
+🚨 比价失败报警 · 2026-08-04 08:00–08:30 · 本期触发 7 条
+• 系统技术失败 3 条
+   - trace abc123 | v0.3.4 | 比价过程出错
+• 商品识别失败 2 条
+   - trace abc200 | v0.5.1 | 未识别到商品
+• 启动/超时失败 1 条
+   - trace def456 | v0.3.4 | 启动淘宝超时
+• 深度放弃(cancelled) 1 条
+   - trace ghi789 | v0.6.3 | 等待 98s / 26 步后退出
+```
+
+- **截断阈值**：单类型明细超 `MAX_DETAIL_PER_TYPE`（默认 20）条时，只列前 20 条 + 「另有 N 条」；本期总命中超 `MAX_TOTAL`（默认 50）条时降级为只给各类型计数，提示去分析库查（防报警风暴，如 07-14 那种高失败日）。
+
+## 8. 配置项
+
+`app/core/config.py`（`pydantic-settings`）：
+
+| 配置 | 默认 | 说明 |
+|---|---|---|
+| `COMPARE_ALERT_ENABLED` | `False` | 总开关；关时 worker 不启动 |
+| `COMPARE_ALERT_SCAN_INTERVAL_SEC` | `1800` | 扫描间隔，可配 900（15min） |
+| `COMPARE_ALERT_FEISHU_WEBHOOK` | `""` | 群机器人 webhook；空则 worker 仅打日志不外发 |
+| `COMPARE_ALERT_CANCELLED_MS_THRESHOLD` | `90000` | T5 耗时阈值（ms） |
+| `COMPARE_ALERT_CANCELLED_STEP_THRESHOLD` | `30` | T5 步数阈值 |
+| `COMPARE_ALERT_TIMEOUT_KEYWORDS` | `"超时,启动,加载"` | T2 关键词 |
+| `COMPARE_ALERT_UNRECOGNIZED_KEYWORDS` | `"未识别"` | T6 关键词 |
+| `COMPARE_ALERT_BIZ_EXCLUDE_KEYWORDS` | `"未找到,打烊,起送,门店,店内,不配送,这些菜,未入驻,休息"` | T1 的 information 业务词排除 |
+| `COMPARE_ALERT_MAX_DETAIL_PER_TYPE` | `20` | 单类型明细截断 |
+| `COMPARE_ALERT_MAX_TOTAL` | `50` | 本期总命中截断（超则只给计数） |
+| `COMPARE_ALERT_SEND_EMPTY` | `False` | 无命中是否发「本期无异常」简讯 |
+
+> **`business_type` 复核**：当前数据全为 `food`，规则未按 `business_type` 限定。接入 `ecom`/`coupon` 时需复核各规则（尤其 T2/T6 关键词与 T5 阈值是否仍适用）。
+
+## 9. 错误处理与边界
+
+- **worker 单轮异常吞掉不退出**（`except Exception: logger.exception`），仿 heartbeat。
+- **DB / 飞书发送异常**：log，本轮不推进水位，下轮重试补发。
+- **单实例锁**：文件锁 `data/compare_alert.lock`（O_CREAT|O_EXCL + stale 检测）。
+- **NULL 语义**：T5 中 `total_ms`/`step_count` 为 NULL 的 cancelled，SQL 比较 `NULL>90000` 为 false → 不命中（无数据不报，符合预期）。T1 中 `information IS NULL` 时业务词排除不触发（视为非业务）→ 仍属 T1，原因文案兜底「比价过程出错」。
+- **冷启动不回溯历史**：首次启动水位=当前 `max(updated_at)`，避免把历史失败一次性全报。
+
+## 10. 测试策略
+
+仿现有 `tests/` 风格（`TestClient` + monkeypatch 外部依赖，SQLite 临时库）：
+
+- `test_compare_alert_rules.py`：喂各类记录（T1/T6/T2/T5 命中样本 + 业务失败/success/早退 cancelled/running 反例 + T1 业务词误入反例），断言分类与原因文案；覆盖阈值边界（`total_ms=90000` 不命中、`90001` 命中）与 NULL 语义。
+- `test_compare_alert_worker.py`：monkeypatch notifier 与 `SessionLocal`，验证 `updated_at` 水位推进、发送失败不推进、冷启动=max、命中汇总、截断逻辑。
+- `test_feishu_notifier.py`：monkeypatch HTTP，验证消息体格式与发送失败抛异常。
+- 迁移测试：`updated_at` 列 + 索引存在，`onupdate` 在 ORM 更新时刷新。
+
+## 11. 未来增强
+
+1. **T3 running 悬挂告警**：待第 4 节 🔴 回归排查后，作为独立高频告警加回。
+2. **T4 单平台适配失效**：`platforms[].status='failed'` 逐平台维度。
+3. **cancelled 退出上下文埋点**：客户端终止时上报退出阶段、已比出平台数、是否已看到中间结果——让 cancelled 从「只有时机」升级为「可归因」。
+4. **分维度统计**：汇总附带按 `app_version`/`source_platform` 的命中分布，辅助定位回归版本/平台。
+5. **趋势型报警**：记录级之上叠加比率/环比（需另设样本量保护）。
+
+## 附录 A：分析数据来源与复现
+
+- **来源**：线上 PostgreSQL 16 `pg_dump` 单表 `comparison_record`（plain SQL，187MB）。
+- **本地环境**：Docker 容器 `shaguabijia-pg`（postgres:16-alpine），独立分析库 `cr_analysis`（用户 `shaguabijia_app`）。导入：`docker cp` dump 进容器后 `psql -f`（末尾外键引用 `public.user` 报错属预期，单表 dump 无 user 表，不影响数据与索引）。
+- **样本**：3867 行，2026-06-09 ~ 08-04。
+- **关键分布**（供实现期回归对照）：
+
+| 指标 | success | failed | cancelled |
+|---|---|---|---|
+| 数量 | 1501 | 957 | 1394 |
+| step_count 中位 / p90 | 38 / 61 | 22 / 47 | 5 / 31 |
+| total_ms 中位 / p90 / p99 | 113s / 200s / 391s | 77s / 168s / 449s | 24s / 124s / 602s |
+
+- **failed 细分**（合计 957）：T1 系统技术 386（`fail_reason IS NULL` 393 − 业务误入 7）、T2 超时/启动 35、T6 识别失败 96、业务失败 440（含误入的 7 条）。

docs/superpowers/specs/2026-08-05-compare-stuck-detection-design.md

@@ -0,0 +1,168 @@
+# 比价「卡死定位」报警增强设计
+
+- 日期：2026-08-05
+- 分支：feat-compare-fail-alert（延续一期）
+- 关联：`docs/superpowers/specs/2026-08-04-compare-fail-alert-design.md`（一期报警）
+
+## 1. 背景与问题
+
+一期报警对 **cancelled（中途退出）** 的判据（`app/services/compare_alert.py` 的 `classify_record`）：
+
+```
+cancelled 且 (total_ms > 90s 或 step_count > 30步) → T5 深度放弃
+```
+
+这个判据量的是「投入多少」，不是「卡没卡」，两头都错：
+
+- **漏报**：一进平台就卡在登录墙 / 加载失败，5 秒 2 步就退 → 判「不深度」→ 不报。但这是真卡死。
+- **误报**：用户正常挑了 100 秒、点了 40 步，比完价不满意退了 → `>90s` → 报「深度放弃」。但根本没卡。
+
+根因：`total_ms`/`step_count` 是**整场**的量，把「卡在一步反复失败」和「正常深度使用」混为一谈。
+
+### 1.1 数据佐证（真实 trace）
+
+- **卡死例**（`20260804_114703` meituan）：`pipeline_step` 从 `set_address`(step 0-1) → `enter_store`(3-8) → **`add_one_dish`(9 一路到 120+，110+ 帧全困在这一个环节)**。且 `timing.json` 里根本没有 meituan——`step_profiler` 只在平台 `is_done` 时落 timing，卡死平台永不 done。
+- **正常例**（`20260803_165239` eleme）：`set_address`(0) → `enter_store`(1-7) → done，每个环节 ≤7 帧就推进走了。
+
+**卡死的结构特征**：某 `pipeline_step` 连续几十上百帧不变（原地打转）；正常则是逐环节推进、单环节 ≤7 帧。两者空档极大（7 vs 110+），可用一个帧数阈值干净区分，且**不需要大量数据归纳环节语义**。
+
+## 2. 目标
+
+- cancelled 判据：从「整场耗时/帧数阈值」→「trace 末段原地打转」，抓真卡死（含短时卡死）、不误报正常深度使用。
+- **判定与展示一体**：直接报「卡在 平台·环节」。
+- failed 类（T1/T2/T6）：判定不变，best-effort 补卡点定位。
+- 稳：读不到 trace 回退原耗时/帧数保底，**绝不阻断报警发送**。
+
+## 3. 取数：同机直读（不改 pricebot）
+
+app-server 与 pricebot **同机**。trace 落盘在 `{WORK_LOG_DIR}/{dir_name}/`：
+
+- **dir_name 从 `comparison_record.trace_url` 尾段抠**：`trace_url = {base}/traces/{dir_name}/`，尾段就是磁盘目录名，新老格式都对得上，规避从 `trace_id` 反推老格式「首帧时刻」的难题。
+- 只读末段帧的**头部字段**（`pipeline_step` / `detected_page`），不解析后面的无障碍树（`windows`，占单帧 99% 体积）。
+- 不改 pricebot、不需要 `INTERNAL_API_SECRET`、不走网络。
+
+> 备选途径（已否决）：pricebot 加内部接口（要改两仓 + secret）、公网 `trace_url` GET timing.json（本地 SSL 大面积超时 + timing.json 缺卡死平台）。同机直读最优。
+
+## 4. 架构分层
+
+保持判定纯函数、IO 单独成层：
+
+| 模块 | 职责 | 性质 |
+|---|---|---|
+| `services/compare_alert.py`（微调） | failed 判定不变；cancelled 只保留**回退保底**判定（`>90s`/`>30步`） | 纯函数 |
+| `services/trace_stuck.py`（新） | 给定 trace 目录 → 逐平台读末段 → 判「原地打转」→ 返回卡点列表 | 薄 IO + 纯逻辑 |
+| `core/compare_alert_worker.py`（编排） | 先跑纯 `classify_batch` 出候选，再对候选调 `trace_stuck` 增强 | 编排 |
+
+## 5. trace_stuck 模块
+
+### 5.1 卡死判据（逐平台）
+
+对某平台的 `step_*.json` 序列，从**末帧往前**数，连续 `(pipeline_step, detected_page)` 都相同的帧数 ≥ N → 判该平台卡死，卡点 = 该 `pipeline_step`。
+
+- `N = COMPARE_ALERT_STUCK_FRAME_THRESHOLD`（默认 **15**；正常环节 ≤7 帧、卡死 110+ 帧，空档极大）。
+- **「无推进」= `(pipeline_step, detected_page)` 双不变**（页面没跳转、环节没变）。这样能区分：
+  - 「加多菜」：`pipeline_step` 相同但 `detected_page` 在跳（换菜/回菜单）= 推进 → 不判卡死；
+  - 「卡在一步」：两者都不变 = 原地打转 → 卡死。
+- 从末帧往前最多读 `COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES`（默认 **40**）帧，够判 ≥N 即停，防超长 trace 全读。
+
+### 5.2 逐平台聚合（B 方案：不漏）
+
+一条 trace **逐平台**判，所有卡死平台都收集——不只「帧数最多」的那个。因为「帧数最多」会在**卡死平台帧数不是最多**时漏报（如另一平台正常加了 8 道菜跑了 30 帧、卡死平台一进就卡登录 5 帧退），而那恰是短时卡死。多个卡死平台都列进 reason。
+
+### 5.3 接口
+
+```python
+@dataclass(frozen=True)
+class StuckPoint:
+    platform: str
+    pipeline_step: str
+    frames: int          # 末段连续困住的帧数
+
+@dataclass(frozen=True)
+class StuckResult:
+    readable: bool               # trace 是否读到（区分「读不到」与「读到但没卡」）
+    points: list[StuckPoint]     # 卡死平台列表；readable=True 且空 = 确认没卡死
+
+def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> StuckResult:
+    """逐平台判卡死。目录不存在/异常/无平台帧 → StuckResult(readable=False, points=[])。"""
+
+def last_step(trace_dir: Path) -> StuckPoint | None:
+    """failed 用：取帧数最多平台的末帧环节（不要求原地打转）。读不到 → None。"""
+```
+
+## 6. 判定流
+
+### 6.1 cancelled（trace 优先 → 保底）
+
+```
+worker 对 cancelled 候选:
+  res = read_stuck_points(dir)
+  if res.readable and res.points:   # 读到且有卡死平台 → 报卡死（带卡点环节）
+      报 T5, reason="深度放弃", stuck_point = "、".join(f"{平台}·{环节}" for res.points)
+  else:                             # 其余一律回退耗时/步数兜底(见下方 2026-08-06 修订)
+      >90s或>30步 → T5「深度放弃」(卡点列留空); 否则不报
+```
+
+> **2026-08-06 修订(compare-fail-alert 排查)**:原设计「读到且没卡死 → 不报」在线上是死路——
+> **线上 trace 几乎总可读**(WORK_LOG_DIR 已配、同机直读),于是耗时兜底那条分支基本永不触发,
+> `total_ms>90s` 阈值形同虚设,**超长放弃(实测 113s / 516s)一条都报不出来**。改为:只有「判出卡点」
+> 独占带卡点的 T5;**其余(可读没卡点 / 读不到 / 无 trace)一律回退耗时兜底**,超长照报(卡点留空)。
+> 权衡:这会重新引入第 1 节「误报」——用户正常浏览 90s+ 后退出也会报。若噪音大,调高
+> `COMPARE_ALERT_CANCELLED_MS_THRESHOLD`(如 180s/300s)收敛,不动代码。
+
+### 6.2 failed（T1/T2/T6，判定不变 + 附卡点）
+
+```
+worker 对 failed 命中:
+  sp = last_step(dir)                       # 读不到 → None
+  if sp: reason += f"｜卡在 {平台}·{环节}"
+```
+
+`failed` 只取「末帧停在哪」，不要求原地打转（它已失败、末帧即失败点）。intent 阶段就失败（无平台目录，典型 T6）→ 不附，reason 原样。
+
+## 7. 卡点文案映射
+
+`PIPELINE_STEP_LABELS`（小映射表，映射不到原样显示英文、不阻断）：
+
+| pipeline_step | 中文 |
+|---|---|
+| `set_address` | 定位 |
+| `enter_store` | 进店 |
+| `add_one_dish` | 加菜 |
+| …（实现时按 pricebot 实际枚举补全） | |
+
+平台名同样映射（`meituan`→美团、`eleme`→饿了么、`jd_waimai`→京东外卖）。
+
+## 8. 配置（`app/core/config.py`；路径敏感项放 `.env`）
+
+| 配置 | 默认 | 说明 |
+|---|---|---|
+| `COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR` | `""` | pricebot work_logs 绝对路径；**空 = 跳过 trace、全走保底**（行为等同一期） |
+| `COMPARE_ALERT_STUCK_FRAME_THRESHOLD` | `15` | N：末段连续同环节达此帧数判卡死 |
+| `COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES` | `40` | 每平台最多往前读多少帧 |
+| `COMPARE_ALERT_TRACE_MAX_RECORDS` | `30` | 每轮最多对多少条命中记录读 trace（限量） |
+
+## 9. 展示
+
+卡片「失败原因」列下附一行卡点小字，**不新增列**。cancelled 卡死时卡点即 reason 本身；failed 的卡点附在原因后。
+
+> 依赖：本期展示复用一期卡片的「失败原因」列。若一期卡片（schema 2.0 table 组件，当前仍在临时脚本 `scripts/_test_alert_card.py`）尚未固化为正式 `format_alert_card` + `send_feishu_card`，本期实现时一并固化。
+
+## 10. 降级与成本
+
+- 只对命中记录读、限量 `MAX_RECORDS`、每平台只读末段头部字段、单文件读加超时。
+- **任何异常降级**：cancelled 回退保底、failed 不附卡点，绝不阻断报警。
+- `work_log_dir` 未配 → 整个 trace 增强跳过，行为等同一期（纯保底）。
+
+## 11. 测试
+
+- **trace_stuck 单测**：卡死正例（meituan 目录 → 判出 `add_one_dish`）、正常负例（eleme → 不判卡死）、加多菜不误判（`detected_page` 在变）、末段不足 N 帧、读不到目录降级。
+- **worker 集成**：trace 优先命中 vs 读不到回退保底切换；failed 附卡点；限量 `MAX_RECORDS` 生效。
+- fixture 用 tmp 造 `step_*.json`，**只含头部字段**（trace_id/step/platform/pipeline_step/detected_page）即可，不需无障碍树。
+
+## 12. 不做（YAGNI）
+
+- 不改 pricebot（不加内部接口）。
+- 不落库（不加 `comparison_record` 列、不做迁移）。
+- 不做每帧耗时（`timing.json` 缺卡死平台，且报警用不上逐帧耗时）。
+- 不做环节黑白名单 / 语义分类（结构判据已够，且需大数据）。

tests/conftest.py

@@ -31,6 +31,7 @@ os.environ.setdefault("WXPAY_MCH_ID", "test-mch")
 os.environ.setdefault("WXPAY_MCH_SERIAL_NO", "test-serial")
 os.environ.setdefault("WXPAY_PUBLIC_KEY_ID", "test-pubkey-id")
 os.environ.setdefault("RATE_LIMIT_ENABLED", "false")  # 限流内存计数会跨用例累加,测试关掉
+os.environ.setdefault("COMPARE_ALERT_ENABLED", "false")  # 报警 worker 测试不启动(避免 .env 的 true 干扰 test_defaults)
 # 穿山甲发奖回调:测试里开启 + 给个 mock 验签密钥,test 内自签自验闭环
 os.environ.setdefault("PANGLE_CALLBACK_ENABLED", "true")
 os.environ.setdefault("PANGLE_REWARD_SECRET", "test-pangle-secret-only-for-pytest")

tests/test_admin_comparison_outcome.py

@@ -0,0 +1,62 @@
+"""admin 展示口径派生单测：记录级原始结局 → (admin_status, outcome_hint)。"""
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import select
+
+from app.admin.repositories.comparison_outcome import admin_success_sql, derive_admin_outcome
+from app.db.session import SessionLocal
+from app.models.comparison import ComparisonRecord
+
+
+@pytest.mark.parametrize(
+    ("raw_payload", "status", "expected"),
+    [
+        # 真实形态：status 已 normalize，细分在 raw_payload.record_status
+        ({"record_status": "success"}, "success", ("success", None)),
+        ({"record_status": "below_minimum"}, "success", ("success", "未满起送")),
+        ({"record_status": "store_closed"}, "failed", ("success", "门店打烊")),
+        ({"record_status": "store_not_found"}, "failed", ("success", "未找到店")),
+        ({"record_status": "items_not_found"}, "failed", ("success", "未找到菜")),
+        ({"record_status": "no_delivery"}, "failed", ("success", "单点不配送")),
+        ({"record_status": "unsupported"}, "failed", ("success", "平台·场景不支持")),
+        ({"record_status": "failed"}, "failed", ("failed", None)),  # 纯技术故障
+        # 空字符串视作缺失，兜到下一级（与 SQL nullif 对齐）
+        ({"record_status": "", "status": "store_closed"}, "failed", ("success", "门店打烊")),
+        # POST 路径：细分在 raw_payload.status
+        ({"status": "store_not_found"}, "failed", ("success", "未找到店")),
+        # 兜底 status 列：raw_payload 缺失（极老记录）或残留细分值
+        (None, "success", ("success", None)),
+        (None, "failed", ("failed", None)),
+        (None, "store_closed", ("success", "门店打烊")),  # 迁移未覆盖的残留
+        # 生命周期态优先，不看结局
+        ({}, "cancelled", ("cancelled", None)),
+        ({"record_status": "success"}, "running", ("running", None)),
+    ],
+)
+def test_derive_admin_outcome(raw_payload, status, expected):
+    assert derive_admin_outcome(raw_payload, status) == expected
+
+
+def test_admin_success_sql_matches_python_on_empty_string() -> None:
+    """record_status 为空串时，SQL 侧(nullif)与 Python 侧(or)都应兜到 status 列结局、判为成功。"""
+    db = SessionLocal()
+    try:
+        rec = ComparisonRecord(
+            trace_id="outcome-empty-record-status",
+            status="failed",
+            raw_payload={"record_status": "", "status": "store_closed"},
+        )
+        db.add(rec)
+        db.flush()
+        matched = db.execute(
+            select(ComparisonRecord.id).where(
+                ComparisonRecord.trace_id == "outcome-empty-record-status",
+                admin_success_sql(),
+            )
+        ).scalar_one_or_none()
+        assert matched is not None  # SQL 侧判成功
+        assert derive_admin_outcome(rec.raw_payload, rec.status) == ("success", "门店打烊")  # Python 侧一致
+    finally:
+        db.rollback()
+        db.close()

tests/test_admin_read.py

@@ -82,21 +82,25 @@ def test_dashboard_period_comparison_is_aggregated_by_backend(
     admin_client: TestClient, admin_token: str
 ) -> None:
     created_at = datetime(2037, 1, 15, 12)
+    # (trace_id, status 列, record_status, total_ms, cost)
     rows = [
-        ("dashboard-aggregate-success", "success", 101, 0.1),
-        ("dashboard-aggregate-failed", "failed", 200, 0.2),
-        ("dashboard-aggregate-cancelled", "cancelled", 300, 0.3),
-        ("dashboard-aggregate-running", "running", 400, 0.4),
+        ("dashboard-success", "success", "success", 101, 0.1),
+        ("dashboard-below-min", "success", "below_minimum", 120, 0.2),
+        ("dashboard-store-not-found", "failed", "store_not_found", 130, 0.0),
+        ("dashboard-failed", "failed", "failed", 200, 0.3),
+        ("dashboard-cancelled", "cancelled", None, 300, 0.4),
+        ("dashboard-running", "running", None, 400, 0.5),
     ]
     db = SessionLocal()
     try:
-        for trace_id, status, total_ms, llm_cost_yuan in rows:
+        for trace_id, status, record_status, total_ms, llm_cost_yuan in rows:
             db.add(
                 ComparisonRecord(
                     trace_id=trace_id,
                     status=status,
                     total_ms=total_ms,
                     llm_cost_yuan=llm_cost_yuan,
+                    raw_payload={"record_status": record_status} if record_status else None,
                     created_at=created_at,
                 )
             )
@@ -111,14 +115,15 @@ def test_dashboard_period_comparison_is_aggregated_by_backend(
     )
     assert response.status_code == 200, response.text
     comparison = response.json()["period"]["comparison"]
-    assert comparison["total"] == 4
-    assert comparison["completed"] == 2
+    assert comparison["total"] == 6
+    # admin 成功 = success + below_minimum + store_not_found = 3
+    assert comparison["success"] == 3
+    # completed = 3 成功 + 1 纯 failed = 4
+    assert comparison["completed"] == 4
     assert comparison["cancelled"] == 1
-    assert comparison["success"] == 1
-    assert comparison["success_rate"] == 0.3333
-    assert comparison["median_duration_ms"] == 151
-    assert comparison["p95_duration_ms"] == 195
-    assert comparison["token_cost_total_yuan"] == pytest.approx(1.0)
+    # 分母 = total - cancelled = 5
+    assert comparison["success_rate"] == 0.6
+    assert comparison["token_cost_total_yuan"] == pytest.approx(1.5)
 
 
 def test_dashboard_coupon_success_rate_excludes_abandoned_sessions(
@@ -1202,3 +1207,55 @@ def test_period_signin_boost_moves_to_reward_video_without_double_count(
     assert coins["reward_video_coin_total"] == 300
     assert coins["regular_task_coin_total"] == 50
     assert coins["signin_boost_coin_total"] == 200
+
+
+def test_comparison_records_expose_admin_outcome(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """列表 / 详情下发 admin_status + outcome_hint：外部缺失=成功⚠，纯故障=失败。"""
+    db = SessionLocal()
+    try:
+        user = user_repo.upsert_user_for_login(
+            db, phone="13800009051", register_channel="sms"
+        )
+        # 未找到店：status 已 normalize 成 failed，细分在 raw_payload.record_status
+        gap_rec = ComparisonRecord(
+            user_id=user.id, trace_id="admin-outcome-store-not-found",
+            status="failed", store_name="缺店排查店ZZZ",
+            raw_payload={"record_status": "store_not_found"},
+        )
+        db.add(gap_rec)
+        # 纯技术故障
+        db.add(ComparisonRecord(
+            user_id=user.id, trace_id="admin-outcome-tech-failed",
+            status="failed", store_name="缺店排查店ZZZ",
+            raw_payload={"record_status": "failed"},
+        ))
+        db.commit()
+        uid = user.id
+        gap_id = gap_rec.id
+    finally:
+        db.close()
+
+    resp = admin_client.get(
+        "/admin/api/comparison-records",
+        params={"user_id": uid, "store": "缺店排查店ZZZ"},
+        headers=_auth(admin_token),
+    )
+    assert resp.status_code == 200, resp.text
+    by_trace = {it["trace_id"]: it for it in resp.json()["items"]}
+
+    gap = by_trace["admin-outcome-store-not-found"]
+    assert gap["admin_status"] == "success"
+    assert gap["outcome_hint"] == "未找到店"
+
+    tech = by_trace["admin-outcome-tech-failed"]
+    assert tech["admin_status"] == "failed"
+    assert tech["outcome_hint"] is None
+
+    detail = admin_client.get(
+        f"/admin/api/comparison-records/{gap_id}", headers=_auth(admin_token)
+    )
+    assert detail.status_code == 200, detail.text
+    assert detail.json()["admin_status"] == "success"
+    assert detail.json()["outcome_hint"] == "未找到店"

tests/test_compare_alert_config.py

@@ -0,0 +1,17 @@
+"""比价报警配置默认值与关键词解析。"""
+from __future__ import annotations
+
+from app.core.config import settings
+
+
+def test_defaults() -> None:
+    assert settings.COMPARE_ALERT_ENABLED is False
+    assert settings.COMPARE_ALERT_SCAN_INTERVAL_SEC == 900
+    assert settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD == 90000
+    assert settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD == 30
+
+
+def test_keyword_parsing() -> None:
+    assert settings.compare_alert_timeout_keywords == ("超时", "启动", "加载")
+    assert settings.compare_alert_unrecognized_keywords == ("未识别",)
+    assert "打烊" in settings.compare_alert_biz_exclude_keywords

tests/test_compare_alert_fallback.py

@@ -0,0 +1,62 @@
+"""classify_cancelled_fallback / make_hit 单测。"""
+from app.services.compare_alert import classify_cancelled_fallback, make_hit
+
+
+class _Rec:
+    def __init__(self, **kw):
+        self.trace_id = kw.get("trace_id", "t")
+        self.status = kw.get("status", "cancelled")
+        self.total_ms = kw.get("total_ms")
+        self.step_count = kw.get("step_count")
+        self.fail_reason = kw.get("fail_reason")
+        self.information = kw.get("information")
+        self.app_version = kw.get("app_version")
+        self.created_at = kw.get("created_at")
+        self.trace_url = kw.get("trace_url")
+        self.user_id = kw.get("user_id")
+
+
+def test_fallback_deep_by_ms():
+    hit = classify_cancelled_fallback(
+        _Rec(total_ms=95000, step_count=5),
+        cancelled_ms_threshold=90000, cancelled_step_threshold=30,
+    )
+    assert hit is not None and hit.alert_type == "T5" and "深度放弃" in hit.reason
+
+
+def test_fallback_deep_by_step():
+    hit = classify_cancelled_fallback(
+        _Rec(total_ms=1000, step_count=35),
+        cancelled_ms_threshold=90000, cancelled_step_threshold=30,
+    )
+    assert hit is not None and hit.alert_type == "T5"
+
+
+def test_fallback_shallow_none():
+    hit = classify_cancelled_fallback(
+        _Rec(total_ms=5000, step_count=3),
+        cancelled_ms_threshold=90000, cancelled_step_threshold=30,
+    )
+    assert hit is None
+
+
+def test_make_hit_carries_fields():
+    hit = make_hit(_Rec(trace_id="tx", app_version="0.6.0"), "T5", "卡在 美团·加菜")
+    assert hit.trace_id == "tx"
+    assert hit.reason == "卡在 美团·加菜"
+    assert hit.app_version == "0.6.0"
+
+
+def test_make_hit_carries_total_ms_and_step():
+    hit = make_hit(_Rec(trace_id="tx", total_ms=602000, step_count=157), "T5", "深度放弃")
+    assert hit.total_ms == 602000
+    assert hit.step_count == 157
+
+
+def test_make_hit_total_ms_step_default_none():
+    # rec 没有这两个属性时安全降级为 None(不报错)
+    class _Bare:
+        trace_id = "b"; status = "cancelled"; reason = None
+        app_version = None; created_at = None; trace_url = None; user_id = None
+    hit = make_hit(_Bare(), "T1", "技术失败")
+    assert hit.total_ms is None and hit.step_count is None

tests/test_compare_alert_format.py

@@ -0,0 +1,598 @@
+"""AlertHit[] → 飞书消息:分组 / 截断 / 含关键词(text + post 两种格式)。"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from app.services.compare_alert import AlertHit
+from app.services.compare_alert_format import ALERT_KEYWORD, format_alert_message, format_alert_post
+
+
+def _hits(n, alert_type="T1"):
+    return [
+        AlertHit(
+            trace_id=f"t{i}",
+            alert_type=alert_type,
+            reason="比价过程出错",
+            app_version="v0.3.4",
+            created_at=None,
+            trace_url=None,
+            user_id=None,
+        )
+        for i in range(n)
+    ]
+
+
+# ---- format_alert_message(纯文本,保留原有测试) ----
+
+def test_contains_keyword_and_count():
+    msg = format_alert_message(
+        _hits(2), window_label="2026-08-04 08:00–08:30",
+        max_detail_per_type=20, max_total=50,
+    )
+    assert ALERT_KEYWORD in msg
+    assert "本期触发 2 条" in msg
+    assert "系统技术失败 2 条" in msg
+    assert "t0" in msg and "v0.3.4" in msg
+
+
+def test_group_by_type():
+    hits = _hits(1, "T1") + _hits(1, "T6") + _hits(1, "T5")
+    msg = format_alert_message(hits, window_label="w", max_detail_per_type=20, max_total=50)
+    assert "系统技术失败 1 条" in msg
+    assert "商品识别失败 1 条" in msg
+    assert "深度放弃(cancelled) 1 条" in msg
+
+
+def test_per_type_truncation():
+    msg = format_alert_message(_hits(25), window_label="w", max_detail_per_type=20, max_total=50)
+    assert msg.count("t0") == 1
+    assert "另有 5 条" in msg
+
+
+def test_total_truncation_counts_only():
+    msg = format_alert_message(_hits(60), window_label="w", max_detail_per_type=20, max_total=50)
+    assert "系统技术失败 60 条" in msg
+    assert "t0" not in msg
+    assert "分析库" in msg
+
+
+# ---- format_alert_post(富文本 post) ----
+
+def _hits_with_meta(n, alert_type="T1", *, user_id=None, trace_url=None, created_at=None):
+    return [
+        AlertHit(
+            trace_id=f"tr{i}",
+            alert_type=alert_type,
+            reason="比价过程出错",
+            app_version="v1.2.3",
+            created_at=created_at or datetime(2026, 8, 4, 10, 30),
+            trace_url=trace_url,
+            user_id=user_id,
+        )
+        for i in range(n)
+    ]
+
+
+def test_post_title_contains_keyword():
+    hits = _hits_with_meta(1)
+    title, content = format_alert_post(
+        hits, window_label="2026-08-04 10:00", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    assert ALERT_KEYWORD in title
+
+
+def test_post_content_is_list():
+    hits = _hits_with_meta(2)
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    assert isinstance(content, list)
+    assert len(content) >= 1
+    # 每个段落是 list[dict]
+    for para in content:
+        assert isinstance(para, list)
+        for elem in para:
+            assert "tag" in elem
+
+
+def test_post_summary_count():
+    hits = _hits_with_meta(3, "T1") + _hits_with_meta(2, "T6")
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    # 摘要第一段含合计数和类型计数
+    first_para_text = "".join(e.get("text", "") for e in content[0])
+    assert "合计 5 条" in first_para_text
+    assert "系统技术失败 3" in first_para_text
+    assert "商品识别失败 2" in first_para_text
+
+
+def test_post_phone_map_applied():
+    hits = _hits_with_meta(1, user_id=42)
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={42: "13800138000"}, max_detail_per_type=20, max_total=50,
+    )
+    # 找明细行(非摘要非表头)中含手机号
+    all_text = " ".join(
+        e.get("text", "") for para in content for e in para
+    )
+    assert "13800138000" in all_text
+
+
+def test_post_no_user_id_shows_dash():
+    hits = _hits_with_meta(1, user_id=None)
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    all_text = " ".join(e.get("text", "") for para in content for e in para)
+    assert "｜ - ｜" in all_text
+
+
+def test_post_trace_url_becomes_a_element():
+    hits = _hits_with_meta(1, trace_url="https://trace.example.com/tr0")
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    # 找 tag=a 的元素
+    a_elements = [e for para in content for e in para if e.get("tag") == "a"]
+    assert len(a_elements) == 1
+    assert a_elements[0]["href"] == "https://trace.example.com/tr0"
+    assert a_elements[0]["text"] == "trace"
+
+
+def test_post_no_trace_url_shows_trace_id_prefix():
+    hits = _hits_with_meta(1, trace_url=None)
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    # 无 trace_url 时:tag=text, text=trace_id[:16]
+    detail_texts = [e.get("text", "") for para in content for e in para if e.get("tag") == "text"]
+    # trace_id 是 tr0,截 16 位
+    assert any("tr0" in t for t in detail_texts)
+
+
+def test_post_total_truncation_no_detail():
+    hits = _hits_with_meta(60)
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    # 超 max_total:只有摘要+截断提示,无表头、无明细
+    all_text = " ".join(e.get("text", "") for para in content for e in para)
+    assert "合计 60 条" in all_text
+    assert "时间 ｜ 手机号" not in all_text
+    assert "tr0" not in all_text
+    assert "分析库" in all_text
+
+
+def test_post_per_type_truncation():
+    hits = _hits_with_meta(25)
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    all_text = " ".join(e.get("text", "") for para in content for e in para)
+    assert "另有 5 条" in all_text
+
+
+def test_post_created_at_formatting():
+    hits = [
+        AlertHit(
+            trace_id="tx1",
+            alert_type="T1",
+            reason="测试",
+            app_version="v2.0",
+            created_at=datetime(2026, 8, 4, 10, 30),
+            trace_url=None,
+            user_id=None,
+        )
+    ]
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    all_text = " ".join(e.get("text", "") for para in content for e in para)
+    assert "08-04 10:30" in all_text
+
+
+def test_post_no_created_at_shows_dash():
+    hits = [
+        AlertHit(
+            trace_id="tx2",
+            alert_type="T1",
+            reason="测试",
+            app_version=None,
+            created_at=None,
+            trace_url=None,
+            user_id=None,
+        )
+    ]
+    title, content = format_alert_post(
+        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
+    )
+    all_text = " ".join(e.get("text", "") for para in content for e in para)
+    # 无 created_at 时间显示 "-"
+    assert "- ｜" in all_text
+
+
+# ---- format_alert_card(schema 2.0 卡片) ----
+
+from app.services.compare_alert_format import format_alert_card  # noqa: E402
+
+
+def _card_hits(n, alert_type="T1", *, total_ms=None, step_count=None, user_id=None,
+               trace_url=None, created_at=None):
+    return [
+        AlertHit(
+            trace_id=f"card{i}",
+            alert_type=alert_type,
+            reason="比价过程出错",
+            app_version="v1.5.0",
+            created_at=created_at or datetime(2026, 8, 4, 10, 30),
+            trace_url=trace_url,
+            user_id=user_id,
+            total_ms=total_ms,
+            step_count=step_count,
+        )
+        for i in range(n)
+    ]
+
+
+def test_card_schema_and_header():
+    card = format_alert_card(
+        _card_hits(1),
+        window_label="2026-08-04 10:00",
+        phone_map={},
+        interval_min=15,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    assert card["schema"] == "2.0"
+    assert card["header"]["template"] == "red"
+    assert ALERT_KEYWORD in card["header"]["title"]["content"]
+
+
+def test_card_body_markdown_summary():
+    card = format_alert_card(
+        _card_hits(3),
+        window_label="2026-08-04 10:00",
+        phone_map={},
+        interval_min=15,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    elements = card["body"]["elements"]
+    md = elements[0]
+    assert md["tag"] == "markdown"
+    assert "近 15 分钟" in md["content"]
+    assert "合计 3 条" in md["content"]
+
+
+def test_card_has_table_seven_columns():
+    card = format_alert_card(
+        _card_hits(2),
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    elements = card["body"]["elements"]
+    # 第二个元素是 table
+    assert len(elements) >= 2
+    table = elements[1]
+    assert table["tag"] == "table"
+    cols = table["columns"]
+    assert len(cols) == 7
+    display_names = [c["display_name"] for c in cols]
+    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"
+    # 其余列 data_type 均为 text
+    for c in cols:
+        if c["name"] != "trace":
+            assert c["data_type"] == "text"
+
+
+def test_card_cost_cell_format():
+    """cost 格式: total_ms 和 step_count 均有值时 '{Ns} / {M步}'"""
+    hits = [
+        AlertHit(
+            trace_id="t1",
+            alert_type="T1",
+            reason="r",
+            app_version="v1",
+            created_at=datetime(2026, 8, 4, 10, 0),
+            trace_url=None,
+            user_id=None,
+            total_ms=602_000,
+            step_count=157,
+        )
+    ]
+    card = format_alert_card(
+        hits,
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    row = table["rows"][0]
+    assert row["cost"] == "602s / 157步"
+
+
+def test_card_cost_only_ms():
+    hits = [
+        AlertHit(
+            trace_id="t2",
+            alert_type="T1",
+            reason="r",
+            app_version="v1",
+            created_at=None,
+            trace_url=None,
+            user_id=None,
+            total_ms=30_000,
+            step_count=None,
+        )
+    ]
+    card = format_alert_card(
+        hits,
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    row = table["rows"][0]
+    assert row["cost"] == "30s"
+
+
+def test_card_cost_only_steps():
+    hits = [
+        AlertHit(
+            trace_id="t3",
+            alert_type="T1",
+            reason="r",
+            app_version="v1",
+            created_at=None,
+            trace_url=None,
+            user_id=None,
+            total_ms=None,
+            step_count=42,
+        )
+    ]
+    card = format_alert_card(
+        hits,
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    row = table["rows"][0]
+    assert row["cost"] == "42步"
+
+
+def test_card_cost_none_when_both_missing():
+    hits = _card_hits(1, total_ms=None, step_count=None)
+    card = format_alert_card(
+        hits,
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    row = table["rows"][0]
+    assert row["cost"] == "-"
+
+
+def test_card_total_truncation_no_table():
+    """超 max_total 时只有 markdown 摘要，没有 table 元素"""
+    card = format_alert_card(
+        _card_hits(60),
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    elements = card["body"]["elements"]
+    assert len(elements) == 1
+    assert elements[0]["tag"] == "markdown"
+    assert "comparison_record" in elements[0]["content"]
+
+
+def test_card_empty_hits():
+    """空 hits 返回含「本期无异常」的卡片（无 table）"""
+    card = format_alert_card(
+        [],
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    assert card["schema"] == "2.0"
+    elements = card["body"]["elements"]
+    assert len(elements) == 1
+    assert elements[0]["tag"] == "markdown"
+    assert "本期无异常" in elements[0]["content"]
+
+
+def test_card_rows_count_respects_per_type_limit():
+    """每类型最多 max_detail_per_type 条"""
+    hits = _card_hits(25, "T1")
+    card = format_alert_card(
+        hits,
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    assert len(table["rows"]) == 20
+
+
+def test_card_trace_url_becomes_markdown_link():
+    hits = _card_hits(1, trace_url="https://trace.example.com/t0")
+    card = format_alert_card(
+        hits,
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    row = table["rows"][0]
+    assert row["trace"] == "[链接](https://trace.example.com/t0)"
+
+
+def test_card_no_trace_url_shows_trace_id_prefix():
+    hits = _card_hits(1, trace_url=None)
+    card = format_alert_card(
+        hits,
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    row = table["rows"][0]
+    # trace_id = "card0"[:12]
+    assert row["trace"] == "card0"
+
+
+def test_card_phone_from_map():
+    hits = _card_hits(1, user_id=7)
+    card = format_alert_card(
+        hits,
+        window_label="w",
+        phone_map={7: "13912345678"},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    row = table["rows"][0]
+    assert row["phone"] == "13912345678"
+
+
+def test_card_version_not_truncated():
+    """版本完整显示，不缩写"""
+    hits = [
+        AlertHit(
+            trace_id="tv1",
+            alert_type="T1",
+            reason="r",
+            app_version="v2.15.3-release",
+            created_at=None,
+            trace_url=None,
+            user_id=None,
+        )
+    ]
+    card = format_alert_card(
+        hits,
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    row = table["rows"][0]
+    assert row["ver"] == "v2.15.3-release"
+
+
+def test_card_table_has_page_size_and_header_style():
+    card = format_alert_card(
+        _card_hits(1),
+        window_label="w",
+        phone_map={},
+        interval_min=10,
+        max_detail_per_type=20,
+        max_total=50,
+    )
+    table = card["body"]["elements"][1]
+    assert "page_size" in table
+    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"] == "-"
+
+
+def test_card_shows_criteria_legend_with_config_thresholds():
+    """卡片摘要含「判据」说明:深度放弃带当前配置阈值、技术失败说明任一平台 status=failed。"""
+    hits = _card_hits(1, "T5", total_ms=95000) + _card_hits(1, "T1")
+    card = format_alert_card(
+        hits, window_label="w", phone_map={}, interval_min=15,
+        max_detail_per_type=20, max_total=50,
+        cancelled_ms_threshold=90000, cancelled_step_threshold=30,
+    )
+    md = card["body"]["elements"][0]["content"]
+    assert "判据" in md
+    assert "耗时>90s 或 步数>30" in md          # 深度放弃判据带配置值
+    assert "任一平台 status=failed" in md        # 技术失败判据
+
+
+def test_card_criteria_reflects_custom_thresholds():
+    """判据里的阈值随配置变化(不是写死 90/30)。"""
+    card = format_alert_card(
+        _card_hits(1, "T5", total_ms=200000),
+        window_label="w", phone_map={}, interval_min=15,
+        max_detail_per_type=20, max_total=50,
+        cancelled_ms_threshold=180000, cancelled_step_threshold=45,
+    )
+    md = card["body"]["elements"][0]["content"]
+    assert "耗时>180s 或 步数>45" in md

tests/test_compare_alert_migration.py

@@ -0,0 +1,46 @@
+"""updated_at 列存在、回填 = created_at、onupdate 在 ORM 更新时刷新。"""
+from __future__ import annotations
+
+import time
+
+from sqlalchemy import inspect
+
+from app.db.session import SessionLocal, engine
+from app.models.comparison import ComparisonRecord
+
+
+def test_updated_at_column_and_index_exist() -> None:
+    insp = inspect(engine)
+    cols = {c["name"] for c in insp.get_columns("comparison_record")}
+    assert "updated_at" in cols
+    idx_names = {i["name"] for i in insp.get_indexes("comparison_record")}
+    # 迁移环境手动建索引名为 ix_comparison_updated;
+    # 测试环境 create_all() 按 SQLAlchemy 命名约定生成 ix_comparison_record_updated_at。
+    # 两种环境都验通过即可。
+    assert (
+        "ix_comparison_updated" in idx_names
+        or "ix_comparison_record_updated_at" in idx_names
+    ), f"updated_at index not found; available indexes: {idx_names}"
+
+
+def test_onupdate_refreshes_updated_at() -> None:
+    db = SessionLocal()
+    try:
+        rec = ComparisonRecord(trace_id="alert-updated-at-onupdate", status="running")
+        db.add(rec)
+        db.commit()
+        db.refresh(rec)
+        first = rec.updated_at
+        assert first is not None
+        time.sleep(1.1)  # SQLite CURRENT_TIMESTAMP 秒级精度,睡过 1 秒才看得出变化
+        rec.status = "failed"
+        db.commit()
+        db.refresh(rec)
+        assert rec.updated_at > first
+    finally:
+        db.rollback()
+        db.query(ComparisonRecord).filter(
+            ComparisonRecord.trace_id == "alert-updated-at-onupdate"
+        ).delete()
+        db.commit()
+        db.close()

tests/test_compare_alert_rules.py

@@ -0,0 +1,92 @@
+"""比价报警规则分类:记录 → AlertHit | None(纯函数,不碰 DB)。"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from app.services.compare_alert import classify_record
+
+KW = dict(
+    cancelled_ms_threshold=90000,
+    cancelled_step_threshold=30,
+    timeout_keywords=("超时", "启动", "加载"),
+    unrecognized_keywords=("未识别",),
+    biz_exclude_keywords=("未找到", "打烊", "起送", "门店", "店内", "不配送", "这些菜", "未入驻", "休息"),
+)
+
+
+def _rec(**kw):
+    base = dict(
+        status="failed", fail_reason=None, information=None,
+        total_ms=None, step_count=None, trace_id="t", app_version=None, business_type="food",
+    )
+    base.update(kw)
+    return SimpleNamespace(**base)
+
+
+@pytest.mark.parametrize(
+    ("rec", "expected_type"),
+    [
+        (_rec(status="failed", fail_reason=None, information="比价过程出错，请稍后重试"), "T1"),
+        (_rec(status="failed", fail_reason=None, information=None), "T1"),
+        (_rec(status="failed", fail_reason=None, information="美团外卖门店已打烊，无法比价"), None),
+        (_rec(status="failed", fail_reason="未识别到商品"), "T6"),
+        (_rec(status="failed", fail_reason="启动淘宝超时, 请稍后重试"), "T2"),
+        (_rec(status="failed", fail_reason="淘宝闪购店内未找到这些菜品"), None),
+        (_rec(status="cancelled", total_ms=98000, step_count=10), "T5"),
+        (_rec(status="cancelled", total_ms=20000, step_count=31), "T5"),
+        (_rec(status="cancelled", total_ms=20000, step_count=5), None),
+        (_rec(status="cancelled", total_ms=None, step_count=None), None),
+        (_rec(status="success"), None),
+        (_rec(status="running"), None),
+    ],
+)
+def test_classify_record_type(rec, expected_type):
+    hit = classify_record(rec, **KW)
+    assert (hit.alert_type if hit else None) == expected_type
+
+
+def test_t5_boundary_exclusive():
+    assert classify_record(_rec(status="cancelled", total_ms=90000, step_count=30), **KW) is None
+    assert classify_record(_rec(status="cancelled", total_ms=90001, step_count=30), **KW).alert_type == "T5"
+
+
+def test_reason_texts():
+    t1 = classify_record(_rec(status="failed", fail_reason=None, information="  比价过程出错  "), **KW)
+    assert t1.reason == "技术失败·比价过程出错"
+    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 == "深度放弃"
+
+
+# ---- failed 混合单:业务 headline 盖住某平台真技术失败(线上 trace 20260806_144544)----
+
+def test_failed_business_headline_masks_target_technical_failure_reports_t1():
+    # fail_reason 被派生成京东业务原因(未找到),却盖住淘宝 status=failed 的真技术崩溃 → 补判 T1。
+    rec = _rec(
+        status="failed", fail_reason="京东此店内未找到这些菜品",
+        information="比价过程出错，请稍后重试",
+        raw_payload={"platform_results": {
+            "meituan": {"status": "source", "is_source": True},
+            "jd_waimai": {"status": "items_not_found", "reason": "京东此店内未找到这些菜品", "is_source": False},
+            "taobao_flash": {"status": "failed", "reason": "比价过程出错，请稍后重试", "is_source": False},
+        }},
+    )
+    hit = classify_record(rec, **KW)
+    assert hit is not None
+    assert hit.alert_type == "T1"
+
+
+def test_failed_target_failed_but_business_reason_no_false_positive():
+    # pricebot 把打烊漏标成 status=failed,但 reason 是业务话术 → 不算技术崩溃,不报(不误报)。
+    rec = _rec(
+        status="failed", fail_reason="京东此店内未找到这些菜品",
+        information="比价过程出错，请稍后重试",
+        raw_payload={"platform_results": {
+            "jd_waimai": {"status": "items_not_found", "reason": "京东此店内未找到这些菜品", "is_source": False},
+            "taobao_flash": {"status": "failed", "reason": "门店已打烊，无法比价", "is_source": False},
+        }},
+    )
+    assert classify_record(rec, **KW) is None

tests/test_compare_alert_stuck_worker.py

@@ -0,0 +1,146 @@
+"""build_hits 集成测:cancelled trace 优先/保底切换、failed 附卡点、限量。"""
+import json
+from pathlib import Path
+
+from app.core.compare_alert_worker import _fmt_stuck, build_hits
+from app.services.trace_stuck import StuckPoint
+
+
+class _Rec:
+    def __init__(self, **kw):
+        self.trace_id = kw.get("trace_id", "t")
+        self.status = kw.get("status", "cancelled")
+        self.total_ms = kw.get("total_ms")
+        self.step_count = kw.get("step_count")
+        self.fail_reason = kw.get("fail_reason")
+        self.information = kw.get("information")
+        self.app_version = kw.get("app_version")
+        self.created_at = kw.get("created_at")
+        self.trace_url = kw.get("trace_url")
+        self.user_id = kw.get("user_id")
+
+
+def _frame(pdir: Path, idx: int, step: str, page: str) -> None:
+    pdir.mkdir(parents=True, exist_ok=True)
+    body = {"pipeline_step": step, "detected_page": page, "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,
+    timeout_keywords=("超时",), unrecognized_keywords=("未识别",), biz_exclude_keywords=(),
+)
+
+
+def test_cancelled_stuck_reports_via_trace(tmp_path):
+    for i in range(18):
+        _frame(tmp_path / "20260804_x" / "meituan", i, "add_one_dish", "meal_detail_popup")
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_x/",
+               total_ms=5000, step_count=3)  # 保底不会中,靠 trace 判卡死
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
+    assert len(hits) == 1
+    assert hits[0].alert_type == "T5"
+    assert hits[0].reason == "深度放弃"
+    assert "美团·加菜" in hits[0].stuck_point
+    assert "帧" in hits[0].stuck_point
+    assert "meal_detail_popup" in hits[0].stuck_point  # 卡点带末帧页面
+
+
+def test_cancelled_readable_not_stuck_long_duration_reports(tmp_path):
+    # trace 可读但没判出原地卡点:仍过耗时/步数阈值兜底,超长(>90s)照报 T5;没卡点也用末帧标"退出前在哪屏"。
+    # (线上 trace 几乎总可读,若不回退则 total_ms 阈值形同虚设、超长放弃永不报——见 compare-fail-alert 排查。)
+    p = tmp_path / "20260804_y" / "eleme"
+    _frame(p, 0, "set_address", "home")
+    for i in range(1, 6):
+        _frame(p, i, "enter_store", "store")
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_y/",
+               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].reason == "深度放弃"
+    assert hits[0].stuck_point == "饿了么·进店·store"  # 无卡点 → 用末帧(平台·环节·页面)标退出前在哪屏
+
+
+def test_cancelled_readable_not_stuck_short_no_report(tmp_path):
+    # trace 可读没卡点、且耗时/步数都没超阈值 → 正常早退,不报(兜底阈值把住,不误报)。
+    p = tmp_path / "20260804_ys" / "eleme"
+    _frame(p, 0, "set_address", "home")
+    for i in range(1, 6):
+        _frame(p, i, "enter_store", "store")
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_ys/",
+               total_ms=5000, step_count=3)
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
+    assert hits == []
+
+
+def test_cancelled_unreadable_falls_back(tmp_path):
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/nope/",
+               total_ms=95000, step_count=3)
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
+    assert len(hits) == 1
+    assert hits[0].alert_type == "T5" and "深度放弃" in hits[0].reason
+
+
+def test_no_work_log_dir_uses_fallback(tmp_path):
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/y/",
+               total_ms=95000, step_count=3)
+    hits = build_hits([rec], work_log_dir="", **_KW)
+    assert len(hits) == 1 and "深度放弃" in hits[0].reason
+
+
+def test_failed_gets_stuck_point_appended(tmp_path):
+    for i in range(20):
+        _frame(tmp_path / "20260804_f" / "meituan", i, "add_one_dish", "meal_detail_popup")
+    rec = _Rec(status="failed", fail_reason="启动超时",
+               trace_url="https://x/traces/20260804_f/")
+    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].stuck_point
+    assert "卡在" not in hits[0].reason
+
+
+def test_max_trace_reads_zero_skips_trace(tmp_path):
+    for i in range(18):
+        _frame(tmp_path / "20260804_z" / "meituan", i, "add_one_dish", "meal_detail_popup")
+    rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_z/",
+               total_ms=95000, step_count=3)
+    kw = {**_KW, "max_trace_reads": 0}
+    hits = build_hits([rec], work_log_dir=str(tmp_path), **kw)
+    # 没读 trace → 回退保底 → deep(95s) → 深度放弃
+    assert len(hits) == 1 and "深度放弃" in hits[0].reason
+
+
+def test_shared_reads_budget_across_branches(tmp_path):
+    # cancelled 和 failed 共用 max_trace_reads 预算;预算=1 时 cancelled 先消耗,failed 拿不到卡点
+    for i in range(18):
+        _frame(tmp_path / "20260804_c" / "meituan", i, "add_one_dish", "meal_detail_popup")
+    for i in range(20):
+        _frame(tmp_path / "20260804_d" / "meituan", i, "add_one_dish", "meal_detail_popup")
+    cancelled = _Rec(status="cancelled", trace_url="https://x/traces/20260804_c/",
+                     total_ms=5000, step_count=3)
+    failed = _Rec(status="failed", fail_reason="启动超时",
+                  trace_url="https://x/traces/20260804_d/")
+    kw = {**_KW, "max_trace_reads": 1}
+    hits = build_hits([cancelled, failed], work_log_dir=str(tmp_path), **kw)
+    assert len(hits) == 2
+    # 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帧"

tests/test_compare_alert_worker.py

@@ -0,0 +1,93 @@
+"""扫描 worker 的同步核心 _scan_and_alert:水位读写 / 冷启动 / 发送成功推进、失败不推进。
+
+monkeypatch 掉真正的飞书发送;用真实 SQLite(SessionLocal)插入几条记录。
+"""
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from app.core import compare_alert_worker as w
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+
+WM_KEY = w.WATERMARK_KEY
+
+
+@pytest.fixture()
+def clean_db():
+    db = SessionLocal()
+    db.query(ComparisonRecord).delete()
+    db.query(AppConfig).filter(AppConfig.key == WM_KEY).delete()
+    db.commit()
+    yield db
+    db.query(ComparisonRecord).delete()
+    db.query(AppConfig).filter(AppConfig.key == WM_KEY).delete()
+    db.commit()
+    db.close()
+
+
+def _add(db, trace, status, **kw):
+    rec = ComparisonRecord(trace_id=trace, status=status, **kw)
+    db.add(rec)
+    db.commit()
+    db.refresh(rec)
+    return rec
+
+
+def _card_title(card: dict) -> str:
+    return card.get("header", {}).get("title", {}).get("content", "")
+
+
+def _card_md_content(card: dict) -> str:
+    elements = card.get("body", {}).get("elements", [])
+    return elements[0].get("content", "") if elements else ""
+
+
+def test_cold_start_sets_watermark_no_alert(clean_db, monkeypatch):
+    # 冷启动:表非空 → 水位=当前 max(updated_at),不回溯已有历史失败、不报
+    _add(clean_db, "old-fail", "failed", fail_reason=None, information="比价过程出错")
+    sent = []
+    monkeypatch.setattr(w, "_send_card", lambda card: sent.append(card))
+    w._scan_and_alert()
+    assert sent == []  # 冷启动不报历史
+    row = clean_db.get(AppConfig, WM_KEY)
+    assert row is not None  # 水位已初始化
+
+
+def test_alerts_on_new_failed_and_advances(clean_db, monkeypatch):
+    # seed 一条 + 冷启动建水位;sleep 1.1s 拉开时间(SQLite 秒级精度,否则新记录同秒、追不上水位)
+    _add(clean_db, "seed", "running")
+    monkeypatch.setattr(w, "_send_card", lambda card: None)
+    w._scan_and_alert()  # 冷启动,水位=seed.updated_at
+    time.sleep(1.1)
+    sent = []
+    monkeypatch.setattr(w, "_send_card", lambda card: sent.append(card))
+    _add(clean_db, "new-fail", "failed", fail_reason=None, information="比价过程出错")
+    w._scan_and_alert()
+    assert len(sent) == 1
+    card = sent[0]
+    # title 含关键词
+    assert "比价失败报警" in _card_title(card)
+    # markdown 摘要含"系统技术失败"
+    assert "系统技术失败 1" in _card_md_content(card)
+
+
+def test_send_failure_does_not_advance_watermark(clean_db, monkeypatch):
+    _add(clean_db, "seed2", "running")
+    monkeypatch.setattr(w, "_send_card", lambda card: None)
+    w._scan_and_alert()  # 冷启动建水位
+    wm_before = clean_db.get(AppConfig, WM_KEY).value
+    time.sleep(1.1)
+    _add(clean_db, "fail-send", "failed", fail_reason=None, information="比价过程出错")
+
+    def boom(card):
+        raise w.feishu_notifier.FeishuNotifyError("down")
+
+    monkeypatch.setattr(w, "_send_card", boom)
+    w._scan_and_alert()  # 发送失败
+    clean_db.expire_all()
+    wm_after = clean_db.get(AppConfig, WM_KEY).value
+    assert wm_after == wm_before  # 未推进,下轮补发

tests/test_compare_daily_limit.py

@@ -140,7 +140,7 @@ def test_compare_start_rejects_101st_beijing_day_attempt(client) -> None:
     )
 
     assert response.status_code == 429
-    assert response.json()["detail"] == "今日已比价超过100次，请明天再试"
+    assert response.json()["detail"] == "今日比价额度用完啦，明天再来吧～"
     with SessionLocal() as db:
         assert db.scalar(
             select(func.count(ComparisonRecord.id)).where(

tests/test_comparison_admin_summary.py

@@ -1,4 +1,4 @@
-"""比价记录页后端分页与概览聚合。"""
+"""比价记录页后端概览聚合（admin 口径：外部缺失记为成功）。"""
 from __future__ import annotations
 
 from datetime import UTC, date, datetime
@@ -7,13 +7,14 @@ import pytest
 from sqlalchemy.dialects import postgresql
 
 from app.admin.repositories import queries
+from app.admin.repositories.comparison_outcome import admin_success_sql
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 
 
 def test_postgresql_duration_summary_uses_ordered_set_aggregates() -> None:
     stmt = queries._comparison_duration_aggregate_stmt(
-        [], "success", (0.05, 0.5, 0.95, 0.99)
+        [], admin_success_sql(), (0.05, 0.5, 0.95, 0.99)
     )
     sql = str(
         stmt.compile(
@@ -21,82 +22,106 @@ def test_postgresql_duration_summary_uses_ordered_set_aggregates() -> None:
             compile_kwargs={"literal_binds": True},
         )
     )
-
     assert sql.count("percentile_cont") == 4
-    assert "comparison_record.status IN ('success', 'below_minimum')" in sql
+    # 口径按原始结局 coalesce，而非直接读 status 列
+    assert "coalesce" in sql.lower()
 
 
-def test_summary_uses_only_success_durations_and_filters_beijing_date() -> None:
+def test_summary_counts_external_gaps_as_success() -> None:
     db = SessionLocal()
     try:
+        # (trace_id, status 列[已 normalize], record_status[原始结局], total_ms, cost, saved)
         rows = [
-            ("summary-success-a", "success", 1000, 1.0, 100),
-            ("summary-success-b", "success", 3000, 2.0, 0),
-            ("summary-failed", "failed", 100_000, 3.0, 0),
-            ("summary-cancelled", "cancelled", 5000, 4.0, 0),
-            ("summary-below-minimum", "below_minimum", 5000, None, 0),
-            ("summary-store-closed", "store_closed", 200_000, None, 0),
-            ("summary-running", "running", 4000, None, 0),
+            ("sum-success", "success", "success", 1000, 1.0, 100),
+            ("sum-below-min", "success", "below_minimum", 2000, 2.0, 0),
+            ("sum-store-closed", "failed", "store_closed", 3000, None, 0),
+            ("sum-store-not-found", "failed", "store_not_found", 4000, None, 0),
+            ("sum-failed", "failed", "failed", 100_000, 3.0, 0),
+            ("sum-cancelled", "cancelled", None, 5000, None, 0),
+            ("sum-running", "running", None, 6000, None, 0),
         ]
-        for trace_id, status, total_ms, cost, saved in rows:
+        for trace_id, status, record_status, total_ms, cost, saved in rows:
             db.add(ComparisonRecord(
                 trace_id=trace_id,
                 status=status,
                 total_ms=total_ms,
                 llm_cost_yuan=cost,
                 saved_amount_cents=saved,
+                raw_payload={"record_status": record_status} if record_status else None,
                 created_at=datetime(2038, 1, 15, 12, tzinfo=UTC),
             ))
-        db.add(ComparisonRecord(
-            trace_id="summary-outside-day",
-            status="success",
-            total_ms=999_999,
-            created_at=datetime(2038, 1, 16, 16, tzinfo=UTC),
-        ))
         db.flush()
 
         summary = queries.comparison_records_summary(
             db, date_from=date(2038, 1, 15), date_to=date(2038, 1, 15)
         )
 
+        # admin 成功 = success + below_minimum + store_closed + store_not_found = 4
         assert summary["started"] == 7
-        assert summary["completed"] == 5
-        assert summary["success"] == 3
-        assert summary["success_rate"] == pytest.approx(3 / 6)
-        assert summary["avg_token_cost"] == pytest.approx(2.5)
-        assert summary["lower_price_rate"] == pytest.approx(1 / 3)
-        assert summary["avg_duration_ms"] == 3000
-        assert summary["p5_duration_ms"] == 1200
-        assert summary["p50_duration_ms"] == 3000
-        assert summary["p95_duration_ms"] == 4800
-        assert summary["p99_duration_ms"] == 4960
+        assert summary["success"] == 4
+        assert summary["completed"] == 5           # 4 成功 + 1 纯 failed
         assert summary["cancelled"] == 1
-        assert summary["cancelled_rate"] == pytest.approx(1 / 7)
+        assert summary["success_rate"] == pytest.approx(4 / 6)  # 分母 started - cancelled
+        assert summary["avg_token_cost"] == pytest.approx(2.0)  # (1+2+3)/3
+        assert summary["lower_price_rate"] == pytest.approx(1 / 4)  # 仅 sum-success saved>0
+        # 耗时统计集 = admin 成功的 total_ms [1000,2000,3000,4000]
+        assert summary["avg_duration_ms"] == 2500
+        assert summary["p5_duration_ms"] == 1150
+        assert summary["p50_duration_ms"] == 2500
+        assert summary["p95_duration_ms"] == 3850
+        assert summary["p99_duration_ms"] == 3970
         assert summary["cancelled_p50_ms"] == 5000
-
-        items, _next_cursor, total = queries.list_comparison_records(
-            db, date_from=date(2038, 1, 15), date_to=date(2038, 1, 15), limit=20
-        )
-        assert total == 7
-        assert {item.trace_id for item in items} == {row[0] for row in rows}
-
-        success_items, _next_cursor, success_total = queries.list_comparison_records(
-            db, status="success", date_from=date(2038, 1, 15), date_to=date(2038, 1, 15)
-        )
-        assert success_total == 3
-        assert {item.status for item in success_items} == {"success", "below_minimum"}
-
-        failed_items, _next_cursor, failed_total = queries.list_comparison_records(
-            db, status="failed", date_from=date(2038, 1, 15), date_to=date(2038, 1, 15)
-        )
-        assert failed_total == 2
-        assert {item.status for item in failed_items} == {"failed", "store_closed"}
-
-        running_items, _next_cursor, running_total = queries.list_comparison_records(
-            db, status="running", date_from=date(2038, 1, 15), date_to=date(2038, 1, 15)
-        )
-        assert running_total == 1
-        assert running_items[0].status == "running"
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_list_status_filter_uses_admin_outcome() -> None:
+    """列表「状态」筛选走 admin 口径：筛成功含 6 类、筛失败仅纯技术故障；并验日期边界排除。"""
+    db = SessionLocal()
+    try:
+        rows = [
+            ("flt-success", "success", "success"),
+            ("flt-below-min", "success", "below_minimum"),
+            ("flt-store-not-found", "failed", "store_not_found"),
+            ("flt-failed", "failed", "failed"),
+            ("flt-cancelled", "cancelled", None),
+        ]
+        for trace_id, status, record_status in rows:
+            db.add(ComparisonRecord(
+                trace_id=trace_id,
+                status=status,
+                raw_payload={"record_status": record_status} if record_status else None,
+                created_at=datetime(2039, 3, 10, 12, tzinfo=UTC),
+            ))
+        # 日期边界：窗口外一条 admin 成功记录（次日），应被日期筛选排除
+        db.add(ComparisonRecord(
+            trace_id="flt-out-of-window",
+            status="success",
+            raw_payload={"record_status": "success"},
+            created_at=datetime(2039, 3, 11, 12, tzinfo=UTC),
+        ))
+        db.flush()
+
+        succ, _c1, succ_total = queries.list_comparison_records(
+            db, status="success", date_from=date(2039, 3, 10), date_to=date(2039, 3, 10)
+        )
+        # 3 条当天 admin 成功（含 store_not_found）；窗口外 flt-out-of-window 被日期排除
+        assert succ_total == 3
+        assert {it.trace_id for it in succ} == {
+            "flt-success", "flt-below-min", "flt-store-not-found",
+        }
+
+        fail, _c2, fail_total = queries.list_comparison_records(
+            db, status="failed", date_from=date(2039, 3, 10), date_to=date(2039, 3, 10)
+        )
+        assert fail_total == 1
+        assert {it.trace_id for it in fail} == {"flt-failed"}
+
+        _canc, _c3, canc_total = queries.list_comparison_records(
+            db, status="cancelled", date_from=date(2039, 3, 10), date_to=date(2039, 3, 10)
+        )
+        assert canc_total == 1
     finally:
         db.rollback()
         db.close()

tests/test_feishu_notifier.py

@@ -0,0 +1,169 @@
+"""飞书群机器人发送:text/post 消息体格式 / 关键词失败 / 网络失败,均 monkeypatch httpx 不真发。"""
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from app.integrations import feishu_notifier
+
+# ---- send_feishu_text ----
+
+def test_send_posts_text_payload(monkeypatch):
+    captured = {}
+
+    def fake_post(url, json, timeout):
+        captured["url"] = url
+        captured["json"] = json
+        return httpx.Response(200, json={"code": 0, "msg": "success"})
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    feishu_notifier.send_feishu_text("https://open.feishu.cn/hook/xxx", "比价失败报警 · test", timeout=5.0)
+    assert captured["url"] == "https://open.feishu.cn/hook/xxx"
+    assert captured["json"] == {"msg_type": "text", "content": {"text": "比价失败报警 · test"}}
+
+
+def test_send_raises_on_keyword_rejection(monkeypatch):
+    def fake_post(url, json, timeout):
+        return httpx.Response(200, json={"code": 19024, "msg": "Key Words Not Found"})
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    with pytest.raises(feishu_notifier.FeishuNotifyError):
+        feishu_notifier.send_feishu_text("https://hook", "无关键词文本", timeout=5.0)
+
+
+def test_send_raises_on_http_error(monkeypatch):
+    def fake_post(url, json, timeout):
+        return httpx.Response(500, text="boom")
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    with pytest.raises(feishu_notifier.FeishuNotifyError):
+        feishu_notifier.send_feishu_text("https://hook", "比价失败报警", timeout=5.0)
+
+
+# ---- send_feishu_post ----
+
+def test_send_post_payload_structure(monkeypatch):
+    """send_feishu_post 发出 msg_type=post 的 payload,结构符合飞书 post 格式。"""
+    captured = {}
+
+    def fake_post(url, json, timeout):
+        captured["url"] = url
+        captured["json"] = json
+        return httpx.Response(200, json={"code": 0, "msg": "success"})
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    content = [[{"tag": "text", "text": "合计 3 条"}]]
+    feishu_notifier.send_feishu_post(
+        "https://open.feishu.cn/hook/yyy",
+        "🚨 比价失败报警 · 2026-08-04",
+        content,
+        timeout=5.0,
+    )
+    assert captured["url"] == "https://open.feishu.cn/hook/yyy"
+    payload = captured["json"]
+    assert payload["msg_type"] == "post"
+    zh_cn = payload["content"]["post"]["zh_cn"]
+    assert zh_cn["title"] == "🚨 比价失败报警 · 2026-08-04"
+    assert zh_cn["content"] == content
+
+
+def test_send_post_raises_on_code_nonzero(monkeypatch):
+    """send_feishu_post code!=0 时抛 FeishuNotifyError。"""
+    def fake_post(url, json, timeout):
+        return httpx.Response(200, json={"code": 19024, "msg": "Key Words Not Found"})
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    with pytest.raises(feishu_notifier.FeishuNotifyError):
+        feishu_notifier.send_feishu_post(
+            "https://hook", "🚨 比价失败报警", [[{"tag": "text", "text": "test"}]], timeout=5.0
+        )
+
+
+def test_send_post_raises_on_http_error(monkeypatch):
+    """send_feishu_post 非 2xx 抛 FeishuNotifyError。"""
+    def fake_post(url, json, timeout):
+        return httpx.Response(500, text="internal server error")
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    with pytest.raises(feishu_notifier.FeishuNotifyError):
+        feishu_notifier.send_feishu_post(
+            "https://hook", "🚨 比价失败报警", [[{"tag": "text", "text": "test"}]], timeout=5.0
+        )
+
+
+def test_send_post_raises_on_network_error(monkeypatch):
+    """send_feishu_post 网络异常抛 FeishuNotifyError。"""
+    def fake_post(url, json, timeout):
+        raise httpx.ConnectError("connection refused")
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    with pytest.raises(feishu_notifier.FeishuNotifyError):
+        feishu_notifier.send_feishu_post(
+            "https://hook", "🚨 比价失败报警", [[{"tag": "text", "text": "test"}]], timeout=5.0
+        )
+
+
+# ---- send_feishu_card ----
+
+def test_send_card_payload_structure(monkeypatch):
+    """send_feishu_card 发出 msg_type=interactive + card 字段的 payload。"""
+    captured = {}
+
+    def fake_post(url, json, timeout):
+        captured["url"] = url
+        captured["json"] = json
+        return httpx.Response(200, json={"code": 0, "msg": "success"})
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    card = {
+        "schema": "2.0",
+        "header": {"title": {"tag": "plain_text", "content": "🚨 比价失败报警 · test"}, "template": "red"},
+        "body": {"elements": [{"tag": "markdown", "content": "近 15 分钟"}]},
+    }
+    feishu_notifier.send_feishu_card("https://open.feishu.cn/hook/zzz", card, timeout=5.0)
+    assert captured["url"] == "https://open.feishu.cn/hook/zzz"
+    assert captured["json"]["msg_type"] == "interactive"
+    assert captured["json"]["card"] is card
+
+
+def test_send_card_uses_default_timeout(monkeypatch):
+    """send_feishu_card 默认 timeout=10.0。"""
+    captured = {}
+
+    def fake_post(url, json, timeout):
+        captured["timeout"] = timeout
+        return httpx.Response(200, json={"code": 0, "msg": "success"})
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    feishu_notifier.send_feishu_card("https://hook", {"schema": "2.0"})
+    assert captured["timeout"] == 10.0
+
+
+def test_send_card_raises_on_code_nonzero(monkeypatch):
+    """send_feishu_card code!=0 时抛 FeishuNotifyError。"""
+    def fake_post(url, json, timeout):
+        return httpx.Response(200, json={"code": 19024, "msg": "Key Words Not Found"})
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    with pytest.raises(feishu_notifier.FeishuNotifyError):
+        feishu_notifier.send_feishu_card("https://hook", {"schema": "2.0"})
+
+
+def test_send_card_raises_on_http_error(monkeypatch):
+    """send_feishu_card 非 2xx 抛 FeishuNotifyError。"""
+    def fake_post(url, json, timeout):
+        return httpx.Response(500, text="boom")
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    with pytest.raises(feishu_notifier.FeishuNotifyError):
+        feishu_notifier.send_feishu_card("https://hook", {"schema": "2.0"})
+
+
+def test_send_card_raises_on_network_error(monkeypatch):
+    """send_feishu_card 网络异常抛 FeishuNotifyError。"""
+    def fake_post(url, json, timeout):
+        raise httpx.ConnectError("connection refused")
+
+    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
+    with pytest.raises(feishu_notifier.FeishuNotifyError):
+        feishu_notifier.send_feishu_card("https://hook", {"schema": "2.0"})

tests/test_trace_stuck.py

@@ -0,0 +1,174 @@
+"""trace_stuck 单测:用 tmp 造 step_*.json(只含头部字段)验证卡死判据。"""
+import json
+from pathlib import Path
+
+from app.services.trace_stuck import (
+    StuckPoint,
+    dir_name_from_trace_url,
+    last_step,
+    read_stuck_points,
+)
+
+
+def _frame(pdir: Path, idx: int, step: str, page: str) -> None:
+    """造一帧 step json:头部放 pipeline_step/detected_page,尾部塞大 windows 模拟真实。"""
+    pdir.mkdir(parents=True, exist_ok=True)
+    body = {
+        "trace_id": "t", "step": idx, "platform": pdir.name,
+        "pipeline_step": step, "detected_page": page,
+        "windows": [{"nodes": ["x" * 200]}],
+    }
+    (pdir / f"step_{idx:03d}.json").write_text(
+        json.dumps(body, ensure_ascii=False), encoding="utf-8"
+    )
+
+
+def test_stuck_when_tail_repeats_same_step(tmp_path):
+    pdir = tmp_path / "meituan"
+    for i in range(20):
+        _frame(pdir, i, "add_one_dish", "meal_detail_popup")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.readable is True
+    assert res.points == [StuckPoint("meituan", "add_one_dish", 20, detected_page="meal_detail_popup")]
+
+
+def test_not_stuck_when_progressing(tmp_path):
+    pdir = tmp_path / "eleme"
+    _frame(pdir, 0, "set_address", "home")
+    for i in range(1, 8):
+        _frame(pdir, i, "enter_store", "store")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.readable is True
+    assert res.points == []
+
+
+def test_adding_many_dishes_not_stuck_when_page_changes(tmp_path):
+    # add_one_dish 重复但 detected_page 在跳(换菜)=推进,不判卡死
+    pdir = tmp_path / "meituan"
+    for i in range(20):
+        _frame(pdir, i, "add_one_dish", "menu" if i % 2 == 0 else "dish_popup")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.points == []
+
+
+def test_below_threshold_not_stuck(tmp_path):
+    pdir = tmp_path / "meituan"
+    for i in range(10):  # < 15
+        _frame(pdir, i, "add_one_dish", "meal_detail_popup")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.points == []
+
+
+def test_missing_dir_not_readable(tmp_path):
+    res = read_stuck_points(tmp_path / "nope", threshold=15, max_tail=40)
+    assert res.readable is False
+    assert res.points == []
+
+
+def test_empty_dir_no_platform_frames_not_readable(tmp_path):
+    (tmp_path / "emptysub").mkdir()
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.readable is False
+
+
+def test_per_platform_one_stuck_one_normal(tmp_path):
+    m = tmp_path / "meituan"
+    for i in range(18):
+        _frame(m, i, "add_one_dish", "meal_detail_popup")
+    e = tmp_path / "eleme"
+    _frame(e, 0, "set_address", "home")
+    for i in range(1, 6):
+        _frame(e, i, "enter_store", "store")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.readable is True
+    assert res.points == [StuckPoint("meituan", "add_one_dish", 18, detected_page="meal_detail_popup")]
+
+
+def test_last_step_returns_busiest_platform_last_env(tmp_path):
+    m = tmp_path / "meituan"
+    for i in range(20):
+        _frame(m, i, "add_one_dish", "meal_detail_popup")
+    e = tmp_path / "eleme"
+    for i in range(3):
+        _frame(e, i, "enter_store", "store")
+    sp = last_step(tmp_path)
+    assert sp == StuckPoint("meituan", "add_one_dish", 20, detected_page="meal_detail_popup")
+
+
+def test_stuck_point_label_includes_page():
+    # label 带页面(平台/环节中文 + 页面原始英文);无页面时只到环节
+    assert StuckPoint("meituan", "add_one_dish", 20, detected_page="meal_detail_popup").label() \
+        == "美团·加菜·meal_detail_popup"
+    assert StuckPoint("meituan", "add_one_dish", 20).label() == "美团·加菜"
+
+
+def test_read_stuck_points_returns_last_frame(tmp_path):
+    # res.last = 帧数最多平台的末帧(带 detected_page),供"退出前在哪屏"用
+    m = tmp_path / "meituan"
+    for i in range(6):
+        _frame(m, i, "enter_store", "store")
+    _frame(m, 6, "add_one_dish", "menu")  # 末帧换到 menu
+    e = tmp_path / "eleme"
+    _frame(e, 0, "set_address", "home")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert res.last == StuckPoint("meituan", "add_one_dish", 7, detected_page="menu")
+    assert res.points == []  # 没卡死,但末帧照样有
+
+
+def test_dir_name_from_trace_url():
+    assert dir_name_from_trace_url("https://x/traces/20260804_1_abc/") == "20260804_1_abc"
+    assert dir_name_from_trace_url("https://x/traces/20260804_1_abc") == "20260804_1_abc"
+    assert dir_name_from_trace_url("") is None
+    assert dir_name_from_trace_url(None) is None
+
+
+def test_corrupt_frame_does_not_crash(tmp_path):
+    # 截断的 UTF-8(pricebot 被 SIGTERM 打断的末帧)不应抛 UnicodeDecodeError
+    pdir = tmp_path / "meituan"
+    pdir.mkdir()
+    (pdir / "step_000.json").write_bytes(b'{"pipeline_step": "add\xff')
+    res = read_stuck_points(tmp_path, threshold=1, max_tail=40)
+    assert res.points == []  # 抠不出字段 → 不判卡死;关键是没崩溃
+
+
+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"
+    )
+
+
+def test_stuck_ms_computed_from_timestamps(tmp_path):
+    # 末段 16 帧都卡在 add_one_dish,timestamp 从 :00 到 :30(每帧+2s) → 卡住 30s
+    pdir = tmp_path / "meituan"
+    for i in range(16):
+        _frame_ts(pdir, i, "add_one_dish", "meal_detail_popup",
+                  f"2026-08-04T12:00:{i * 2:02d}.000000")
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert len(res.points) == 1
+    sp = res.points[0]
+    assert sp.frames == 16
+    assert sp.stuck_ms == 30000  # (第15帧:30 - 第0帧:00) = 30s
+
+
+def test_stuck_ms_none_when_timestamp_missing(tmp_path):
+    # 帧无 timestamp 字段 → stuck_ms 为 None(不报错)
+    pdir = tmp_path / "meituan"
+    for i in range(16):
+        _frame(pdir, i, "add_one_dish", "meal_detail_popup")  # 现有 _frame,无 timestamp
+    res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
+    assert len(res.points) == 1
+    assert res.points[0].stuck_ms is None
+
+
+def test_stuck_ms_none_when_clock_goes_backwards(tmp_path):
+    # 帧 timestamp 非单调(时钟回退):step_000=:30 ... step_015=:00 → 末帧早于段首 → 负时长 → 降级 None
+    pdir = tmp_path / "meituan"
+    for i in range(16):
+        _frame_ts(pdir, i, "add_one_dish", "meal_detail_popup",
+                  f"2026-08-04T12:00:{(30 - i * 2):02d}.000000")
+    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

tests/test_withdraw_tiers.py

@@ -1,9 +1,9 @@
 """福利页(coin_cash)提现档位规则测试(7-9提现ui对齐)。
 
 规则(2026-07-09 拍板):
-  - 档位 0.1/0.3(新人,历史一次性,免广告)+ 0.5(日3次)/10/20(日1次)。
+  - 档位 0.1/0.3(新人,历史一次性,免广告)+ 0.5(日3次)/10/20/100(日1次)。
   - 计次口径"发起就算":当天创建的单不论最终状态(含被拒)都占名额;新人档任何状态都算用过。
-  - 常规三档每天只能选一个;新人档不参与该互斥,两个新人档同天可各提一次。
+  - 常规四档每天只能选一个;新人档不参与该互斥,两个新人档同天可各提一次。
   - invite_cash 无档位概念:tiers 为空、下单不走档位闸(邀请页行为不变)。
 wxpay 调用全部 monkeypatch;现金余额 DB 直灌(同 test_withdraw.py 套路)。
 """
@@ -74,15 +74,18 @@ def _tiers(client, token: str, source: str = "coin_cash") -> list[dict]:
 
 
 def test_withdraw_info_tiers_full_and_invite_empty(client, monkeypatch) -> None:
-    """新用户 coin_cash 下发 5 档(新人角标齐);invite_cash tiers 为空。"""
+    """新用户 coin_cash 下发 6 档(新人角标齐);invite_cash tiers 为空。"""
     token = _login(client, "13800006001")
     tiers = _tiers(client, token)
-    assert [t["amount_cents"] for t in tiers] == [10, 30, 50, 1000, 2000]
-    assert [t["label"] for t in tiers] == ["0.1", "0.3", "0.5", "10", "20"]
+    assert [t["amount_cents"] for t in tiers] == [10, 30, 50, 1000, 2000, 10000]
+    assert [t["label"] for t in tiers] == ["0.1", "0.3", "0.5", "10", "20", "100"]
     assert tiers[0]["badge"] == "新人福利" and tiers[0]["is_newbie"] is True
     assert tiers[1]["badge"] == "新人福利" and tiers[1]["is_newbie"] is True
     assert all(t["available"] for t in tiers)
     assert tiers[2]["remaining_today"] == 3  # 0.5 日 3 次
+    # daily_limit 一起下发:客户端只有在 daily_limit>1 且 remaining_today<daily_limit
+    # (= 今天已提过)时才画「今日还可提N次」角标,光看 remaining_today 分不出单次档没提的情况。
+    assert [t["daily_limit"] for t in tiers] == [1, 1, 3, 1, 1, 1]
     assert _tiers(client, token, source="invite_cash") == []
 
 
@@ -101,9 +104,9 @@ def test_newbie_tiers_independent_and_once_forever(client, monkeypatch) -> None:
     amounts = [t["amount_cents"] for t in tiers]
     assert 10 not in amounts          # 0.1 消失
     assert 30 in amounts              # 0.3 还在,同天仍可提
-    # 新人档不参与"选一个额度":常规三档全部仍可提
+    # 新人档不参与"选一个额度":常规四档全部仍可提
     regular = {t["amount_cents"]: t for t in tiers if not t["is_newbie"]}
-    assert all(regular[a]["available"] for a in (50, 1000, 2000))
+    assert all(regular[a]["available"] for a in (50, 1000, 2000, 10000))
 
     r = _withdraw(client, token, 30)  # 同天 0.3 照提
     assert r.status_code == 200, r.text
@@ -115,7 +118,7 @@ def test_newbie_tiers_independent_and_once_forever(client, monkeypatch) -> None:
 
 
 def test_regular_daily_select_one_tier(client, monkeypatch) -> None:
-    """当天提过 0.5 → 10/20 置灰 other_tier_selected,下单 409;0.5 还能继续提(3 次内)。"""
+    """当天提过 0.5 → 10/20/100 置灰 other_tier_selected,下单 409;0.5 还能继续提(3 次内)。"""
     _patch_userinfo(monkeypatch, "openid_tier_3")
     token = _login(client, "13800006003")
     _seed_balances(client, token, "13800006003", cash=10_000)
@@ -129,6 +132,7 @@ def test_regular_daily_select_one_tier(client, monkeypatch) -> None:
     assert tiers[50]["available"] and tiers[50]["remaining_today"] == 2
     assert not tiers[1000]["available"] and tiers[1000]["disabled_reason"] == "other_tier_selected"
     assert not tiers[2000]["available"] and tiers[2000]["disabled_reason"] == "other_tier_selected"
+    assert not tiers[10000]["available"] and tiers[10000]["disabled_reason"] == "other_tier_selected"
 
     r = _withdraw(client, token, 1000)  # 选一额度互斥 → 409
     assert r.status_code == 409, r.text