fix(analytics): unreplied is_send守卫防漂移假行+真预算注释+501候选分支测试

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

yunke-server/app/analytics/metrics.py

@@ -460,6 +460,9 @@ async def churn(
 
 _VALID_TAIL_TYPES = ("text", "image", "voice", "video", "file", "link")
 
+# 候选量阈值:≤此值决赛圈全量点查(total 精确);超过则按页点查(total=候选数)
+_FINALIZE_ALL_MAX = 500
+
 
 def _norm_phone(p: str | None) -> str | None:
     """归一:+86/空格/横线剥掉(spec B1 同款规则)。"""
@@ -479,12 +482,16 @@ async def unreplied(
     """B2 该回未回(spec §3.4)。候选=汇总表最新日行;决赛圈点查类型/预览/通话豁免。
 
     候选资格看「最新日行」(row_number 先于方向/超时过滤——员工次日已回的
-    旧未回日不复活);决赛圈再点查尾部消息有效类型(排除 unknown/divider 等)、
-    剔除已删/拉黑客户、通话豁免(该客户 phone 任一晚于最后消息的通话=已回应,
+    旧未回日不复活);决赛圈再点查尾部消息:须员工未回(is_send 非 True——
+    stats/message 漂移时以 message 实尾为准,写钩子吞重算失败是已记录状态;
+    is_send=None 保留=与统计同向保守)且有效类型(排除 unknown/divider 等),
+    剔除已删/拉黑客户,通话豁免(该客户 phone 任一晚于最后消息的通话=已回应,
     不限定通话员工——客户被回应过即可,spec:尾部=max(最后消息,最后通话))。
+    无 customer 行的 peer(voice-only)保留在清单——豁免不可判定,保守方向
+    是有意为之。
 
-    total 口径:候选 ≤500 时决赛圈全过滤后分页(total 精确);>500 时按候选
-    分页(total=候选数,注释口径——决赛圈过滤可能让个别页少几行)。
+    total 口径:候选 ≤ _FINALIZE_ALL_MAX 时决赛圈全过滤后分页(total 精确);
+    超过则按候选分页(total=候选数,注释口径——决赛圈过滤可能让个别页少几行)。
     """
     now = _now_ms()
     cutoff = now - hours * 3600 * 1000
@@ -542,7 +549,7 @@ async def unreplied(
                 ).order_by(Message.timestamp.desc()).limit(1)
             )
         ).scalar_one_or_none()
