fix(auto-exchange): 「当天已兑」标记持久化到 app_config，防非0点重启重复补扫 (#223)

概览
把 0 点自动兑金币 worker 的「当天是否已兑」判定,从内存变量 last_run 改为持久化到 app_config(key auto_exchange.last_run_date)。修掉「进程重启即忘 → 每次非 0 点部署都全量补扫、把 0 点后才达标用户在非 0 点兑现金」的 bug。改 1 个实现文件 + 1 个文档 + 新增 1 个测试(+171 / −11)。

---------

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

app/core/compare_alert_worker.py

@@ -127,13 +127,16 @@ def build_hits(
                 hit = _dc_replace(make_hit(rec, "T5", "深度放弃"), stuck_point=stuck)
             else:
                 # trace 判出卡点 → 上面带卡点报。其余一律回退耗时/步数兜底:可读但没判出卡点、
-                # 读不到、无 trace,都过 total_ms/step 阈值——超长放弃照报(卡点列留空),不再因
-                # 「trace 可读但不原地卡」把超长放弃整条吞掉(线上 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:
@@ -226,6 +229,8 @@ def _scan_and_alert() -> None:
                 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:

app/core/daily_exchange_worker.py

@@ -5,9 +5,14 @@
 `wallet.daily_auto_exchange`。
 
 健壮性:
-- **逐用户幂等**:当天已有 exchange_in 流水的用户跳过(见 wallet._has_exchange_in_on),
-  故启动补跑 / 多次唤醒 / 进程重启都安全,不会重复兑。
-- **当天首跑即补**:进程起来时若当天还没兑过,立即兑一轮(等价原 timer 的 Persistent 补跑)。
+- **持久化「当天已兑」标记**(app_config `auto_exchange.last_run_date`):当天已兑则整轮跳过,
+  **标记跨进程重启不丢** → 同一北京日内多次启动/部署不再重复补扫。这是修 bug 的关键:原来用
+  内存变量记「今天跑过没」,进程重启即归零,导致每次非 0 点部署都全量补扫一遍、把 0 点后才达标
+  的用户在**非 0 点**兑成现金(用户反馈的「非 0 点也出现金币转现金记录」)。
+- **漏了 0 点即补**:标记 < 今天(如 0 点服务器宕机)时标记 != today → 照常补跑一轮(保留原
+  timer 的 Persistent 语义)。这类补跑确实落非 0 点,但仅限「真漏了 0 点」的罕见场景,非常态。
+- **逐用户幂等**:当天已有 exchange_in 流水的用户跳过(见 wallet._has_exchange_in_on),做兜底
+  防同机多进程 / 标记写入失败的竞态,不会重复兑。
 - **同机多进程互斥**:文件锁保证多 worker 只有一个实际跑(防跨进程并发导致 TOCTOU 双兑)。
 - **开关**:settings.AUTO_EXCHANGE_ENABLED=false 时不启动(与脚本/原 timer 同一开关)。
 """
@@ -27,10 +32,14 @@ from sqlalchemy.exc import SQLAlchemyError
 from app.core import rewards
 from app.core.config import settings
 from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
 from app.repositories import wallet as wallet_repo
 
 logger = logging.getLogger("shagua.daily_exchange")
 _LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "daily_exchange.lock"
+# 「上次成功自动兑的北京日」标记,持久化在 app_config(仿 compare_alert 水位,不进 CONFIG_DEFS——
+# 它是 worker 内部运行状态,非运营可配项)。用它替代内存 last_run:重启不丢 → 同日不重复补扫。
+LAST_RUN_KEY = "auto_exchange.last_run_date"
 
 
 def _touch_lock() -> None:
@@ -72,9 +81,44 @@ def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
                 _LOCK_PATH.unlink()
 
 
-def _exchange_once() -> dict:
+def _read_last_run(db) -> date | None:
+    """读持久化的「上次成功自动兑的北京日」标记(app_config,跨进程重启不丢)。脏值当作没跑过。"""
+    row = db.get(AppConfig, LAST_RUN_KEY)
+    if row is None or not row.value:
+        return None
+    try:
+        return date.fromisoformat(row.value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _write_last_run(db, day: date) -> None:
+    """落库「当天已兑」标记。用专用 key 直接写 AppConfig(仿 compare_alert 水位,不走 set_value)。"""
+    iso = day.isoformat()
+    row = db.get(AppConfig, LAST_RUN_KEY)
+    if row is None:
+        db.add(AppConfig(key=LAST_RUN_KEY, value=iso, updated_by_admin_id=None))
+    else:
+        row.value = iso
+    db.commit()
+
+
+def _exchange_if_due(db, today: date) -> dict | None:
+    """当天未兑过(持久化标记 != today)才跑一轮并记标记;已兑过返回 None(整轮跳过)。
+
+    标记写在 daily_auto_exchange 之后:即便中途崩,标记仍是旧值 → 下次重跑,逐用户幂等会跳过已兑的、
+    补完剩下的(安全)。真漏了 0 点(标记 < today)时标记 != today,仍会补跑,保留原「当天首跑即补」。
+    """
+    if _read_last_run(db) == today:
+        return None
+    result = wallet_repo.daily_auto_exchange(db)
+    _write_last_run(db, today)
+    return result
+
+
+def _exchange_once(today: date) -> dict | None:
     with SessionLocal() as db:
-        return wallet_repo.daily_auto_exchange(db)
+        return _exchange_if_due(db, today)
 
 
 async def _run_loop() -> None:
@@ -89,16 +133,15 @@ async def _run_loop() -> None:
 
 async def _run_locked_loop(interval: int) -> None:
     logger.info("daily auto-exchange worker started interval=%ss", interval)
-    # 本进程上次跑过的北京日;None=尚未跑过本进程(启动即补当天)。
-    last_run: date | None = None
     try:
         while True:
             try:
                 _touch_lock()
                 today = rewards.cn_today()
-                if last_run != today:
-                    result = await asyncio.to_thread(_exchange_once)
-                    last_run = today
+                # 「今天是否已兑」以持久化标记为准(见 _exchange_if_due),不再用内存变量 →
+                # 进程重启不会把当天当「没跑过」重复补扫;已兑当天返回 None(整轮跳过)。
+                result = await asyncio.to_thread(_exchange_once, today)
+                if result is not None:
                     logger.info("daily auto-exchange done date=%s result=%s", today, result)
             except SQLAlchemyError:
                 logger.exception("daily auto-exchange db error")

app/services/compare_alert.py

@@ -64,6 +64,31 @@ def classify_cancelled_fallback(
     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,
     *,
@@ -87,7 +112,11 @@ def classify_record(
             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])
-        return None
+        # 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,

app/services/compare_alert_format.py

@@ -180,12 +180,26 @@ _TABLE_COLUMNS = [
     {"name": "phone",  "display_name": "手机号",  "data_type": "text"},
     {"name": "cost",   "display_name": "用时",    "data_type": "text"},
     {"name": "reason", "display_name": "失败原因", "data_type": "text"},
-    {"name": "stuck",  "display_name": "卡点",    "data_type": "text"},
+    {"name": "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],
     *,
@@ -194,6 +208,8 @@ def format_alert_card(
     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 发送）。
 
@@ -225,6 +241,15 @@ def format_alert_card(
         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"

app/services/trace_stuck.py

@@ -46,17 +46,22 @@ class StuckPoint:
     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)
-        return f"{p}·{s}"
+        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:
@@ -108,7 +113,7 @@ def _platform_stuck(
     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)
+    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:
@@ -118,6 +123,8 @@ def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> Stuc
             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
@@ -128,9 +135,15 @@ def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> Stuc
             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)
+        return StuckResult(readable=True, points=points, last=last)
     except OSError:
         return StuckResult(readable=False, points=[])
 
@@ -151,9 +164,9 @@ def last_step(trace_dir: Path) -> StuckPoint | None:
         if best is None:
             return None
         _, platform, step_files = best
-        ps, _pg, _ts = _read_head(step_files[-1])
+        ps, pg, _ts = _read_head(step_files[-1])
         if ps is None:
             return None
-        return StuckPoint(platform, ps, len(step_files))  # stuck_ms=None(failed 不算时长)
+        return StuckPoint(platform, ps, len(step_files), detected_page=pg)  # stuck_ms=None(failed 不算时长)
     except OSError:
         return None

deploy/daily-exchange.md

@@ -5,9 +5,13 @@
 
 ## 现状(2026-06 起):已改为 App 进程内任务,不再需要 systemd timer
 「0 点自动兑换」现由 **App 进程内后台任务** `app.core.daily_exchange_worker` 负责:App 一起来就
-每 10 分钟检查、跨过北京 0 点自动跑一轮(逐用户幂等、文件锁互斥、重启补跑当天遗漏)。**无需再装 / 启用
+每 10 分钟检查、跨过北京 0 点自动跑一轮(逐用户幂等、文件锁互斥)。**无需再装 / 启用
 `daily-exchange.timer`**。
 
+- **「当天已兑」标记持久化在 app_config**(key=`auto_exchange.last_run_date`):当天兑过后,同一北京日
+  内进程重启 / 部署**不再重复补扫**——避免把 0 点后才达标的用户在非 0 点兑现金。只有真漏了 0 点
+  (标记 < 今天,如 0 点服务器宕机)才会在重启后补跑一轮。
+
 - 开关仍是 `.env` 的 `AUTO_EXCHANGE_ENABLED`(false → worker 不启动);间隔由 `AUTO_EXCHANGE_CHECK_INTERVAL_SEC`(默认 600s=10min)控。
 - 下方 systemd timer / 脚本属**遗留 + 手动应急**:逻辑同一套且幂等,可手动 `--once` 补跑;
   但**不要再 `enable` timer 与进程内 worker 并存**(虽幂等不会双兑,纯属多余)。

tests/test_compare_alert_format.py

@@ -282,8 +282,8 @@ def test_card_has_table_seven_columns():
     cols = table["columns"]
     assert len(cols) == 7
     display_names = [c["display_name"] for c in cols]
-    assert display_names == ["时间", "手机号", "用时", "失败原因", "卡点", "版本", "trace"]
-    # 「卡点」列在「失败原因」后、「版本」前
+    assert display_names == ["时间", "手机号", "用时", "失败原因", "末帧", "版本", "trace"]
+    # 「末帧」列在「失败原因」后、「版本」前
     names = [c["name"] for c in cols]
     reason_idx = names.index("reason")
     stuck_idx = names.index("stuck")
@@ -570,3 +570,29 @@ def test_card_stuck_point_shown_in_row():
     )
     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_rules.py

@@ -59,3 +59,34 @@ def test_reason_texts():
     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

@@ -46,10 +46,11 @@ def test_cancelled_stuck_reports_via_trace(tmp_path):
     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 可读但没判出原地卡点:仍过耗时/步数阈值兜底,超长(>90s)照报 T5;没卡点也用末帧标"退出前在哪屏"。
     # (线上 trace 几乎总可读,若不回退则 total_ms 阈值形同虚设、超长放弃永不报——见 compare-fail-alert 排查。)
     p = tmp_path / "20260804_y" / "eleme"
     _frame(p, 0, "set_address", "home")
@@ -61,7 +62,7 @@ def test_cancelled_readable_not_stuck_long_duration_reports(tmp_path):
     assert len(hits) == 1
     assert hits[0].alert_type == "T5"
     assert hits[0].reason == "深度放弃"
-    assert hits[0].stuck_point is None  # 没卡点 → 卡片卡点列显 "-"
+    assert hits[0].stuck_point == "饿了么·进店·store"  # 无卡点 → 用末帧(平台·环节·页面)标退出前在哪屏
 
 
 def test_cancelled_readable_not_stuck_short_no_report(tmp_path):

tests/test_daily_auto_exchange.py

@@ -0,0 +1,113 @@
+"""0 点自动兑金币 worker:持久化「当天已兑」标记,防同日重启重复补扫。
+
+背景(路径 B bug):worker 原用内存变量 last_run 记「今天跑过没」,进程重启即归零 → 每次
+非 0 点部署/重启都会全量补扫一遍,把 0 点后才达标的用户在**非 0 点**兑成现金(用户反馈的
+「非 0 点也出现金币转现金记录」)。修复:标记持久化到 app_config,跨重启不丢。
+"""
+from __future__ import annotations
+
+from datetime import date
+
+import pytest
+from sqlalchemy import delete, select, update
+
+from app.core import daily_exchange_worker as w
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.user import User
+from app.models.wallet import CashTransaction, CoinAccount
+from app.repositories import wallet as wallet_repo
+
+_PHONE_SEQ = [0]
+
+
+@pytest.fixture(autouse=True)
+def _isolate_exchange_state():
+    """daily_auto_exchange 全表扫描 + app_config 标记会跨用例泄漏(SQLite 测试库 session 级共享,
+    commit 后 rollback 是 no-op),故每个用例先清零所有余额 + 清掉自动兑标记,保证干净起步。"""
+    db = SessionLocal()
+    try:
+        db.execute(update(CoinAccount).values(
+            coin_balance=0, cash_balance_cents=0, invite_cash_balance_cents=0))
+        db.execute(delete(AppConfig).where(AppConfig.key == w.LAST_RUN_KEY))
+        db.commit()
+    finally:
+        db.close()
+    yield
+
+
+def _new_user(db, *, coin: int) -> int:
+    """建一个 User + 指定金币余额的 CoinAccount,返回 user_id。"""
+    _PHONE_SEQ[0] += 1
+    # 198 段避免撞其他测试文件的固定手机号 / username UNIQUE
+    u = User(phone=f"198{_PHONE_SEQ[0]:08d}", username=f"ax{_PHONE_SEQ[0]}", status="active")
+    db.add(u)
+    db.flush()
+    acc = wallet_repo.get_or_create_account(db, u.id, commit=False)
+    acc.coin_balance = coin
+    acc.total_coin_earned = coin
+    db.flush()
+    return u.id
+
+
+def test_first_run_converts_and_records_marker() -> None:
+    """首轮(标记为空)→ 到分全额兑 + 落库当天标记。"""
+    db = SessionLocal()
+    try:
+        uid = _new_user(db, coin=300)
+        db.commit()
+
+        today = date(2026, 8, 6)
+        result = w._exchange_if_due(db, today)
+
+        assert result is not None and result["converted"] >= 1
+        acc = db.get(CoinAccount, uid)
+        assert acc.coin_balance == 0 and acc.cash_balance_cents > 0   # 300 为整分,无零头
+        assert w._read_last_run(db) == today                          # 标记落库
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_same_day_restart_does_not_resweep() -> None:
+    """路径 B 回归:0 点兑过后,同日进程重启(下午部署)不得再补扫当天新达标用户。"""
+    db = SessionLocal()
+    try:
+        today = date(2026, 8, 6)
+        first = _new_user(db, coin=200)
+        db.commit()
+        assert w._exchange_if_due(db, today) is not None
+        assert db.get(CoinAccount, first).coin_balance == 0          # 首轮兑掉
+
+        # 0 点后才达标的用户(白天攒够;或 0 点是零头、白天赚够)
+        late = _new_user(db, coin=500)
+        db.commit()
+
+        # 模拟同一北京日内进程重启 → 持久化标记 == today → 整轮跳过,不碰 late
+        assert w._exchange_if_due(db, today) is None
+        acc = db.get(CoinAccount, late)
+        assert acc.coin_balance == 500 and acc.cash_balance_cents == 0
+        assert db.execute(select(CashTransaction).where(
+            CashTransaction.user_id == late,
+            CashTransaction.biz_type == "exchange_in",
+        )).first() is None                                            # late 无任何兑现金流水
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_new_day_runs_again() -> None:
+    """跨到北京新的一天(或真漏了 0 点,标记 < today)→ 照常补跑。"""
+    db = SessionLocal()
+    try:
+        w._write_last_run(db, date(2026, 8, 6))   # 昨天已兑
+        late = _new_user(db, coin=500)
+        db.commit()
+
+        result = w._exchange_if_due(db, date(2026, 8, 7))
+        assert result is not None and result["converted"] >= 1
+        assert db.get(CoinAccount, late).coin_balance == 0
+        assert w._read_last_run(db) == date(2026, 8, 7)
+    finally:
+        db.rollback()
+        db.close()

tests/test_trace_stuck.py

@@ -29,7 +29,7 @@ def test_stuck_when_tail_repeats_same_step(tmp_path):
         _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)]
+    assert res.points == [StuckPoint("meituan", "add_one_dish", 20, detected_page="meal_detail_popup")]
 
 
 def test_not_stuck_when_progressing(tmp_path):
@@ -81,7 +81,7 @@ def test_per_platform_one_stuck_one_normal(tmp_path):
         _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)]
+    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):
@@ -92,7 +92,27 @@ def test_last_step_returns_busiest_platform_last_env(tmp_path):
     for i in range(3):
         _frame(e, i, "enter_store", "store")
     sp = last_step(tmp_path)
-    assert sp == StuckPoint("meituan", "add_one_dish", 20)
+    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():