fix(management): first_reply_fact 同毫秒并列主键冲突+定序抖动——reply_msg_id 进 PK/按(ts,msg_id)排序

评审修正(代码质量评审 Critical#1 + Important#2/#3):
- reply_msg_id 进复合 PK(模型+migration b7c8d9e0f1a2 原地改,本地
  down/up 重建,表空未部署):同毫秒 F,T,F,T 交换的两个样本
  (thread_end_ts, reply_ts) 完全并列不再撞主键(原会中断整个
  recompute_all 事务);兼作下钻证据定位回复消息。
- load_conversation 增查 msg_id 并按 (ts, msg_id) 排序,
  compute_conversation 协议改 3 元组——SQL 等键顺序未定义,原实现
  同毫秒 [F,T] 归属随查询抖动(0/1 answered 摇摆)。
- recompute_pairs 每 500 会话显式 flush+进度日志(内存上界对齐
  summary.recompute_all 惯例,不再依赖隐式 autoflush)。
- 测试:旧 9 个更新协议;新增同毫秒 F,T,F,T 回归/并列确定性/
  空会话幂等清空,fr_sum 总量断言并入双串测试;
  test_management_models 构造补新主键列。

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

yunke-server/app/db/models.py

@@ -1021,6 +1021,9 @@ class FirstReplyFact(Base):
     peer_id: Mapped[str] = mapped_column(String(128), primary_key=True)
     thread_end_ts: Mapped[int] = mapped_column(BigInteger, primary_key=True)
     reply_ts: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    # 回复消息 msg_id:同毫秒 F,T,F,T 交换的 (thread_end_ts, reply_ts) 并列
+    # 去重(一条回复恰好闭合一个串);兼作下钻证据定位回复消息
+    reply_msg_id: Mapped[str] = mapped_column(String(64), primary_key=True)
     # 串末条所在上海自然日 yyyymmdd(冗余物化,窗口/趋势索引)
     day: Mapped[int] = mapped_column(Integer, nullable=False)
     fr_ms: Mapped[int] = mapped_column(BigInteger, nullable=False)

yunke-server/app/storage/first_reply_facts.py

@@ -8,11 +8,17 @@
 compute_conversation 同时输出每日串聚合(threads/answered/fr_sum_ms,
 按串末条归属日)——employee_metric_daily 的串类列与 fact 行**同源**
 (回复率分子分母一致,v2 §1.1 #5)。
+
+时序确定性:同毫秒并列(ts 相同的 F/T 消息)按 msg_id 定序(wechat
+msgSvrId 服务端单调,近似真实到达序)——SQL 等键返回顺序未定义,不做
+定序会导致串归属随查询抖动;fact PK 含 reply_msg_id,同毫秒 F,T,F,T
+交换的两个样本不撞主键。
 """
 from __future__ import annotations
 
 import logging
 from collections import namedtuple
+from collections.abc import Iterable
 
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -22,8 +28,9 @@ from app.storage.summary import day_key
 
 logger = logging.getLogger(__name__)
 
-# 一个被回串的首响样本;day = 串末条客户消息所在上海自然日
+# 一个被回串的首响样本;day = 串末条客户消息所在上海自然日,
-FactSample = namedtuple("FactSample", "thread_end_ts reply_ts fr_ms day")
+# reply_msg_id = 闭合该串的员工回复消息 msg_id(PK 并列去重 + 下钻证据)
+FactSample = namedtuple("FactSample", "thread_end_ts reply_ts fr_ms day reply_msg_id")
 # 某会话某日的串聚合(分子分母同源)
 DayAgg = namedtuple("DayAgg", "threads answered fr_sum_ms")
 
@@ -38,19 +45,23 @@ class _DayAcc:
 
 
 def compute_conversation(rows: list) -> tuple[list[FactSample], dict[int, DayAgg]]:
-    """跨日精确串算法。rows = [(ts, is_send)] 该会话全历史,**须按 ts 升序**。
+    """跨日精确串算法。rows = [(ts, is_send, msg_id)] 该会话全历史,
+    **须按 (ts, msg_id) 升序**(load_conversation 负责排序;同毫秒并列以
+    msg_id 定序,见模块 docstring)。
 
     - 串 = 客户连续消息(is_send=False)不被员工消息(is_send=True)打断的
       段;is_send=None 跳过(不打断也不起串)——与 _thread_stats 判定一致。
     - 每个被回串一个样本:fr_ms = 员工首条回复 ts − 串**末条**客户消息 ts,
-      归属 day = 串末条所在上海自然日(跨夜串不再丢失)。
+      归属 day = 串末条所在上海自然日(跨夜串不再丢失);reply_msg_id =
+      该回复消息 msg_id——一条回复恰好闭合一个串,同毫秒 F,T,F,T 交换的
+      (thread_end_ts, reply_ts) 并列靠它去重,兼作下钻证据。
     - 会话末尾仍未闭合的串:计入当日 threads(回复率分母),不产生样本。
     """
     facts: list[FactSample] = []
     acc: dict[int, _DayAcc] = {}
     in_thread = False
     end_ts: int | None = None
-    for ts, is_send in rows:
+    for ts, is_send, msg_id in rows:
         if is_send is False:
             in_thread = True
             end_ts = int(ts)
@@ -58,7 +69,7 @@ def compute_conversation(rows: list) -> tuple[list[FactSample], dict[int, DayAgg
             if in_thread and end_ts is not None:
                 fr = int(ts) - end_ts
                 d = day_key(end_ts)
-                facts.append(FactSample(end_ts, int(ts), fr, d))
+                facts.append(FactSample(end_ts, int(ts), fr, d, msg_id))
                 a = acc.setdefault(d, _DayAcc())
                 a.threads += 1
                 a.answered += 1
@@ -76,13 +87,19 @@ def compute_conversation(rows: list) -> tuple[list[FactSample], dict[int, DayAgg
 async def load_conversation(
     session: AsyncSession, channel: str, owner: str, peer: str
 ) -> list:
-    """拉取该 1:1 会话全历史 (timestamp, is_send),Python 侧排序(串算法
+    """拉取该 1:1 会话全历史 (timestamp, is_send, msg_id),Python 侧按
-    需严格时序——与 summary.recompute_days 同惯例,避免依赖 SQL 排序)。
+    (ts, msg_id) 排序——串算法需严格时序且同毫秒并列须定序(SQL 等键顺序
+    未定义,不排序会随查询抖动;与 summary.recompute_days 同惯例,避免
+    依赖 SQL 排序)。friend 过滤口径与 summary._peer_filter 的 friend 分支
+    同源:talker_id 原值精确匹配 + room_id 为空(peer_id 与 talker_id 同源
+    同形态,不做截断/归一化)。
 
     公有命名:employee_metrics 宽表生成复用同一加载口径(同源原则)。"""
     rows = (
         await session.execute(
-            sa.select(Message.timestamp, Message.is_send).where(
+            sa.select(
+                Message.timestamp, Message.is_send, Message.msg_id
+            ).where(
                 Message.channel == channel,
                 Message.owner_account_id == owner,
                 Message.talker_id == peer,
@@ -91,7 +108,7 @@ async def load_conversation(
             )
         )
     ).all()
-    return sorted(rows, key=lambda r: r[0])
+    return sorted(rows, key=lambda r: (r[0], r[2]))
 
 
 async def recompute_pairs(
@@ -99,8 +116,12 @@ async def recompute_pairs(
 ) -> int:
     """重算给定 friend 会话的 fact 行:按会话 delete+insert,幂等。
 
-    pairs = [(channel, owner_account_id, peer_id)]。返回会话数。
+    pairs = [(channel, owner_account_id, peer_id)]。返回会话数。每 500 个
+    会话 flush 一次约束 session 内存(与 summary.recompute_all 同惯例——
+    全量回填百万级 fact 行不能攒到最后),同时打进度日志(回填分钟级,
+    可观测)。commit 由调用方负责。
     """
+    n = 0
     for channel, owner, peer in sorted(pairs):
         rows = await load_conversation(session, channel, owner, peer)
         facts, _agg = compute_conversation(rows)
@@ -119,15 +140,25 @@ async def recompute_pairs(
                     peer_id=peer,
                     thread_end_ts=f.thread_end_ts,
                     reply_ts=f.reply_ts,
+                    reply_msg_id=f.reply_msg_id,
                     day=f.day,
                     fr_ms=f.fr_ms,
                 )
             )
-    return len(pairs)
+        n += 1
+        if n % 500 == 0:
+            await session.flush()
+            logger.info("first_reply_facts: 已重算 %d 个会话", n)
+    return n
 
 
 async def all_friend_pairs(session: AsyncSession) -> list[tuple[str, str, str]]:
-    """全部 1:1 会话(friend 行 distinct)——全量回填用。"""
+    """全部 1:1 会话(friend 行 distinct)——全量回填用。
+
+    依赖 msg_daily_stats 已回填(先 stats 后 facts):friend 会话清单取自
+    stats 表而非 message 全表 distinct——写路径钩子保证 stats 常新,代价
+    远低于扫 message。
+    """
     rows = (
         await session.execute(
             sa.select(
@@ -141,7 +172,7 @@ async def all_friend_pairs(session: AsyncSession) -> list[tuple[str, str, str]]:
 
 
 async def pairs_active_in_days(
-    session: AsyncSession, days
+    session: AsyncSession, days: Iterable[int]
 ) -> list[tuple[str, str, str]]:
     """窗口内有消息的 friend 会话(4h 滚动重算用;批次F 接调度)。"""
     rows = (

yunke-server/migrations/versions/b7c8d9e0f1a2_management_phase1_tables.py

@@ -26,6 +26,9 @@ _TABLES = [
         sa.Column("peer_id", sa.String(length=128), primary_key=True),
         sa.Column("thread_end_ts", sa.BigInteger(), primary_key=True),
         sa.Column("reply_ts", sa.BigInteger(), primary_key=True),
+        # 同毫秒 F,T,F,T 交换的 (thread_end_ts, reply_ts) 并列去重(一条回复
+        # 恰好闭合一个串,reply_msg_id 唯一定位);兼作下钻证据定位回复消息
+        sa.Column("reply_msg_id", sa.String(length=64), primary_key=True),
         sa.Column("day", sa.Integer(), nullable=False),
         sa.Column("fr_ms", sa.BigInteger(), nullable=False),
     )),

yunke-server/tests/test_first_reply_facts.py

@@ -5,6 +5,10 @@
 2. 多日未回串只计一次、归属末条日(日切会按天重复计——分母膨胀);
 3. 同源:threads_answered ≡ fact 行数(按归属日);
 4. is_send=None 不参与串判定(与 _thread_stats 一致)。
+
+评审修正补判据:
+5. 同毫秒 F,T,F,T 交换 → reply_msg_id 进 PK 去重,不撞主键(原崩溃回归);
+6. 同毫秒并列按 (ts, msg_id) 定序,结果确定不随 SQL 等键序抖动。
 """
 import datetime as dt
 from zoneinfo import ZoneInfo
@@ -30,7 +34,8 @@ def _ts(*args) -> int:
 def test_overnight_thread_recovered():
     """客户 23:00 发→员工次日 9:00 回:fr_ms≈10h,归属 D1(v2 §1.1 #1)。"""
     facts, agg = compute_conversation(
-        [(_ts(2026, 9, 1, 23, 0), False), (_ts(2026, 9, 2, 9, 0), True)]
+        [(_ts(2026, 9, 1, 23, 0), False, "f1"),
+         (_ts(2026, 9, 2, 9, 0), True, "t1")]
     )
     assert len(facts) == 1
     assert facts[0].fr_ms == 10 * 3600 * 1000
@@ -41,8 +46,8 @@ def test_overnight_thread_recovered():
 def test_multiday_open_thread_counts_once_at_end_day():
     """客户消息跨 D1/D2 无回复:1 个串归属 D2;日切口径会 D1、D2 各计 1。"""
     facts, agg = compute_conversation(
-        [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 20, 0), False),
+        [(_ts(2026, 9, 1, 10, 0), False, "f1"), (_ts(2026, 9, 1, 20, 0), False, "f2"),
-         (_ts(2026, 9, 2, 9, 0), False)]
+         (_ts(2026, 9, 2, 9, 0), False, "f3")]
     )
     assert facts == []
     assert 20260901 not in agg
@@ -52,26 +57,27 @@ def test_multiday_open_thread_counts_once_at_end_day():
 def test_reply_closes_thread_and_opens_new():
     """员工回复后客户再发=新串;串内多条客户消息取末条为 end。"""
     facts, agg = compute_conversation(
-        [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 10, 5), False),
+        [(_ts(2026, 9, 1, 10, 0), False, "f1"), (_ts(2026, 9, 1, 10, 5), False, "f2"),
-         (_ts(2026, 9, 1, 10, 30), True),          # 回复1:fr=30-10:05=25min
+         (_ts(2026, 9, 1, 10, 30), True, "t1"),          # 回复1:fr=30-10:05=25min
-         (_ts(2026, 9, 1, 11, 0), False), (_ts(2026, 9, 1, 12, 0), True)]
+         (_ts(2026, 9, 1, 11, 0), False, "f3"),
-    )
+         (_ts(2026, 9, 1, 12, 0), True, "t2")])           # 回复2:fr=12:00-11:00=60min
     assert len(facts) == 2
     assert facts[0].fr_ms == 25 * 60 * 1000
     assert agg[20260901].threads == 2 and agg[20260901].answered == 2
