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

描述

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

---------

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

alembic/versions/coin_transaction_trace_id.py

@@ -0,0 +1,60 @@
+"""coin_transaction.trace_id (金币记录按会话聚合比价/领券看广告金币)
+
+Revision ID: coin_transaction_trace_id
+Revises: savings_record_trace_id
+Create Date: 2026-08-07
+
+比价/领券信息流发奖时把本场 trace_id 一并写入 coin_transaction;金币变动记录接口按
+trace_id 把一次比价/领券的多条广告金币聚合成一条。历史行从 ad_feed_reward_record
+回填(ref_id == client_event_id)。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'coin_transaction_trace_id'
+down_revision: Union[str, Sequence[str], None] = 'savings_record_trace_id'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # SQLite 下 ADD COLUMN(可空) 与 CREATE INDEX 均原生支持,无需 batch_alter_table。
+    op.add_column(
+        'coin_transaction',
+        sa.Column('trace_id', sa.String(length=64), nullable=True),
+    )
+    op.create_index(
+        op.f('ix_coin_transaction_trace_id'),
+        'coin_transaction',
+        ['trace_id'],
+        unique=False,
+    )
+    # 历史回填:从 ad_feed_reward_record 按 ref_id==client_event_id 补 trace_id。仅比价/领券两类、
+    # 仅当前为空、且广告行确有 trace_id 时补(EXISTS 守护);`trace_id IS NULL` 保证重跑幂等。
+    # 相关子查询 SQLite/PG 通用。两类 biz_type 在应用侧为 rewards.FEED_AD_SESSION_BIZ_TYPES,
+    # 此处按「迁移不可变」原则硬编码历史快照(勿改为 import 应用常量)。
+    # 大表(prod PG)如需可改按 id 区间分批;此处一次性 UPDATE。
+    op.execute(
+        """
+        UPDATE coin_transaction SET trace_id = (
+            SELECT r.trace_id FROM ad_feed_reward_record r
+            WHERE r.client_event_id = coin_transaction.ref_id)
+        WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon')
+          AND trace_id IS NULL
+          AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
+                      WHERE r2.client_event_id = coin_transaction.ref_id
+                        AND r2.trace_id IS NOT NULL)
+        """
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        op.f('ix_coin_transaction_trace_id'),
+        table_name='coin_transaction',
+    )
+    op.drop_column('coin_transaction', 'trace_id')

app/core/compare_alert_worker.py

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

app/core/rewards.py

@@ -26,6 +26,16 @@ SIGNIN_REWARDS: tuple[int, ...] = (
 SIGNIN_CYCLE_LEN: int = len(SIGNIN_REWARDS)
 
 
+# ===== 信息流广告「按会话聚合」的 biz_type =====
+# 比价 / 领券等候期看的信息流广告,每条各写一条 coin_transaction;这两类在「金币变动记录」
+# 里按 trace_id 聚合成一条展示(见 repositories/wallet.list_coin_transactions)。其余类型
+# (reward_video / guide_video / 通用 feed_ad_reward / signin / task_* 等)不聚合。
+FEED_AD_SESSION_BIZ_TYPES: tuple[str, str] = (
+    "feed_ad_reward_comparison",
+    "feed_ad_reward_coupon",
+)
+
+
 def signin_reward(cycle_day: int) -> int:
     """cycle_day 取值 1..SIGNIN_CYCLE_LEN。"""
     return SIGNIN_REWARDS[cycle_day - 1]

app/models/ad_feed_reward.py

@@ -36,8 +36,8 @@ class AdFeedRewardRecord(Base):
     # 点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页)。比价与领券共用同一信息流
     # 代码位,slot_id/our_code_id 分不出,只能客户端各调用点显式打标;NULL=历史/未升级客户端=未分类。
     feed_scene: Mapped[str | None] = mapped_column(String(16), nullable=True)
-    # 本次比价 trace_id(仅 comparison 场景由客户端带上):把这场广告金币归属到对应比价记录,
-    # 比价记录页按 trace_id 聚合本场实发金币显示「比价赚 N 金币」。领券/福利/旧客户端 = NULL。
+    # 本次会话 trace_id:比价一直带;领券自 2026-07-15(客户端 a98cab8)起也带;福利/旧客户端 = NULL。
+    # 用途:比价记录页按 trace_id 聚合「比价赚 N 金币」;金币记录列表把一次比价/领券的多条广告金币聚合成一条。
     trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     # 来源(广告收益报表用):我们的应用环境 prod/test + 我们配置的代码位 104xxx,由客户端 feed-reward 上报带上。旧数据为 NULL。
     app_env: Mapped[str | None] = mapped_column(String(16), nullable=True)

app/models/wallet.py

@@ -84,6 +84,9 @@ class CoinTransaction(Base):
     # 关联业务 id(签到日期、任务 key 等),可空
     ref_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
     remark: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    # 会话键:仅比价/领券信息流发奖(feed_ad_reward_comparison/coupon)时写入本场 trace_id,
+    # 金币记录列表据此把一次比价/领券的多条广告金币聚合成一条。其余类型 = NULL。
+    trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
 
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), index=True, nullable=False

app/repositories/ad_feed_reward.py

@@ -219,7 +219,7 @@ def grant_feed_reward(
     crud_wallet.grant_coins(
         db, user_id, coin,
         biz_type=reward_biz, ref_id=client_event_id,
-        remark=reward_remark,
+        remark=reward_remark, trace_id=trace_id,
     )
     rec = AdFeedRewardRecord(
         client_event_id=client_event_id,

app/repositories/wallet.py

@@ -10,9 +10,10 @@ import logging
 import re
 import unicodedata
 import uuid
+from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import func, select, update
+from sqlalchemy import String, and_, case, cast, func, literal, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -170,6 +171,7 @@ def grant_coins(
     biz_type: str,
     ref_id: str | None = None,
     remark: str | None = None,
+    trace_id: str | None = None,
 ) -> tuple[CoinAccount, CoinTransaction]:
     """金币变动入口(正数入账 / 负数出账)。更新余额 + 写流水,不 commit。
 
@@ -189,6 +191,7 @@ def grant_coins(
         biz_type=biz_type,
         ref_id=ref_id,
         remark=remark,
+        trace_id=trace_id,
         created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),  # 存北京 wall-clock(客户端原样切片显示)
     )
     db.add(txn)
