feat(welfare): 完成不活跃清零-邀请金不清 + 预警编排 + worker + admin口径统一

- 清零改两桶(金币 + 折算现金);邀请奖励金物理隔离、不清(产品红线),仅快照入审计
- 预警编排 select_warn_candidates / run_warn_once / run_once(分档 + streak 去重);
  notifier.warn 去掉 invite 参数(邀请金不清、不预警)
- 预警逐用户 try/except 隔离 + run_once 保证预警故障绝不阻塞清零(接真推送前必备)
- 每日 worker inactivity_reset_worker(文件锁 + 北京日守卫 + RUN_HOUR 门槛 + 总闸默认关)
  + main.py lifespan 接线
- admin 最近活跃口径改用共享 activity(移除 last_login_at、纳入 home_view 占位)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/admin/repositories/queries.py

@@ -11,7 +11,6 @@ from zoneinfo import ZoneInfo
 from sqlalchemy import Select, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session
 
-from app.admin.repositories.stats import COMPARE_START_EVENT, COUPON_START_EVENT
 from app.core import rewards
 from app.core.config import settings
 from app.models.ad_feed_reward import AdFeedRewardRecord
@@ -32,10 +31,7 @@ from app.models.wallet import (
     InviteCashTransaction,
     WithdrawOrder,
 )
-from app.repositories import ad_ecpm
-
-# 「最近活跃」计入的行为事件(与大盘 DAU/留存活跃口径一致:开始比价 + 开始领券)
-_ACTIVE_EVENTS = (COMPARE_START_EVENT, COUPON_START_EVENT)
+from app.repositories import activity, ad_ecpm
 
 # 折算成可提现现金时,非广告金币来源的排除集(广告单独统计、人工调整不算"赚取")
 _NON_TASK_BIZ_TYPES = ("reward_video", "feed_ad_reward", "admin_grant", "admin_deduct")
@@ -88,49 +84,6 @@ def offset_paginate(
     return items, next_cursor, total
 
 
-def _last_active_parts():
-    """「最近活跃」的两个按 user_id 预聚合派生表(最近开始比价/领券事件、最近领券发起)。
-
-    活跃口径与大盘 DAU/留存一致(2026-07-05 产品定:进入 App≈登录 last_login_at +
-    发起比价 real_compare_start + 发起领券 real_coupon_start/claim_started)。
-    用 LEFT JOIN 预聚合而非相关标量子查询:后者在 PG 上对 users 每行各跑一个 SubPlan
-    (排序键、range 筛选、offset_paginate 的 count 三处叠加),埋点表大了会拖垮列表接口;
-    预聚合借 analytics_event.event 索引只扫两类 start 事件,每次查询聚合一次。
-    """
-    ev_agg = (
-        select(
-            AnalyticsEvent.user_id.label("user_id"),
-            func.max(AnalyticsEvent.created_at).label("last_at"),
-        )
-        .where(
-            AnalyticsEvent.user_id.is_not(None),
-            AnalyticsEvent.event.in_(_ACTIVE_EVENTS),
-        )
-        .group_by(AnalyticsEvent.user_id)
-        .subquery()
-    )
-    eng_agg = (
-        select(
-            CouponPromptEngagement.user_id.label("user_id"),
-            func.max(CouponPromptEngagement.created_at).label("last_at"),
-        )
-        .where(
-            CouponPromptEngagement.user_id.is_not(None),
-            CouponPromptEngagement.engage_type == "claim_started",
-        )
-        .group_by(CouponPromptEngagement.user_id)
-        .subquery()
-    )
-    return ev_agg, eng_agg
-
-
-def _norm_utc(dt: datetime | None) -> datetime | None:
-    """naive 视为 UTC 补 tzinfo(SQLite 读回 naive、PG 读回 aware,混着 max() 会 TypeError)。"""
-    if dt is None:
-        return None
-    return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)
-
-
 def _attach_last_active(db: Session, users: list[User]) -> None:
     """给本页用户瞬态挂 last_active_at(非 DB 列,供 AdminUserListItem from_attributes 读)。
 
@@ -144,7 +97,7 @@ def _attach_last_active(db: Session, users: list[User]) -> None:
             select(AnalyticsEvent.user_id, func.max(AnalyticsEvent.created_at))
             .where(
                 AnalyticsEvent.user_id.in_(uids),
-                AnalyticsEvent.event.in_(_ACTIVE_EVENTS),
+                AnalyticsEvent.event.in_(activity.ACTIVE_EVENTS),
             )
             .group_by(AnalyticsEvent.user_id)
         ).all()
@@ -161,9 +114,9 @@ def _attach_last_active(db: Session, users: list[User]) -> None:
     )
     for u in users:
         candidates = [
-            _norm_utc(u.last_login_at),
-            _norm_utc(ev_map.get(u.id)),
-            _norm_utc(eng_map.get(u.id)),
+            activity.norm_utc(u.created_at),      # baseline 由 last_login_at 改为 created_at(登录不算活跃)
+            activity.norm_utc(ev_map.get(u.id)),
+            activity.norm_utc(eng_map.get(u.id)),
         ]
         u.last_active_at = max((c for c in candidates if c is not None), default=None)
 
@@ -191,16 +144,12 @@ def list_users(
     (口径见 [_last_active_expr])。**offset 分页**(cursor=offset):任意列排序下游标语义统一,
     代价是翻页期间数据变动可能错位一条——admin 低频场景可接受(同 [list_all_withdraw_orders])。
     日期入参统一转 tz-aware UTC 比较(列为 timestamptz,见 _as_utc)。"""