+    assert agg[20260901].fr_sum_ms == 85 * 60 * 1000  # 25min + 60min
 
 
 def test_unknown_direction_ignored():
     facts, _agg = compute_conversation(
-        [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 10, 10), None),
+        [(_ts(2026, 9, 1, 10, 0), False, "f1"), (_ts(2026, 9, 1, 10, 10), None, "u1"),
-         (_ts(2026, 9, 1, 10, 20), True)]
+         (_ts(2026, 9, 1, 10, 20), True, "t1")]
     )
     assert len(facts) == 1  # None 不打断串:仍视为对 10:00 串的回复
 
 
 def test_staff_message_outside_thread_is_noop():
     facts, agg = compute_conversation(
-        [(_ts(2026, 9, 1, 9, 0), True), (_ts(2026, 9, 1, 10, 0), True)]
+        [(_ts(2026, 9, 1, 9, 0), True, "t1"), (_ts(2026, 9, 1, 10, 0), True, "t2")]
     )
     assert facts == [] and agg == {}
 
@@ -84,12 +90,29 @@ def test_unsorted_rows_raise_or_documented():
     assert "升序" in m.compute_conversation.__doc__
 
 
+def test_tie_break_by_msg_id_deterministic():
+    """同毫秒 [F(m1), T(m2)]:按 (ts, msg_id) 定序 F 在前 → 1 个被回串;
+    同一 multiset 无论入参顺序,排序后结果一致(SQL 等键序未定义,不能
+    随查询抖动)。"""
+    same = _ts(2026, 9, 1, 10, 0)
+    rows = [(same, False, "m1"), (same, True, "m2")]
+    ordered = sorted(rows, key=lambda r: (r[0], r[2]))
+    assert ordered[0][2] == "m1"  # F(m1) 先于 T(m2)
+    facts, agg = compute_conversation(ordered)
+    assert len(facts) == 1 and facts[0].reply_msg_id == "m2"
+    assert agg[20260901].answered == 1
+    shuffled = [rows[1], rows[0]]  # 同一 multiset,入参顺序相反
+    facts2, agg2 = compute_conversation(
+        sorted(shuffled, key=lambda r: (r[0], r[2])))
+    assert facts2 == facts and agg2 == agg
+
+
 # ---------------------------- 落库 ----------------------------
 
 async def _seed_conversation(session, *, channel="wechat", owner="wx1",
                              peer="p1", rows):
