fix(management): 宽表生命周期hist改时点语义(≤as_of累积)——历史日dormant不再被未来消息虚增;补正向同源/percentile None判别+防御性flush

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

yunke-server/app/storage/employee_metrics.py

@@ -15,6 +15,7 @@ import logging
 import math
 from bisect import bisect_right
 from collections.abc import Iterable
+from itertools import accumulate
 
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -170,8 +171,10 @@ async def _lifecycle_daily(
 ) -> dict[int, dict[str, dict[str, int]]]:
     """逐日 {day: {user: {state: count}}}——customer 当前归属 + 互动日 sweep。
 
-    数据源:msg_daily_stats friend 行全历史(互动日/条数)。16k 客户 × 窗口
-    天数 bisect,夜间全量约秒级;4h tick 只算当日+3 天。
+    数据源:msg_daily_stats friend 行全历史(互动日/条数)。hist 为 ≤as_of
+    的历史消息数(时点语义:与 last 同一 bisect 位取按日累积数组 cum[idx],
+    as_of 之后的消息不虚增历史日的沉睡门槛)。16k 客户 × 窗口天数
+    bisect+classify,夜间全量实测见批次A核实文档;4h tick 只算当日+3 天。
     """
     rows = (
         await session.execute(
@@ -184,12 +187,13 @@ async def _lifecycle_daily(
             )
         )
     ).all()
-    days_map: dict[str, set[int]] = {}
-    cnt: dict[str, int] = {}
+    # peer → {day: 消息条数};days_list 与累积数组 cum 对齐:
+    # cum[i] = 该客户 ≤days_list[i] 的历史消息数(hist 时点语义)
+    day_msgs: dict[str, dict[int, int]] = {}
     for peer, day, snd, rcv in rows:
-        days_map.setdefault(peer, set()).add(day)
-        cnt[peer] = cnt.get(peer, 0) + int(snd or 0) + int(rcv or 0)
-    interaction = {p: sorted(ds) for p, ds in days_map.items()}
+        dm = day_msgs.setdefault(peer, {})
+        dm[day] = dm.get(day, 0) + int(snd or 0) + int(rcv or 0)
+    interaction = {p: sorted(dm) for p, dm in day_msgs.items()}
 
     custs = (
         await session.execute(
@@ -212,14 +216,15 @@ async def _lifecycle_daily(
         if uid is None:
             continue
         first = days_list[0]
-        hist = cnt.get(peer, 0)
+        cum = list(accumulate(day_msgs[peer][d] for d in days_list))
         for d in days:
             idx = bisect_right(days_list, d) - 1
             if idx < 0:
                 continue
             last = days_list[idx]
             prev = days_list[idx - 1] if idx > 0 else None
-            state = classify(first, last, prev, d, hist)
+            hist_asof = cum[idx]
+            state = classify(first, last, prev, d, hist_asof)
             if state:
                 out[d].setdefault(uid, {}).setdefault(state, 0)
                 out[d][uid][state] += 1
@@ -345,4 +350,7 @@ async def recompute_metric_days(
                     )
                 )
                 written += 1
+            # 防御性 flush:防 autoflush=False 类配置下逐日 add 的 ORM 对象
+            # 攒到事务尾(夜间全量约 52k 行;与 Task3 周期性 flush 同教训)
+            await session.flush()
     return written