fix(alembic): 合并多 head(#65 device 迁移 + comparison_token_cols)

#65 的 device 迁移链与 comparison_token_cols 都从 ceb286289426 分叉成双 head,alembic upgrade head 报 Multiple head revisions;加 merge 迁移 4dc2af7ebe74 合并(空 DDL 仅接链)。

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

.env.example

@@ -27,6 +27,11 @@ JG_PRIVATE_KEY_PATH=./secrets/jverify_rsa_private.pem
 JG_VERIFY_ENDPOINT=https://api.verification.jpush.cn/v1/web/loginTokenVerify
 JG_REQUEST_TIMEOUT_SEC=15
 
+# ===== 无障碍保护存活监控(pull 后置检测;本期不接推送)=====
+HEARTBEAT_MONITOR_ENABLED=true
+HEARTBEAT_TIMEOUT_MINUTES=10
+HEARTBEAT_SCAN_INTERVAL_SEC=60
+
 # ===== 短信 (mock 模式) =====
 # mock = true 时,任意 6 位数字均通过,且 /sms/send 不真发短信(只 log)。
 # 后续接阿里云/腾讯云短信时,改成 false 并填供应商相关 key。
@@ -34,15 +39,6 @@ SMS_MOCK=true
 SMS_CODE_TTL_SEC=300
 SMS_SEND_INTERVAL_SEC=60
 
-# ===== 测试账号(release 包全流程联调用)=====
-# 配一个固定测试手机号,专供无 SIM 卡 / 不走一键登录时打通全流程:该号登录【免短信验证码】
-# (real 模式下也跳过校验)、每次登录【都重走新手引导】,并有【每日登录上限】防被人猜到号后脚本刷。
-# 逻辑见 app/core/test_account.py,与其他业务解耦。
-# ⚠️ 留空 = 关闭整功能(生产默认);要启用才填号(如 11111111111)。改完重启生效,随时可清空停用。
-TEST_ACCOUNT_PHONE=
-# 该测试号每日最多登录次数,当日超过即拒绝(429),次日归零。
-TEST_ACCOUNT_DAILY_LIMIT=500
-
 # ===== 美团联盟 CPS =====
 # 美团联盟后台 → 媒体管理 拿到的 appkey 和密钥
 MT_CPS_APP_KEY=

alembic/versions/4dc2af7ebe74_merge_device_65_and_comparison_token_.py

@@ -0,0 +1,26 @@
+"""merge device#65 and comparison_token_cols heads
+
+Revision ID: 4dc2af7ebe74
+Revises: comparison_token_cols, ad_feed_reward_trace_id
+Create Date: 2026-06-23 17:57:54.083531
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '4dc2af7ebe74'
+down_revision: Union[str, Sequence[str], None] = ('comparison_token_cols', 'ad_feed_reward_trace_id')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

alembic/versions/ad_feed_reward_trace_id.py

@@ -0,0 +1,42 @@
+"""ad_feed_reward_record.trace_id (比价看广告金币归属到比价记录)
+
+Revision ID: ad_feed_reward_trace_id
+Revises: device_kill_alert_pending
+Create Date: 2026-06-21
+
+比价等候期信息流广告(feed_ad_reward)结算时,客户端带上本场比价 trace_id 落此列;
+比价记录页按 trace_id 聚合本场实发金币,显示「比价赚 N 金币」。领券/福利/旧客户端 = NULL。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'ad_feed_reward_trace_id'
+down_revision: Union[str, Sequence[str], None] = 'device_kill_alert_pending'
+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(
+        'ad_feed_reward_record',
+        sa.Column('trace_id', sa.String(length=64), nullable=True),
+    )
+    op.create_index(
+        op.f('ix_ad_feed_reward_record_trace_id'),
+        'ad_feed_reward_record',
+        ['trace_id'],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        op.f('ix_ad_feed_reward_record_trace_id'),
+        table_name='ad_feed_reward_record',
+    )
+    op.drop_column('ad_feed_reward_record', 'trace_id')

alembic/versions/comparison_token_cols.py

@@ -0,0 +1,37 @@
+"""add input_tokens / output_tokens to comparison_record
+
+比价记录增加本次 LLM 累计 token(server 从 llm_calls[].usage 累加),
+供 admin 比价记录页展示 token 数与估算成本。
+
+Revision ID: comparison_token_cols
+Revises: feedback_review_fields
+Create Date: 2026-06-23 00:40:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "comparison_token_cols"
+down_revision: Union[str, Sequence[str], None] = "feedback_review_fields"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "comparison_record",
+        sa.Column("input_tokens", sa.Integer(), nullable=True),
+    )
+    op.add_column(
+        "comparison_record",
+        sa.Column("output_tokens", sa.Integer(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("comparison_record", "output_tokens")
+    op.drop_column("comparison_record", "input_tokens")

alembic/versions/device_kill_alert_pending.py

@@ -0,0 +1,39 @@
+"""device.kill_alert_pending (无障碍掉线「后置检测」待提醒标记)
+
+Revision ID: device_kill_alert_pending
+Revises: device_liveness_table
+Create Date: 2026-06-19
+
+后置检测 pull 版(spec/accessibility-liveness-pull-prompt.md):worker 检出心跳掉线即置此标记,
+客户端下次进 App 拉取到 → 弹「开启自启动」引导,交互后 ack 清回。与 liveness_state 解耦,故单列。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'device_kill_alert_pending'
+down_revision: Union[str, Sequence[str], None] = 'device_liveness_table'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # server_default=sa.false() 跨方言: SQLite 渲染 0 / PG 渲染 false(别用 sa.text("0"),PG 布尔严格会报错),
+    # 给存量行回填 False。
+    with op.batch_alter_table('device_liveness', schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                'kill_alert_pending',
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.false(),
+            )
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('device_liveness', schema=None) as batch_op:
+        batch_op.drop_column('kill_alert_pending')

alembic/versions/device_table.py

@@ -0,0 +1,53 @@
+"""device table (无障碍保护存活检测 + 极光推送)
+
+Revision ID: device_liveness_table
+Revises: ceb286289426
+Create Date: 2026-06-15 12:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'device_liveness_table'
+down_revision: Union[str, Sequence[str], None] = 'ceb286289426'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'device_liveness',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('user_id', sa.Integer(), nullable=False),
+        sa.Column('device_id', sa.String(length=128), nullable=False),
+        sa.Column('registration_id', sa.String(length=64), nullable=True),
+        sa.Column('platform', sa.String(length=16), nullable=False),
+        sa.Column('app_version', sa.String(length=32), nullable=True),
+        sa.Column('ever_protected', sa.Boolean(), nullable=False),
+        sa.Column('last_heartbeat_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('last_report_protection_on', sa.Boolean(), nullable=False),
+        sa.Column('liveness_state', sa.String(length=16), nullable=False),
+        sa.Column('notified_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint('user_id', 'device_id', name='uq_device_liveness_user_device'),
+    )
+    with op.batch_alter_table('device_liveness', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_device_liveness_user_id'), ['user_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_device_liveness_device_id'), ['device_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_device_liveness_last_heartbeat_at'), ['last_heartbeat_at'], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('device_liveness', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_device_liveness_last_heartbeat_at'))
+        batch_op.drop_index(batch_op.f('ix_device_liveness_device_id'))
+        batch_op.drop_index(batch_op.f('ix_device_liveness_user_id'))
+
+    op.drop_table('device_liveness')

app/admin/repositories/cps.py

@@ -474,44 +474,53 @@ def group_order_daily(
 
 
 def group_wx_users(db: Session, *, group_id: int, limit: int = 200) -> list[dict]:
-    """群内微信用户(首次来源 first_group_id=group_id)+ 每人在该群的 visit/copy 次数(领券画像)。
+    """该群来过的微信用户:在本群有带 openid 点击的所有用户 + 本群 visit/copy 次数 + 画像。
 
-    按首次进入倒序。copy 次数 = 该用户在该群点了几次「复制口令」(领券意向),visit = 进落地页次数。
+    用「本群点击」判定归属,而非 cps_wx_user.first_group_id —— 同一用户会在多个群活动,
+    first_group_id 只记首次授权群,会漏掉"先在别处授权、cookie 已存、又来本群"的用户。
+    copy=本群点「复制口令」次数(领券意向),visit=进落地页次数。按本群首次点击倒序。
+    昵称/头像来自授权画像;美团/京东群只 302 跳转、无 userinfo 触发点,故多为空。
     """
-    users = (
-        db.execute(
-            select(CpsWxUser)
-            .where(CpsWxUser.first_group_id == group_id)
-            .order_by(desc(CpsWxUser.first_seen))
-            .limit(limit)
-        )
-        .scalars()
-        .all()
-    )
-    if not users:
-        return []
-    openids = [u.openid for u in users]
-    stat: dict[str, dict] = {}
     rows = db.execute(
         select(CpsClick.openid, CpsClick.event_type, func.count())
         .where(CpsClick.group_id == group_id)
-        .where(CpsClick.openid.in_(openids))
+        .where(CpsClick.openid.is_not(None))
         .group_by(CpsClick.openid, CpsClick.event_type)
     ).all()
+    if not rows:
+        return []
+    stat: dict[str, dict] = {}
     for openid, event_type, cnt in rows:
         s = stat.setdefault(openid, {"visit": 0, "copy": 0})
         if event_type == "copy":
             s["copy"] = cnt
         else:
             s["visit"] = cnt
-    return [
+    openids = list(stat.keys())
+    # 本群每个 openid 的首次点击时间(排序 + 展示"首次进入")
+    first_seen = dict(
+        db.execute(
+            select(CpsClick.openid, func.min(CpsClick.clicked_at))
+            .where(CpsClick.group_id == group_id)
+            .where(CpsClick.openid.is_not(None))
+            .group_by(CpsClick.openid)
+        ).all()
+    )
+    # 关联画像(未授权 userinfo 的昵称/头像为空)
+    users = {
+        u.openid: u
+        for u in db.execute(select(CpsWxUser).where(CpsWxUser.openid.in_(openids))).scalars().all()
+    }
+    result = [
         {
-            "openid": u.openid,
-            "nickname": u.nickname,
-            "headimgurl": u.headimgurl,
-            "first_seen": u.first_seen,
-            "visit_count": stat.get(u.openid, {}).get("visit", 0),
-            "copy_count": stat.get(u.openid, {}).get("copy", 0),
+            "openid": openid,
+            "nickname": users[openid].nickname if openid in users else None,
+            "headimgurl": users[openid].headimgurl if openid in users else None,
+            "first_seen": first_seen.get(openid),
+            "visit_count": stat[openid]["visit"],
+            "copy_count": stat[openid]["copy"],
         }
-        for u in users
+        for openid in openids
     ]
+    result.sort(key=lambda x: x["first_seen"], reverse=True)
+    return result[:limit]

app/admin/schemas/comparison.py

@@ -33,6 +33,8 @@ class AdminComparisonListItem(BaseModel):
     step_count: int | None = None
     llm_call_count: int | None = None
     retry_count: int | None = None
+    input_tokens: int | None = None      # Σ usage.prompt_tokens(server 派生)
+    output_tokens: int | None = None     # Σ usage.completion_tokens(server 派生)
     device_model: str | None = None
     rom_vendor: str | None = None
     rom_name: str | None = None

app/api/v1/ad.py

@@ -32,6 +32,7 @@ from app.schemas.ad import (
     EcpmReportOut,
     FeedRewardIn,
     FeedRewardOut,
+    FeedRewardUnitsOut,
     PangleCallbackOut,
     RewardNoShowIn,
     RewardNoShowOut,
@@ -373,6 +374,7 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
         adn=payload.adn,
         slot_id=payload.slot_id,
         feed_scene=payload.feed_scene,
+        trace_id=payload.trace_id,
         app_env=payload.app_env,
         our_code_id=payload.our_code_id,
         aborted=payload.aborted,
@@ -390,6 +392,21 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
     )
 
 
+@router.get(
+    "/feed-reward/units",
+    response_model=FeedRewardUnitsOut,
+    summary="查账号累计信息流发奖份数(前端算因子2 LT、做实时金币进度条用)",
+    dependencies=[Depends(rate_limit(120, 60, "ad-feed-units"))],
+)
+def feed_reward_units(user: CurrentUser, db: DbSession) -> FeedRewardUnitsOut:
+    """返回账号累计已发奖份数(因子2 LT 基线)。
+
+    信息流金币进度条:前端 show 时拉一次,之后每满 10 秒一份、逐份用 (granted_units + offset)
+    查 LT 因子,精确复刻发奖公式 → 进度条数值 ≈ 实际到账。
+    """
+    return FeedRewardUnitsOut(granted_units=crud_feed.granted_unit_total(db, user.id))
+
+
 @router.post(
     "/reward-noshow",
     response_model=RewardNoShowOut,

app/api/v1/auth.py

@@ -15,7 +15,6 @@ import logging
 from fastapi import APIRouter, Depends, HTTPException, status
 
 from app.api.deps import CurrentUser, DbSession
-from app.core import test_account
 from app.core.ratelimit import rate_limit
 from app.core.security import TokenError, decode_token, issue_token_pair
 from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
@@ -39,17 +38,13 @@ logger = logging.getLogger("shagua.auth")
 router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
 
 
-def _login_response(
-    user, *, onboarding_completed: bool, force_onboarding: bool = False
-) -> TokenWithUser:
-    """登录类接口共用的响应组装。onboarding_completed 据登录请求的 device_id 算好后传入;
-    force_onboarding 仅测试号置 True(让应用内重新登录也重走引导)。"""
+def _login_response(user, *, onboarding_completed: bool) -> TokenWithUser:
+    """登录类接口共用的响应组装。onboarding_completed 据登录请求的 device_id 算好后传入。"""
     tokens = issue_token_pair(user.id)
     return TokenWithUser(
         **tokens,
         user=UserOut.model_validate(user),
         onboarding_completed=onboarding_completed,
-        force_onboarding=force_onboarding,
     )
 
 
@@ -87,12 +82,6 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
     dependencies=[Depends(rate_limit(10, 60, "sms-send"))],  # 同 IP 每分钟≤10 次(防一 IP 刷不同号)
 )
 def sms_send(req: SmsSendRequest) -> SmsSendResponse:
-    # 测试账号:不真发短信(号码非真实手机号,真发会失败/浪费),直接放行让客户端进入填码界面。
-    # 真正的"免验证码"在 /sms/login 跳过校验;每日上限也在 login 处算(send 不耗额度)。
-    if test_account.is_test_account(req.phone):
-        logger.info("test_account sms_send short-circuit (不真发)")
-        return SmsSendResponse(sent=True, mock=True, cooldown_sec=0)
-
     try:
         cooldown = send_code(req.phone)
     except SmsError as e:
@@ -110,19 +99,6 @@ def sms_send(req: SmsSendRequest) -> SmsSendResponse:
     dependencies=[Depends(rate_limit(20, 60, "sms-login"))],  # 防撞库爆破(另有单码失败次数上限)
 )
 def sms_login(req: SmsLoginRequest, db: DbSession) -> TokenWithUser:
-    # 测试账号:免验证码登录 + 每日上限 + 每次都走新手引导(详见 app/core/test_account.py)。
-    # 放在最前面:命中即不校验验证码;先扣当日额度,超限直接拒,挡住有人猜到号后脚本刷。
-    if test_account.is_test_account(req.phone):
-        if not test_account.try_consume_quota():
-            raise HTTPException(status_code=429, detail="测试账号今日使用次数已达上限,请明天再试")
-        user = user_repo.upsert_user_for_login(db, phone=req.phone, register_channel="sms")
-        if user.status != "active":
-            raise HTTPException(status_code=403, detail="account disabled")
-        logger.info("sms_login test_account ok user_id=%d phone=%s", user.id, mask_phone(req.phone))
-        # onboarding 恒 false + force_onboarding=true:每次登录都重走引导(含应用内退出后重登),
-        # 不读/不写 onboarding_completion 表。
-        return _login_response(user, onboarding_completed=False, force_onboarding=True)
-
     if not verify_code(req.phone, req.code):
         raise HTTPException(status_code=400, detail="invalid sms code")
 

app/api/v1/compare_record.py

@@ -77,8 +77,15 @@ def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
         rec.llm_calls = calls
         rec.llm_call_count = len(calls)
         rec.retry_count = sum(1 for c in calls if c.get("error"))
+        # token 累加(usage 已被 pricebot llm_client 归一为 prompt/completion_tokens;
+        # error 的调用 usage 可能为 None,or {} 兜底)
+        rec.input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
+        rec.output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
         db.commit()
-        logger.info("backfill llm_calls trace=%s n=%d", trace_id, len(calls))
+        logger.info(
+            "backfill llm_calls trace=%s n=%d in_tok=%d out_tok=%d",
+            trace_id, len(calls), rec.input_tokens, rec.output_tokens,
+        )
     except Exception as e:  # noqa: BLE001 best-effort
         logger.warning("backfill llm_calls failed trace=%s: %s", trace_id, e)
     finally:

app/api/v1/device.py

@@ -0,0 +1,96 @@
+"""设备注册 / 心跳 endpoint(无障碍保护存活检测)。
+
+路由前缀 /api/v1/device,需 Bearer 鉴权(设备绑登录用户)。
+  POST /register   注册设备 / 更新 registration_id(App 前台、拿到 push token 时调)
+  POST /heartbeat  上报心跳(无障碍服务存活时周期调,刷新存活)
+
+后端 heartbeat_monitor_worker 据此发现心跳超时的设备并极光推送告警。
+见 spec: spec/accessibility-liveness-push.md。
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter
+
+from app.api.deps import CurrentUser, DbSession
+from app.repositories import device as device_repo
+from app.schemas.device import (
+    DeviceOut,
+    DeviceRegisterRequest,
+    HeartbeatRequest,
+    LivenessAckRequest,
+    LivenessOut,
+    OkResponse,
+)
+
+logger = logging.getLogger("shagua.device")
+
+router = APIRouter(prefix="/api/v1/device", tags=["device"])
+
+
+@router.post("/register", response_model=DeviceOut, summary="注册设备/更新推送token")
+def register_device(
+    req: DeviceRegisterRequest,
+    user: CurrentUser,
+    db: DbSession,
+) -> DeviceOut:
+    device = device_repo.register_or_update(
+        db,
+        user_id=user.id,
+        device_id=req.device_id,
+        registration_id=req.registration_id,
+        platform=req.platform,
+        app_version=req.app_version,
+    )
+    logger.info(
+        "device register user_id=%d device_id=%s reg=%s",
+        user.id,
+        req.device_id,
+        bool(req.registration_id),
+    )
+    return DeviceOut.model_validate(device)
+
+
+@router.post("/heartbeat", response_model=OkResponse, summary="上报心跳")
+def report_heartbeat(
+    req: HeartbeatRequest,
+    user: CurrentUser,
+    db: DbSession,
+) -> OkResponse:
+    device_repo.touch_heartbeat(
+        db,
+        user_id=user.id,
+        device_id=req.device_id,
+        accessibility_enabled=req.accessibility_enabled,
+        registration_id=req.registration_id,
+    )
+    return OkResponse()
+
+
+@router.get("/liveness", response_model=LivenessOut, summary="查询本机掉线告警(后置检测)")
+def get_liveness(
+    device_id: str,
+    user: CurrentUser,
+    db: DbSession,
+) -> LivenessOut:
+    """客户端进 App 时拉取: 本机是否被判定掉线过(kill_alert_pending)。
+
+    设备从未注册过 → 返回默认(无告警)。命中 → 客户端弹「开启自启动」引导, 之后调 /liveness/ack 清。
+    见 spec: spec/accessibility-liveness-pull-prompt.md。
+    """
+    device = device_repo.get_device(db, user_id=user.id, device_id=device_id)
+    if device is None:
+        return LivenessOut()
+    return LivenessOut.model_validate(device)
+
+
+@router.post("/liveness/ack", response_model=OkResponse, summary="确认已提醒(清掉线告警)")
+def ack_liveness(
+    req: LivenessAckRequest,
+    user: CurrentUser,
+    db: DbSession,
+) -> OkResponse:
+    """客户端已弹过「开启自启动」引导 → 清掉「待提醒」标记(下次真掉线 worker 再置)。"""
+    device_repo.ack_kill_alert(db, user_id=user.id, device_id=req.device_id)
+    return OkResponse()

app/core/config.py

@@ -66,6 +66,11 @@ class Settings(BaseSettings):
     JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify"
     JG_REQUEST_TIMEOUT_SEC: int = 15
 
+    # 无障碍保护存活监控后台任务(pull 后置检测;本期不接推送)
+    HEARTBEAT_MONITOR_ENABLED: bool = True       # 总开关
+    HEARTBEAT_TIMEOUT_MINUTES: int = 10          # 多久没心跳算掉线(≈3 个客户端心跳周期)
+    HEARTBEAT_SCAN_INTERVAL_SEC: int = 60        # 扫描周期
+
     # ===== 短信 =====
     SMS_MOCK: bool = True
     SMS_CODE_TTL_SEC: int = 300
@@ -79,19 +84,6 @@ class Settings(BaseSettings):
     SMS_DAILY_LIMIT_PER_PHONE: int = 10   # 单手机号每日发送上限(防刷 + 控费)
     SMS_MAX_VERIFY_ATTEMPTS: int = 5      # 单个验证码最多校验失败次数,超过即作废(防爆破)
 
-    # ===== 测试账号(release 包全流程联调用)=====
-    # 配一个固定测试手机号,专供无 SIM 卡 / 不走一键登录时打通全流程:该号登录【免短信验证码】
-    # (real 模式下也跳过校验)、每次登录【强制重走新手引导】,并设【每日使用次数上限】防被人
-    # 猜到号后脚本滥用。两个值都能随时改 .env。逻辑全在 app/core/test_account.py,与其他业务解耦。
-    # ⚠️ TEST_ACCOUNT_PHONE 留空 = 整个功能关闭(生产默认安全;要启用才显式填号)。
-    TEST_ACCOUNT_PHONE: str = ""           # 测试手机号(11 位,如 11111111111);空=关闭整功能
-    TEST_ACCOUNT_DAILY_LIMIT: int = 500    # 该测试号每日最多登录次数,当日超过即拒绝登录
-
-    @property
-    def test_account_phone(self) -> str:
-        """规整后的测试手机号(去空白);空串=功能关闭。"""
-        return self.TEST_ACCOUNT_PHONE.strip()
-
     # ===== 美团联盟 CPS =====
     # 未配置时所有 /api/v1/meituan/* 接口 200 返空(优雅降级),不影响登录/领券等其他业务。
     MT_CPS_APP_KEY: str = ""

app/core/heartbeat_monitor_worker.py

@@ -0,0 +1,148 @@
+"""无障碍保护存活监控后台任务。
+
+周期扫描「曾经保护过、当前 alive、心跳超时」的设备 = App 被彻底杀掉/无障碍已停(心跳断了),
+**命中即在服务器终端打印告警**(本期先不接推送,工程量大,用终端打印代替真实通知);并把状态机
+推进到 notified 防每轮重复打印(心跳恢复时由 repositories.device.touch_heartbeat 重置回 alive)。
+结构仿 withdraw_reconcile_worker(单实例锁 + asyncio 轮询 + 优雅退出)。
+
+见 spec: spec/accessibility-liveness-push.md。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import os
+import time
+from collections.abc import Iterator
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.repositories import device as device_repo
+
+logger = logging.getLogger("shagua.heartbeat_monitor")
+_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "heartbeat_monitor.lock"
+
+
+def _touch_lock() -> None:
+    with contextlib.suppress(FileNotFoundError):
+        os.utime(_LOCK_PATH, None)
+
+
+@contextlib.contextmanager
+def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
+    """同机多进程保护:同一时间只允许一个监控 worker 运行。"""
+    _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd: int | None = None
+    try:
+        try:
+            fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+        except FileExistsError:
+            try:
+                age = time.time() - _LOCK_PATH.stat().st_mtime
+            except FileNotFoundError:
+                age = stale_after_sec + 1
+            if age > stale_after_sec:
+                with contextlib.suppress(FileNotFoundError):
+                    _LOCK_PATH.unlink()
+                try:
+                    fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+                except FileExistsError:
+                    fd = None
+
+        if fd is None:
+            yield False
+            return
+
+        os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
+        yield True
+    finally:
+        if fd is not None:
+            os.close(fd)
+            with contextlib.suppress(FileNotFoundError):
+                _LOCK_PATH.unlink()
+
+
+def _silent_seconds(last: datetime | None) -> int | None:
+    """距上次心跳的秒数(兼容 sqlite 取回的 naive datetime)。"""
+    if last is None:
+        return None
+    ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow()
+    return int((ref - last).total_seconds())
+
+
+def _scan_once(timeout_minutes: int) -> dict:
+    """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备,在**服务器终端打印**告警代替真实推送。
+
+    本期不接推送(极光/厂商通道工程量大),只做服务端掉线检测:命中即 logger.warning 打印到终端,
+    并把状态机推进到 notified 防每轮重复打印(心跳恢复时 touch_heartbeat 会重置回 alive)。
+    """
+    notified = 0
+    with SessionLocal() as db:
+        overdue = device_repo.list_overdue(db, timeout_minutes=timeout_minutes)
+        for device in overdue:
+            silent = _silent_seconds(device.last_heartbeat_at)
+            logger.warning(
+                "[掉线检测] user_id=%s device_id=%s 已 %s 秒无心跳(阈值 %d 分钟)"
+                " → 判定 App 已被杀/无障碍已停。【已置 kill_alert_pending: 用户下次进 App 将弹「开启自启动」引导(后置检测);推送本期未接】",
+                device.user_id,
+                device.device_id,
+                silent if silent is not None else "?",
+                timeout_minutes,
+            )
+            device_repo.mark_notified(db, device_id_pk=device.id)
+            notified += 1
+        return {"checked": len(overdue), "notified": notified}
+
+
+async def _run_loop() -> None:
+    interval = max(10, int(settings.HEARTBEAT_SCAN_INTERVAL_SEC))
+    timeout_minutes = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES))
+    lock_stale_after = max(interval * 3, 600)
+    with _single_instance_lock(lock_stale_after) as lock_acquired:
+        if not lock_acquired:
+            logger.warning("heartbeat monitor skipped: another worker owns lock")
+            return
+        await _run_locked_loop(interval, timeout_minutes)
+
+
+async def _run_locked_loop(interval: int, timeout_minutes: int) -> None:
+    logger.info(
+        "heartbeat monitor started interval=%ss timeout=%sm",
+        interval,
+        timeout_minutes,
+    )
+    try:
+        while True:
+            try:
+                _touch_lock()
+                result = await asyncio.to_thread(_scan_once, timeout_minutes)
+                if result["notified"]:
+                    logger.info("heartbeat monitor result=%s", result)
+            except SQLAlchemyError:
+                logger.exception("heartbeat monitor db error")
+            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
+                logger.exception("heartbeat monitor unexpected error")
+            await asyncio.sleep(interval)
+    except asyncio.CancelledError:
+        logger.info("heartbeat monitor stopped")
+        raise
+
+
+def start_heartbeat_monitor() -> asyncio.Task | None:
+    if not settings.HEARTBEAT_MONITOR_ENABLED:
+        logger.info("heartbeat monitor disabled")
+        return None
+    return asyncio.create_task(_run_loop(), name="heartbeat-monitor")
+
+
+async def stop_heartbeat_monitor(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task

app/core/test_account.py

@@ -1,80 +0,0 @@
-"""测试账号:免验证码登录 + 每次重走新手引导 + 每日使用上限。
-
-为打通 **release 包全流程**(无 SIM 卡、不走极光一键登录)而设的一个固定测试手机号。
-对这个号(且仅这个号)放开三件事:
-  1. **免短信验证码**:`/sms/send` 不真发、`/sms/login` 跳过 `verify_code`,real 模式下同样生效
-     —— 测试时直接填任意验证码即可登入,不用等真短信。
-  2. **每次都走新手引导**:登录响应 `onboarding_completed` 恒为 false,不读/不依赖
-     onboarding_completion 表,所以反复登录都能体验引导。
-  3. **每日使用次数上限**:防被人猜到这个号后写脚本一直刷。当天登录数超过上限即拒绝(429),
-     次日自动归零。
-
-手机号与上限都在 .env 配(`TEST_ACCOUNT_PHONE` / `TEST_ACCOUNT_DAILY_LIMIT`),随时可改。
-`TEST_ACCOUNT_PHONE` 留空 = 整个功能关闭(生产默认态),`is_test_account()` 对任何号都返回
-False,登录/短信回到原逻辑,零影响。
-
-计数存**进程内存**(单 worker 够用,与 sms.py 同款约定):重启清零、多 worker 不共享。作为
-一个测试号的粗粒度防滥用闸够用;且因所有登录都落同一个 phone → 同一个 user,滥用面天然只
-限于这一个账号。要严格持久化/跨进程再迁 DB 或 Redis(同 sms.py 的技术债)。
-"""
-from __future__ import annotations
-
-import logging
-from datetime import datetime
-from threading import Lock
-
-from app.core.config import settings
-
-logger = logging.getLogger("shagua.test_account")
-
-# 进程内每日计数:(date_str, 当日已登录次数)。单 worker 有效,重启清零(见模块 docstring)。
-_lock = Lock()
-_usage: tuple[str, int] = ("", 0)
-
-
-def _today() -> str:
-    return datetime.now().strftime("%Y-%m-%d")
-
-
-def is_enabled() -> bool:
-    """功能总开关:配了 TEST_ACCOUNT_PHONE 才启用(空=关闭)。"""
-    return bool(settings.test_account_phone)
-
-
-def is_test_account(phone: str) -> bool:
-    """该手机号是否为配置的测试账号。功能关闭时对任何号都返回 False。"""
-    return is_enabled() and phone == settings.test_account_phone
-
-
-def try_consume_quota() -> bool:
-    """测试账号登录时调:当日计数 +1。
-
-    Returns:
-        True  —— 未超当日上限,本次放行(计数已 +1);
-        False —— 已达上限,本次拒绝(计数不变)。
-
-    线程安全;跨天自动归零。上限取自 settings.TEST_ACCOUNT_DAILY_LIMIT(可在 .env 改)。
-    """
-    global _usage
-    limit = settings.TEST_ACCOUNT_DAILY_LIMIT
-    with _lock:
-        today = _today()
-        day, cnt = _usage
-        if day != today:  # 跨天归零
-            cnt = 0
-        if cnt >= limit:
-            _usage = (today, cnt)  # 已满,保持不变
-            logger.warning(
-                "测试账号 %s 今日登录数已达上限 %d,拒绝", settings.test_account_phone, limit
-            )
-            return False
-        _usage = (today, cnt + 1)
-        logger.info("测试账号 %s 第 %d/%d 次登录", settings.test_account_phone, cnt + 1, limit)
-        return True
-
-
-def _reset_for_test() -> None:
-    """仅供单测:清空进程内计数,隔离用例间状态。"""
-    global _usage
-    with _lock:
-        _usage = ("", 0)

app/main.py

@@ -20,6 +20,7 @@ from app.api.v1.compare import router as compare_router
 from app.api.v1.compare_milestone import router as compare_milestone_router
 from app.api.v1.compare_record import router as compare_record_router
 from app.api.v1.coupon import router as coupon_router
+from app.api.v1.device import router as device_router
 from app.api.v1.cps_redirect import router as cps_redirect_router
 from app.api.internal.app_version import router as internal_app_version_router
 from app.api.internal.launch_confirm import router as internal_launch_confirm_router
@@ -38,6 +39,10 @@ from app.api.v1.user import router as user_router
 from app.api.v1.wallet import router as wallet_router
 from app.api.v1.wxpay import router as wxpay_router
 from app.core.config import settings
+from app.core.heartbeat_monitor_worker import (
+    start_heartbeat_monitor,
+    stop_heartbeat_monitor,
+)
 from app.core.daily_exchange_worker import (
     start_daily_exchange_worker,
     stop_daily_exchange_worker,
@@ -63,10 +68,12 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.DATABASE_URL.split("://", 1)[0],
     )
     reconcile_task = start_withdraw_reconcile_worker()
+    heartbeat_task = start_heartbeat_monitor()
     daily_exchange_task = start_daily_exchange_worker()
     try:
         yield
     finally:
+        await stop_heartbeat_monitor(heartbeat_task)
         await stop_withdraw_reconcile_worker(reconcile_task)
         await stop_daily_exchange_worker(daily_exchange_task)
         logger.info("shutting down")
@@ -100,6 +107,7 @@ app.include_router(user_router)
 app.include_router(feedback_router)
 app.include_router(invite_router)
 app.include_router(coupon_router)
+app.include_router(device_router)
 app.include_router(compare_router)
 app.include_router(compare_record_router)
 app.include_router(compare_milestone_router)

app/models/__init__.py

@@ -12,6 +12,7 @@ from app.models.cps_link import CpsClick, CpsLink  # noqa: F401
 from app.models.cps_order import CpsOrder  # noqa: F401
 from app.models.cps_wx_user import CpsWxUser  # noqa: F401
 from app.models.comparison_milestone import ComparisonMilestoneClaim  # noqa: F401
+from app.models.device import DeviceLiveness  # noqa: F401
 from app.models.coupon_state import (  # noqa: F401
     CouponClaimRecord,
     CouponDailyCompletion,

app/models/ad_feed_reward.py

@@ -33,6 +33,9 @@ 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: 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)
     our_code_id: Mapped[str | None] = mapped_column(String(64), nullable=True)

app/models/comparison.py

@@ -124,6 +124,9 @@ class ComparisonRecord(Base):
     step_count: Mapped[int | None] = mapped_column(Integer, nullable=True)     # 客户端 agent loop 总步数
     llm_call_count: Mapped[int | None] = mapped_column(Integer, nullable=True) # 本次 LLM 调用次数(server 从 llm_calls 算)
     retry_count: Mapped[int | None] = mapped_column(Integer, nullable=True)    # LLM 失败重试次数(server 从 llm_calls error 算)
+    # 本次 LLM 累计 token(server 从 llm_calls[].usage 累加;旧记录/未采集为 None)
+    input_tokens: Mapped[int | None] = mapped_column(Integer, nullable=True)   # Σ usage.prompt_tokens
+    output_tokens: Mapped[int | None] = mapped_column(Integer, nullable=True)  # Σ usage.completion_tokens
     # 每次 LLM 调用明细 [{scene,model,input_messages,output,usage,latency_ms,error}];
     # server 收上报后按 trace_id 同机拉 pricebot 落库(见 compare_record 端点)。旧记录/未采集为 None。
     llm_calls: Mapped[list | None] = mapped_column(_JSON, nullable=True)

app/models/device.py

@@ -0,0 +1,91 @@
+"""设备表(无障碍保护存活检测 + 极光推送)。
+
+每条 = 一个用户的一台设备(per-install,device_id 由客户端 DeviceId.get() 生成)。
+客户端的无障碍服务存活时周期上报心跳刷新 last_heartbeat_at;App 前台/登录时上报
+registration_id(极光推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、
+现在心跳超时」的设备,通过极光推送提醒用户重开无障碍。
+
+liveness_state 状态机(防刷屏,一次掉线只推一条):
+  unknown → alive(收到 service 心跳)→ silent/notified(扫描发现超时并已推送)
+  心跳恢复时 handler 重置回 alive。
+见 spec: spec/accessibility-liveness-push.md。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class DeviceLiveness(Base):
+    # 表名不叫 device:device 易被当成「设备信息(品牌/型号/系统)」表;本表实为**无障碍存活监控状态**
+    # (心跳 last_heartbeat_at + liveness_state + kill_alert_pending + 推送目标 registration_id),故名 device_liveness。
+    __tablename__ = "device_liveness"
+    __table_args__ = (
+        UniqueConstraint("user_id", "device_id", name="uq_device_liveness_user_device"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("user.id"), index=True, nullable=False
+    )
+    # 客户端 DeviceId.get() 生成的 per-install id(如 device_Pixel_ab12cd34)
+    device_id: Mapped[str] = mapped_column(String(128), index=True, nullable=False)
+    # 极光推送 registration id;拿到才填(JCollectionAuth 同意后才下发)
+    registration_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android")
+    app_version: Mapped[str | None] = mapped_column(String(32), nullable=True)
+
+    # 收到过 service 心跳即 true(=该设备开过无障碍,功能对它有意义)
+    ever_protected: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
+    # 最近一次 service 心跳时间(存活证明);超时即视为保护掉线
+    last_heartbeat_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), index=True, nullable=True
+    )
+    # 最近一次上报的无障碍开关状态(观测用)
+    last_report_protection_on: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
+    # unknown / alive / silent / notified
+    liveness_state: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="unknown"
+    )
+    # 最近一次推送告警时间
+    notified_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    # 掉线告警「待客户端提醒」标记(后置检测 pull 版, 见 spec accessibility-liveness-pull-prompt.md)。
+    # 与 liveness_state 解耦: worker 检出掉线即置 True; touch_heartbeat(心跳恢复)不动它
+    # → 规避「服务随 App 重启先发心跳、state 被重置回 alive → 客户端进 App 漏看」竞态; 只由客户端 ack 清。
+    kill_alert_pending: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<DeviceLiveness id={self.id} user_id={self.user_id} "
+            f"device_id={self.device_id} state={self.liveness_state}>"
+        )

app/repositories/ad_feed_reward.py

@@ -42,21 +42,20 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
     ).scalar_one()
 
 
-def _unit_reward_total(db: Session, user_id: int, ecpm: str, unit_count: int) -> int:
-    """按每个 10 秒单位逐份计算奖励,LT 使用**账号累计**奖励份序号(不按天重置)。"""
-    if unit_count <= 0:
-        return 0
-    existing_units = db.execute(
-        select(func.coalesce(func.sum(AdFeedRewardRecord.unit_count), 0))
-        .where(
-            AdFeedRewardRecord.user_id == user_id,
-            AdFeedRewardRecord.status == "granted",
-        )
-    ).scalar_one()
-    total = 0
-    for offset in range(1, unit_count + 1):
-        total += rewards.calculate_ad_reward_coin(ecpm, int(existing_units) + offset)
-    return total
+def granted_unit_total(db: Session, user_id: int) -> int:
+    """账号累计已发奖**条**数(COUNT status=granted),不按天重置 = 因子2(LT)基线。
+
+    口径:**一条广告 = 一份额度(一个单次公式值)**,故"已发条数"即"已发份数"。供前端拉取后精确复刻
+    金币公式做实时金币进度条(前端因子2 用 granted_unit_total + 本场已结算条数 + 1 取档,与本表发奖一致)。
+    """
+    return int(
+        db.execute(
+            select(func.count()).select_from(AdFeedRewardRecord).where(
+                AdFeedRewardRecord.user_id == user_id,
+                AdFeedRewardRecord.status == "granted",
+            )
+        ).scalar_one()
+    )
 
 
 def grant_feed_reward(
@@ -70,20 +69,21 @@ def grant_feed_reward(
     adn: str | None = None,
     slot_id: str | None = None,
     feed_scene: str | None = None,
+    trace_id: str | None = None,
     app_env: str | None = None,
     our_code_id: str | None = None,
     aborted: bool = False,
 ) -> AdFeedRewardRecord:
-    """比价/领券一整场信息流(轮播多条)结束后结算奖励。client_event_id 幂等,同号重试不重复发。
+    """**每条**信息流广告(客户端每条各上报一次)结算奖励。client_event_id 幂等,同号重试不重复发。
 
-    发奖规则:**比价全程不关广告才发**,金额按整场**总观看时长**折份(每 10 秒 1 份)。
-    - aborted=True(用户中途 ✕ 关闭):整场不发,记 status='closed_early' 留痕(原因可查)。
-    - 总时长不足 10 秒(unit_count==0):记 status='too_short' 不发。
+    发奖规则:**一条广告 = 一个单次公式值**(rewards.calculate_ad_reward_coin),因子2(LT)按账号累计
+    **条**数递进;看满一份时长(unit_count>=1, 即 ≥10 秒)才发,**不逐份累加**。
+    - aborted=True(用户中途 ✕ 关闭这条):本条不发,记 status='closed_early'。
+    - 时长不足 10 秒(unit_count==0):记 status='too_short' 不发。
     - 命中当日条数上限:记 status='capped' 不发。
-    duration_seconds 是整场累计秒数。一期 eCPM/时长均由客户端上报,故服务端两道硬闸防刷:时长钳到
-    FEED_MAX_DURATION_SECONDS 限单场份数,eCPM 在 rewards.calculate_ad_reward_coin 内钳到
-    AD_ECPM_MAX_FEN 限单份金额;叠加每日 get_ad_daily_limit 条数上限,把单用户日产出锁进有限区间。
-    feed_scene:点位场景(comparison/coupon/welfare),仅做归类落库,不参与发奖计算。
+    duration_seconds 是**这一条**的观看秒数。服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS、
+    eCPM 在 calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN;叠加每日 get_ad_daily_limit 条数上限。
+    feed_scene:点位场景(comparison/coupon/welfare),仅归类落库,不参与计算。
     """
     existing = _find_by_event(db, client_event_id)
     if existing is not None:
@@ -107,6 +107,7 @@ def grant_feed_reward(
             adn=adn,
             slot_id=slot_id,
             feed_scene=feed_scene,
+            trace_id=trace_id,
             app_env=app_env,
             our_code_id=our_code_id,
             coin=0,
@@ -126,6 +127,7 @@ def grant_feed_reward(
             adn=adn,
             slot_id=slot_id,
             feed_scene=feed_scene,
+            trace_id=trace_id,
             app_env=app_env,
             our_code_id=our_code_id,
             coin=0,
@@ -146,6 +148,7 @@ def grant_feed_reward(
             adn=adn,
             slot_id=slot_id,
             feed_scene=feed_scene,
+            trace_id=trace_id,
             app_env=app_env,
             our_code_id=our_code_id,
             coin=0,
@@ -153,12 +156,14 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    coin = _unit_reward_total(db, user_id, ecpm, unit_count)
+    # 一条广告 = 一个「单次公式值」(因子2 按账号累计**条**数, 即第 existing_ads+1 条);看满一份(unit_count>=1)即发,不逐份累加。
+    existing_ads = granted_unit_total(db, user_id)
+    coin = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
     if coin > 0:
         crud_wallet.grant_coins(
             db, user_id, coin,
             biz_type="feed_ad_reward", ref_id=client_event_id,
-            remark=f"信息流广告奖励 {unit_count}份",
+            remark="信息流广告奖励",
         )
     rec = AdFeedRewardRecord(
         client_event_id=client_event_id,
@@ -171,6 +176,7 @@ def grant_feed_reward(
         adn=adn,
         slot_id=slot_id,
         feed_scene=feed_scene,
+        trace_id=trace_id,
         app_env=app_env,
         our_code_id=our_code_id,
         coin=coin,

app/repositories/comparison.py

@@ -11,6 +11,7 @@ from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.core.rewards import CN_TZ
+from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.comparison import ComparisonRecord
 from app.models.savings import SavingsRecord
 from app.schemas.compare_record import ComparisonRecordIn
@@ -155,6 +156,29 @@ def _ordered_shop_names(db: Session, user_id: int) -> set[str]:
     return {s for s in rows if s}
 
 
+def _ad_coins_by_trace(db: Session, user_id: int, trace_ids: list[str]) -> dict[str, int]:
+    """各 trace_id「看广告赚的金币」:按 trace_id 聚合该用户已发放(granted)的信息流广告金币。
+
+    广告金币来自比价等候期信息流(biz_type=feed_ad_reward);客户端结算时把本场 trace_id 带上,
+    落到 ad_feed_reward_record.trace_id。此处按 trace_id 求和 = 这次比价看广告实发的金币。
+    trace_id 仅比价场景客户端带(领券/福利/旧客户端为 NULL),故按 trace_id 过滤天然只算比价广告。
+    空集合直接返回(避免 IN () 非法)。
+    """
+    if not trace_ids:
+        return {}
+    rows = db.execute(
+        select(
+            AdFeedRewardRecord.trace_id,
+            func.coalesce(func.sum(AdFeedRewardRecord.coin), 0),
+        ).where(
+            AdFeedRewardRecord.user_id == user_id,
+            AdFeedRewardRecord.status == "granted",
+            AdFeedRewardRecord.trace_id.in_(trace_ids),
+        ).group_by(AdFeedRewardRecord.trace_id)
+    ).all()
+    return {tid: int(coin) for tid, coin in rows if tid}
+
+
 def list_records(
     db: Session,
     user_id: int,
@@ -162,7 +186,7 @@ def list_records(
     limit: int = 20,
     cursor: int | None = None,
 ) -> tuple[list[ComparisonRecord], int | None]:
-    """比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记(瞬态,不写库)。"""
+    """比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记 + 「看广告赚的金币」(瞬态,不写库)。"""
     stmt = select(ComparisonRecord).where(ComparisonRecord.user_id == user_id)
     if cursor is not None:
         stmt = stmt.where(ComparisonRecord.id < cursor)
@@ -172,10 +196,13 @@ def list_records(
     next_cursor = items[-1].id if len(items) == limit else None
 
     # 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。
-    # ordered 非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。
+    # ordered / ad_coins_earned 均非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。
     ordered_shops = _ordered_shop_names(db, user_id)
+    # 「本次比价看广告赚的金币」:按本页 trace_id 一次性聚合(同 ordered 范式)。
+    ad_coins = _ad_coins_by_trace(db, user_id, [it.trace_id for it in items])
     for it in items:
         it.ordered = bool(it.store_name and it.store_name in ordered_shops)
+        it.ad_coins_earned = ad_coins.get(it.trace_id, 0)
 
     return items, next_cursor
 
@@ -210,10 +237,13 @@ def get_stats(db: Session, user_id: int) -> tuple[int, int]:
 
 
 def get_record(db: Session, user_id: int, record_id: int) -> ComparisonRecord | None:
-    """取单条(限本人,避免越权读他人记录)。"""
-    return db.execute(
+    """取单条(限本人,避免越权读他人记录)。附「看广告赚的金币」瞬态属性(同 list_records)。"""
+    rec = db.execute(
         select(ComparisonRecord).where(
             ComparisonRecord.id == record_id,
             ComparisonRecord.user_id == user_id,
         )
     ).scalar_one_or_none()
+    if rec is not None:
+        rec.ad_coins_earned = _ad_coins_by_trace(db, user_id, [rec.trace_id]).get(rec.trace_id, 0)
+    return rec

app/repositories/device.py

@@ -0,0 +1,124 @@
+"""device 表读写(设备注册 / 心跳 / 超时扫描)。"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.device import DeviceLiveness
+
+
+def _get(db: Session, *, user_id: int, device_id: str) -> DeviceLiveness | None:
+    stmt = select(DeviceLiveness).where(
+        DeviceLiveness.user_id == user_id, DeviceLiveness.device_id == device_id
+    )
+    return db.execute(stmt).scalar_one_or_none()
+
+
+def register_or_update(
+    db: Session,
+    *,
+    user_id: int,
+    device_id: str,
+    registration_id: str | None,
+    platform: str = "android",
+    app_version: str | None = None,
+) -> DeviceLiveness:
+    """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。"""
+    device = _get(db, user_id=user_id, device_id=device_id)
+    if device is None:
+        device = DeviceLiveness(
+            user_id=user_id,
+            device_id=device_id,
+            registration_id=registration_id,
+            platform=platform or "android",
+            app_version=app_version,
+        )
+        db.add(device)
+    else:
+        if registration_id:
+            device.registration_id = registration_id
+        if platform:
+            device.platform = platform
+        if app_version:
+            device.app_version = app_version
+    db.commit()
+    db.refresh(device)
+    return device
+
+
+def touch_heartbeat(
+    db: Session,
+    *,
+    user_id: int,
+    device_id: str,
+    accessibility_enabled: bool,
+    registration_id: str | None,
+) -> DeviceLiveness:
+    """处理一次心跳(心跳也能自注册)。
+
+    service 心跳或 accessibility_enabled=true 时,刷新存活并把状态机重置回 alive、
+    清掉 notified_at(掉线恢复 → 下次再断才会再推一条)。
+    """
+    now = datetime.now(timezone.utc)
+    device = _get(db, user_id=user_id, device_id=device_id)
+    if device is None:
+        device = DeviceLiveness(user_id=user_id, device_id=device_id)
+        db.add(device)
+
+    if registration_id:
+        device.registration_id = registration_id
+    device.last_report_protection_on = accessibility_enabled
+
+    if accessibility_enabled:
+        device.last_heartbeat_at = now
+        device.ever_protected = True
+        device.liveness_state = "alive"
+        device.notified_at = None
+
+    db.commit()
+    db.refresh(device)
+    return device
+
+
+def list_overdue(db: Session, *, timeout_minutes: int) -> list[DeviceLiveness]:
+    """掉线设备:曾经保护过、当前 alive、心跳超时。
+
+    本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。
+    """
+    cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
+    stmt = select(DeviceLiveness).where(
+        DeviceLiveness.ever_protected.is_(True),
+        DeviceLiveness.liveness_state == "alive",
+        DeviceLiveness.last_heartbeat_at.is_not(None),
+        DeviceLiveness.last_heartbeat_at < cutoff,
+    )
+    return list(db.execute(stmt).scalars().all())
+
+
+def mark_notified(db: Session, *, device_id_pk: int) -> None:
+    """标记掉线已检出(状态机进入 notified,避免每轮重复处理)+ 置「待客户端提醒」标记。
+
+    kill_alert_pending 与 state 解耦(见 spec accessibility-liveness-pull-prompt.md):
+    心跳恢复(touch_heartbeat)只重置 state、不动此标记,故客户端下次进 App 必能拉到这次掉线。
+    """
+    device = db.get(DeviceLiveness, device_id_pk)
+    if device is not None:
+        device.liveness_state = "notified"
+        device.notified_at = datetime.now(timezone.utc)
+        device.kill_alert_pending = True
+        db.commit()
+
+
+def get_device(db: Session, *, user_id: int, device_id: str) -> DeviceLiveness | None:
+    """按 (user_id, device_id) 取设备(liveness 查询用)。"""
+    return _get(db, user_id=user_id, device_id=device_id)
+
+
+def ack_kill_alert(db: Session, *, user_id: int, device_id: str) -> None:
+    """客户端已弹过「开启自启动」引导 → 清「待提醒」标记(幂等; 下次真掉线 worker 再置)。"""
+    device = _get(db, user_id=user_id, device_id=device_id)
+    if device is not None and device.kill_alert_pending:
+        device.kill_alert_pending = False
+        db.commit()

app/schemas/ad.py

@@ -141,6 +141,12 @@ class FeedRewardIn(BaseModel):
         description="点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页);"
         "比价与领券共用同一信息流代码位,需客户端在各调用点显式标注,缺省=未分类",
     )
+    trace_id: str | None = Field(
+        None,
+        max_length=64,
+        description="本次比价 trace_id(比价场景必带):把这场广告金币归属到对应比价记录,"
+        "比价记录页据此显示「比价赚 N 金币」。领券/福利场景为空",
+    )
     app_env: str | None = Field(
         None, max_length=16, description="我们的穿山甲应用环境:prod(傻瓜比价正式) / test(测试应用)"
     )
@@ -160,6 +166,14 @@ class FeedRewardOut(BaseModel):
     daily_limit: int = Field(..., description="每日信息流展示次数上限")
 
 
+class FeedRewardUnitsOut(BaseModel):
+    granted_units: int = Field(
+        ...,
+        description="账号累计已发奖的信息流**条**数(一条广告=一份额度),不按天重置;"
+        "前端据此算因子2(LT)做实时金币进度条,与后端发奖公式对齐",
+    )
+
+
 class RewardNoShowIn(BaseModel):
     """激励视频展示了但用户提前关闭/跳过、未触发发奖——上报留痕(不发金币)。
 

app/schemas/auth.py

@@ -48,12 +48,6 @@ class TokenWithUser(TokenPair):
         description="该 设备+账号 是否已走完新手引导(据登录请求里的 device_id 计算)。"
         "true → 客户端登录后直接进首页,跳过引导。",
     )
-    force_onboarding: bool = Field(
-        False,
-        description="是否强制本次登录走新手引导(仅测试号置 true)。客户端据此让【应用内重新登录】"
-        "(退出后再登)也重走引导,普通号为 false、按原逻辑回原页/首页。与 onboarding_completed "
-        "分工:后者管 App 启动路由,本字段管应用内登录后的去向。",
-    )
 
 
 # ===== 极光一键登录 =====

app/schemas/compare_record.py

@@ -51,6 +51,13 @@ class ComparisonResultIn(BaseModel):
     # 跟 coupon_saved 并存, 是更丰富的明细; 空=没抠到 → 前端回退单券路径。
     # 必须显式声明: 落库走 model_dump(), pydantic 默认丢未知字段, 不声明这行会被悄悄吞掉。
     applied_coupons: list[AppliedCouponIn] = Field(default_factory=list)
+    # 逐平台状态(客户端上报前把 done.params 的 comparison_results + platform_results.status 合并写入):
+    #   success / below_minimum / store_closed / store_not_found / items_not_found / unsupported / failed。
+    # 「比价记录」页据此对无价平台显示对应文案(未满起送价/门店打烊/店家未入驻/比价失败)。
+    # 必须显式声明: 落库走 model_dump(), 不声明会被 pydantic 静默丢弃 → 记录页拿不到 status。
+    status: str | None = None
+    # 门店打烊原因(price 为 None 时带): 与 status="store_closed" 等价的更早信号, 一并落库供前端兜底判打烊。
+    store_closed: str | None = None
 
 
 class ComparisonRecordIn(BaseModel):
@@ -73,6 +80,8 @@ class ComparisonRecordIn(BaseModel):
     # unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
     # admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是
     # 按平台 id 索引的 dict 而非 list,只作 debug 展示)。
+    # ⚠️ 早先误声明成 list → 每条成功记录 422 被拒、记录页只剩失败/终止
+    # (2026-06-17 引入 platform_results 上报后复现);改 dict 后修复。
     platform_results: dict = Field(default_factory=dict)
     skipped_dish_count: int | None = None
     skipped_dish_names: list[str] = Field(default_factory=list)
@@ -138,6 +147,9 @@ class ComparisonRecordOut(BaseModel):
     # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
     # 由 list_records 动态算出挂在 ORM 实例上(非 DB 列),from_attributes 读出;缺省 False。
     ordered: bool = False
+    # 「本次比价看广告赚的金币」(feed_ad_reward,按 trace_id 聚合 granted 金币;非 DB 列,
+    # list_records/get_record 动态挂在 ORM 实例上,from_attributes 读出)。0=没赚到 → 记录页不显示金币标。
+    ad_coins_earned: int = 0
     created_at: datetime
 
 

app/schemas/device.py

@@ -0,0 +1,48 @@
+"""设备注册 / 心跳相关 schema(无障碍保护存活检测)。"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class DeviceRegisterRequest(BaseModel):
+    device_id: str
+    registration_id: str | None = None
+    platform: str = "android"
+    app_version: str | None = None
+
+
+class HeartbeatRequest(BaseModel):
+    device_id: str
+    source: str = "service"          # service | app
+    accessibility_enabled: bool = True
+    registration_id: str | None = None
+
+
+class DeviceOut(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    device_id: str
+    registration_id: str | None
+    ever_protected: bool
+    liveness_state: str
+    last_heartbeat_at: datetime | None
+    updated_at: datetime | None
+
+
+class OkResponse(BaseModel):
+    ok: bool = True
+
+
+class LivenessOut(BaseModel):
+    """本机掉线告警状态(后置检测 pull 版)。客户端只需这一个布尔判断要不要弹「开启自启动」引导,
+    故只返回 kill_alert_pending(不暴露设备详情 / 内部 liveness_state 等)。从未注册过 → 默认 False(无告警)。"""
+    model_config = ConfigDict(from_attributes=True)
+
+    kill_alert_pending: bool = False
+
+
+class LivenessAckRequest(BaseModel):
+    device_id: str

tests/test_test_account.py

@@ -1,165 +0,0 @@
-"""测试账号(app/core/test_account.py)行为测试。
-
-覆盖四件事:
-  - 免验证码登录(mock 与 real 模式都跳过校验);
-  - 每次登录都重走新手引导(onboarding_completed 恒 false,即便完成表已有记录);
-  - 每日登录上限(超过即 429),且只算 login、不算 send;
-  - 功能关闭(TEST_ACCOUNT_PHONE 留空)时对任何号都不生效,回到原逻辑。
-"""
-from __future__ import annotations
-
-import pytest
-
-from app.core import test_account
-from app.integrations import sms
-
-# 测试自注入的固定号:各用例用 monkeypatch 把它塞进 settings.TEST_ACCOUNT_PHONE,
-# 再断言行为。它与 .env 配的值【相互独立】——测试不读 .env(保证确定性 / CI 无 .env 也能跑),
-# 写成和 .env 同值只为直观;换任意 11 位号测试照样全过,二者无需保持一致。
-TEST_PHONE = "11111111111"
-
-
-@pytest.fixture()
-def enabled(monkeypatch):
-    """启用测试账号功能并清空当日计数;monkeypatch 与 _reset 保证用例间隔离。"""
-    monkeypatch.setattr(test_account.settings, "TEST_ACCOUNT_PHONE", TEST_PHONE)
-    test_account._reset_for_test()
-    yield
-    test_account._reset_for_test()
-
-
-# ============================ 开关 / 识别 ============================
-
-def test_disabled_when_phone_unset(monkeypatch) -> None:
-    """TEST_ACCOUNT_PHONE 留空 = 整功能关闭,对测试号也不特殊处理。"""
-    monkeypatch.setattr(test_account.settings, "TEST_ACCOUNT_PHONE", "")
-    assert test_account.is_enabled() is False
-    assert test_account.is_test_account(TEST_PHONE) is False
-
-
-def test_only_configured_phone_is_test_account(enabled) -> None:
-    assert test_account.is_test_account(TEST_PHONE) is True
-    assert test_account.is_test_account("13800138000") is False
-
-
-def test_phone_is_whitespace_trimmed(monkeypatch) -> None:
-    """.env 里手机号前后带空格也能正确识别。"""
-    monkeypatch.setattr(test_account.settings, "TEST_ACCOUNT_PHONE", f"  {TEST_PHONE}  ")
-    assert test_account.is_test_account(TEST_PHONE) is True
-
-
-# ============================ 免验证码 + 新手引导 ============================
-
-def test_login_skips_code_and_returns_tokens(client, enabled) -> None:
-    """测试号任意验证码即可登入(不用等真短信),拿到 JWT,onboarding 为 false。"""
-    r = client.post(
-        "/api/v1/auth/sms/login",
-        json={"phone": TEST_PHONE, "code": "000000", "device_id": "dev-a"},
-    )
-    assert r.status_code == 200, r.text
-    data = r.json()
-    assert data["user"]["phone"] == TEST_PHONE
-    assert data["user"]["register_channel"] == "sms"
-    assert data["onboarding_completed"] is False
-    assert data["force_onboarding"] is True   # 测试号:应用内重登也强制重走引导
-    assert "access_token" in data and "refresh_token" in data
-
-
-def test_onboarding_always_false_even_if_completed(client, enabled) -> None:
-    """即便 onboarding_completion 表已有 (该用户, 该设备) 记录,测试号再登仍返回 false。"""
-    from app.db.session import SessionLocal
-    from app.repositories import onboarding as onboarding_repo
-
-    device = "dev-onb"
-    r = client.post(
-        "/api/v1/auth/sms/login",
-        json={"phone": TEST_PHONE, "code": "123456", "device_id": device},
-    )
-    uid = r.json()["user"]["id"]
-
-    db = SessionLocal()
-    try:
-        onboarding_repo.mark_completed(db, user_id=uid, device_id=device)
-        assert onboarding_repo.is_completed(db, user_id=uid, device_id=device) is True
-    finally:
-        db.close()
-
-    r2 = client.post(
-        "/api/v1/auth/sms/login",
-        json={"phone": TEST_PHONE, "code": "123456", "device_id": device},
-    )
-    assert r2.json()["onboarding_completed"] is False  # 仍重走引导
-
-
-# ============================ real 模式跳过 ============================
-
-def test_login_skips_verify_in_real_mode(client, enabled, monkeypatch) -> None:
-    """real 模式(SMS_MOCK=false)普通号要真校验;测试号必须跳过 → 任意码仍 200。"""
-    monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
-    r = client.post(
-        "/api/v1/auth/sms/login",
-        json={"phone": TEST_PHONE, "code": "9999"},
-    )
-    assert r.status_code == 200, r.text
-
-
-def test_send_short_circuits_in_real_mode(client, enabled, monkeypatch) -> None:
-    """real 模式下测试号 /sms/send 不真打极光(否则缺号/缺 key 会 503),直接 200 放行进填码界面。"""
-    monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
-    r = client.post("/api/v1/auth/sms/send", json={"phone": TEST_PHONE})
-    assert r.status_code == 200, r.text
-    body = r.json()
-    assert body["sent"] is True
-    assert body["cooldown_sec"] == 0
-
-
-# ============================ 每日上限 ============================
-
-def test_daily_limit_blocks_after_threshold(client, enabled, monkeypatch) -> None:
-    """当日登录数到上限后再登 → 429。"""
-    monkeypatch.setattr(test_account.settings, "TEST_ACCOUNT_DAILY_LIMIT", 2)
-    test_account._reset_for_test()
-
-    for _ in range(2):
-        r = client.post("/api/v1/auth/sms/login", json={"phone": TEST_PHONE, "code": "123456"})
-        assert r.status_code == 200, r.text
-
-    r = client.post("/api/v1/auth/sms/login", json={"phone": TEST_PHONE, "code": "123456"})
-    assert r.status_code == 429
-    assert "上限" in r.json()["detail"]
-
-
-def test_send_does_not_consume_quota(client, enabled, monkeypatch) -> None:
-    """额度只算 login、不算 send:狂发 send 后那唯一一次 login 仍能成功。"""
-    monkeypatch.setattr(test_account.settings, "TEST_ACCOUNT_DAILY_LIMIT", 1)
-    test_account._reset_for_test()
-
-    for _ in range(3):
-        assert client.post("/api/v1/auth/sms/send", json={"phone": TEST_PHONE}).status_code == 200
-
-    assert client.post("/api/v1/auth/sms/login", json={"phone": TEST_PHONE, "code": "1234"}).status_code == 200
-    # 第二次 login 才超限
-    assert client.post("/api/v1/auth/sms/login", json={"phone": TEST_PHONE, "code": "1234"}).status_code == 429
-
-
-# ============================ 计数单元逻辑 ============================
-
-def test_quota_resets_across_day(enabled, monkeypatch) -> None:
-    """跨天自动归零。"""
-    monkeypatch.setattr(test_account.settings, "TEST_ACCOUNT_DAILY_LIMIT", 1)
-    monkeypatch.setattr(test_account, "_today", lambda: "2026-06-23")
-    assert test_account.try_consume_quota() is True
-    assert test_account.try_consume_quota() is False  # 当天超限
-
-    monkeypatch.setattr(test_account, "_today", lambda: "2026-06-24")
-    assert test_account.try_consume_quota() is True   # 次日归零放行
-
-
-def test_non_test_phone_unaffected_when_enabled(client, enabled) -> None:
-    """功能开启时,普通号走原 mock 流程(任意 6 位通过),不受测试号逻辑影响。"""
-    phone = "13812341234"
-    client.post("/api/v1/auth/sms/send", json={"phone": phone})
-    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456", "device_id": "d1"})
-    assert r.status_code == 200, r.text
-    assert r.json()["user"]["phone"] == phone
-    assert r.json()["force_onboarding"] is False   # 普通号不强制重走,按原逻辑