-    for i, (ts, is_send) in enumerate(rows):
+    for ts, is_send, msg_id in rows:
-        session.add(Message(msg_id=f"{peer}-{i}", channel=channel,
+        session.add(Message(msg_id=msg_id, channel=channel,
                             owner_account_id=owner, talker_id=peer,
                             timestamp=ts, is_send=is_send, type="text",
                             text="x"))
@@ -102,23 +125,43 @@ async def _seed_conversation(session, *, channel="wechat", owner="wx1",
 async def test_recompute_pairs_persists_and_idempotent(make_session):
     s = make_session
     await _seed_conversation(
-        s, rows=[(_ts(2026, 9, 1, 23, 0), False), (_ts(2026, 9, 2, 9, 0), True)])
+        s, rows=[(_ts(2026, 9, 1, 23, 0), False, "p1-f0"),
+                 (_ts(2026, 9, 2, 9, 0), True, "p1-t1")])
     n = await recompute_pairs(s, [("wechat", "wx1", "p1")])
     await s.commit()
     assert n == 1
     rows = (await s.execute(sa.select(FirstReplyFact))).scalars().all()
     assert len(rows) == 1 and rows[0].fr_ms == 10 * 3600 * 1000
+    assert rows[0].reply_msg_id == "p1-t1"
     # 幂等:重跑不重复
     await recompute_pairs(s, [("wechat", "wx1", "p1")])
     await s.commit()
     assert len((await s.execute(sa.select(FirstReplyFact))).scalars().all()) == 1
 
 
+async def test_same_ms_ftft_two_distinct_facts(make_session):
+    """同毫秒 F,T,F,T 交换:两个样本 (thread_end_ts, reply_ts) 完全并列
+    ——reply_msg_id 进 PK 去重后恰好 2 行(评审复现的主键冲突回归)。"""
+    s = make_session
+    same = _ts(2026, 9, 1, 10, 0)
+    await _seed_conversation(s, rows=[
+        (same, False, "m1"), (same, True, "m2"),
+        (same, False, "m3"), (same, True, "m4"),
+    ])
+    n = await recompute_pairs(s, [("wechat", "wx1", "p1")])
+    await s.commit()
+    assert n == 1
+    rows = (await s.execute(sa.select(FirstReplyFact))).scalars().all()
+    assert len(rows) == 2
+    assert {r.reply_msg_id for r in rows} == {"m2", "m4"}
+
+
 async def test_recompute_pairs_replaces_stale(make_session):
     """会话追加消息后重算,旧样本被替换(delete+insert 语义)。"""
     s = make_session
     await _seed_conversation(
-        s, rows=[(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 11, 0), True)])
+        s, rows=[(_ts(2026, 9, 1, 10, 0), False, "p1-f0"),
+                 (_ts(2026, 9, 1, 11, 0), True, "p1-t1")])
     await recompute_pairs(s, [("wechat", "wx1", "p1")])
     await s.commit()
     s.add(Message(msg_id="p1-new", channel="wechat", owner_account_id="wx1",
@@ -131,6 +174,24 @@ async def test_recompute_pairs_replaces_stale(make_session):
     assert len(rows) == 1  # 新串未回,只剩旧样本
 
 
+async def test_recompute_pairs_empty_conversation(make_session):
+    """stats 行在但 message 无合格行(空会话/全被过滤):删旧行不插新——
+    delete+insert 语义对空会话同样成立(幂等清空)。"""
+    s = make_session
+    s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
+                        peer_kind="friend", peer_id="p1", day=20260901))
+    # 预置历史残留 fact,重算后应被清掉
+    s.add(FirstReplyFact(channel="wechat", owner_account_id="wx1", peer_id="p1",
+                         thread_end_ts=_ts(2026, 9, 1, 10, 0),
+                         reply_ts=_ts(2026, 9, 1, 11, 0), reply_msg_id="stale",
+                         day=20260901, fr_ms=3600 * 1000))
+    await s.commit()
+    n = await recompute_pairs(s, [("wechat", "wx1", "p1")])
+    await s.commit()
+    assert n == 1
+    assert (await s.execute(sa.select(FirstReplyFact))).scalars().all() == []
+
+
 async def test_all_friend_pairs_only_friend(make_session):
     s = make_session
     s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",

yunke-server/tests/test_management_models.py

@@ -21,7 +21,7 @@ async def test_first_reply_fact_roundtrip(make_session):
     s = make_session
     s.add(FirstReplyFact(channel="wechat", owner_account_id="wx1", peer_id="p1",
                          thread_end_ts=1_700_000_000_000,
-                         reply_ts=1_700_000_600_000,
+                         reply_ts=1_700_000_600_000, reply_msg_id="m1",
                          day=20260901, fr_ms=600_000))
     await s.commit()
     val = (await s.execute(sa.select(FirstReplyFact.fr_ms))).scalar_one()
@@ -31,7 +31,8 @@ async def test_first_reply_fact_roundtrip(make_session):
 async def test_first_reply_fact_unique_sample(make_session):
     s = make_session
     kwargs = {"channel": "wechat", "owner_account_id": "wx1", "peer_id": "p1",
-              "thread_end_ts": 1, "reply_ts": 2, "day": 20260901, "fr_ms": 1}
+              "thread_end_ts": 1, "reply_ts": 2, "reply_msg_id": "m1",
+              "day": 20260901, "fr_ms": 1}
     s.add(FirstReplyFact(**kwargs))
     await s.commit()
     s.add(FirstReplyFact(**kwargs))