@@ -257,24 +260,94 @@ def grant_invite_cash(
     return acc, txn
 
 
+@dataclass(frozen=True)
+class CoinLedgerRow:
+    """`list_coin_transactions` 的返回行。广告类按 trace_id 聚合后的展示行,
+    **非 ORM 对象**(避免把聚合后的 amount 误写回底层流水)。"""
+    id: int
+    amount: int
+    balance_after: int
+    biz_type: str
+    ref_id: str | None
+    remark: str | None
+    created_at: datetime
+    merged_count: int
+
+
 def list_coin_transactions(
     db: Session,
     user_id: int,
     *,
     limit: int = 20,
     cursor: int | None = None,
-) -> tuple[list[CoinTransaction], int | None]:
-    """金币流水分页(按 id 倒序,游标式)。
+) -> tuple[list[CoinLedgerRow], int | None]:
+    """金币流水分页(游标式,按 id 倒序)。
 
-    cursor 为上一页最后一条的 id;返回 (本页列表, next_cursor)。
-    next_cursor 为 None 表示没有下一页。
+    比价/领券两类信息流广告(rewards.FEED_AD_SESSION_BIZ_TYPES)且带 trace_id 的行,
+    按 trace_id 聚合成一条(一次比价/领券 = 一行):金额合计、代表行取组内最新一条
+    (MAX(id))的余额/时间、merged_count=组内条数。其余每条一行。
+
+    分组必须在该用户**全量**行上算真实 rep_id 后再按 rep_id 过滤——**不可**把 CTE 输入
+    裁成 id<cursor,否则会话行交错跨游标时会算出与上页重复的「残组」。
+
+    cursor 为上一页最后一条的 id(即其组 rep_id);返回 (本页列表, next_cursor)。
     """
-    stmt = select(CoinTransaction).where(CoinTransaction.user_id == user_id)
-    if cursor is not None:
-        stmt = stmt.where(CoinTransaction.id < cursor)
-    stmt = stmt.order_by(CoinTransaction.id.desc()).limit(limit)
+    ct = CoinTransaction
+    group_key = case(
+        (
+            and_(
+                ct.biz_type.in_(rewards.FEED_AD_SESSION_BIZ_TYPES),
+                ct.trace_id.is_not(None),
+            ),
+            literal("T:") + ct.trace_id,
+        ),
+        else_=literal("I:") + cast(ct.id, String),
+    ).label("group_key")
 
-    items = list(db.execute(stmt).scalars().all())
+    grp = (
+        select(
+            group_key,
+            func.max(ct.id).label("rep_id"),
+            func.sum(ct.amount).label("total_amount"),
+            func.count().label("merged_count"),
+        )
+        .where(ct.user_id == user_id)
+        .group_by(group_key)
+        .cte("grp")
+    )
+
+    stmt = (
+        select(
+            ct.id,
+            grp.c.total_amount.label("amount"),
+            ct.balance_after,
+            ct.biz_type,
+            ct.ref_id,
+            ct.remark,
+            ct.created_at,
+            grp.c.merged_count,
+        )
+        .select_from(grp)
+        .join(ct, ct.id == grp.c.rep_id)
+    )
+    if cursor is not None:
+        stmt = stmt.where(grp.c.rep_id < cursor)
+    stmt = stmt.order_by(grp.c.rep_id.desc()).limit(limit)
+
+    rows = db.execute(stmt).all()
+    items = [
+        CoinLedgerRow(
+            id=r.id,
+            amount=int(r.amount),
+            balance_after=r.balance_after,
+            biz_type=r.biz_type,
+            ref_id=r.ref_id,
+            remark=r.remark,
+            created_at=r.created_at,
+            merged_count=int(r.merged_count),
+        )
+        for r in rows
+    ]
     next_cursor = items[-1].id if len(items) == limit else None
     return items, next_cursor
 

app/schemas/welfare.py

@@ -30,6 +30,7 @@ class CoinTransactionOut(BaseModel):
     ref_id: str | None = None
     remark: str | None = None
     created_at: datetime
+    merged_count: int = Field(1, description="本行合并的底层流水条数(比价/领券按会话聚合;未合并=1)")
 
 
 class CoinTransactionPage(BaseModel):

app/services/trace_stuck.py

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

docs/superpowers/plans/2026-08-07-coin-ledger-aggregate-ad-rewards.md

@@ -0,0 +1,761 @@
+# 金币记录按会话汇总比价/领券看广告金币 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:** App「金币变动记录」里，一次比价 / 一次领券连续看广告获得的多条金币，按 `trace_id` 聚合成一条展示。
+
+**Architecture:** 纯后端。给 `coin_transaction` 加 `trace_id` 列并在发奖时写入；`GET /api/v1/wallet/coin-transactions` 改为按 `trace_id` 分组的游标分页（比价/领券两类 feed 广告按会话合并，其余每条一行）；历史行从 `ad_feed_reward_record` 回填。领券自 2026-07-15 起结算已带 trace_id，无需改 Android。仅动 App 用户接口，admin 审计接口不变。
+
+**Tech Stack:** FastAPI · SQLAlchemy 2.0 · Alembic · Pydantic v2 · pytest（SQLite）。
+
+**Spec:** `docs/superpowers/specs/2026-08-07-coin-ledger-aggregate-ad-rewards-design.md`
+
+**约定：** dev / pytest 均 **SQLite**，prod PostgreSQL——所有 SQL 保持 SQLite/PG 通用（不用 PG 专有语法）。金额单位为金币（非分）。
+
+---
+
+## File Structure
+
+| 文件 | 改动 | 职责 |
+|---|---|---|
+| `app/core/rewards.py` | 加常量 `FEED_AD_SESSION_BIZ_TYPES` | 「按会话聚合」的两类 biz_type 单一来源（查询 + 写入 + 测试共用） |
+| `app/models/wallet.py` | `CoinTransaction` 加 `trace_id` 列 | 金币流水会话键 |
+| `app/repositories/wallet.py` | `grant_coins` 加 `trace_id` 参；`list_coin_transactions` 改分组分页；加 `CoinLedgerRow` | 写入透传 + 聚合读取 |
+| `app/repositories/ad_feed_reward.py` | `grant_feed_reward` 调 `grant_coins` 时传 `trace_id` | 把本场 trace_id 写进金币流水 |
+| `app/schemas/welfare.py` | `CoinTransactionOut` 加 `merged_count` | 下发合并条数 |
+| `app/models/ad_feed_reward.py` | 更正 `trace_id` 注释 | 文档（领券自 2026-07-15 也带） |
+| `alembic/versions/coin_transaction_trace_id.py` | 新建迁移 | 加列 + 索引 + 回填历史 |
+| `tests/test_welfare.py` | 追加测试 | 覆盖写入 / 回填 / 发奖透传 / 聚合 / 分页防残组 |
+
+---
+
+## Task 1: 常量 + `CoinTransaction.trace_id` 列 + `grant_coins` 透传
+
+**Files:**
+- Modify: `app/core/rewards.py`（加常量）
+- Modify: `app/models/wallet.py:86`（`remark` 后加 `trace_id` 列）
+- Modify: `app/repositories/wallet.py:165-196`（`grant_coins` 加参数 + 写入）
+- Test: `tests/test_welfare.py`（追加）
+
+- [ ] **Step 1: 写失败测试**
+
+追加到 `tests/test_welfare.py` 末尾：
+
+```python
+def test_grant_coins_persists_trace_id(client) -> None:
+    """grant_coins 传 trace_id 落库;不传则为 None。"""
+    phone = "13800002001"
+    _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _, txn1 = crud_wallet.grant_coins(
+            db, user.id, 5, biz_type="feed_ad_reward_comparison",
+            ref_id="evt1", remark="比价奖励", trace_id="trace-A",
+        )
+        _, txn2 = crud_wallet.grant_coins(
+            db, user.id, 30, biz_type="signin", remark="每日签到奖励",
+        )
+        db.commit()
+        assert txn1.trace_id == "trace-A"
+        assert txn2.trace_id is None
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `pytest tests/test_welfare.py::test_grant_coins_persists_trace_id -q`
+Expected: FAIL —— `TypeError: grant_coins() got an unexpected keyword argument 'trace_id'`
+
+- [ ] **Step 3: 加常量**
+
+在 `app/core/rewards.py` 的签到常量段之后（约第 32 行 `SIGNIN_CYCLE_LEN` 之后）插入：
+
+```python
+# ===== 信息流广告「按会话聚合」的 biz_type =====
+# 比价 / 领券等候期看的信息流广告,每条各写一条 coin_transaction;这两类在「金币变动记录」
+# 里按 trace_id 聚合成一条展示(见 repositories/wallet.list_coin_transactions)。其余类型
+# (reward_video / guide_video / 通用 feed_ad_reward / signin / task_* 等)不聚合。
+FEED_AD_SESSION_BIZ_TYPES: tuple[str, str] = (
+    "feed_ad_reward_comparison",
+    "feed_ad_reward_coupon",
+)
+```
+
+- [ ] **Step 4: 加模型列**
+
+`app/models/wallet.py`，把 `remark` 行（第 86 行）后面补一列：
+
+```python
+    remark: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    # 会话键:仅比价/领券信息流发奖(feed_ad_reward_comparison/coupon)时写入本场 trace_id,
+    # 金币记录列表据此把一次比价/领券的多条广告金币聚合成一条。其余类型 = NULL。
+    trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
+```
+
+（`String` 在该文件已导入。`index=True` 会在测试 `create_all` 时自动建 `ix_coin_transaction_trace_id`。）
+
+- [ ] **Step 5: `grant_coins` 加参数 + 写入**
+
+`app/repositories/wallet.py` 的 `grant_coins`：签名加 `trace_id`，构造 `CoinTransaction` 时带上。
+
+签名（第 165-173 行）改为：
+
+```python
+def grant_coins(
+    db: Session,
+    user_id: int,
+    amount: int,
+    *,
+    biz_type: str,
+    ref_id: str | None = None,
+    remark: str | None = None,
+    trace_id: str | None = None,
+) -> tuple[CoinAccount, CoinTransaction]:
+```
+
+`CoinTransaction(...)` 构造（第 185-193 行）改为：
+
+```python
+    txn = CoinTransaction(
+        user_id=user_id,
+        amount=amount,
+        balance_after=acc.coin_balance,
+        biz_type=biz_type,
+        ref_id=ref_id,
+        remark=remark,
+        trace_id=trace_id,
+        created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),  # 存北京 wall-clock(客户端原样切片显示)
+    )
+```
+
+- [ ] **Step 6: 跑测试确认通过**
+
+Run: `pytest tests/test_welfare.py::test_grant_coins_persists_trace_id -q`
+Expected: PASS
+
+- [ ] **Step 7: 回归 + lint**
+
+Run: `pytest tests/test_welfare.py -q && ruff check app/core/rewards.py app/models/wallet.py app/repositories/wallet.py tests/test_welfare.py`
+Expected: 原有用例仍 PASS，无新增 lint。
+
+- [ ] **Step 8: 提交**
+
+```bash
+git add app/core/rewards.py app/models/wallet.py app/repositories/wallet.py tests/test_welfare.py
+git commit -m "feat(wallet): coin_transaction 增 trace_id 列 + grant_coins 透传
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 2: 迁移（加列 + 索引 + 回填历史）
+
+**Files:**
+- Create: `alembic/versions/coin_transaction_trace_id.py`
+- Test: `tests/test_welfare.py`（追加回填语义测试）
+
+- [ ] **Step 1: 写失败测试（回填语义）**
+
+在 `tests/test_welfare.py` 顶部 import 区补两行（若尚无）：
+
+```python
+from sqlalchemy import text
+from app.models.ad_feed_reward import AdFeedRewardRecord
+```
+
+追加测试：
+
+```python
+def test_backfill_coin_trace_id_from_ad_record(client) -> None:
+    """回填:coin_transaction(trace_id 空)按 ref_id==client_event_id 从 ad_feed_reward_record 补 trace_id;
+    只补比价/领券两类,无关类型与无匹配的不动。SQL 与迁移 coin_transaction_trace_id 保持同步。"""
+    phone = "13800002002"
+    _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        # 广告表:两条带 trace_id 的记录(模拟客户端已上报)
+        db.add(AdFeedRewardRecord(
+            client_event_id="evt-cmp", user_id=user.id, reward_date="2026-08-07",
+            duration_seconds=10, unit_count=1, ecpm_raw="1000",
+            feed_scene="comparison", trace_id="trace-CMP", coin=5, status="granted",
+        ))
+        db.add(AdFeedRewardRecord(
+            client_event_id="evt-cpn", user_id=user.id, reward_date="2026-08-07",
+            duration_seconds=10, unit_count=1, ecpm_raw="1000",
+            feed_scene="coupon", trace_id="trace-CPN", coin=7, status="granted",
+        ))
+        db.commit()
+        # 老金币流水:trace_id 全空(模拟改动前),ref_id 指向上面的广告事件
+        _, c1 = crud_wallet.grant_coins(db, user.id, 5, biz_type="feed_ad_reward_comparison", ref_id="evt-cmp", remark="比价奖励")
+        _, c2 = crud_wallet.grant_coins(db, user.id, 7, biz_type="feed_ad_reward_coupon", ref_id="evt-cpn", remark="领券奖励")
+        _, c3 = crud_wallet.grant_coins(db, user.id, 30, biz_type="signin", remark="每日签到奖励")
+        db.commit()
+        assert c1.trace_id is None and c2.trace_id is None
+
+        # 执行与迁移 upgrade() 等价的回填 SQL(务必与迁移保持一致)
+        db.execute(text(
+            """
+            UPDATE coin_transaction SET trace_id = (
+                SELECT r.trace_id FROM ad_feed_reward_record r
+                WHERE r.client_event_id = coin_transaction.ref_id)
+            WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon')
+              AND trace_id IS NULL
+              AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
+                          WHERE r2.client_event_id = coin_transaction.ref_id
+                            AND r2.trace_id IS NOT NULL)
+            """
+        ))
+        db.commit()
+        db.refresh(c1); db.refresh(c2); db.refresh(c3)
+        assert c1.trace_id == "trace-CMP"   # 比价补上
+        assert c2.trace_id == "trace-CPN"   # 领券补上
+        assert c3.trace_id is None          # 签到不动
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `pytest tests/test_welfare.py::test_backfill_coin_trace_id_from_ad_record -q`
+Expected: FAIL —— `sqlite3.OperationalError: no such column: trace_id`（Task 1 已加列则此步应改为直接 PASS；若已 PASS 说明 create_all 已含列，跳到 Step 4 建迁移）。
+
+> 说明：本测试验证的是回填 SQL 的 JOIN 语义，不经 Alembic。Task 1 完成后列已在测试库存在，测试可能直接 PASS——这是预期的（回填 SQL 本身是正确逻辑）。真正要新建的产物是迁移文件（Step 4），供 dev/prod 使用。
+
+- [ ] **Step 3: 跑测试确认通过**
+
+Run: `pytest tests/test_welfare.py::test_backfill_coin_trace_id_from_ad_record -q`
+Expected: PASS
+
+- [ ] **Step 4: 建迁移文件**
+
+Create `alembic/versions/coin_transaction_trace_id.py`：
+
+```python
+"""coin_transaction.trace_id (金币记录按会话聚合比价/领券看广告金币)
+
+Revision ID: coin_transaction_trace_id
+Revises: comparison_updated_at
+Create Date: 2026-08-07
+
+比价/领券信息流发奖时把本场 trace_id 一并写入 coin_transaction;金币变动记录接口按
+trace_id 把一次比价/领券的多条广告金币聚合成一条。历史行从 ad_feed_reward_record
+回填(ref_id == client_event_id)。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'coin_transaction_trace_id'
+down_revision: Union[str, Sequence[str], None] = 'comparison_updated_at'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # SQLite 下 ADD COLUMN(可空) 与 CREATE INDEX 均原生支持,无需 batch_alter_table。
+    op.add_column(
+        'coin_transaction',
+        sa.Column('trace_id', sa.String(length=64), nullable=True),
+    )
+    op.create_index(
+        op.f('ix_coin_transaction_trace_id'),
+        'coin_transaction',
+        ['trace_id'],
+        unique=False,
+    )
+    # 历史回填:从 ad_feed_reward_record 按 ref_id==client_event_id 补 trace_id。仅比价/领券两类、
+    # 仅当前为空、且广告行确有 trace_id 时补(EXISTS 守护);`trace_id IS NULL` 保证重跑幂等。
+    # 相关子查询 SQLite/PG 通用。两类 biz_type 在应用侧为 rewards.FEED_AD_SESSION_BIZ_TYPES,
+    # 此处按「迁移不可变」原则硬编码历史快照(勿改为 import 应用常量)。
+    # 大表(prod PG)如需可改按 id 区间分批;此处一次性 UPDATE。
+    op.execute(
+        """
+        UPDATE coin_transaction SET trace_id = (
+            SELECT r.trace_id FROM ad_feed_reward_record r
+            WHERE r.client_event_id = coin_transaction.ref_id)
+        WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon')
+          AND trace_id IS NULL
+          AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
+                      WHERE r2.client_event_id = coin_transaction.ref_id
+                        AND r2.trace_id IS NOT NULL)
+        """
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        op.f('ix_coin_transaction_trace_id'),
+        table_name='coin_transaction',
+    )
+    op.drop_column('coin_transaction', 'trace_id')
+```
+
+- [ ] **Step 5: 验证迁移可正反向应用（用一次性 scratch SQLite，不碰 dev 库）**
+
+Run:
+```bash
+DATABASE_URL="sqlite:///./data/_migcheck.db" python -m alembic upgrade head && \
+DATABASE_URL="sqlite:///./data/_migcheck.db" python -m alembic downgrade -1 && \
+DATABASE_URL="sqlite:///./data/_migcheck.db" python -m alembic upgrade head && \
+rm -f ./data/_migcheck.db
+```
+Expected: 三条 alembic 命令均无报错，最终 head 落在 `coin_transaction_trace_id`；`rm` 清掉临时库。
+
+- [ ] **Step 6: lint**
+
+Run: `ruff check alembic/versions/coin_transaction_trace_id.py tests/test_welfare.py`
+Expected: 无新增 lint。
+
+- [ ] **Step 7: 提交**
+
+```bash
+git add alembic/versions/coin_transaction_trace_id.py tests/test_welfare.py
+git commit -m "feat(wallet): 迁移加 coin_transaction.trace_id + 索引 + 回填历史
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 3: `grant_feed_reward` 把 trace_id 写进金币流水
+
+**Files:**
+- Modify: `app/repositories/ad_feed_reward.py:219-223`（`grant_coins` 调用加 `trace_id`）
+- Test: `tests/test_welfare.py`（追加）
+
+- [ ] **Step 1: 写失败测试**
+
+在 `tests/test_welfare.py` 顶部 import 区补（若尚无）：
+
+```python
+from sqlalchemy import select
+from app.models.wallet import CoinTransaction
+from app.repositories import ad_feed_reward as crud_feed
+```
+
+追加测试：
+
+```python
+def test_grant_feed_reward_sets_coin_trace_id(client) -> None:
+    """grant_feed_reward(comparison) 把 trace_id 透传给 grant_coins,coin_transaction 带上本场 trace_id。"""
+    phone = "13800002003"
+    _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        rec = crud_feed.grant_feed_reward(
+            db, user.id,
+            client_event_id="evt-fr-1", ecpm="1000", duration_seconds=10,
+            feed_scene="comparison", trace_id="trace-FR", display_coin=5,
+        )
+        assert rec.status == "granted", rec.status
+        txn = db.execute(
+            select(CoinTransaction).where(
+                CoinTransaction.user_id == user.id,
+                CoinTransaction.ref_id == "evt-fr-1",
+            )
+        ).scalar_one()
+        assert txn.biz_type == "feed_ad_reward_comparison"
+        assert txn.trace_id == "trace-FR"
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `pytest tests/test_welfare.py::test_grant_feed_reward_sets_coin_trace_id -q`
+Expected: FAIL —— `AssertionError: assert None == 'trace-FR'`（`grant_coins` 尚未收到 trace_id）。
+
+- [ ] **Step 3: 传 trace_id**
+
+`app/repositories/ad_feed_reward.py` 第 219-223 行 `grant_coins` 调用改为：
+
+```python
+    crud_wallet.grant_coins(
+        db, user_id, coin,
+        biz_type=reward_biz, ref_id=client_event_id,
+        remark=reward_remark, trace_id=trace_id,
+    )
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `pytest tests/test_welfare.py::test_grant_feed_reward_sets_coin_trace_id -q`
+Expected: PASS
+
+- [ ] **Step 5: lint**
+
+Run: `ruff check app/repositories/ad_feed_reward.py tests/test_welfare.py`
+Expected: 无新增 lint。
+
+- [ ] **Step 6: 提交**
+
+```bash
+git add app/repositories/ad_feed_reward.py tests/test_welfare.py
+git commit -m "feat(ad-feed): 发奖把本场 trace_id 写入金币流水
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 4: 聚合查询 —— `list_coin_transactions` 分组 + `CoinLedgerRow` + `merged_count`
+
+**Files:**
+- Modify: `app/repositories/wallet.py`（imports + 加 `CoinLedgerRow` + 重写 `list_coin_transactions`）
+- Modify: `app/schemas/welfare.py:23-32`（`CoinTransactionOut` 加 `merged_count`）
+- Test: `tests/test_welfare.py`（追加 3 个聚合测试 + 一个 `_seed_coin` 帮手）
+
+- [ ] **Step 1: 写失败测试**
+
+在 `tests/test_welfare.py` 追加帮手 + 测试：
+
+```python
+def _seed_coin(db, user_id, amount, biz_type, *, trace_id=None, ref_id=None, remark=None):
+    crud_wallet.grant_coins(
+        db, user_id, amount, biz_type=biz_type, ref_id=ref_id, remark=remark, trace_id=trace_id
+    )
+
+
+def test_coin_transactions_aggregate_by_trace(client) -> None:
+    """一次比价的多条广告金币聚合成一条:金额合计、merged_count=条数、balance_after 取最后一条。"""
+    phone = "13800002004"
+    token = _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")
+        _seed_coin(db, user.id, 3, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")
+        _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e3", remark="比价奖励")
+        db.commit()
+    r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token))
+    assert r.status_code == 200, r.text
+    items = r.json()["items"]
+    assert len(items) == 1
+    row = items[0]
+    assert row["biz_type"] == "feed_ad_reward_comparison"
+    assert row["amount"] == 12          # 5+3+4
+    assert row["merged_count"] == 3
+    assert row["balance_after"] == 12   # 末条到账后余额(本用户从 0 起)
+
+
+def test_coin_transactions_distinct_traces_stay_separate(client) -> None:
+    """不同 trace(两次比价 / 一次领券)各成一条;不同会话不合并。"""
+    phone = "13800002005"
+    token = _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="cmpA", ref_id="a1", remark="比价奖励")
+        _seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b1", remark="比价奖励")
+        _seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b2", remark="比价奖励")
+        _seed_coin(db, user.id, 7, "feed_ad_reward_coupon", trace_id="cpnC", ref_id="c1", remark="领券奖励")
+        db.commit()
+    items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"]
+    assert len(items) == 3
+    assert sorted(i["amount"] for i in items) == [5, 7, 12]
+    cpn = next(i for i in items if i["biz_type"] == "feed_ad_reward_coupon")
+    assert cpn["amount"] == 7 and cpn["merged_count"] == 1
+
+
+def test_coin_transactions_non_session_rows_stay_per_row(client) -> None:
+    """签到 / 无 trace 的通用信息流各自一行,不被聚合;夹在比价广告中间的签到不影响比价聚合。"""
+    phone = "13800002006"
+    token = _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")
+        _seed_coin(db, user.id, 30, "signin", remark="每日签到奖励")            # 夹在两条比价广告中间
+        _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")
+        _seed_coin(db, user.id, 8, "feed_ad_reward", trace_id=None, ref_id="w1", remark="信息流广告奖励")
+        db.commit()
+    items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"]
+    assert len(items) == 3
+    cmp_row = next(i for i in items if i["biz_type"] == "feed_ad_reward_comparison")
+    assert cmp_row["amount"] == 9 and cmp_row["merged_count"] == 2
+    signin_row = next(i for i in items if i["biz_type"] == "signin")
+    assert signin_row["amount"] == 30 and signin_row["merged_count"] == 1
+    feed_row = next(i for i in items if i["biz_type"] == "feed_ad_reward")
+    assert feed_row["amount"] == 8 and feed_row["merged_count"] == 1
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `pytest tests/test_welfare.py::test_coin_transactions_aggregate_by_trace -q`
+Expected: FAIL —— 未聚合，返回 3 条 / `merged_count` 字段缺失（`KeyError` 或 `len(items)==3`）。
+
+- [ ] **Step 3: schema 加 `merged_count`**
+
+`app/schemas/welfare.py` 的 `CoinTransactionOut`（第 23-32 行）在 `created_at` 后加一行：
+
+```python
+class CoinTransactionOut(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    amount: int = Field(..., description="正=入账,负=出账")
+    balance_after: int
+    biz_type: str
+    ref_id: str | None = None
+    remark: str | None = None
+    created_at: datetime
+    merged_count: int = Field(1, description="本行合并的底层流水条数(比价/领券按会话聚合;未合并=1)")
+```
+
+- [ ] **Step 4: 补 imports + 加 `CoinLedgerRow`**
+
+`app/repositories/wallet.py` 顶部：把 `from sqlalchemy import func, select, update`（第 15 行）改为：
+
+```python
+from sqlalchemy import String, and_, case, cast, func, literal, select, update
+```
+
+在 stdlib import 区（约第 7-13 行）加一行：
+
+```python
+from dataclasses import dataclass
+```
+
+在 `list_coin_transactions` 之前加返回行类型：
+
+```python
+@dataclass(frozen=True)
+class CoinLedgerRow:
+    """`list_coin_transactions` 的返回行。广告类按 trace_id 聚合后的展示行,
+    **非 ORM 对象**(避免把聚合后的 amount 误写回底层流水)。"""
+    id: int
+    amount: int
+    balance_after: int
+    biz_type: str
+    ref_id: str | None
+    remark: str | None
+    created_at: datetime
+    merged_count: int
+```
+
+- [ ] **Step 5: 重写 `list_coin_transactions`**
+
+`app/repositories/wallet.py` 第 260-279 行整体替换为：
+
+```python
+def list_coin_transactions(
+    db: Session,
+    user_id: int,
+    *,
+    limit: int = 20,
+    cursor: int | None = None,
+) -> tuple[list[CoinLedgerRow], int | None]:
+    """金币流水分页(游标式,按 id 倒序)。
+
+    比价/领券两类信息流广告(rewards.FEED_AD_SESSION_BIZ_TYPES)且带 trace_id 的行,
+    按 trace_id 聚合成一条(一次比价/领券 = 一行):金额合计、代表行取组内最新一条
+    (MAX(id))的余额/时间、merged_count=组内条数。其余每条一行。
+
+    分组必须在该用户**全量**行上算真实 rep_id 后再按 rep_id 过滤——**不可**把 CTE 输入
+    裁成 id<cursor,否则会话行交错跨游标时会算出与上页重复的「残组」。
+
+    cursor 为上一页最后一条的 id(即其组 rep_id);返回 (本页列表, next_cursor)。
+    """
+    ct = CoinTransaction
+    group_key = case(
+        (
+            and_(
+                ct.biz_type.in_(rewards.FEED_AD_SESSION_BIZ_TYPES),
+                ct.trace_id.is_not(None),
+            ),
+            literal("T:") + ct.trace_id,
+        ),
+        else_=literal("I:") + cast(ct.id, String),
+    ).label("group_key")
+
+    grp = (
+        select(
+            group_key,
+            func.max(ct.id).label("rep_id"),
+            func.sum(ct.amount).label("total_amount"),
+            func.count().label("merged_count"),
+        )
+        .where(ct.user_id == user_id)
+        .group_by(group_key)
+        .cte("grp")
+    )
+
+    stmt = (
+        select(
+            ct.id,
+            grp.c.total_amount.label("amount"),
+            ct.balance_after,
+            ct.biz_type,
+            ct.ref_id,
+            ct.remark,
+            ct.created_at,
+            grp.c.merged_count,
+        )
+        .select_from(grp)
+        .join(ct, ct.id == grp.c.rep_id)
+    )
+    if cursor is not None:
+        stmt = stmt.where(grp.c.rep_id < cursor)
+    stmt = stmt.order_by(grp.c.rep_id.desc()).limit(limit)
+
+    rows = db.execute(stmt).all()
+    items = [
+        CoinLedgerRow(
+            id=r.id,
+            amount=int(r.amount),
+            balance_after=r.balance_after,
+            biz_type=r.biz_type,
+            ref_id=r.ref_id,
+            remark=r.remark,
+            created_at=r.created_at,
+            merged_count=int(r.merged_count),
+        )
+        for r in rows
+    ]
+    next_cursor = items[-1].id if len(items) == limit else None
+    return items, next_cursor
+```
+
+（端点 `app/api/v1/wallet.py` 无需改：`CoinTransactionOut.model_validate(it)` 对 `CoinLedgerRow` dataclass 按 `from_attributes` 读取即可。）
+
+- [ ] **Step 6: 跑三个聚合测试确认通过**
+
+Run: `pytest tests/test_welfare.py -k "coin_transactions_aggregate_by_trace or distinct_traces_stay_separate or non_session_rows_stay_per_row" -q`
+Expected: 3 PASS
+
+- [ ] **Step 7: 全量回归 + lint**
+
+Run: `pytest tests/test_welfare.py -q && ruff check app/repositories/wallet.py app/schemas/welfare.py tests/test_welfare.py`
+Expected: 原有用例(含 `test_signin_flow` / `test_exchange_flow` 等对 coin-transactions 的断言)仍 PASS；无新增 lint。
+
+- [ ] **Step 8: 提交**
+
+```bash
+git add app/repositories/wallet.py app/schemas/welfare.py tests/test_welfare.py
+git commit -m "feat(wallet): 金币记录按 trace_id 聚合比价/领券看广告金币
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 5: 分页防残组回归测试
+
+**Files:**
+- Test: `tests/test_welfare.py`（追加）
+
+- [ ] **Step 1: 写回归测试**
+
+追加：
+
+```python
+def test_coin_transactions_pagination_no_phantom_regroup(client) -> None:
+    """交错跨游标不产生残组:会话广告成员被其它记录隔开、rep_id 在游标上、成员在游标下时,
+    翻到下一页该会话不得以「残组」重复出现(锁死 spec §11 反面优化警示)。"""
+    phone = "13800002007"
+    token = _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")   # id=n+1
+        _seed_coin(db, user.id, 30, "signin", remark="每日签到奖励")                                              # id=n+2
+        _seed_coin(db, user.id, 40, "signin", remark="每日签到奖励")                                              # id=n+3
+        _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")   # id=n+4 = t1 的 rep
+        db.commit()
+    # 第 1 页(limit=2):按 rep 降序 = [t1(rep=n+4, 合计 9), signin(n+3, 40)]
+    p1 = client.get("/api/v1/wallet/coin-transactions?limit=2", headers=_auth(token)).json()
+    assert len(p1["items"]) == 2
+    assert p1["items"][0]["biz_type"] == "feed_ad_reward_comparison"
+    assert p1["items"][0]["amount"] == 9 and p1["items"][0]["merged_count"] == 2
+    assert p1["items"][1]["biz_type"] == "signin" and p1["items"][1]["amount"] == 40
+    assert p1["next_cursor"] is not None
+    # 第 2 页:只剩另一条 signin(30);t1 的 rep 在游标上,成员虽在游标下也不得成残组重复
+    p2 = client.get(
+        f"/api/v1/wallet/coin-transactions?limit=2&cursor={p1['next_cursor']}",
+        headers=_auth(token),
+    ).json()
+    assert len(p2["items"]) == 1
+    assert p2["items"][0]["biz_type"] == "signin" and p2["items"][0]["amount"] == 30
+    assert all(i["biz_type"] != "feed_ad_reward_comparison" for i in p2["items"])
+    assert p2["next_cursor"] is None
+```
+
+- [ ] **Step 2: 跑测试确认通过（锁定不变量）**
+
+Run: `pytest tests/test_welfare.py::test_coin_transactions_pagination_no_phantom_regroup -q`
+Expected: PASS（Task 4 的全量分组实现已正确；若 FAIL 说明 CTE 被错误地按 cursor 裁剪，回到 Task 4 Step 5 修正）。
+
+- [ ] **Step 3: lint**
+
+Run: `ruff check tests/test_welfare.py`
+Expected: 无新增 lint。
+
+- [ ] **Step 4: 提交**
+
+```bash
+git add tests/test_welfare.py
+git commit -m "test(wallet): 锁死金币记录分组分页不产生残组
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 6: 更正 `ad_feed_reward_record.trace_id` 过时注释
+
+**Files:**
+- Modify: `app/models/ad_feed_reward.py:39-40`
+
+- [ ] **Step 1: 改注释**
+
+`app/models/ad_feed_reward.py` 第 39-40 行两行注释替换为：
+
+```python
+    # 本次会话 trace_id:比价一直带;领券自 2026-07-15(客户端 a98cab8)起也带;福利/旧客户端 = NULL。
+    # 用途:比价记录页按 trace_id 聚合「比价赚 N 金币」;金币记录列表把一次比价/领券的多条广告金币聚合成一条。
+    trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
+```
+
+- [ ] **Step 2: 冒烟 + lint**
+
+Run: `python -c "import app.models.ad_feed_reward" && ruff check app/models/ad_feed_reward.py`
+Expected: 无报错、无 lint。
+
+- [ ] **Step 3: 提交**
+
+```bash
+git add app/models/ad_feed_reward.py
+git commit -m "docs(ad-feed): 更正 trace_id 注释(领券自 2026-07-15 也带)
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## 收尾验证
+
+- [ ] **全量测试**：`pytest -q`（对齐 memory「preexisting-test-lint-debt」：先前已知失败数不变、无本改动引入的新失败）。
+- [ ] **lint**：`ruff check .`（无本改动引入的新问题）。
+- [ ] **迁移落 dev**：`python -m alembic upgrade head`（把 dev SQLite 迁到最新，确认 `./run.sh` 正常）。
+
+---
+
+## Self-Review（写完计划后自查）
+
+**Spec 覆盖：**
+- §4 加列 → Task 1 Step 4 + Task 2。✓
+- §5 写入透传 → Task 1 Step 5 + Task 3。✓
+- §6 聚合查询（CASE 分组键、MAX(id) 代表/游标、biz_type 门控、常量）→ Task 4 Step 4-5 + Task 1 常量。✓
+- §7 `merged_count` 下发 → Task 4 Step 3。✓
+- §8 回填（幂等 + EXISTS 守护 + 通用 SQL）→ Task 2 Step 4。✓
+- §9 注释更正 → Task 6。✓
+- §10 App 端无需改 → 端点未改（Task 4 Step 5 注）。✓
+- §11 反面优化警示（残组）→ Task 4 docstring + Task 5 回归测试。✓
+- §3 只改用户接口、admin 不动 → 全程只碰 `crud_wallet.list_coin_transactions`，未触 `app/admin`。✓
+- §12 测试（分组/隔离/空 trace/分页/回填）→ Task 2/4/5。✓
+
+**占位扫描：** 无 TBD / TODO；每个代码步骤均给出完整代码与确切命令。✓
+
+**类型一致：** `grant_coins(..., trace_id=...)`（Task 1）↔ `grant_feed_reward` 调用（Task 3）↔ 查询 `rewards.FEED_AD_SESSION_BIZ_TYPES`（Task 4）↔ 迁移硬编码同两值（Task 2，有意快照）一致；`CoinLedgerRow` 字段（Task 4 Step 4）↔ `CoinTransactionOut` 字段（Task 4 Step 3）逐一对应（id/amount/balance_after/biz_type/ref_id/remark/created_at/merged_count）。✓

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

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

docs/superpowers/specs/2026-08-07-coin-ledger-aggregate-ad-rewards-design.md

@@ -0,0 +1,190 @@
+# 金币记录：比价/领券看广告金币按会话汇总成一条 设计
+
+- 日期：2026-08-07
+- 分支：feat/coin-ledger-aggregate-ad-rewards
+- 相关：`app/repositories/wallet.py`（发币/流水）、`app/repositories/ad_feed_reward.py`（信息流发奖）、`app/api/v1/wallet.py`（金币流水接口）、Android `CoinHistoryViewModel`（列表渲染）
+
+## 1. 背景与问题
+
+App「金币变动记录」列表按**每看一次广告一条**展示。一次比价的等候期会连续看多条信息流广告，一次领券流程同理，于是列表里一次比价/领券会刷出一长串「比价奖励 +x」「领券奖励 +x」，淹没其它记录、观感差。
+
+需求：**一次比价的所有看广告金币合并成一条、一次领券的合并成一条**，金额取该次会话合计。
+
+## 2. 现状（关键事实）
+
+- 发币唯一入口 `crud_wallet.grant_coins`（`app/repositories/wallet.py:165`）：每条广告写一条 `CoinTransaction`。金币账本 `coin_transaction` 与现金账本物理分离。
+- 信息流发奖 `grant_feed_reward`（`app/repositories/ad_feed_reward.py:71`）按点位场景拆 `biz_type`：
+  - `feed_ad_reward_comparison`（比价，remark「比价奖励」）
+  - `feed_ad_reward_coupon`（领券，remark「领券奖励」）
+  - `feed_ad_reward`（通用/福利/旧端）
+- **trace_id 已全链贯通到结算**：
+  - 比价：客户端结算上报早已带 trace_id。
+  - 领券：`CouponForegroundService.reportFeedReward` 自 **2026-07-15**（commit `a98cab8`）起带 `feedScene="coupon" + traceId=sessionTraceId`。
+  - 两者的 trace_id 都落到了 `ad_feed_reward_record.trace_id`（`grant_feed_reward` 写入）。
+- **唯一缺口**：`grant_feed_reward` 调 `grant_coins` 时**没把 trace_id 透传下去**，`coin_transaction` 表本身**没有 trace_id 列** → 金币流水层无法按会话分组。
+- 列表接口 `GET /api/v1/coin-transactions`（`app/api/v1/wallet.py:61` → `list_coin_transactions` `app/repositories/wallet.py:260`）：按 `id` 倒序游标分页。
+- 展示文案权在客户端：Android `CoinHistoryViewModel.coinTitle` 按 `bizType` 直显固定标题（比价奖励/领券奖励），后端 `remark` 只作兜底。
+
+结论：这是**纯后端**改动，领券不需要额外 Android 改动即可生效。
+
+## 3. 目标与范围
+
+- 目标：金币记录列表里，同一次比价的多条「比价奖励」合并成一条、同一次领券的多条「领券奖励」合并成一条；金额为该次会话合计。
+- 会话键：`trace_id`（一次比价/一次领券 = 一个 trace_id）。与「比价记录页」现有的 trace_id 聚合口径一致。
+- **范围**：只圈两类 `biz_type` —— `feed_ad_reward_comparison`、`feed_ad_reward_coupon`。激励视频 `reward_video`、引导视频 `guide_video`、通用信息流 `feed_ad_reward`、签到、任务、兑换等**一律不动**，仍每条一行。
+- 聚合位置：**后端**（`/coin-transactions` 直接返回合并后的行）。客户端聚合被否决（见 §11）。
+- **只改 App 用户接口**：聚合仅作用于 `crud_wallet.list_coin_transactions`（`GET /api/v1/wallet/coin-transactions`）。**admin 接口不动**——`app/admin/routers/wallet.py` 走独立的 `queries.list_all_coin_transactions`（跨用户、可按 `biz_type` 筛），审计/客服必须能看到**每一条**广告发币，保持每条一行。两者物理分离，聚合天然不波及 admin，此处显式声明防误改。
+- 历史：**加列 + 一次性回填**，历史记录也合并。
+
+## 4. 数据模型改动
+
+`coin_transaction` 新增列：
+
+| 列 | 类型 | 约束 | 说明 |
+|---|---|---|---|
+| `trace_id` | `String(64)` | nullable, index | 会话键；仅比价/领券信息流发奖时写入，其余为 NULL |
+
+- 新增 Alembic 迁移（`render_as_batch` 兼容 SQLite）：加列 + 索引 `ix_coin_transaction_trace_id`。
+- 可选复合索引 `(user_id, trace_id)` 辅助分组扫描（视线上量级决定，MVP 可先只加单列索引）。
+
+## 5. 写入路径改动
+
+- `grant_coins(...)` 增加可选参数 `trace_id: str | None = None`，写入 `CoinTransaction.trace_id`。默认 None → 其余调用方（签到/任务/兑换/激励视频）零改动、保持 NULL。
+- `grant_feed_reward` 调 `grant_coins` 时透传 `trace_id=trace_id`（比价/领券自然有值；welfare/旧端为 None）。
+- 幂等不变：`client_event_id` 仍是幂等键；`grant_coins` 仍不 commit，由 `_commit_record` 同事务提交。
+
+## 6. 读取/聚合查询（核心）
+
+`list_coin_transactions` 改为「分组游标分页」。用 CTE 先分组、再 JOIN 回代表行取展示字段。
+
+分组规则：
+
+```
+group_key =
+  若 biz_type ∈ (feed_ad_reward_comparison, feed_ad_reward_coupon) 且 trace_id 非空
+      → 'T:' || trace_id          # 同一会话所有广告归一组
+  否则
+      → 'I:' || id                # 每条自成一组(行为等同现状)
+```
+
+每组取：`rep_id = MAX(id)`、`total = SUM(amount)`、`merged_count = COUNT(*)`；再 JOIN `coin_transaction` 取 `rep_id` 那条的 `balance_after / biz_type / ref_id / remark / created_at / trace_id`。
+
+等价 SQL（SQLite / PostgreSQL 通用）：
+
+```sql
+WITH grp AS (
+  SELECT
+    CASE WHEN biz_type IN ('feed_ad_reward_comparison','feed_ad_reward_coupon')
+              AND trace_id IS NOT NULL
+         THEN 'T:' || trace_id
+         ELSE 'I:' || CAST(id AS TEXT) END AS group_key,
+    MAX(id)     AS rep_id,
+    SUM(amount) AS total_amount,
+    COUNT(*)    AS merged_count
+  FROM coin_transaction
+  WHERE user_id = :uid
+  GROUP BY group_key
+)
+SELECT ct.id, grp.total_amount AS amount, ct.balance_after,
+       ct.biz_type, ct.ref_id, ct.remark, ct.trace_id, ct.created_at,
+       grp.merged_count
+FROM grp
+JOIN coin_transaction ct ON ct.id = grp.rep_id
+WHERE (:cursor IS NULL OR grp.rep_id < :cursor)
+ORDER BY grp.rep_id DESC
+LIMIT :limit;
+```
+
+返回行字段口径：
+
+- `id` = 组内 `MAX(id)`：做列表 key + 下一页游标。组间不重复（每个 id 只属一组），`rep_id DESC` 是全序，游标 `rep_id < cursor` 干净。
+- `amount` = 组内合计（该次会话总金币；这些广告行金额恒正）。
+- `balance_after` / `created_at` / `remark` / `ref_id` = 代表行（最后一条）的值。`balance_after` 即该次会话最后一条广告到账后的余额，正确。
+- `merged_count` = 合并条数（未合并 = 1）。
+- `next_cursor` = 本页最后一行的 `id`（= 其组 rep_id），够 limit 才给，否则 None。
+
+非分组类型、`trace_id` 为空的旧广告行 → 各自成组（`'I:'||id`），行为与现状完全一致。
+
+> 说明（分组安全性）：分组键按 `biz_type` **门控**——通用 `feed_ad_reward`（福利/旧端）即便偶带 trace_id 也走 `'I:'||id` 保持每条一行，只有比价/领券两类才按 trace_id 合并。trace_id 由后端按会话签发、全局唯一（一个 trace = 一次比价**或**一次领券），故仅按 trace_id 分组不会把两类混并，代表行的 `biz_type` 唯一确定展示标题。
+>
+> 实现约定：这两类 biz_type 抽成模块常量 `FEED_AD_SESSION_BIZ_TYPES`，**查询 / 回填 / 测试共用一处**，避免字符串散落三地漂移。
+
+## 7. 接口契约
+
+`GET /api/v1/coin-transactions` 响应新增 `merged_count`（`CoinTransactionOut` 加字段，Pydantic 默认 1）：
+
+```jsonc
+{
+  "items": [
+    { "id": 1520, "amount": 12, "balance_after": 3380,
+      "biz_type": "feed_ad_reward_comparison", "ref_id": "evt_...",
+      "remark": "比价奖励", "created_at": "2026-08-07T12:03:11",
+      "merged_count": 5 },              // 本次比价看了 5 条广告，合计 +12
+    { "id": 1512, "amount": 30, "balance_after": 3368,
+      "biz_type": "signin", "remark": "每日签到奖励",
+      "merged_count": 1 }
+  ],
+  "next_cursor": 1490
+}
+```
+
+`trace_id` 是否在响应里下发：**可选**，MVP 不下发（客户端用不到，标题按 biz_type、金额已合计）。若后续要做「点开看明细」再补。
+
+## 8. 历史回填
+
+迁移里一次性把历史 `coin_transaction.trace_id` 从 `ad_feed_reward_record` 补齐（相关子查询 UPDATE，SQLite/PG 通用）：
+
+```sql
+UPDATE coin_transaction SET trace_id = (
+  SELECT r.trace_id FROM ad_feed_reward_record r
+  WHERE r.client_event_id = coin_transaction.ref_id)
+WHERE biz_type IN ('feed_ad_reward_comparison','feed_ad_reward_coupon')
+  AND trace_id IS NULL
+  AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
+              WHERE r2.client_event_id = coin_transaction.ref_id
+                AND r2.trace_id IS NOT NULL);
+```
+
+- 关联键：`coin_transaction.ref_id == ad_feed_reward_record.client_event_id`（`grant_feed_reward` 里两者同源，见 `app/repositories/ad_feed_reward.py:221`）。`client_event_id` 有唯一约束 → 相关子查询至多一行，无「子查询多行」风险。
+- 只补两类 biz_type；只在广告行确有 trace_id 时补（`EXISTS` 守护）。
+- **幂等**：`AND trace_id IS NULL` 保证重跑不重复改。
+- **大表守护**：`coin_transaction` 线上若很大，单条全表 UPDATE 在 PG 上是一次性长事务/行锁；必要时按 `id` 区间**分批**（每批数万行）跑。SQLite（dev）无所谓。
+- 2026-07-15 前的领券老行 `ad_feed_reward_record.trace_id` 本就为空 → 回填后仍为空 → 保持每条一行（可接受，会随时间自然淘汰）。
+
+## 9. 顺手更正
+
+更新 `app/models/ad_feed_reward.py:39` 的 `trace_id` 注释：现「仅 comparison 场景由客户端带上……领券/福利/旧客户端 = NULL」已过时，改为「比价一直带；领券自 2026-07-15 起也带；福利/旧客户端仍 NULL」。
+
+## 10. App 端
+
+- **核心无需改**：接口返回合并后的行，`CoinHistoryViewModel.coinTitle` 仍按 `bizType` 显示「比价奖励/领券奖励」，金额显示合计值。
+- **可选增强（后续、非本次必须）**：客户端读到 `merged_count > 1` 时把标题渲染成「比价奖励 ×5」或副行「看了 5 条广告」。做不做都不影响本功能生效。
+
+## 11. 边界与取舍
+
+- **时间交错重排**：一次比价的广告中间夹了签到等其它记录时，合并行锚定在「该会话最后一条广告」的 id 上，夹在中间的签到会排到合并行之后。这是折叠的固有效果，可接受。
+- **余额不连续**：相邻两展示行的 `balance_after` 之间存在被折叠的中间变动，属聚合固有现象；每行 `balance_after` 仍是该点真实运行余额。
+- **迟到广告**：会话已过、同 trace 的迟到广告到账后并入该组、金额变大、合并行上移（因 rep_id 变大）；用户未刷新则短暂 stale。会话结束后广告即时结算，迟到罕见，可接受。
+- **单条会话**：一次会话只看了 1 条广告 → 组内 1 行、`merged_count=1`、展示与现状一致，无需特判。
+- **查询成本**：分组 CTE 每页对该用户**全部** `coin_transaction` 行分组后再分页，O(N)/页，比原「`id < cursor LIMIT n`」的索引 seek 重。金币记录非热路径（偶尔打开、翻几页）；粗估重度用户万级行/年、单页分组数毫秒级（估算，非实测），可接受。
+  - **索引口径**：分组按 CASE 表达式（`'T:'||trace_id` / `'I:'||id`）在内存做，`(user_id, trace_id)` 复合索引**并不加速**该 group-by；真正需要的只是「按 user_id 取该用户全部行」，现有 `ix_coin_transaction_user_id` 已够。故本设计**只加 `trace_id` 单列索引**（回填 / 潜在按 trace 查用），不加 `(user_id, trace_id)`；若要 index-only 可选覆盖索引 `(user_id, id, biz_type, trace_id, amount)`，MVP 不加。
+  - ⚠️ **反面优化警示**：**不要**把 CTE 输入裁成 `WHERE id < :cursor` 求快。会话行若被其它记录交错、成员跨越游标（部分成员 `id < cursor`、但组的 `rep_id ≥ cursor`），裁剪后会算出一个 rep_id 更小的「残组」→ 与上一页已展示的整组**重复出现**。分组必须在该用户全量行上算出真实 `rep_id` 后，再 `rep_id < cursor` 过滤。
+  - **未来读优化（YAGNI，暂不做）**：若量级压力显现，加只读会话聚合投影表 `coin_ad_session_agg(user_id, trace_id, total_coin, cnt, last_txn_id, last_created_at)`，写广告流水时 upsert，列表用「非广告行 UNION 该投影」走索引分页。这是**读优化**（明细账本仍每条一行、审计不动），与 §13 否决的「写时账本聚合」不同。
+
+## 12. 测试
+
+- **分组**：同 trace 多条比价广告 → 一行（`amount` 合计 / `merged_count=N` / `balance_after` 取最后 / `id=MAX`）；两次不同 trace 的比价 → 两行；领券同 trace → 一行。
+- **隔离**：比价广告序列中间夹一条签到 → 比价折叠成一行、签到独立成行；`reward_video` / `signin` / `exchange_out` → 仍 1:1，`merged_count=1`。
+- **空 trace_id**：`feed_ad_reward`（通用）或 trace_id 为 NULL 的广告行 → 每条一行。
+- **分页**：多组跨页时按 `rep_id` 游标翻页不重不漏；一个大组（跨越 limit 边界的多条底层行）不被拆成两页；`next_cursor` 到底返回 None。
+- **交错跨游标不重复（防残组回归）**：造一个会话，其广告成员 id 被一条非广告行隔开（部分成员在游标下、`rep_id` 在游标上），翻到下一页时该组**不得**再以「残组」重复出现——锁死 §11 的反面优化警示。
+- **回填迁移**：造 `coin_transaction`（比价/领券行，ref_id 指向带 trace_id 的 `ad_feed_reward_record`）+ 无关行，跑迁移后仅两类被正确补 trace_id、无关行不动。
+- **既有测试**：`/coin-transactions` 现有用例（非广告类型 1:1、分页）保持通过。
+
+## 13. 不做（YAGNI）
+
+- 不做写时聚合（会破坏实时到账 / 幂等 / 余额连续性）。
+- 不做客户端聚合（SWR 缓存 + 游标分页下，一个会话可能跨页，跨页分组脆弱）。
+- 不动激励视频 / 引导视频 / 通用信息流 / 签到等其它类型。
+- 不动「比价记录页」——它的「比价赚 N 金币」是独立查询（直接聚合 `ad_feed_reward_record`），不受本改动影响。
+- MVP 不下发 trace_id、不做「点开看明细」。

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

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

tests/test_compare_alert_stuck_worker.py

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

tests/test_trace_stuck.py

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

tests/test_welfare.py

@@ -4,7 +4,7 @@
 """
 from __future__ import annotations
 
-import json
+from sqlalchemy import select, text
 
 from app.core.config import settings
 from app.core.rewards import (
@@ -17,6 +17,9 @@ from app.core.rewards import (
 )
 from app.db.session import SessionLocal
 from app.integrations import pangle
+from app.models.ad_feed_reward import AdFeedRewardRecord
+from app.models.wallet import CoinTransaction
+from app.repositories import ad_feed_reward as crud_feed
 from app.repositories import wallet as crud_wallet
 from app.repositories.user import get_user_by_phone
 
@@ -384,3 +387,208 @@ def test_endpoints_require_auth(client) -> None:
     assert client.get("/api/v1/savings/summary").status_code == 401
     assert client.get("/api/v1/savings/battle").status_code == 401
     assert client.get("/api/v1/savings/records").status_code == 401
+
+
+def test_grant_coins_persists_trace_id(client) -> None:
+    """grant_coins 传 trace_id 落库;不传则为 None。"""
+    phone = "13800002001"
+    _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _, txn1 = crud_wallet.grant_coins(
+            db, user.id, 5, biz_type="feed_ad_reward_comparison",
+            ref_id="evt1", remark="比价奖励", trace_id="trace-A",
+        )
+        _, txn2 = crud_wallet.grant_coins(
+            db, user.id, 30, biz_type="signin", remark="每日签到奖励",
+        )
+        db.commit()
+        assert txn1.trace_id == "trace-A"
+        assert txn2.trace_id is None
+
+
+def test_backfill_coin_trace_id_from_ad_record(client) -> None:
+    """回填:coin_transaction(trace_id 空)按 ref_id==client_event_id 从 ad_feed_reward_record 补 trace_id;
+    只补比价/领券两类,无关类型与无匹配的不动。SQL 与迁移 coin_transaction_trace_id 保持同步。"""
+    phone = "13800002002"
+    _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        db.add(AdFeedRewardRecord(
+            client_event_id="evt-cmp", user_id=user.id, reward_date="2026-08-07",
+            duration_seconds=10, unit_count=1, ecpm_raw="1000",
+            feed_scene="comparison", trace_id="trace-CMP", coin=5, status="granted",
+        ))
+        db.add(AdFeedRewardRecord(
+            client_event_id="evt-cpn", user_id=user.id, reward_date="2026-08-07",
+            duration_seconds=10, unit_count=1, ecpm_raw="1000",
+            feed_scene="coupon", trace_id="trace-CPN", coin=7, status="granted",
+        ))
+        # 广告行存在但 trace_id 为空(2026-07-15 前的老比价广告):回填必须跳过、保持 NULL(EXISTS 守护)。
+        db.add(AdFeedRewardRecord(
+            client_event_id="evt-null", user_id=user.id, reward_date="2026-08-07",
+            duration_seconds=10, unit_count=1, ecpm_raw="1000",
+            feed_scene="comparison", trace_id=None, coin=5, status="granted",
+        ))
+        db.commit()
+        _, c1 = crud_wallet.grant_coins(db, user.id, 5, biz_type="feed_ad_reward_comparison", ref_id="evt-cmp", remark="比价奖励")
+        _, c2 = crud_wallet.grant_coins(db, user.id, 7, biz_type="feed_ad_reward_coupon", ref_id="evt-cpn", remark="领券奖励")
+        _, c3 = crud_wallet.grant_coins(db, user.id, 30, biz_type="signin", remark="每日签到奖励")
+        _, c4 = crud_wallet.grant_coins(db, user.id, 5, biz_type="feed_ad_reward_comparison", ref_id="evt-null", remark="比价奖励")
+        db.commit()
+        assert c1.trace_id is None and c2.trace_id is None
+
+        db.execute(text(
+            """
+            UPDATE coin_transaction SET trace_id = (
+                SELECT r.trace_id FROM ad_feed_reward_record r
+                WHERE r.client_event_id = coin_transaction.ref_id)
+            WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon')
+              AND trace_id IS NULL
+              AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
+                          WHERE r2.client_event_id = coin_transaction.ref_id
+                            AND r2.trace_id IS NOT NULL)
+            """
+        ))
+        db.commit()
+        db.refresh(c1)
+        db.refresh(c2)
+        db.refresh(c3)
+        db.refresh(c4)
+        assert c1.trace_id == "trace-CMP"
+        assert c2.trace_id == "trace-CPN"
+        assert c3.trace_id is None          # 无关类型不动
+        assert c4.trace_id is None          # 广告行 trace_id 为空 → EXISTS 守护跳过、保持 NULL
+
+
+def test_grant_feed_reward_sets_coin_trace_id(client) -> None:
+    """grant_feed_reward(comparison) 把 trace_id 透传给 grant_coins,coin_transaction 带上本场 trace_id。"""
+    phone = "13800002003"
+    _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        rec = crud_feed.grant_feed_reward(
+            db, user.id,
+            client_event_id="evt-fr-1", ecpm="1000", duration_seconds=10,
+            feed_scene="comparison", trace_id="trace-FR", display_coin=5,
+        )
+        assert rec.status == "granted", rec.status
+        txn = db.execute(
+            select(CoinTransaction).where(
+                CoinTransaction.user_id == user.id,
+                CoinTransaction.ref_id == "evt-fr-1",
+            )
+        ).scalar_one()
+        assert txn.biz_type == "feed_ad_reward_comparison"
+        assert txn.trace_id == "trace-FR"
+
+
+def _seed_coin(db, user_id, amount, biz_type, *, trace_id=None, ref_id=None, remark=None):
+    crud_wallet.grant_coins(
+        db, user_id, amount, biz_type=biz_type, ref_id=ref_id, remark=remark, trace_id=trace_id
+    )
+
+
+def test_coin_transactions_aggregate_by_trace(client) -> None:
+    """一次比价的多条广告金币聚合成一条:金额合计、merged_count=条数、balance_after 取最后一条。"""
+    phone = "13800002004"
+    token = _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")
+        _seed_coin(db, user.id, 3, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")
+        _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e3", remark="比价奖励")
+        db.commit()
+    r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token))
+    assert r.status_code == 200, r.text
+    items = r.json()["items"]
+    assert len(items) == 1
+    row = items[0]
+    assert row["biz_type"] == "feed_ad_reward_comparison"
+    assert row["amount"] == 12
+    assert row["merged_count"] == 3
+    assert row["balance_after"] == 12
+
+
+def test_coin_transactions_distinct_traces_stay_separate(client) -> None:
+    """不同 trace(两次比价 / 一次领券)各成一条;不同会话不合并。"""
+    phone = "13800002005"
+    token = _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="cmpA", ref_id="a1", remark="比价奖励")
+        _seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b1", remark="比价奖励")
+        _seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b2", remark="比价奖励")
+        _seed_coin(db, user.id, 7, "feed_ad_reward_coupon", trace_id="cpnC", ref_id="c1", remark="领券奖励")
+        db.commit()
+    items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"]
+    assert len(items) == 3
+    assert sorted(i["amount"] for i in items) == [5, 7, 12]
+    cpn = next(i for i in items if i["biz_type"] == "feed_ad_reward_coupon")
+    assert cpn["amount"] == 7 and cpn["merged_count"] == 1
+
+
+def test_coin_transactions_non_session_rows_stay_per_row(client) -> None:
+    """签到 / 无 trace 的通用信息流各自一行,不被聚合;夹在比价广告中间的签到不影响比价聚合。"""
+    phone = "13800002006"
+    token = _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")
+        _seed_coin(db, user.id, 30, "signin", remark="每日签到奖励")
+        _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")
+        _seed_coin(db, user.id, 8, "feed_ad_reward", trace_id=None, ref_id="w1", remark="信息流广告奖励")
+        db.commit()
+    items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"]
+    assert len(items) == 3
+    cmp_row = next(i for i in items if i["biz_type"] == "feed_ad_reward_comparison")
+    assert cmp_row["amount"] == 9 and cmp_row["merged_count"] == 2
+    signin_row = next(i for i in items if i["biz_type"] == "signin")
+    assert signin_row["amount"] == 30 and signin_row["merged_count"] == 1
+    feed_row = next(i for i in items if i["biz_type"] == "feed_ad_reward")
+    assert feed_row["amount"] == 8 and feed_row["merged_count"] == 1
+
+
+def test_coin_transactions_pagination_no_phantom_regroup(client) -> None:
+    """交错跨游标不产生残组:会话广告成员被其它记录隔开、rep_id 在游标上、成员在游标下时,
+    翻到下一页该会话不得以「残组」重复出现(锁死 spec §11 反面优化警示)。"""
+    phone = "13800002007"
+    token = _login(client, phone)
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")   # id=n+1
+        _seed_coin(db, user.id, 30, "signin", remark="每日签到奖励")                                              # id=n+2
+        _seed_coin(db, user.id, 40, "signin", remark="每日签到奖励")                                              # id=n+3
+        _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")   # id=n+4 = t1 的 rep
+        db.commit()
+        t1_ids = db.execute(
+            select(CoinTransaction.id)
+            .where(CoinTransaction.user_id == user.id, CoinTransaction.trace_id == "t1")
+            .order_by(CoinTransaction.id)
+        ).scalars().all()
+    # 第 1 页(limit=2):按 rep 降序 = [t1(rep=n+4, 合计 9), signin(n+3, 40)]
+    p1 = client.get("/api/v1/wallet/coin-transactions?limit=2", headers=_auth(token)).json()
+    assert len(p1["items"]) == 2
+    assert p1["items"][0]["biz_type"] == "feed_ad_reward_comparison"
+    assert p1["items"][0]["amount"] == 9 and p1["items"][0]["merged_count"] == 2
+    assert p1["items"][1]["biz_type"] == "signin" and p1["items"][1]["amount"] == 40
+    assert p1["next_cursor"] is not None
+    # 前置条件:游标(第1页末条 rep)必须严格落在 t1 两成员 id 之间,才是真正的「跨游标」场景;
+    # 否则用例会退化成非跨游标、悄悄测错形状仍通过,失去回归守护意义。
+    assert t1_ids[0] < p1["next_cursor"] < t1_ids[-1]
+    # 第 2 页:只剩另一条 signin(30);t1 的 rep 在游标上,成员虽在游标下也不得成残组重复
+    p2 = client.get(
+        f"/api/v1/wallet/coin-transactions?limit=2&cursor={p1['next_cursor']}",
+        headers=_auth(token),
+    ).json()
+    assert len(p2["items"]) == 1
+    assert p2["items"][0]["biz_type"] == "signin" and p2["items"][0]["amount"] == 30
+    assert all(i["biz_type"] != "feed_ad_reward_comparison" for i in p2["items"])
+    assert p2["next_cursor"] is None