-    # 最近活跃 = max(最近登录, 最近行为事件, 最近领券发起)。PG 用 GREATEST;SQLite 标量 max()
-    # 任一参数 NULL 即返回 NULL,故 LEFT JOIN 未命中侧 coalesce 到 last_login_at 兜底
-    # (注册即登录,该列恒非空)。派生表 1:1(按 user_id 聚合),outerjoin 不会放大行数,
-    # offset_paginate 的 count 不受影响。
-    ev_agg, eng_agg = _last_active_parts()
-    greatest = func.greatest if db.get_bind().dialect.name == "postgresql" else func.max
-    last_active = greatest(
-        User.last_login_at,
-        func.coalesce(ev_agg.c.last_at, User.last_login_at),
-        func.coalesce(eng_agg.c.last_at, User.last_login_at),
+    # 最近活跃 = max(注册时间, 最近行为事件, 最近领券发起)。baseline 由 last_login_at 改为 created_at
+    #(登录不代表在用 App;口径统一到 activity.py,含 home_view + 比价 + 领券,见 activity.ACTIVE_EVENTS)。
+    # 未命中侧 coalesce 到 created_at(恒非空基线)。派生表 1:1,outerjoin 不放大行数。
+    ev_agg, eng_agg = activity.last_active_subqueries(db)
+    last_active = activity.last_active_expr(
+        User.created_at, ev_agg, eng_agg, db.get_bind().dialect.name
     )
     stmt = (
         select(User)

app/core/inactivity_reset_worker.py

@@ -0,0 +1,143 @@
+"""15 天不活跃清零的进程内每日任务。
+
+仿 daily_exchange_worker:App 启动自带,每 `INACTIVITY_RESET_CHECK_INTERVAL_SEC` 醒一次,
+跨进北京新的一天且到达 `INACTIVITY_RESET_RUN_HOUR`(默认 3 点)后跑一轮 `run_once`(预警 + 清零)。
+
+健壮性:
+- **逐用户幂等**:清完余额=0 次日不再匹配;预警按 streak 去重。启动补跑 / 多次唤醒 / 重启都安全。
+- **同机多进程互斥**:文件锁保证多 worker 只有一个实际跑。
+- **开关**:settings.INACTIVITY_RESET_ENABLED=false 时不启动(默认关,灰度验证后再开)。
+
+⚠️ 这是不可逆批量资金操作(清空金币 + 折算现金,**邀请现金不清**)。口径见
+app.repositories.inactivity / app.repositories.activity。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import os
+import time
+from collections.abc import Iterator
+from datetime import date, datetime
+from pathlib import Path
+
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.core.config import settings
+from app.core.rewards import CN_TZ, cn_today
+from app.db.session import SessionLocal
+from app.integrations.notifier import get_notifier
+from app.repositories import inactivity as inactivity_repo
+
+logger = logging.getLogger("shagua.inactivity")
+_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "inactivity_reset.lock"
+
+
+def _cn_today() -> date:
+    return cn_today()
+
+
+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 _run_once_entry() -> dict:
+    """跑一轮(预警 + 清零)。独立开 Session。"""
+    notifier = get_notifier(settings.INACTIVITY_NOTIFY_CHANNEL)
+    with SessionLocal() as db:
+        return inactivity_repo.run_once(
+            db,
+            notifier=notifier,
+            reset_days=settings.INACTIVITY_RESET_DAYS,
+            warn_stages=settings.inactivity_warn_stages,
+            today=_cn_today(),
+        )
+
+
+async def _run_loop() -> None:
+    interval = max(60, int(settings.INACTIVITY_RESET_CHECK_INTERVAL_SEC))
+    lock_stale_after = max(interval * 3, 1800)
+    with _single_instance_lock(lock_stale_after) as lock_acquired:
+        if not lock_acquired:
+            logger.warning("inactivity reset skipped: another worker owns lock")
+            return
+        await _run_locked_loop(interval)
+
+
+async def _run_locked_loop(interval: int) -> None:
+    logger.info(
+        "inactivity reset worker started interval=%ss run_hour=%s",
+        interval,
+        settings.INACTIVITY_RESET_RUN_HOUR,
+    )
+    # 本进程上次跑过的北京日;None=尚未跑过本进程(当天到点即补)。
+    last_run: date | None = None
+    try:
+        while True:
+            try:
+                _touch_lock()
+                today = _cn_today()
+                hour = datetime.now(CN_TZ).hour
+                if last_run != today and hour >= int(settings.INACTIVITY_RESET_RUN_HOUR):
+                    result = await asyncio.to_thread(_run_once_entry)
+                    last_run = today
+                    logger.info("inactivity reset done date=%s result=%s", today, result)
+            except SQLAlchemyError:
+                logger.exception("inactivity reset db error")
+            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
+                logger.exception("inactivity reset unexpected error")
+            await asyncio.sleep(interval)
+    except asyncio.CancelledError:
+        logger.info("inactivity reset worker stopped")
+        raise
+
+
+def start_inactivity_reset_worker() -> asyncio.Task | None:
+    if not settings.INACTIVITY_RESET_ENABLED:
+        logger.info("inactivity reset disabled (INACTIVITY_RESET_ENABLED=false)")
+        return None
+    return asyncio.create_task(_run_loop(), name="inactivity-reset")
+
+
+async def stop_inactivity_reset_worker(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task

app/integrations/notifier.py

@@ -15,9 +15,10 @@ logger = logging.getLogger("shagua.inactivity")
 class InactivityNotifier(Protocol):
     channel: str
 
-    def warn(self, *, user_id: int, coin: int, cash_cents: int, invite_cash_cents: int,
+    def warn(self, *, user_id: int, coin: int, cash_cents: int,
              stage: int, days_until_reset: int) -> str:
-        """发预警,返回状态:'sent' / 'failed' / 'placeholder'。"""
+        """发预警(只涉及会被清的金币 + 折算现金;邀请现金不清、不预警)。
+        返回状态:'sent' / 'failed' / 'placeholder'。"""
         ...
 
 
@@ -26,12 +27,11 @@ class LogNotifier:
 
     channel = "log"
 
-    def warn(self, *, user_id: int, coin: int, cash_cents: int, invite_cash_cents: int,
+    def warn(self, *, user_id: int, coin: int, cash_cents: int,
              stage: int, days_until_reset: int) -> str:
         logger.warning(
-            "[inactivity-warn] user=%s coin=%s cash_cents=%s invite_cash_cents=%s "
-            "stage=T-%s days_until_reset=%s",
-            user_id, coin, cash_cents, invite_cash_cents, stage, days_until_reset,
+            "[inactivity-warn] user=%s coin=%s cash_cents=%s stage=T-%s days_until_reset=%s",
+            user_id, coin, cash_cents, stage, days_until_reset,
         )
         return "placeholder"
 

app/main.py

@@ -49,6 +49,10 @@ from app.core.heartbeat_monitor_worker import (
     start_heartbeat_monitor,
     stop_heartbeat_monitor,
 )
+from app.core.inactivity_reset_worker import (
+    start_inactivity_reset_worker,
+    stop_inactivity_reset_worker,
+)
 from app.core.logging import setup_logging
 from app.core.pricebot_client import aclose_pricebot_client, get_pricebot_client
 from app.core.withdraw_reconcile_worker import (
@@ -80,12 +84,14 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     reconcile_task = start_withdraw_reconcile_worker()
     heartbeat_task = start_heartbeat_monitor()
     daily_exchange_task = start_daily_exchange_worker()
+    inactivity_task = start_inactivity_reset_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)
+        await stop_inactivity_reset_worker(inactivity_task)
         await aclose_pricebot_client()
         logger.info("shutting down")
 

app/models/inactivity.py

@@ -1,7 +1,9 @@
 """15 天不活跃清零相关表。
 
-- inactivity_reset_log:每次清零一行,记清零前三桶余额 + 原因 + 判定时活跃时间/不活跃天数,
-  供纠纷排查(需求①)。清零同时另写 3 条钱包流水(biz_type=inactivity_reset),资金流可逐笔回溯。
+- inactivity_reset_log:每次清零一行,记清零前三桶余额快照 + 原因 + 判定时活跃时间/不活跃天数,
+  供纠纷排查(需求①)。清零同时另写 2 条钱包流水(金币 + 折算现金,biz_type=inactivity_reset),
+  资金流可逐笔回溯。**邀请现金是产品红线、不清零**,invite_cash_balance_cents_before 仅为清零时
+  仍保留的邀请现金快照(便于排查、非被清金额;见 wallet.CoinAccount 注释)。
 - inactivity_notification_log:每次预警一行,记推送时余额快照 + 档位 + 通道 + 状态,
   兼作"预警去重"依据(created_at > last_active)与"待推送"占位 outbox(v1 通道=log)。
 

app/repositories/inactivity.py

@@ -5,26 +5,31 @@
 """
 from __future__ import annotations
 
-from datetime import date, datetime, timezone
+import logging
+from datetime import date, datetime
 
 from sqlalchemy import or_, select
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session
 
 from app.core.rewards import CN_TZ
-from app.models.inactivity import InactivityResetLog
+from app.integrations.notifier import InactivityNotifier
+from app.models.inactivity import InactivityNotificationLog, InactivityResetLog
 from app.models.user import User
 from app.models.wallet import CoinAccount
 from app.repositories import activity
 from app.repositories import wallet as wallet_repo
 
+logger = logging.getLogger("shagua.inactivity")
+
 RESET_BIZ_TYPE = "inactivity_reset"
 RESET_REMARK = "15天不活跃清零"
 
+# 清零候选口径:金币或折算现金有余额即入选。**邀请现金不算**——它是产品红线、不清零
+# (见 wallet.CoinAccount 注释),只有邀请现金余额的用户没有可清项,故不入选。
 _ANY_BALANCE = or_(
     CoinAccount.coin_balance > 0,
     CoinAccount.cash_balance_cents > 0,
-    CoinAccount.invite_cash_balance_cents > 0,
 )
 
 
@@ -58,17 +63,19 @@ def _inactive_days(last_active: datetime, today: date) -> int:
 
 
 def select_inactive_users(db: Session, *, cutoff: datetime):
-    """应清零用户:last_active < cutoff 且三桶有余额。返回 Row 列表(值已快照,可跨 commit)。"""
+    """应清零用户:last_active < cutoff 且金币/折算现金有余额(邀请现金不清、不计)。
+    返回 Row 列表(值已快照,可跨 commit)。"""
     stmt, last_active = _base_query(db)
     stmt = stmt.where(_ANY_BALANCE, last_active < activity.as_utc(cutoff))
     return db.execute(stmt).all()
 
 
 def clear_user(db: Session, *, user_id: int, last_active: datetime, inactive_days: int, reason: str) -> bool:
-    """单用户清零(独立事务、行锁)。三桶归零 + 写审计 + 3 条流水。返回是否真清了(有余额)。"""
+    """单用户清零(独立事务、行锁)。金币 + 折算现金归零 + 写审计 + 2 条流水;**邀请现金不清**
+    (产品红线,见 wallet.CoinAccount 注释),仅作快照记入审计。返回是否真清了(有可清余额)。"""
     acc = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True)
     coin, cash, invite = acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents
-    if coin == 0 and cash == 0 and invite == 0:
+    if coin == 0 and cash == 0:  # 邀请现金不清,故不算"有可清余额"
         return False
     log = InactivityResetLog(
         user_id=user_id, coin_balance_before=coin, cash_balance_cents_before=cash,
@@ -82,8 +89,7 @@ def clear_user(db: Session, *, user_id: int, last_active: datetime, inactive_day
         wallet_repo.grant_coins(db, user_id, -coin, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
     if cash:
         wallet_repo.grant_cash(db, user_id, -cash, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
-    if invite:
-        wallet_repo.grant_invite_cash(db, user_id, -invite, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
+    # 邀请现金(invite_cash_balance_cents)刻意不动:两本账物理隔离、邀请金是产品红线。
     db.commit()
     return True
 
@@ -105,3 +111,73 @@ def run_reset_once(db: Session, *, reset_days: int, today: date) -> dict:
             db.rollback()
             stats["failed"] += 1
     return stats
+
+
+def select_warn_candidates(db: Session, *, clear_cutoff: datetime, warn_hi: datetime):
+    """预警候选:clear_cutoff <= last_active < warn_hi 且有可清余额(即已进预警窗、尚未到清零)。"""
+    stmt, last_active = _base_query(db)
+    stmt = stmt.where(
+        _ANY_BALANCE,
+        last_active >= activity.as_utc(clear_cutoff),
+        last_active < activity.as_utc(warn_hi),
+    )
+    return db.execute(stmt).all()
+
+
+def run_warn_once(db: Session, notifier: InactivityNotifier, *,
+                  reset_days: int, warn_stages: list[int], today: date) -> dict:
+    """扫一轮预警。每人取"最紧急的已到达档",按 streak 去重(notification_log.created_at > last_active)。
+    预警只涉及会被清的金币 + 折算现金;邀请现金不清、不预警(仅在 notification_log 记快照)。
+    逐用户 try/except 隔离:单用户通知器抛错 / DB 错不阻断其余,也绝不能拖累后续清零。"""
+    stats = {"warned": 0, "warn_skipped": 0, "warn_failed": 0}
+    if not warn_stages:
+        return stats
+    clear_cutoff = activity.reset_cutoff(reset_days, today)                # 到此即清零,不再预警
+    warn_hi = activity.reset_cutoff(reset_days - max(warn_stages), today)  # 最早预警档边界
+    ascending = sorted(warn_stages)                                       # 最紧急(最小 k)在前
+    for row in select_warn_candidates(db, clear_cutoff=clear_cutoff, warn_hi=warn_hi):
+        idays = _inactive_days(row.last_active, today)
+        stage = next((k for k in ascending if idays >= reset_days - k), None)
+        if stage is None:  # 防御:候选已在预警窗内、stage 必命中,此分支实际不可达
+            continue
+        try:
+            already = db.execute(
+                select(InactivityNotificationLog.id).where(
+                    InactivityNotificationLog.user_id == row.user_id,
+                    InactivityNotificationLog.stage == stage,
+                    InactivityNotificationLog.created_at > activity.as_utc(row.last_active),
+                ).limit(1)
+            ).first()
+            if already:
+                stats["warn_skipped"] += 1
+                continue
+            status = notifier.warn(
+                user_id=row.user_id, coin=row.coin_balance, cash_cents=row.cash_balance_cents,
+                stage=stage, days_until_reset=reset_days - idays,
+            )
+            db.add(InactivityNotificationLog(
+                user_id=row.user_id, stage=stage, inactive_days=idays,
+                coin_balance=row.coin_balance, cash_balance_cents=row.cash_balance_cents,
+                invite_cash_balance_cents=row.invite_cash_balance_cents,  # 快照,不参与"将清"额度
+                channel=notifier.channel, status=status,
+            ))
+            db.commit()
+            stats["warned"] += 1
+        except Exception:  # noqa: BLE001 - 单用户预警失败(通知器抛错/DB 错)隔离,不阻断其余、不拖累清零
+            db.rollback()
+            stats["warn_failed"] += 1
+    return stats
+
+
+def run_once(db: Session, *, notifier: InactivityNotifier, reset_days: int,
+             warn_stages: list[int], today: date) -> dict:
+    """一轮完整任务:先预警(阶段 A)再清零(阶段 B)。返回合并统计。
+    预警整段异常也**绝不阻塞清零**——清零是核心、不可逆资金操作,不能被通知故障拖住。"""
+    try:
+        warn = run_warn_once(db, notifier, reset_days=reset_days, warn_stages=warn_stages, today=today)
+    except Exception:  # noqa: BLE001 - 预警阶段整体失败(如候选查询失败)也要继续清零
+        logger.exception("inactivity warn phase failed; proceeding to reset")
+        db.rollback()
+        warn = {"warned": 0, "warn_skipped": 0, "warn_failed": 0, "warn_phase_error": 1}
+    reset = run_reset_once(db, reset_days=reset_days, today=today)
+    return {**warn, **reset}

tests/test_inactivity_reset.py

@@ -85,9 +85,11 @@ def _isolate_inactivity_state():
 def _new_user(db, *, created_at, coin=0, cash=0, invite=0) -> int:
     """直接建一个 User + CoinAccount,created_at 可控。返回 user_id。"""
     _PHONE_SEQ[0] += 1
-    u = User(phone=f"139{_PHONE_SEQ[0]:08d}", created_at=created_at,
+    # 199 前缀 + 递增序号:共享测试库跨文件累积用户,别的文件用固定手机号(如 test_admin_write
+    # 的 13900000001..),这里用没人用的 199 段避免撞 user.phone / username 的 UNIQUE。
+    u = User(phone=f"199{_PHONE_SEQ[0]:08d}", created_at=created_at,
              last_login_at=created_at, status="active",
-             username=f"2{_PHONE_SEQ[0]:010d}")
+             username=f"inact{_PHONE_SEQ[0]}")
     db.add(u)
     db.flush()
     acc = wallet_repo.get_or_create_account(db, u.id, commit=False)
@@ -142,13 +144,13 @@ def test_log_notifier_returns_placeholder(caplog) -> None:
     from app.integrations.notifier import LogNotifier, get_notifier
     n = get_notifier("log")
     assert isinstance(n, LogNotifier) and n.channel == "log"
-    status = n.warn(user_id=1, coin=10, cash_cents=20, invite_cash_cents=30, stage=7, days_until_reset=8)
+    status = n.warn(user_id=1, coin=10, cash_cents=20, stage=7, days_until_reset=8)
     assert status == "placeholder"
     # 未实现通道回退 LogNotifier(占位)
     assert get_notifier("jpush").channel == "log"
 
 
-def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> None:
+def test_run_reset_clears_coin_and_cash_but_preserves_invite_cash() -> None:
     from sqlalchemy import select
     from app.models.wallet import CoinAccount, CoinTransaction, CashTransaction, InviteCashTransaction
     from app.repositories import inactivity
@@ -168,10 +170,13 @@ def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> None:
         assert stats["cleared"] == 1 and stats["failed"] == 0
 
         acc = db.get(CoinAccount, old)
-        assert (acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents) == (0, 0, 0)
+        # 金币 + 折算现金清零;邀请现金是产品红线,原封不动(见 wallet.CoinAccount 注释)
+        assert (acc.coin_balance, acc.cash_balance_cents) == (0, 0)
+        assert acc.invite_cash_balance_cents == 300
         assert acc.total_coin_earned == 100  # 历史累计不动
 
         log = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == old)).scalar_one()
+        # 审计仍快照三桶余额(邀请现金记为"清零时仍保留"的余额,便于纠纷排查)
         assert (log.coin_balance_before, log.cash_balance_cents_before,
                 log.invite_cash_balance_cents_before) == (100, 200, 300)
         assert log.inactive_days == 22 and log.reason == "inactive_15d"
@@ -181,12 +186,198 @@ def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> None:
         assert ct.amount == -100 and ct.balance_after == 0 and ct.ref_id == str(log.id)
         assert db.execute(select(CashTransaction).where(
             CashTransaction.user_id == old, CashTransaction.biz_type == "inactivity_reset")).scalar_one().amount_cents == -200
+        # 关键:不写邀请现金流水(邀请现金不清)
         assert db.execute(select(InviteCashTransaction).where(
-            InviteCashTransaction.user_id == old, InviteCashTransaction.biz_type == "inactivity_reset")).scalar_one().amount_cents == -300
+            InviteCashTransaction.user_id == old,
+            InviteCashTransaction.biz_type == "inactivity_reset")).first() is None
 
-        # 活跃用户不动;再跑一次幂等(已清零 → 不再匹配)
+        # 活跃用户不动;再跑一次幂等(coin+cash 已 0、邀请现金不算候选 → 不再匹配)
         assert db.get(CoinAccount, fresh).coin_balance == 50
         assert inactivity.run_reset_once(db, reset_days=15, today=today)["cleared"] == 0
     finally:
         db.rollback()
         db.close()
+
+
+def test_user_with_only_invite_cash_is_not_cleared() -> None:
+    """只有邀请现金余额的久不活跃用户:邀请现金是产品红线,不清 → 根本不该被选中。"""
+    from app.models.wallet import CoinAccount
+    from app.repositories import inactivity
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        uid = _new_user(db, created_at=datetime(2026, 1, 10, tzinfo=timezone.utc),
+                        coin=0, cash=0, invite=500)
+        db.commit()
+        stats = inactivity.run_reset_once(db, reset_days=15, today=today)
+        assert stats["cleared"] == 0
+        assert db.get(CoinAccount, uid).invite_cash_balance_cents == 500  # 原封不动
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_run_warn_picks_stage_and_dedups_within_streak() -> None:
+    from app.integrations.notifier import LogNotifier
+    from app.repositories import inactivity
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        # 末次活跃 1/22(距 today 10 天)→ 档 7 命中(idays>=8),档 2 未到(需>=13)
+        uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=100)
+        db.commit()
+
+        stats = inactivity.run_warn_once(db, LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)
+        assert stats["warned"] == 1
+        from sqlalchemy import select
+        rows = db.execute(select(InactivityNotificationLog).where(
+            InactivityNotificationLog.user_id == uid)).scalars().all()
+        assert len(rows) == 1 and rows[0].stage == 7 and rows[0].status == "placeholder"
+        assert rows[0].inactive_days == 10 and rows[0].coin_balance == 100
+
+        # 同一 streak 再跑 → 不重推
+        assert inactivity.run_warn_once(db, LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)["warned"] == 0
+
+        # 无余额用户不预警
+        _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=0)
+        db.commit()
+        assert inactivity.run_warn_once(db, LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)["warned"] == 0
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_run_once_warns_then_resets() -> None:
+    from app.integrations.notifier import LogNotifier
+    from app.models.wallet import CoinAccount
+    from app.repositories import inactivity
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        warn_uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=10)   # 10天→预警
+        clear_uid = _new_user(db, created_at=datetime(2026, 1, 5, tzinfo=timezone.utc), coin=10)   # 27天→清零
+        db.commit()
+        stats = inactivity.run_once(db, notifier=LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)
+        assert stats["warned"] == 1 and stats["cleared"] == 1
+        assert db.get(CoinAccount, clear_uid).coin_balance == 0
+        assert db.get(CoinAccount, warn_uid).coin_balance == 10   # 预警不动钱
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_worker_disabled_returns_none(monkeypatch) -> None:
+    from app.core import inactivity_reset_worker as w
+    from app.core.config import settings
+    monkeypatch.setattr(settings, "INACTIVITY_RESET_ENABLED", False)
+    assert w.start_inactivity_reset_worker() is None
+
+
+def test_worker_run_once_entry_executes(monkeypatch) -> None:
+    """_run_once_entry 用真实 SessionLocal 跑一轮,总闸开时能清掉一个不活跃用户。"""
+    from app.core import inactivity_reset_worker as w
+    from app.core.config import settings
+    from app.models.wallet import CoinAccount
+
+    monkeypatch.setattr(settings, "INACTIVITY_RESET_ENABLED", True)
+    monkeypatch.setattr(settings, "INACTIVITY_RESET_DAYS", 15)
+    monkeypatch.setattr(settings, "INACTIVITY_WARN_DAYS_BEFORE", "")  # 只测清零
+    # 固定"今天"避免依赖真实时钟
+    monkeypatch.setattr(w, "_cn_today", lambda: date(2026, 2, 1))
+
+    db = SessionLocal()
+    try:
+        uid = _new_user(db, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), coin=100)
+        db.commit()
+    finally:
+        db.close()
+
+    stats = w._run_once_entry()
+    assert stats["cleared"] >= 1
+
+    db = SessionLocal()
+    try:
+        assert db.get(CoinAccount, uid).coin_balance == 0
+    finally:
+        db.close()
+
+
+def test_admin_list_users_last_active_ignores_login() -> None:
+    """admin 用户列表 last_active_at 改用共享口径:登录不算活跃(baseline=created_at)、只认活跃事件。"""
+    from app.admin.repositories import queries
+
+    db = SessionLocal()
+    try:
+        created = datetime(2026, 1, 1, tzinfo=timezone.utc)
+        uid = _new_user(db, created_at=created)
+        u = db.get(User, uid)
+        u.last_login_at = datetime(2026, 6, 1, tzinfo=timezone.utc)  # 登录很新、但无任何活跃事件
+        db.commit()
+        phone = db.get(User, uid).phone
+        users, _cursor, _total = queries.list_users(db, phone=phone)
+        item = next(x for x in users if x.id == uid)
+        assert activity.norm_utc(item.last_active_at) == created  # 登录不算 → last_active=created_at
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_run_warn_isolates_notifier_failure_and_does_not_block_reset() -> None:
+    """单用户通知器抛错:预警计 warn_failed、不外抛,且清零(reset)照常执行。"""
+    from app.models.wallet import CoinAccount
+    from app.repositories import inactivity
+
+    class BoomNotifier:
+        channel = "log"
+
+        def warn(self, *, user_id, coin, cash_cents, stage, days_until_reset) -> str:
+            raise RuntimeError("push service down")
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        warn_uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=10)   # 10天→预警
+        clear_uid = _new_user(db, created_at=datetime(2026, 1, 5, tzinfo=timezone.utc), coin=10)   # 27天→清零
+        db.commit()
+
+        stats = inactivity.run_once(db, notifier=BoomNotifier(), reset_days=15,
+                                    warn_stages=[7, 2], today=today)
+        assert stats["warned"] == 0 and stats["warn_failed"] >= 1   # 预警失败被隔离
+        assert stats["cleared"] == 1                                # 关键:清零没被阻塞
+        assert db.get(CoinAccount, clear_uid).coin_balance == 0
+        assert db.get(CoinAccount, warn_uid).coin_balance == 10     # 预警用户不动钱
+        # 预警失败已回滚,不留半条 notification_log
+        from sqlalchemy import select
+        assert db.execute(select(InactivityNotificationLog).where(
+            InactivityNotificationLog.user_id == warn_uid)).first() is None
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_run_once_reset_runs_even_if_warn_phase_throws(monkeypatch) -> None:
+    """预警整段异常(如候选查询失败)也绝不阻塞清零。"""
+    from app.integrations.notifier import LogNotifier
+    from app.models.wallet import CoinAccount
+    from app.repositories import inactivity
+
+    def boom(*a, **k):
+        raise RuntimeError("warn phase blew up")
+
+    monkeypatch.setattr(inactivity, "run_warn_once", boom)
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        clear_uid = _new_user(db, created_at=datetime(2026, 1, 5, tzinfo=timezone.utc), coin=10)
+        db.commit()
+        stats = inactivity.run_once(db, notifier=LogNotifier(), reset_days=15,
+                                    warn_stages=[7, 2], today=today)
+        assert stats["cleared"] == 1
+        assert db.get(CoinAccount, clear_uid).coin_balance == 0
+    finally:
+        db.rollback()
+        db.close()