-        if last_msg is None or last_msg.type not in _VALID_TAIL_TYPES:
+        if last_msg is None or last_msg.is_send or last_msg.type not in _VALID_TAIL_TYPES:
             return None
         cust = await session.get(Customer, peer)
         if cust is not None and (
@@ -575,8 +582,11 @@ async def unreplied(
         }
 
     async def _finalize(rows: list[Any]) -> list[dict]:
-        # 顺序逐个 await——AsyncSession 不能并发执行查询(禁 asyncio.gather);
-        # 每页 ≤100 个点查、每个毫秒级,顺序总耗时百毫秒级可接受
+        # 顺序逐个 await——AsyncSession 不能并发执行查询(禁 asyncio.gather)。
+        # 真预算:≤_FINALIZE_ALL_MAX 分支=全部候选点查(≤500 候选 ×~3 查询
+        # ≈1500 顺序往返,毫秒级/个);>500 分支=仅本页 ≤100 候选。豁免查询
+        # phone_call.call_number 无索引(3k 行小表扫;量大时批量化 IN 或补
+        # 索引——后续优化不在本次)
         out = []
         for o, p, t in rows:
             r = await _finalist(o, p, t)
@@ -584,7 +594,7 @@ async def unreplied(
                 out.append(r)
         return out
 
-    if len(cand) <= 500:
+    if len(cand) <= _FINALIZE_ALL_MAX:
         items = await _finalize(cand)
         # keep 候选顺序(ts asc, peer asc)——顺序循环天然保序
         total = len(items)

yunke-server/tests/test_analytics_churn_unreplied.py

@@ -138,9 +138,6 @@ def test_next_tier_bad_tier_raises_value_error():
     assert metrics._next_tier(90) == 10_000  # 90 档=≥90 的开区间上限
 
 
-_VALID_TYPES = ("text", "image", "voice", "video", "file", "link")
-
-
 async def _seed_unreplied(session):
     now = _NOW
     h3 = 3 * 3600 * 1000
@@ -283,3 +280,63 @@ async def test_unreplied_endpoint_422(make_client):
     assert r.status_code == 422
     r2 = await client.get("/api/analytics/unreplied", params={"channel": "sms"})
     assert r2.status_code == 422
+
+
+async def test_unreplied_stats_message_drift_staff_tail_excluded(
+    make_client, monkeypatch
+):
+    """stats/message 漂移守卫:统计行客户尾向超时,但 message 真实尾部是员工回复 → 不在清单。
+
+    写钩子吞重算失败、夜间重建兜底前 stats 可能滞后(一等状态);决赛圈以
+    message 实尾为准——is_send=True=员工尾部=定义上非待回复(is_send=None
+    保留=与统计同向保守)。
+    """
+    monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW)
+    async def seed(session):
+        now = _NOW
+        h3 = 3 * 3600 * 1000
+        session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
+                                  peer_kind="friend", peer_id="t9", day=20260904,
+                                  send_cnt=5, recv_cnt=3, last_ts=now - h3,
+                                  last_is_send=False))
+        # 真实尾部:员工 1h 前的回复(stats 行漂移,仍停在客户尾向 3h 前)
+        session.add(Message(msg_id="t9m", channel="wechat", owner_account_id="wx1",
+                            talker_id="t9", timestamp=now - 3600 * 1000,
+                            is_send=True, type="text", text="已经回了哈",
+                            fetched_at=dt.datetime(2026, 9, 4, 12, 0)))
+        out = await metrics.unreplied(session, channel="wechat", hours=2,
+                                      page=1, page_size=20)
+        assert all(r["peer_id"] != "t9" for r in out["items"])
+
+    await make_client(seed)
+
+
+async def test_unreplied_large_candidate_branch(make_client, monkeypatch):
+    """候选 > _FINALIZE_ALL_MAX:按候选分页,total=候选数;决赛圈过滤后本页可少几行。
+
+    501 个候选只给 bulk000 一条 message 行 → 第 1 页 items 仅 1 行(而非
+    page_size 100),锁定「total=候选数、items 可能少几行」的文档口径;
+    同时证明大候选分支的决赛圈点查真实执行(有效行保留)。
+    """
+    monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW)
+    async def seed(session):
+        now = _NOW
+        h3 = 3 * 3600 * 1000
+        for i in range(metrics._FINALIZE_ALL_MAX + 1):
+            session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
+                                      peer_kind="friend", peer_id=f"bulk{i:03d}",
+                                      day=20260904, send_cnt=5, recv_cnt=3,
+                                      last_ts=now - h3, last_is_send=False))
+        # ts 全等 → 按 peer asc 决胜,bulk000 排第 1 页首行
+        session.add(Message(msg_id="b0m", channel="wechat", owner_account_id="wx1",
+                            talker_id="bulk000", timestamp=now - h3,
+                            is_send=False, type="text", text="在吗",
+                            fetched_at=dt.datetime(2026, 9, 4, 12, 0)))
+        await session.commit()
+        out = await metrics.unreplied(session, channel="wechat", hours=2,
+                                      page=1, page_size=100)
+        assert out["total"] == metrics._FINALIZE_ALL_MAX + 1  # total=候选数
+        ids = [r["peer_id"] for r in out["items"]]
+        assert ids == ["bulk000"]  # 无 message 行的候选决赛圈全滤=口径内少几行
+
+    await make_client(seed)