fix(staff-stats): 统计tab新客户/删除客户/红包昵称切 link——共享好友不再漏计

全项目同型审计结论的收尾项:统计分析 tab 按 owner 查全局单行档案表,
共享好友(档案挂别的员工名下)不计入——

- _new_customers(新客户):主体 Customer → CustomerAccountLink(add_time
  窗+首条消息兜底 JOIN 平移到 link.owner);数字变大=对齐云客按员工口径
- _del_customers(删除客户):_flag_customers 参数化 model,del 走 link
  (delete_flag per-relation);blocked 保留档案口径(is_block 未镜像进
  link,已知偏差=他人员工名下拉黑不计,注释标明)
- _red_transfer 昵称:link 优先档案兜底(同 sessions._meta 模式)

审计同时确认无需修:群侧(chat_group 复合主键天然多对多)、搜索/长语音
(按 friend 全局直查)、analytics(全局口径)、WA(与微信共用同批修复,
12,578 行 link 已回填,1,967 共享客户在册;无昵称行=WA 联系人本身没存
名字,与云客一致显示号码)。

644 测试绿(新增共享好友新客户判别测试;存量测试种子补 link 行镜像
生产写路径)。

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

docs/superpowers/plans/2026-09-08-management-batch-a-verification.md

@@ -0,0 +1,172 @@
+# 批次A Task 0 核实记录：五项核实 + 回测名单
+
+- **日期**：2026-09-09
+- **方式**：`docker exec -i yunke-mysql-1 mysql -uyunke -pyunke yunke -e "<SQL>"`（容器名以 `docker ps` 实测为准，确认为 `yunke-mysql-1`）
+- **用途**：本记录的全部结论直接约束批次A（数据底座）后续任务的实现细节；偏差处显式标注「影响 Task N」。任务编号引用 [2026-09-08-management-console-batch-a-data-foundation.md](./2026-09-08-management-console-batch-a-data-foundation.md)。
+- **数据快照范围**：`stat_daily_staff` / `msg_daily_stats` 覆盖 20251024 ~ 20260907；`phone_call.start_call_time` 覆盖 2025-10-18 ~ 2026-09-05。
+
+---
+
+## ① 员工主数据（staff 表）
+
+| 指标 | 实测 |
+|---|---|
+| staff 总数 | **41**（与计划假设 ≈41 一致） |
+| `is_on_job=1` | 41（全部在职） |
+| `wechat_account` DISTINCT user_id | 41 |
+| `whatsapp_account` DISTINCT user_id | 20 |
+| 微信幽灵账号（user_id 不在 staff） | **0** |
+| WA 幽灵账号（补充核实） | **0** |
+| `stat_daily_staff` DISTINCT user_id | 40（其 user_id 全部在 staff 内；员工 `FD685A5620184156BE75A8D73FB72F64` 无任何消息统计行） |
+
+**结论**：账号表 → staff 无幽灵映射，`known_user_ids` 过滤（v2 §1.3 #11）零损耗，不会意外丢行。两个渠道的 account_map 产出 user_id 均能落到 staff。**无偏差，Task 5/6 按计划实现即可。**
+
+---
+
+## ② phone_call 列与单位
+
+表结构关键列：`call_id`(PK)、`channel`、`start_call_time`(bigint)、`call_duration`(int)、`call_status`(int)、`incoming_call`(int)、`user_id`、`call_number`、`fetched_at`、`updated_at`。
+
+### channel 分布
+
+| channel | 行数 |
+|---|---|
+| `voice_call` | 2,504 |
+| `wa_voice_call` | 686 |
+| 合计 | **3,190**（与已知 3,190 条一致） |
+
+### call_duration 单位：**秒**（确认）
+
+| 指标 | 实测 |
+|---|---|
+| 总和 | 263,651 秒 = **73.24 小时**（对照已知约 73 小时，吻合） |
+| 均值 | 82.65 秒（对照预期均值 ≈82 秒，吻合） |
+
+### call_status × duration 交叉
+
+| call_status | 行数 | duration=0 | duration>0 | 语义判定 |
+|---|---|---|---|---|
+| 1 | 1,508 | 31 | 1,477 | **接通**（含 31 条 0 秒即挂） |
+| 2 | 1,682 | 1,682 | 0 | **未接通**（未接/拒接/取消，与方向无关） |
+
+### incoming_call 分布
+
+| incoming_call | 行数 | 语义 |
+|---|---|---|
+| 1 | 1,764 | 呼入 |
+| 0 | 1,426 | 呼出 |
+
+### start_call_time：**毫秒 epoch**（确认）
+
+13 位整数；`FROM_UNIXTIME(x/1000)` 解出 max = 2026-09-05 16:00:55（合理近期），按秒解释则超范围。min = 2025-10-18。与 `message.timestamp` 毫秒同口径。
+
+### call_number 形态（按 channel 完全可分）
+
+- `voice_call` → 微信 ID：`wxid_*` 或自定义微信号（`csf7169708`、`yuanye751380`、`Maple_1988` 等）
+- `wa_voice_call` → WA JID：`2348182388888@s.whatsapp.net` 等
+
+### 与计划默认假设的偏差（重要）
+
+1. **计划假设 #4「PhoneCall.channel 是通话类型（phone_call/voice_call）非消息渠道」不完全成立**：实测无 `phone_call`（蜂窝）值，且 `wa_voice_call` **能**从 channel 列判别 WA。计划 Task 5 注释「voice_call 行无法从列判别微信/WA」不成立。**影响 Task 5**：`_followup_exists` 可按渠道收窄——`voice_call` 只查该员工 wechat 账号的 1:1 消息、`wa_voice_call` 只查 whatsapp 账号（当前双渠道任一匹配的写法也不会错——wxid 不会撞 WA JID，仅多一次无效查询；收窄属优化非必需）。
+2. `call_duration` 单位=秒、`start_call_time`=毫秒 epoch：**与计划默认一致，Task 5 无需换算**（`day_bounds_ms` 直接用）。
+3. 接通判定：计划默认 `duration>0`。实测 `call_status=1 AND duration>0` = 1,477 行，`duration>0` 口径与之差 31 行（31 条接通但 0 秒，占 1.0%）。**影响 Task 5**：`answered=dur>0` 口径可用；若追求语义精确可改 `call_status=1`（含 0 秒接通），差异 ≤1%，实现时任选其一并在表注释写明。
+
+---
+
+## ③ customer 归属字段填充率
+
+| 指标 | 实测 |
+|---|---|
+| 总行数 | 16,339 |
+| `user_id` 填充 | 16,339（**100%**） |
+| `owner_account_id` 填充 | 16,339（**100%**） |
+| `delete_flag=1` | 53 |
+
+分渠道：wechat 7,040（owner 填充 7,040）/ whatsapp 9,299（owner 填充 9,299）。
+
+### 两条归属路径一致性（全表校验，非抽样）
+
+`customer.user_id` vs `owner_account_id → wechat_account.wx_id / whatsapp_account.uid → user_id`：
+
+| 渠道 | 校验行数 | 一致 |
+|---|---|---|
+| whatsapp | 9,299 | 9,299 |
+| wechat | 7,040 | 7,040 |
+
+**结论**：两条路径 100% 等价、零冲突。生命周期归属覆盖面完整。**影响 Task 6**：默认 `owner_account_id → account_map → user_id` 路径确认可用；`customer.user_id` 列可直接当冗余等价路径使用（如需省一次 join，实测安全）。
+
+补充观察（**影响 Task 3/6 的 peer 对齐**）：**微信侧主体是无后缀裸 ID，后缀形态是少数**。`customer.friend_id` 实测形态分布（全表）：
+
+| 渠道 | 形态 | 行数 | 占比 |
+|---|---|---|---|
+| wechat | **无后缀裸 ID**（`a_515855826`、`a053136a` 等） | 6,779 | **96.3%** |
+| wechat | `@openim` 后缀 | 261 | 3.7% |
+| whatsapp | `@s.whatsapp.net` | 9,266 | 99.6% |
+| whatsapp | `@lid` 后缀 | 33 | 0.4% |
+
+`message.talker_id` 侧同查（行数口径）：wechat 裸 ID 844,287 / `@openim` 16,873 / `@chatroom` 17,003（群）；whatsapp `@s.whatsapp.net` 72,357 / `@lid` 4,284 / `@g.us` 9,439（群）。即：微信 peer 主键以裸 ID 为准、`@openim` 为少数共存；WA 侧基本全带后缀。Task 3 实现 peer 匹配时按实测形态对齐（两边原值相等才命中），不做猜测性截断/补后缀。
+
+---
+
+## ④ 群占比（msg_daily_stats，量化混合口径偏差）
+
+| channel | peer_kind | 行数 | 消息量(send+recv) | 会话线程(cust_threads) |
+|---|---|---|---|---|
+| wechat | friend | 38,343 | 269,074 | 82,879 |
+| wechat | room | 16,234 | 609,089 | 26,220 |
+| whatsapp | friend | 11,673 | 77,994 | 14,178 |
+| whatsapp | room | 866 | 8,086 | 1,402 |
+| **合计** | | **67,116** | **964,243** | **124,679** |
+
+占比汇总：
+
+| 维度 | room 占比 | 说明 |
+|---|---|---|
+| 微信消息量 | **69.4%**（609,089/878,163） | 微信消息近七成来自群 |
+| WA 消息量 | 9.4%（8,086/86,080） | WA 几乎纯 1:1 |
+| 总消息量 | **64.0%**（617,175/964,243） | |
+| 总会话线程 | **22.2%**（27,622/124,679） | |
+
+**结论**：若现有看板首响/回复率不分 `peer_kind`，消息量维度被群污染 64%、线程维度 22%——污染主要来自微信侧（微信消息 69% 是群消息）。批次A `first_reply_fact` / `employee_metric_daily` 强制 `peer_kind='friend'` 过滤是必要的，不能沿用混合口径。
+
+---
+
+## ⑤ call_followup join 键（call_number ↔ message.talker_id，1:1 消息）
+
+| channel | 分母（call_number 非空） | 命中（talker_id 匹配且非群消息） | 命中率 |
+|---|---|---|---|
+| `voice_call` | 2,504 | 2,378 | **95.0%** |
+| `wa_voice_call`（补充核实） | 686 | 681 | **99.3%** |
+
+**结论**：双渠道命中率均远超 50% 阈值，`call_followup_count` 一期**高置信**，无需降级标记、无需在口径中注记低置信。**影响 Task 5**：24h 跟进判定按计划实现。未命中成因已精确解释（复核补充）：`voice_call` 的 126 条未命中**恰好等于** `call_number` 带 `@chatroom` 后缀的 126 条群通话（实测这 126 条对 1:1 消息全部 0 命中——join 限定 `room_id` 为空，群通话本就不该匹配）；剔除群通话后 1:1 可匹配通话命中率为 **2,378/2,378 = 100%**。`wa_voice_call` 侧后缀分布为 `@s.whatsapp.net` 676 / `@g.us`（群）9 / `@lid` 1，未命中 5 条**全部为** `@g.us` 群通话（逐条列出验证）。
+
+---
+
+## 回测员工名单
+
+选取依据：消息量/WA 量/活跃群排名 + 低样本端（见文末各候选分渠道数据）。覆盖：微信为主、WA 为主、群运营为主、高负载、低样本（前三类与高负载有合并）。
+
+| # | user_id | 名字 | 覆盖类型 | 关键数据（20251024~20260907） |
+|---|---|---|---|---|
+| 1 | `369D257EEC9A4357935FDE19DDD99CAA` | 王云青 | **高负载 + 群运营为主 + 微信为主**（合并） | 消息量 #1（162,740：微信 162,371 / WA 369）；活跃群 #1（4,036）；通话 431+9 |
+| 2 | `9C0B195A2C234C829FF507E011B4B460` | 田瑞志 | **WA 为主** | WA 消息 9,986 / 总 11,597 = 86%；WA 通话 28 |
+| 3 | `F99F6557C2904A7E8F4C5A3BD8A597A1` | 张士蓓 | **双通道混合**（WA 量 #1，群 #2） | 微信 91,936 + WA 12,529；活跃群 2,273；通话最多（471 微信 + 97 WA） |
+| 4 | `DC48798A5EA24DBB8B31A89CB858E08B` | 15639270965 | **纯微信无 WA 行**（验证 WA 缺失路径） | 微信 78,680，`stat_daily_staff` 无 whatsapp 行；通话 3 |
+| 5 | `946C08639BF84DFEBBF5EC2A607AE1FA` | 18539430257 | **低样本**（验证空数据路径） | 消息 8 条，无群、无通话、无 WA |
+
+> 名单中的 #1/#3 建议作为 Task 8「本地全量回填实测」的 2 名回测员工（数据最厚、双通道+通话+群全齐）。
+
+---
+
+## 偏差汇总（对后续任务的约束清单）
+
+| # | 偏差/确认 | 影响 |
+|---|---|---|
+| 1 | `phone_call.channel` 实测仅 `voice_call`（微信）/`wa_voice_call`（WA）两值，**无蜂窝 `phone_call` 值**；channel 可判别渠道，计划「无法从列判别微信/WA」结论修正 | Task 5（followup 可按渠道收窄，属优化非必需） |
+| 2 | `call_duration` 单位=**秒**（263,651s=73.24h≈已知 73h）；`start_call_time`=**毫秒 epoch** | Task 5（默认口径正确，无需换算） |
+| 3 | 接通判定 `dur>0` vs `call_status=1` 差 31 行（1.0%），前者少计 31 条 0 秒接通 | Task 5（任选其一，注释写明） |
+| 4 | `customer.owner_account_id` 填充 **100%**，与 `user_id` 列全表 100% 等价 | Task 6（默认 owner 路径确认；可直接用 user_id 列省 join） |
+| 5 | 微信侧 peer ID **主体为无后缀裸 ID**（customer.friend_id 6,779/7,040=96.3%；message.talker_id 行数口径 844,287 裸 vs 16,873 `@openim`），后缀形态（`@openim` 等）为少数；WA 侧基本全带后缀（`@s.whatsapp.net`/`@lid`） | Task 3/6（peer 对齐按实测形态原值匹配，勿截断/勿补后缀） |
+| 6 | 群消息占总消息量 64%（微信侧 69%） | Task 3/6（friend 过滤必须，佐证 spec 口径） |
+| 7 | call_number↔talker_id 命中率 voice_call 95.0% / wa_voice_call 99.3% | Task 5（call_followup_count 高置信，不降级） |
+| 8 | staff 41 行无幽灵账号；`stat_daily_staff` 仅 40 名员工有行 | Task 5/6（known_user_ids 过滤零损耗） |

docs/superpowers/plans/2026-09-08-yunke-management-console-implementation-v2.md

@@ -39,16 +39,17 @@
 | 6 | 周期分位数若 Python 拉全窗口 fact 行（月窗最坏 ~8 万行 ≈10MB 传输）不可取 | 改 SQL 窗口函数 `PERCENT_RANK() OVER (PARTITION BY user_id ORDER BY fr_ms)`（MySQL 8.4 / SQLite 3.25+ 双方言支持），传输降为员工数×2 行 |
 | 7 | fact 4h 增量成本 | 滚动 3 天=近 3 天活跃会话**全历史重算**（最大 friend 会话 5.2 万条 ≈1-2s × 几十活跃会话，4h 一次可接受）；一期取简单正确，预留「尾部增量」优化（历史已闭合串不变） |
 | 8 | full_sync_all.sh 步数变化 | 12→**14** 步：第 13 步 fact 全量重建、第 14 步宽表全量重建（宽表必须在 fact 之后） |
+| 9 | **通话列渠道错位**：v2 初稿把通话四列放进 (day, channel, user) 宽表，但 `phone_call.channel` 是**通话类型**（phone_call/voice_call）非消息渠道（models.py docstring 自证「calls are not tied to a messaging channel」）——channel=all 聚合时通话四列双计 | `app/db/models.py` PhoneCall 定义（批次A计划定案） | 通话拆出独立小表 `employee_call_daily`（day, user_id 主键，无渠道维），宽表删去通话四列 |
 
 ### 1.3 P3 级（补充明确/简化）
 
 | # | 项 | 结论 |
 |---|---|---|
-| 9 | 员工主数据来源未核实 | 列入 Task 0 核实（organization 路由背后是独立员工表还是账号表派生） |
-| 10 | 幽灵员工 | account_map 映射不上的 owner（离职/异常账号），宽表生成时直接**过滤**，不进管理域员工列表（评分场景不适用 analytics 的「读侧标注」策略） |
-| 11 | 功能开关降级 | 现状单 admin 账号无角色可分，v1 Task 13 的「功能开关」是过度设计；改为**上线顺序控制**（本地先行→服务器后上），RBAC 二/三期（与 v1 §15 一致） |
-| 12 | fr_ms 边界 | 按 ts 排序保证非负；超长样本（>30 天）保留在 fact，评分按异常裁剪原则处理（上游整合计划 §6.5「极端值不直接放大分数」）：>30 天样本参与分布展示，但不作为 0 分依据，单独列入超时明细 |
-| 13 | 五维分布图 | 一期只画 3 个可用维度（客户经营/服务质量/有效负载）+ 2 个「待数据」占位（行动执行/风险合规），不暗示完整五维 |
+| 10 | 员工主数据来源未核实 | 列入 Task 0 核实（默认 = `staff` 表，user_id 主键 + is_on_job，models.py 已有） |
+| 11 | 幽灵员工 | account_map 映射不上的 owner（离职/异常账号），宽表生成时直接**过滤**，不进管理域员工列表（评分场景不适用 analytics 的「读侧标注」策略） |
+| 12 | 功能开关降级 | 现状单 admin 账号无角色可分，v1 Task 13 的「功能开关」是过度设计；改为**上线顺序控制**（本地先行→服务器后上），RBAC 二/三期（与 v1 §15 一致） |
+| 13 | fr_ms 边界 | 按 ts 排序保证非负；超长样本（>30 天）保留在 fact，评分按异常裁剪原则处理（上游整合计划 §6.5「极端值不直接放大分数」）：>30 天样本参与分布展示，但不作为 0 分依据，单独列入超时明细 |
+| 14 | 五维分布图 | 一期只画 3 个可用维度（客户经营/服务质量/有效负载）+ 2 个「待数据」占位（行动执行/风险合规），不暗示完整五维 |
 
 ---
 
@@ -71,7 +72,7 @@
 
 ---
 
-## 3. 数据库设计：5 张新表
+## 3. 数据库设计：6 张新表
 
 ### 3.1 `first_reply_fact` 首响事实表（本设计新增）
 
@@ -122,12 +123,19 @@ PK: (day, channel, user_id, metric_version)
 生命周期 4 列（距今天数口径，当日持有客户状态分布，按 customer 当前归属）:
   new_customer_count, cooling_customer_count,
   dormant_customer_count, reactivated_customer_count
-通话 4 列（phone_call 按 user_id+day 直聚，user_id 已是员工 ID 无需转换）:
-  call_count, call_seconds, answered_call_count, call_followup_count
 元数据:
   data_complete, computed_at
 ```
 
+**通话小表 `employee_call_daily`**（§1.2 #9 拆出，无渠道维）：
+
+```text
+PK: (day, user_id)
+列: call_count, call_seconds, answered_call_count, call_followup_count, computed_at
+数据源: phone_call 按 user_id+start_call_time 日聚合；接通=call_duration>0；
+        followup=通话后 24h 内该员工任一账号对该客户有消息（Task 0 核实 join 键）
+```
+
 **相对 v1 的删减与更名**：
 
 - 6 列可派生不落：`reply_2h/24h_count`、`unreplied_2h/24h_count`（fact 表 fr_ms≤阈值计数 + cust_threads 派生）、`first_reply_count/first_reply_sum_ms`（≡ threads_answered/fr_sum_ms）；
@@ -170,7 +178,8 @@ message（按会话全历史，跨日精确算法）        [新 storage/first_r
       │
       └─→ employee_metric_daily             [新 storage/employee_metrics.py：
              msg 计数(msg_daily_stats,friend) + 串类/分位(fact 同源)
-             + customer 生命周期 + phone_call]
+             + customer 生命周期]
+phone_call → employee_call_daily              [通话小表，无渠道维，§1.2 #9]
               → 期聚合/四对比（上期/个人基线4周/团队中位数）
                   → 试算评分 + 异常规则      [management/]
                       → 周/月快照（不可变）  [management/snapshots]

yunke-server/app/analytics/lifecycle.py

@@ -0,0 +1,71 @@
+"""客户生命周期判定(管理域,批次A)——距今天数口径(v2 §1.1 #3 决议)。
+
+阈值取已确认默认(v1 §21,用户拍板):新客 7 天、沉睡 30 天;历史互动
+门槛 MIN_HISTORY_MSGS=5 对齐 analytics churn real 的 min_msgs=5。
+analytics 现有页(churn/客户资产)**暂不切换**到此模块——两套数字的
+差异在 metric-definitions 标注(v2 §5);后续统一时再接,避免动既有
+测试。
+
+本模块为纯叶子节点,禁止 import app.storage(防止 storage→analytics
+消费时成环)。
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+
+NEW_DAYS = 7
+DORMANT_DAYS = 30
+MIN_HISTORY_MSGS = 5
+
+STATES = ("new", "active", "cooling", "dormant", "reactivated")
+
+
+def _to_date(day: int) -> dt.date:
+    return dt.date(day // 10000, (day // 100) % 100, day % 100)
+
+
+def day_add(day: int, delta_days: int) -> int:
+    """yyyymmdd 键加减天数。"""
+    d = _to_date(day) + dt.timedelta(days=delta_days)
+    return d.year * 10000 + d.month * 100 + d.day
+
+
+def days_between(day_a: int, day_b: int) -> int:
+    """b − a 的天数(输入 yyyymmdd 键)。"""
+    return (_to_date(day_b) - _to_date(day_a)).days
+
+
+def classify(
+    first_day: int | None,
+    last_day: int | None,
+    prev_day: int | None,
+    as_of: int,
+    hist_msgs: int,
+) -> str | None:
+    """as_of 日的客户状态;输入均为互动日键(yyyymmdd,上海自然日)。
+
+    - first_day: 该客户全历史首次互动日;last_day: ≤as_of 的最近互动日;
+      prev_day: last_day 之前的一次互动日(无则 None);hist_msgs: 历史总消息数。
+    - 返回 None = 未激活(first/last 缺)或历史不足的长期静默(不计入任何列)。
+
+    判定优先级:new → reactivated → active → cooling → dormant。
+    """
+    if first_day is None or last_day is None:
+        return None
+    if first_day >= day_add(as_of, -(NEW_DAYS - 1)):
+        return "new"
+    if (
+        last_day == as_of
+        and prev_day is not None
+        and days_between(prev_day, last_day) > DORMANT_DAYS
+    ):
+        return "reactivated"
+    gap = days_between(last_day, as_of)
+    if gap < NEW_DAYS:
+        return "active"
+    if gap < DORMANT_DAYS:
+        return "cooling"
+    if hist_msgs >= MIN_HISTORY_MSGS:
+        return "dormant"
+    return None

yunke-server/app/api/routers/export.py

@@ -19,7 +19,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.deps import get_session
-from app.db.models import Customer, Message
+from app.db.models import Customer, CustomerAccountLink, Message
 
 router = APIRouter(prefix="/api/export", tags=["export"])
 
@@ -209,22 +209,40 @@ async def export_chat_html_zip(
             status_code=422, detail=f"一次最多导出 {ZIP_MAX_CONVS} 个会话"
         )
 
-    # 文件名显示名:批量取一次 Customer(备注 > 昵称 > friend_id)
-    cust_rows = (await session.execute(
+    # 文件名显示名:批量取一次(备注 > 昵称 > friend_id)。优先
+    # customer_account_link per-owner 快照——共享客户的档案行可能挂在别的
+    # 员工名下,按 owner 查档案会查空、zip 文件名回退成 wxid(2026-09-09,
+    # 同 sessions._meta 修复);link 没覆盖的回退档案。
+    link_rows = (await session.execute(
         sa.select(
-            Customer.friend_id,
-            Customer.friend_nickname,
-            Customer.friend_remark,
+            CustomerAccountLink.friend_id,
+            CustomerAccountLink.friend_nickname,
+            CustomerAccountLink.friend_remark,
         ).where(
-            Customer.channel == channel,
-            Customer.owner_account_id == owner_account_id,
-            Customer.friend_id.in_(ids),
+            CustomerAccountLink.channel == channel,
+            CustomerAccountLink.owner_account_id == owner_account_id,
+            CustomerAccountLink.friend_id.in_(ids),
         )
     )).all()
     names = {
         r.friend_id: (r.friend_remark or r.friend_nickname or r.friend_id)
-        for r in cust_rows
+        for r in link_rows
     }
+    missing = [t for t in ids if t not in names]
+    if missing:
+        cust_rows = (await session.execute(
+            sa.select(
+                Customer.friend_id,
+                Customer.friend_nickname,
+                Customer.friend_remark,
+            ).where(
+                Customer.channel == channel,
+                Customer.owner_account_id == owner_account_id,
+                Customer.friend_id.in_(missing),
+            )
+        )).all()
+        for r in cust_rows:
+            names[r.friend_id] = (r.friend_remark or r.friend_nickname or r.friend_id)
 
     buf = io.BytesIO()
     with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:

yunke-server/app/api/routers/sessions.py

@@ -22,7 +22,13 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.deps import get_session
 from app.api.schemas import SessionListOut, SessionOut
-from app.db.models import ChatGroup, Customer, Message, MsgDailyStats
+from app.db.models import (
+    ChatGroup,
+    Customer,
+    CustomerAccountLink,
+    Message,
+    MsgDailyStats,
+)
 from app.storage.summary import day_key, has_rows as _summary_has_rows
 
 router = APIRouter(prefix="/api", tags=["sessions"])
@@ -117,7 +123,13 @@ async def _summary_rows(
 
 
 async def _meta(session: AsyncSession, owner: str, channel: str, friend_ids: list[str], room_ids: list[str]):
-    """查 customer/chat_group 表合并昵称/头像;好友行额外带 delete_flag(is_del 过滤用)。
+    """查 customer_account_link(主)/customer(兜底)/chat_group 合并昵称/头像。
+
+    好友名优先读 ``customer_account_link`` 的 per-owner 快照——共享客户的
+    档案行(customer 全局单行,last-wins)可能挂在别的员工名下,按 owner 查
+    档案会查空,会话列表就回退显示 wxid(2026-09-09 用户报「有昵称却显示
+    微信号」);link 表每账号一行,回填后都有昵称。link 没覆盖的(空表过渡/
+    老数据)回退档案路径。
 
     friends 值为 ``(name, avatar, delete_flag)``;群无删除标记,groups 值仍为
     ``(name, avatar)``。
@@ -126,12 +138,29 @@ async def _meta(session: AsyncSession, owner: str, channel: str, friend_ids: lis
     if friend_ids:
         for fid, remark, nick, real, avatar, delete_flag in (await session.execute(
             sa.select(
-                Customer.friend_id, Customer.friend_remark, Customer.friend_nickname,
-                Customer.real_name, Customer.head_image_url, Customer.delete_flag,
+                CustomerAccountLink.friend_id, CustomerAccountLink.friend_remark,
+                CustomerAccountLink.friend_nickname, CustomerAccountLink.real_name,
+                CustomerAccountLink.head_image_url, CustomerAccountLink.delete_flag,
+            )
+            .where(
+                CustomerAccountLink.channel == channel,
+                CustomerAccountLink.owner_account_id == owner,
+                CustomerAccountLink.friend_id.in_(friend_ids),
             )
-            .where(Customer.channel == channel, Customer.owner_account_id == owner, Customer.friend_id.in_(friend_ids))
         )).all():
             friends[fid] = (remark or nick or real or fid, avatar, delete_flag)
+        # link 未覆盖的好友回退档案(按 owner 过滤;共享好友在此仍可能查空,
+        # 上层以 wxid 兜底显示)
+        missing = [fid for fid in friend_ids if fid not in friends]
+        if missing:
+            for fid, remark, nick, real, avatar, delete_flag in (await session.execute(
+                sa.select(
+                    Customer.friend_id, Customer.friend_remark, Customer.friend_nickname,
+                    Customer.real_name, Customer.head_image_url, Customer.delete_flag,
+                )
+                .where(Customer.channel == channel, Customer.owner_account_id == owner, Customer.friend_id.in_(missing))
+            )).all():
+                friends[fid] = (remark or nick or real or fid, avatar, delete_flag)
     groups: dict[str, tuple[str, str | None]] = {}
     if room_ids:
         for rid, name, avatar in (await session.execute(

yunke-server/app/api/routers/staff_chat_stats.py

@@ -20,7 +20,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.deps import get_session
 from app.api.schemas import StaffChatCustomerOut, StaffChatStatsOut, StaffChatStatsSummary
-from app.db.models import Customer, Message, MsgDailyStats
+from app.db.models import Customer, CustomerAccountLink, Message, MsgDailyStats
 from app.storage.summary import day_key, has_rows as _summary_has_rows
 
 router = APIRouter(prefix="/api", tags=["staff-chat-stats"])
@@ -144,22 +144,43 @@ async def _aggregate_summary(
 
 
 async def _customer_names(session: AsyncSession, owner: str, talker_ids: list[str]) -> dict[str, tuple[str, str, str]]:
+    """客户名/头像解析:优先 customer_account_link per-owner 快照,档案兜底。
+
+    共享客户的档案行(customer 全局单行,last-wins)可能挂在别的员工名下,
+    按 owner 查档案会查空、明细列表回退显示 wxid(2026-09-09,同 sessions
+    ._meta 修复)。
+    """
     if not talker_ids:
         return {}
-    stmt = sa.select(
-        Customer.friend_id,
-        Customer.friend_remark,
-        Customer.friend_nickname,
-        Customer.real_name,
-        Customer.head_image_url,
-    ).where(
-        Customer.channel == "wechat",
-        Customer.owner_account_id == owner,
-        Customer.friend_id.in_(talker_ids),
-    )
     info: dict[str, tuple[str, str, str]] = {}
-    for fid, remark, nick, real, head in (await session.execute(stmt)).all():
+    link_stmt = sa.select(
+        CustomerAccountLink.friend_id,
+        CustomerAccountLink.friend_remark,
+        CustomerAccountLink.friend_nickname,
+        CustomerAccountLink.real_name,
+        CustomerAccountLink.head_image_url,
+    ).where(
+        CustomerAccountLink.channel == "wechat",
+        CustomerAccountLink.owner_account_id == owner,
+        CustomerAccountLink.friend_id.in_(talker_ids),
+    )
+    for fid, remark, nick, real, head in (await session.execute(link_stmt)).all():
         info[fid] = (remark or nick or real or fid, real or "", head or "")
+    missing = [t for t in talker_ids if t not in info]
+    if missing:
+        stmt = sa.select(
+            Customer.friend_id,
+            Customer.friend_remark,
+            Customer.friend_nickname,
+            Customer.real_name,
+            Customer.head_image_url,
+        ).where(
+            Customer.channel == "wechat",
+            Customer.owner_account_id == owner,
+            Customer.friend_id.in_(missing),
+        )
+        for fid, remark, nick, real, head in (await session.execute(stmt)).all():
+            info[fid] = (remark or nick or real or fid, real or "", head or "")
     return info
 
 

yunke-server/app/api/routers/staff_stats.py

@@ -40,7 +40,14 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.deps import get_session
 from app.api.schemas import StaffStatsCountsOut, StaffStatsDetailOut
-from app.db.models import ChatGroup, Customer, Message, PhoneCall, WechatAccount
+from app.db.models import (
+    ChatGroup,
+    Customer,
+    CustomerAccountLink,
+    Message,
+    PhoneCall,
+    WechatAccount,
+)
 
 router = APIRouter(prefix="/api", tags=["staff-stats"])
 
@@ -96,50 +103,57 @@ async def _new_customers(
 ) -> tuple[int, list[dict[str, Any]]]:
     """新客户:add_time 窗内;add_time IS NULL 用首条消息时间兜底。
 
+    读 ``customer_account_link``(2026-09-09):customer 档案全局单行、
+    归属 last-wins,共享客户在其他员工名下会漏计;link 每账号一行,与
+    云客按员工统计的口径一致(数字会比档案口径大)。
+
     无 ``detail`` 分支——计数模式有意复用明细物化(窗内新增客户通常
     几十行,``len()`` 即可;COUNT 需 direct+fallback 双查询,不值)。
     """
-    base = [Customer.channel == "wechat", Customer.owner_account_id == wxid]
+    base = [
+        CustomerAccountLink.channel == "wechat",
+        CustomerAccountLink.owner_account_id == wxid,
+    ]
     # 兜底口径:每个 add_time 为空的客户,取其与该员工首条消息 min(timestamp)
     # (LEFT JOIN 走 ix_message_channel_owner_talker_time,HAVING 在 SQL 侧过滤窗)。
     fallback = (
         sa.select(
-            Customer.friend_id,
-            Customer.friend_nickname,
-            Customer.friend_remark,
-            Customer.head_image_url,
+            CustomerAccountLink.friend_id,
+            CustomerAccountLink.friend_nickname,
+            CustomerAccountLink.friend_remark,
+            CustomerAccountLink.head_image_url,
             sa.func.min(Message.timestamp).label("first_ts"),
         )
         .outerjoin(
             Message,
             sa.and_(
                 Message.channel == "wechat",
-                Message.owner_account_id == Customer.owner_account_id,
-                Message.talker_id == Customer.friend_id,
+                Message.owner_account_id == CustomerAccountLink.owner_account_id,
+                Message.talker_id == CustomerAccountLink.friend_id,
                 Message.timestamp.is_not(None),
             ),
         )
-        .where(*base, Customer.add_time.is_(None))
-        # GROUP BY 显式列出全部裸选列:friend_id 是 Customer 唯一 PK,其余列
-        # 函数依赖于它→分组结果不变;MySQL 8 FD 推断可容忍裸列,但显式列表
-        # 在 ONLY_FULL_GROUP_BY 下部署安全(SQLite 同样兼容)。
+        .where(*base, CustomerAccountLink.add_time.is_(None))
+        # GROUP BY 显式列出全部裸选列:friend_id 是 link 复合 PK 的尾列,
+        # (owner,friend) 已由 WHERE 钉死,其余列函数依赖于它→分组结果不变;
+        # MySQL ONLY_FULL_GROUP_BY 部署安全(SQLite 同样兼容)。
         .group_by(
-            Customer.friend_id,
-            Customer.friend_nickname,
-            Customer.friend_remark,
-            Customer.head_image_url,
+            CustomerAccountLink.friend_id,
+            CustomerAccountLink.friend_nickname,
+            CustomerAccountLink.friend_remark,
+            CustomerAccountLink.head_image_url,
         )
         .having(sa.func.min(Message.timestamp) >= start)
         .having(sa.func.min(Message.timestamp) <= end)
     )
     # 直接口径:add_time 本身在窗内。
     direct = sa.select(
-        Customer.friend_id,
-        Customer.friend_nickname,
-        Customer.friend_remark,
-        Customer.head_image_url,
-        Customer.add_time,
-    ).where(*base, Customer.add_time >= start, Customer.add_time <= end)
+        CustomerAccountLink.friend_id,
+        CustomerAccountLink.friend_nickname,
+        CustomerAccountLink.friend_remark,
+        CustomerAccountLink.head_image_url,
+        CustomerAccountLink.add_time,
+    ).where(*base, CustomerAccountLink.add_time >= start, CustomerAccountLink.add_time <= end)
 
     rows: list[dict[str, Any]] = [
         {
@@ -237,19 +251,43 @@ async def _red_transfer(
     talker_ids = [m.talker_id for m in msgs if m.talker_id]
     nick_map: dict[str, str] = {}
     if talker_ids:
+        # 昵称优先 link per-owner 快照(共享好友档案行可能挂别的员工名下,
+        # 2026-09-09 同 sessions._meta),档案兜底
         nick_map = {
             fid: nick
             for fid, nick in (
                 await session.execute(
-                    sa.select(Customer.friend_id, Customer.friend_nickname).where(
-                        Customer.channel == "wechat",
-                        Customer.owner_account_id == wxid,
-                        Customer.friend_id.in_(talker_ids),
+                    sa.select(
+                        CustomerAccountLink.friend_id,
+                        CustomerAccountLink.friend_nickname,
+                    ).where(
+                        CustomerAccountLink.channel == "wechat",
+                        CustomerAccountLink.owner_account_id == wxid,
+                        CustomerAccountLink.friend_id.in_(talker_ids),
                     )
                 )
             ).all()
             if nick
         }
+        missing = [t for t in talker_ids if t not in nick_map]
+        if missing:
+            nick_map.update(
+                {
+                    fid: nick
+                    for fid, nick in (
+                        await session.execute(
+                            sa.select(
+                                Customer.friend_id, Customer.friend_nickname
+                            ).where(
+                                Customer.channel == "wechat",
+                                Customer.owner_account_id == wxid,
+                                Customer.friend_id.in_(missing),
+                            )
+                        )
+                    ).all()
+                    if nick
+                }
+            )
     rows: list[dict[str, Any]] = []
     for m in msgs:
         raw_name = (m.raw_json or {}).get("name") if m.raw_json else None
@@ -273,15 +311,20 @@ async def _flag_customers(
     flag_cond: sa.ColumnElement,
     *,
     detail: bool,
+    model: type,
 ) -> tuple[int, list[dict[str, Any]]]:
     """del_customers / blocked_customers 共用:现存量口径,忽略时间窗。
 
+    ``model`` 由调用方决定(两者列名同形):del 走 ``CustomerAccountLink``
+    (delete_flag per-relation,共享好友不漏),blocked 走 ``Customer``
+    档案(is_block 未镜像进 link,2026-09-09)。
+
     无时间字段→明细行 time 用 updated_at 毫秒(仅展示;update 触发即变)。
     """
-    conds = [Customer.channel == "wechat", Customer.owner_account_id == wxid, flag_cond]
+    conds = [model.channel == "wechat", model.owner_account_id == wxid, flag_cond]
     count = (
         await session.execute(
-            sa.select(sa.func.count()).select_from(Customer).where(*conds)
+            sa.select(sa.func.count()).select_from(model).where(*conds)
         )
     ).scalar_one()
     if not detail:
@@ -289,13 +332,13 @@ async def _flag_customers(
     rows = (
         await session.execute(
             sa.select(
-                Customer.friend_id,
-                Customer.friend_nickname,
-                Customer.friend_remark,
-                Customer.updated_at,
+                model.friend_id,
+                model.friend_nickname,
+                model.friend_remark,
+                model.updated_at,
             )
             .where(*conds)
-            .order_by(Customer.updated_at.desc(), Customer.friend_id.asc())
+            .order_by(model.updated_at.desc(), model.friend_id.asc())
         )
     ).all()
     return count, [
@@ -311,18 +354,26 @@ async def _flag_customers(
 async def _del_customers(
     session: AsyncSession, wxid: str, start: int, end: int, detail: bool
 ) -> tuple[int, list[dict[str, Any]]]:
-    """删除客户:delete_flag=1 现存量(start/end 按口径忽略)。"""
+    """删除客户:delete_flag=1 现存量(start/end 按口径忽略)。
+
+    读 link(per-relation delete_flag,共享好友不漏计,2026-09-09)。
+    """
     return await _flag_customers(
-        session, wxid, Customer.delete_flag == 1, detail=detail
+        session, wxid, CustomerAccountLink.delete_flag == 1,
+        detail=detail, model=CustomerAccountLink,
     )
 
 
 async def _blocked_customers(
     session: AsyncSession, wxid: str, start: int, end: int, detail: bool
 ) -> tuple[int, list[dict[str, Any]]]:
-    """流失(拉黑)客户:is_block=1 现存量(start/end 按口径忽略)。"""
+    """流失(拉黑)客户:is_block=1 现存量(start/end 按口径忽略)。
+
+    保持档案口径:is_block 未镜像进 link 表,切了会漏;已知偏差=共享好友
+    在其他员工名下的拉黑不计入(待 link 补列后切换)。
+    """
     return await _flag_customers(
-        session, wxid, Customer.is_block.is_(True), detail=detail
+        session, wxid, Customer.is_block.is_(True), detail=detail, model=Customer
     )
 
 

yunke-server/app/db/models.py

@@ -999,11 +999,247 @@ class SensitiveScanState(Base):
     updated_at: Mapped[dt.datetime] = _updated_at()
 
 
+# ==========================================================================
+# 经营管理后台(2026-09-08 v2 spec,批次A)——fact/宽表/通话小表 + 三张业务表
+# ==========================================================================
+
+
+class FirstReplyFact(Base):
+    """首响事实表(管理域)——一行 = 一个被回的 1:1 客户消息串(v2 §3.1)。
+
+    跨日精确串口径:按会话**全历史**计算,fr_ms = 员工首条回复 ts − 串末条
+    客户消息 ts,day = 串末条所在上海自然日。区别于 msg_daily_stats 首响
+    三列的日切近似(跨夜串整条丢失,v2 §1.1 #1)。只写 1:1(friend)会话
+    ——群串语义不成立。由回填 CLI / 4h 滚动 / 夜间全量重算,**不挂写钩子**
+    (写钩子零改动)。用途:精确 P50/P90、2h/24h 回复率分子、下钻证据。
+    """
+
+    __tablename__ = "first_reply_fact"
+
+    channel: Mapped[str] = mapped_column(String(16), primary_key=True)
+    owner_account_id: Mapped[str] = mapped_column(String(128), primary_key=True)
+    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)
+
+    __table_args__ = (Index("ix_fr_fact_day_channel", "day", "channel"),)
+
+
+class EmployeeMetricDaily(Base):
+    """员工×日管理指标宽表(v2 §3.2)——1:1 口径,管理域专用。
+
+    列来源(v2 §1.1 #2/#5):消息计数/负载列自 msg_daily_stats
+    (peer_kind='friend' 过滤 + account_map 归并);串类三列与首响日值由
+    first_reply_facts.compute_conversation 同源输出(串末条归属日)——
+    回复率分子分母同口径。metric_version 只影响派生列(生命周期),
+    默认 'v1'。account_map 映射不上或不在 staff 的 owner(幽灵员工)
+    过滤不落表(v2 §1.3 #11)。通话在 employee_call_daily(无渠道维)。
+    """
+
+    __tablename__ = "employee_metric_daily"
+
+    day: Mapped[int] = mapped_column(Integer, primary_key=True)
+    channel: Mapped[str] = mapped_column(String(16), primary_key=True)
+    user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    metric_version: Mapped[str] = mapped_column(
+        String(16), primary_key=True, default="v1", server_default="v1"
+    )
+    # 消息计数列(1:1 口径)
+    msg_sent: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    msg_recv: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    # 负载列(active_rooms 为群负载,唯一计入宽表的群维度)
+    active_friends: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    active_rooms: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    # 串类列(与 first_reply_fact 同源,按串末条归属日)
+    cust_threads: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    threads_answered: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    fr_sum_ms: Mapped[int] = mapped_column(
+        BigInteger, nullable=False, default=0, server_default=sa.text("0")
+    )
+    # 首响日值(趋势图用;评分用周期值从 fact 表窗口函数精确计算)
+    first_reply_p50_ms: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    first_reply_p90_ms: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    # 生命周期 4 列(距今天数口径,当日持有客户状态分布,当前归属)
+    new_customer_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    cooling_customer_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    dormant_customer_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    reactivated_customer_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    # fact 是否已覆盖当日(宽表生成前置条件;False=串类列可能滞后)
+    data_complete: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False, server_default=sa.text("0")
+    )
+    computed_at: Mapped[dt.datetime] = mapped_column(
+        sa.DateTime, nullable=False, server_default=sa.func.now()
+    )
+
+
+class EmployeeCallDaily(Base):
+    """员工×日通话小表(v2 §1.2 #9)——通话无消息渠道维度,独立于宽表。
+
+    phone_call.channel 是通话类型(phone_call/voice_call)非消息渠道,
+    塞进 (day,channel,user) 宽表会在 channel=all 聚合时双计。接通 =
+    call_duration > 0;followup = 通话后 24h 内该员工任一账号对该客户
+    有消息(Task 0 已核实:1:1 通话可匹配率 100%,@chatroom 群通话除外)。
+    """
+
+    __tablename__ = "employee_call_daily"
+
+    day: Mapped[int] = mapped_column(Integer, primary_key=True)
+    user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    call_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    call_seconds: Mapped[int] = mapped_column(
+        BigInteger, nullable=False, default=0, server_default=sa.text("0")
+    )
+    answered_call_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    call_followup_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    computed_at: Mapped[dt.datetime] = mapped_column(
+        sa.DateTime, nullable=False, server_default=sa.func.now()
+    )
+
+
+class EmployeeScoreConfig(Base):
+    """评分配置版本(v1 §10.4/v2 §3.3)——config_json 存维度/指标/权重/
+    方向/阈值/样本门槛;激活前须过权重总和与阈值校验(批次B scoring 实现)。
+    MVP 无可视化编辑器,受控 CLI 种子维护。
+    JSON 列(config_json 等)修改须整列重赋值,原地修改不会提交。"""
+
+    __tablename__ = "employee_score_config"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    version: Mapped[str] = mapped_column(String(32), unique=True, nullable=False)
+    name: Mapped[str] = mapped_column(String(128), nullable=False)
+    # draft | active | retired
+    status: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="draft", server_default="draft"
+    )
+    effective_from: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    effective_to: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    config_json: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
+    created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    approved_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    created_at: Mapped[dt.datetime] = mapped_column(
+        sa.DateTime, nullable=False, server_default=sa.func.now()
+    )
+    activated_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True)
+
+
+class EmployeeScoreSnapshot(Base):
+    """周/月考核快照(v1 §8/v2 §3.3)——不可变:冻结后业务层禁改评分字段;
+    重算走 supersedes_id 替代链。同批次×员工唯一。"""
+
+    __tablename__ = "employee_score_snapshot"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    snapshot_batch_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    # week | month | custom
+    period_type: Mapped[str] = mapped_column(String(8), nullable=False)
+    period_start: Mapped[int] = mapped_column(Integer, nullable=False)
+    period_end: Mapped[int] = mapped_column(Integer, nullable=False)
+    user_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    score_config_version: Mapped[str] = mapped_column(String(32), nullable=False)
+    metric_version: Mapped[str] = mapped_column(String(16), nullable=False)
+    total_score: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    grade: Mapped[str | None] = mapped_column(String(2), nullable=True)
+    # scored | insufficient_sample | incomplete_data
+    score_status: Mapped[str] = mapped_column(
+        String(24), nullable=False, default="scored", server_default="scored"
+    )
+    dimension_scores_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
+    metric_details_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
+    data_as_of: Mapped[int] = mapped_column(Integer, nullable=False)
+    generated_at: Mapped[dt.datetime] = mapped_column(
+        sa.DateTime, nullable=False, server_default=sa.func.now()
+    )
+    frozen_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True)
+    frozen_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    supersedes_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+
+    __table_args__ = (
+        sa.UniqueConstraint("snapshot_batch_id", "user_id",
+                            name="uq_score_snapshot_batch_user"),
+        Index("ix_score_snapshot_period", "period_type", "period_start"),
+    )
+
+
+class ManagementFollowup(Base):
+    """管理辅导闭环(v1 §7.3/v2 §3.3)——来源指标→问题→管理动作→结果。"""
+
+    __tablename__ = "management_followup"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    employee_user_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    # metric | anomaly | review | manual
+    source_type: Mapped[str] = mapped_column(String(16), nullable=False)
+    source_code: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    source_period_start: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    source_period_end: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    source_value_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
+    reason: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
+    action_type: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    owner_manager_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    # open | in_progress | resolved | cancelled
+    status: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="open", server_default="open"
+    )
+    due_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True)
+    result: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
+    employee_feedback: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
+    created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    created_at: Mapped[dt.datetime] = mapped_column(
+        sa.DateTime, nullable=False, server_default=sa.func.now()
+    )
+    updated_at: Mapped[dt.datetime] = _updated_at()
+    closed_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True)
+
+    __table_args__ = (
+        Index("ix_followup_emp_status_due", "employee_user_id", "status", "due_at"),
+        Index("ix_followup_owner_status", "owner_manager_id", "status"),
+    )
+
+
 __all__ = [
     "ChatGroup",
     "Customer",
     "Department",
+    "EmployeeCallDaily",
+    "EmployeeMetricDaily",
+    "EmployeeScoreConfig",
+    "EmployeeScoreSnapshot",
+    "FirstReplyFact",
     "GroupMember",
+    "ManagementFollowup",
     "Media",
     "Message",
     "MomentsPost",

yunke-server/app/storage/first_reply_facts.py

@@ -0,0 +1,195 @@
+"""first_reply_fact 维护层(管理域,批次A)——跨日精确串算法。
+
+区别于 app.storage.summary._thread_stats 的日切近似(v2 §1.1 #1):
+按会话**全历史**计算,跨夜串(客户 23:00 → 员工次日 9:00 回)不再丢失。
+**不挂写钩子**(v2 §3.1):由回填 CLI / 4h 滚动 / 夜间全量重算,写钩子
+零改动。只处理 1:1(friend)会话——群串语义不成立。
+
+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
+
+from app.db.models import FirstReplyFact, Message, MsgDailyStats
+from app.storage.summary import day_key
+
+logger = logging.getLogger(__name__)
+
+# 一个被回串的首响样本;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")
+
+
+class _DayAcc:
+    __slots__ = ("answered", "fr_sum_ms", "threads")
+
+    def __init__(self) -> None:
+        self.threads = 0
+        self.answered = 0
+        self.fr_sum_ms = 0
+
+
+def compute_conversation(rows: list) -> tuple[list[FactSample], dict[int, DayAgg]]:
+    """跨日精确串算法。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 = 串末条所在上海自然日(跨夜串不再丢失);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, msg_id in rows:
+        if is_send is False:
+            in_thread = True
+            end_ts = int(ts)
+        elif is_send is True:
+            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, msg_id))
+                a = acc.setdefault(d, _DayAcc())
+                a.threads += 1
+                a.answered += 1
+                a.fr_sum_ms += fr
+            in_thread = False
+            end_ts = None
+        # is_send=None:跳过
+    if in_thread and end_ts is not None:  # 会话末尾未闭合串
+        a = acc.setdefault(day_key(end_ts), _DayAcc())
+        a.threads += 1
+    return facts, {d: DayAgg(a.threads, a.answered, a.fr_sum_ms)
+                   for d, a in acc.items()}
+
+
+async def load_conversation(
+    session: AsyncSession, channel: str, owner: str, peer: str
+) -> list:
+    """拉取该 1:1 会话全历史 (timestamp, is_send, msg_id),Python 侧按
+    (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, Message.msg_id
+            ).where(
+                Message.channel == channel,
+                Message.owner_account_id == owner,
+                Message.talker_id == peer,
+                sa.or_(Message.room_id.is_(None), Message.room_id == ""),
+                Message.timestamp.is_not(None),
+            )
+        )
+    ).all()
+    return sorted(rows, key=lambda r: (r[0], r[2]))
+
+
+async def recompute_pairs(
+    session: AsyncSession, pairs: list[tuple[str, str, str]]
+) -> int:
+    """重算给定 friend 会话的 fact 行:按会话 delete+insert,幂等。
+
+    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)
+        await session.execute(
+            sa.delete(FirstReplyFact).where(
+                FirstReplyFact.channel == channel,
+                FirstReplyFact.owner_account_id == owner,
+                FirstReplyFact.peer_id == peer,
+            )
+        )
+        for f in facts:
+            session.add(
+                FirstReplyFact(
+                    channel=channel,
+                    owner_account_id=owner,
+                    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,
+                )
+            )
+        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)——全量回填用。
+
+    依赖 msg_daily_stats 已回填(先 stats 后 facts):friend 会话清单取自
+    stats 表而非 message 全表 distinct——写路径钩子保证 stats 常新,代价
+    远低于扫 message。
+    """
+    rows = (
+        await session.execute(
+            sa.select(
+                MsgDailyStats.channel,
+                MsgDailyStats.owner_account_id,
+                MsgDailyStats.peer_id,
+            ).where(MsgDailyStats.peer_kind == "friend")
+        )
+    ).all()
+    return sorted({(r[0], r[1], r[2]) for r in rows})
+
+
+async def pairs_active_in_days(
+    session: AsyncSession, days: Iterable[int]
+) -> list[tuple[str, str, str]]:
+    """窗口内有消息的 friend 会话(4h 滚动重算用;批次F 接调度)。"""
+    rows = (
+        await session.execute(
+            sa.select(
+                MsgDailyStats.channel,
+                MsgDailyStats.owner_account_id,
+                MsgDailyStats.peer_id,
+            ).where(
+                MsgDailyStats.peer_kind == "friend",
+                MsgDailyStats.day.in_(list(days)),
+            )
+        )
+    ).all()
+    return sorted({(r[0], r[1], r[2]) for r in rows})
+
+
+async def recompute_all(session: AsyncSession) -> int:
+    """全量重建(回填/夜间自愈,幂等)。"""
+    return await recompute_pairs(session, await all_friend_pairs(session))

yunke-server/migrations/versions/b7c8d9e0f1a2_management_phase1_tables.py

@@ -0,0 +1,164 @@
+"""经营管理后台批次A:6 张新表(fact/宽表/通话小表/评分配置/快照/辅导)
+
+Revision ID: b7c8d9e0f1a2
+Revises: f7a8b9c0d1e2
+Create Date: 2026-09-08
+
+spec: docs/superpowers/plans/2026-09-08-yunke-management-console-implementation-v2.md §3。
+幂等:sa.inspect 先查,存在即跳过;SQLite/MySQL 通用。回填走 CLI
+(first-reply-facts / employee-metrics),migration 不扫数据。
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "b7c8d9e0f1a2"
+down_revision: str | Sequence[str] | None = "f7a8b9c0d1e2"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_TABLES = [
+    ("first_reply_fact", lambda: op.create_table(
+        "first_reply_fact",
+        sa.Column("channel", sa.String(length=16), primary_key=True),
+        sa.Column("owner_account_id", sa.String(length=128), primary_key=True),
+        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),
+    )),
+    ("employee_metric_daily", lambda: op.create_table(
+        "employee_metric_daily",
+        sa.Column("day", sa.Integer(), primary_key=True),
+        sa.Column("channel", sa.String(length=16), primary_key=True),
+        sa.Column("user_id", sa.String(length=64), primary_key=True),
+        sa.Column("metric_version", sa.String(length=16), primary_key=True,
+                  server_default="v1"),
+        sa.Column("msg_sent", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("msg_recv", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("active_friends", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("active_rooms", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("cust_threads", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("threads_answered", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("fr_sum_ms", sa.BigInteger(), nullable=False, server_default=sa.text("0")),
+        sa.Column("first_reply_p50_ms", sa.BigInteger(), nullable=True),
+        sa.Column("first_reply_p90_ms", sa.BigInteger(), nullable=True),
+        sa.Column("new_customer_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("cooling_customer_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("dormant_customer_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("reactivated_customer_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("data_complete", sa.Boolean(), nullable=False, server_default=sa.text("0")),
+        sa.Column("computed_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+    )),
+    ("employee_call_daily", lambda: op.create_table(
+        "employee_call_daily",
+        sa.Column("day", sa.Integer(), primary_key=True),
+        sa.Column("user_id", sa.String(length=64), primary_key=True),
+        sa.Column("call_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("call_seconds", sa.BigInteger(), nullable=False, server_default=sa.text("0")),
+        sa.Column("answered_call_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("call_followup_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("computed_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+    )),
+    ("employee_score_config", lambda: op.create_table(
+        "employee_score_config",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("version", sa.String(length=32), nullable=False, unique=True),
+        sa.Column("name", sa.String(length=128), nullable=False),
+        sa.Column("status", sa.String(length=16), nullable=False, server_default="draft"),
+        sa.Column("effective_from", sa.Integer(), nullable=True),
+        sa.Column("effective_to", sa.Integer(), nullable=True),
+        sa.Column("config_json", sa.JSON(), nullable=False),
+        sa.Column("created_by", sa.String(length=64), nullable=True),
+        sa.Column("approved_by", sa.String(length=64), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+        sa.Column("activated_at", sa.DateTime(), nullable=True),
+    )),
+    ("employee_score_snapshot", lambda: op.create_table(
+        "employee_score_snapshot",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("snapshot_batch_id", sa.String(length=64), nullable=False),
+        sa.Column("period_type", sa.String(length=8), nullable=False),
+        sa.Column("period_start", sa.Integer(), nullable=False),
+        sa.Column("period_end", sa.Integer(), nullable=False),
+        sa.Column("user_id", sa.String(length=64), nullable=False),
+        sa.Column("score_config_version", sa.String(length=32), nullable=False),
+        sa.Column("metric_version", sa.String(length=16), nullable=False),
+        sa.Column("total_score", sa.Integer(), nullable=True),
+        sa.Column("grade", sa.String(length=2), nullable=True),
+        sa.Column("score_status", sa.String(length=24), nullable=False, server_default="scored"),
+        sa.Column("dimension_scores_json", sa.JSON(), nullable=True),
+        sa.Column("metric_details_json", sa.JSON(), nullable=True),
+        sa.Column("data_as_of", sa.Integer(), nullable=False),
+        sa.Column("generated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+        sa.Column("frozen_at", sa.DateTime(), nullable=True),
+        sa.Column("frozen_by", sa.String(length=64), nullable=True),
+        sa.Column("supersedes_id", sa.Integer(), nullable=True),
+        sa.UniqueConstraint("snapshot_batch_id", "user_id",
+                            name="uq_score_snapshot_batch_user"),
+    )),
+    ("management_followup", lambda: op.create_table(
+        "management_followup",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("employee_user_id", sa.String(length=64), nullable=False),
+        sa.Column("source_type", sa.String(length=16), nullable=False),
+        sa.Column("source_code", sa.String(length=64), nullable=True),
+        sa.Column("source_period_start", sa.Integer(), nullable=True),
+        sa.Column("source_period_end", sa.Integer(), nullable=True),
+        sa.Column("source_value_json", sa.JSON(), nullable=True),
+        sa.Column("reason", sa.Text(), nullable=True),
+        sa.Column("action_type", sa.String(length=32), nullable=True),
+        sa.Column("owner_manager_id", sa.String(length=64), nullable=True),
+        sa.Column("status", sa.String(length=16), nullable=False, server_default="open"),
+        sa.Column("due_at", sa.DateTime(), nullable=True),
+        sa.Column("result", sa.Text(), nullable=True),
+        sa.Column("employee_feedback", sa.Text(), nullable=True),
+        sa.Column("created_by", sa.String(length=64), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+        sa.Column("closed_at", sa.DateTime(), nullable=True),
+    )),
+]
+
+_INDEXES = [
+    ("first_reply_fact", "ix_fr_fact_day_channel", ["day", "channel"]),
+    ("employee_score_snapshot", "ix_score_snapshot_period", ["period_type", "period_start"]),
+    ("management_followup", "ix_followup_emp_status_due",
+     ["employee_user_id", "status", "due_at"]),
+    ("management_followup", "ix_followup_owner_status", ["owner_manager_id", "status"]),
+]
+
+
+def _has_index(bind, table: str, index: str) -> bool:
+    insp = sa.inspect(bind)
+    if not insp.has_table(table):
+        return False
+    return index in {ix["name"] for ix in insp.get_indexes(table)}
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    insp = sa.inspect(bind)
+    for name, create in _TABLES:
+        if not insp.has_table(name):
+            create()
+    # 索引守卫独立于建表(MySQL DDL 逐句自动提交;建表成功但索引失败的
+    # 中断重跑要能补上索引)
+    for table, index, cols in _INDEXES:
+        if not _has_index(bind, table, index):
+            op.create_index(index, table, cols)
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    insp = sa.inspect(bind)
+    for table in ("management_followup", "employee_score_snapshot",
+                  "employee_score_config", "employee_call_daily",
+                  "employee_metric_daily", "first_reply_fact"):
+        if insp.has_table(table):
+            op.drop_table(table)

yunke-server/tests/conftest.py

@@ -112,3 +112,22 @@ async def make_client() -> AsyncIterator[
             await client.aclose()
             await engine.dispose()
         app.dependency_overrides.clear()
+
+
+@pytest.fixture
+async def make_session() -> AsyncIterator[AsyncSession]:
+    """独立内存 SQLite AsyncSession(不挂 API 依赖)——存储层/管理域测试用。
+
+    与 make_client 共用 StaticPool 单连接模式;每个测试独享全新 schema。
+    """
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    async with SessionLocal() as s:
+        yield s
+    await engine.dispose()

yunke-server/tests/test_api_sessions.py

@@ -5,7 +5,7 @@ import time
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.db.models import ChatGroup, Customer, Message
+from app.db.models import ChatGroup, Customer, CustomerAccountLink, Message
 
 OWNER = "wx_owner_s"
 CH = "wechat"
@@ -45,6 +45,40 @@ async def test_sessions_friend_and_group(make_client):
     assert g["name"] == "群一号" and g["today_count"] == 1
 
 
+async def test_sessions_shared_friend_name_from_link(make_client):
+    """共享好友昵称读 link 快照(2026-09-09):档案行挂在别的员工名下时,
+    会话列表此前回退显示 wxid;link 表每账号一行,回填后都有昵称。
+    """
+
+    async def seed(session: AsyncSession) -> None:
+        session.add_all([
+            # OWNER 与共享好友 t_shared 有会话
+            Message(msg_id="sh1", channel=CH, owner_account_id=OWNER, talker_id="t_shared",
+                    room_id=None, is_send=True, timestamp=_NOW, type="text", type_raw=1, text="x"),
+            # 档案行(last-wins)挂在别的员工名下——按 owner 查档案必然查空
+            Customer(friend_id="t_shared", channel=CH,
+                     owner_account_id="wx_other", friend_nickname="A**郑"),
+            # OWNER 名下的 per-owner link 快照
+            CustomerAccountLink(channel=CH, owner_account_id=OWNER, friend_id="t_shared",
+                                friend_nickname="A**郑", friend_remark=None),
+        ])
+        await session.commit()
+
+    client = await make_client(seed)
+    r = await client.get("/api/sessions", params={"owner_account_id": OWNER, "channel": CH})
+    assert r.status_code == 200
+    by = {s.get("talker_id"): s for s in r.json()["sessions"]}
+    assert by["t_shared"]["name"] == "A**郑"
+
+
+async def test_sessions_name_falls_back_to_archive_without_link(make_client):
+    """link 未覆盖的好友回退档案路径(空表过渡安全)。"""
+    client = await make_client(_seed)
+    r = await client.get("/api/sessions", params={"owner_account_id": OWNER, "channel": CH})
+    by = {s.get("talker_id"): s for s in r.json()["sessions"]}
+    assert by["t1"]["name"] == "Alice"  # 档案行在 OWNER 名下
+
+
 async def test_sessions_sorted_by_last_chat_time_desc(make_client):
     client = await make_client(_seed)
     body = (await client.get("/api/sessions", params={"owner_account_id": OWNER})).json()

yunke-server/tests/test_api_staff_stats.py

@@ -5,7 +5,14 @@ import time
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.db.models import ChatGroup, Customer, Message, PhoneCall, WechatAccount
+from app.db.models import (
+    ChatGroup,
+    Customer,
+    CustomerAccountLink,
+    Message,
+    PhoneCall,
+    WechatAccount,
+)
 
 OWNER = "wxid_o"
 UID = "U32HEX"  # 员工 userId(32位HEX)——phone_call.user_id 存的是它,不是 wxId!
@@ -39,6 +46,11 @@ async def _seed_counts(s: AsyncSession) -> None:
     # 删除/拉黑(现存量口径,忽略窗口)
     s.add(Customer(friend_id="c9", channel="wechat", owner_account_id=OWNER, delete_flag=1))
     s.add(Customer(friend_id="c8", channel="wechat", owner_account_id=OWNER, is_block=True))
+    # link 行(生产抓取经 upsert_customer 成对写;新客户/删除统计读 link)
+    s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c1", add_time=IN))
+    s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c2"))
+    s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c3", add_time=1_000))
+    s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c9", delete_flag=1))
     # 语音通话: user_id 是 32HEX 非 wxId!另 seed 一条直存 wxId 的(不计,锁死 join 口径)
     s.add(PhoneCall(call_id="pc1", channel="voice_call", user_id=UID,
                     start_call_time=IN + 4, call_number="c1"))
@@ -78,6 +90,9 @@ async def test_counts_default_window_last_7_days(make_client):
         s.add(Customer(friend_id="c_now", channel="wechat", owner_account_id=OWNER, add_time=now))
         s.add(Customer(friend_id="c_old", channel="wechat", owner_account_id=OWNER,
                        add_time=now - 8 * DAY))
+        s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c_now", add_time=now))
+        s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c_old",
+                                  add_time=now - 8 * DAY))
         s.add(PhoneCall(call_id="pc_now", channel="voice_call", user_id=UID,
                         start_call_time=now, call_number="c_now"))
         s.add(PhoneCall(call_id="pc_old", channel="voice_call", user_id=UID,
@@ -127,6 +142,12 @@ async def test_detail_new_customers_fields(make_client):
         s.add(Customer(friend_id="c2", channel="wechat", owner_account_id=OWNER, add_time=None,
                        friend_nickname="兜底客"))
         s.add(Customer(friend_id="c3", channel="wechat", owner_account_id=OWNER, add_time=1_000))
+        s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c1", add_time=IN,
+                                  friend_nickname="牛牛", friend_remark="大客户",
+                                  head_image_url="http://a/1.png"))
+        s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c2",
+                                  friend_nickname="兜底客"))
+        s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c3", add_time=1_000))
         s.add(Message(msg_id="m1", channel="wechat", owner_account_id=OWNER, talker_id="c2",
                       timestamp=IN - 5))
         await s.commit()
@@ -238,6 +259,9 @@ async def test_detail_del_and_blocked_fields(make_client):
         s.add(Customer(friend_id="c_ok", channel="wechat", owner_account_id=OWNER))
         s.add(Customer(friend_id="c_other", channel="wechat", owner_account_id="wxid_other",
                        delete_flag=1))
+        # 删除读 link:c_del 补 link 行;is_block 保持档案(c_blk 不补)
+        s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c_del",
+                                  delete_flag=1, friend_nickname="已删客"))
         s.add(Customer(friend_id="c_fine", channel="wechat", owner_account_id=OWNER,
                        delete_flag=0, is_block=False))
         await s.commit()
@@ -394,3 +418,32 @@ async def test_new_customers_fallback_first_msg_before_window(make_client):
     assert r_count.status_code == r_detail.status_code == 200
     assert r_count.json()["new_customers"] == 0
     assert r_detail.json()["rows"] == []
+
+
+async def test_new_customers_counts_shared_friend_via_link(make_client):
+    """共享好友(档案挂别的员工名下)凭 link 行计入新客户(2026-09-09 口径修复)。
+
+    旧档案口径下 c_shared 对 OWNER 不可见(add_time 在窗内也不计);切 link
+    后按 per-owner 关系计数,与云客按员工统计一致。
+    """
+
+    async def seed(s: AsyncSession) -> None:
+        s.add(WechatAccount(wx_id=OWNER, user_id=UID))
+        # 档案行(last-wins)挂在别的员工名下
+        s.add(Customer(friend_id="c_shared", channel="wechat",
+                       owner_account_id="wxid_other", add_time=IN))
+        # OWNER 名下的 link 关系
+        s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER,
+                                  friend_id="c_shared", add_time=IN))
+        await s.commit()
+
+    client = await make_client(seed)
+    r = await client.get(
+        "/api/staff-stats",
+        params={"wechat_id": OWNER, "kind": "new_customers",
+                "start": WIN_START, "end": WIN_END},
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert [row["talker_id"] for row in body["rows"]] == ["c_shared"]
+    assert len(body["rows"]) == 1

yunke-server/tests/test_first_reply_facts.py

@@ -0,0 +1,202 @@
+"""first_reply_fact 跨日精确串算法判别测试(批次A Task3)。
+
+四个核心判据(v2 §1.1 #1/#5、§9):
+1. 跨夜串必须产生样本(日切口径丢失的场景);
+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
+
+import sqlalchemy as sa
+
+from app.db.models import FirstReplyFact, Message, MsgDailyStats
+from app.storage.first_reply_facts import (
+    all_friend_pairs,
+    compute_conversation,
+    recompute_pairs,
+)
+
+_SH = ZoneInfo("Asia/Shanghai")
+
+
+def _ts(*args) -> int:
+    return int(dt.datetime(*args, tzinfo=_SH).timestamp() * 1000)
+
+
+# ---------------------------- 纯函数 ----------------------------
+
+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, "f1"),
+         (_ts(2026, 9, 2, 9, 0), True, "t1")]
+    )
+    assert len(facts) == 1
+    assert facts[0].fr_ms == 10 * 3600 * 1000
+    assert facts[0].day == 20260901
+    assert agg[20260901].answered == 1
+
+
+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, "f1"), (_ts(2026, 9, 1, 20, 0), False, "f2"),
+         (_ts(2026, 9, 2, 9, 0), False, "f3")]
+    )
+    assert facts == []
+    assert 20260901 not in agg
+    assert agg[20260902].threads == 1 and agg[20260902].answered == 0
+
+
+def test_reply_closes_thread_and_opens_new():
+    """员工回复后客户再发=新串;串内多条客户消息取末条为 end。"""
+    facts, agg = compute_conversation(
+        [(_ts(2026, 9, 1, 10, 0), False, "f1"), (_ts(2026, 9, 1, 10, 5), False, "f2"),
+         (_ts(2026, 9, 1, 10, 30), True, "t1"),          # 回复1:fr=30-10:05=25min
+         (_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, "f1"), (_ts(2026, 9, 1, 10, 10), None, "u1"),
+         (_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, "t1"), (_ts(2026, 9, 1, 10, 0), True, "t2")]
+    )
+    assert facts == [] and agg == {}
+
+
+def test_unsorted_rows_raise_or_documented():
+    """调用方契约:rows 须升序。乱序输入行为未定义——文档断言 + 防御排序
+    在 load_conversation 里做(与 summary.py 惯例一致)。"""
+    # 纯函数不做防御(性能);落库路径负责排序。此处仅锁定契约注释。
+    import app.storage.first_reply_facts as m
+    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 ts, is_send, msg_id in rows:
+        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"))
+    session.add(MsgDailyStats(channel=channel, owner_account_id=owner,
+                              peer_kind="friend", peer_id=peer,
+                              day=20260901, send_cnt=1, recv_cnt=1))
+    await session.commit()
+
+
+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, "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, "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",
+                  talker_id="p1", timestamp=_ts(2026, 9, 1, 12, 0),
+                  is_send=False, type="text", text="y"))
+    await s.commit()
+    await recompute_pairs(s, [("wechat", "wx1", "p1")])
+    await s.commit()
+    rows = (await s.execute(sa.select(FirstReplyFact))).scalars().all()
+    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",
+                        peer_kind="room", peer_id="r1", day=20260901))
+    s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
+                        peer_kind="friend", peer_id="p1", day=20260901))
+    await s.commit()
+    assert await all_friend_pairs(s) == [("wechat", "wx1", "p1")]

yunke-server/tests/test_lifecycle.py

@@ -0,0 +1,72 @@
+"""客户生命周期判定纯函数测试(批次A Task4)。
+
+口径(v2 §1.1 #3 + v1 §21 默认,已用户确认):新客 7 天、沉睡 30 天、
+历史≥5 条才算沉睡(对齐 churn real min_msgs=5)。analytics 现有页
+(churn/客户资产)暂不切换(口径差异 v2 §5 标注),不动 636 既有测试。
+"""
+
+from app.analytics.lifecycle import DORMANT_DAYS, MIN_HISTORY_MSGS, NEW_DAYS, classify, day_add
+
+
+def test_constants_locked_to_confirmed_defaults():
+    assert NEW_DAYS == 7 and DORMANT_DAYS == 30 and MIN_HISTORY_MSGS == 5
+
+
+def test_new_customer():
+    # 首次沟通在 7 天窗口内
+    assert (
+        classify(first_day=20260901, last_day=20260903, prev_day=None, as_of=20260905, hist_msgs=2)
+        == "new"
+    )
+    # 窗口边界:as_of-6=20260830 含当天共 7 天 → new;再早一天 → 非 new
+    assert classify(20260830, 20260830, None, 20260905, 100) == "new"
+    assert classify(20260829, 20260829, None, 20260905, 100) == "cooling"
+
+
+def test_reactivated_requires_gap_over_30d():
+    # 当日回流且前次间隔>30 天
+    assert (
+        classify(
+            first_day=20260101, last_day=20260905, prev_day=20260801, as_of=20260905, hist_msgs=50
+        )
+        == "reactivated"
+    )
+    # 间隔仅 9 天 → 只是 active
+    assert (
+        classify(
+            first_day=20260101, last_day=20260905, prev_day=20260827, as_of=20260905, hist_msgs=50
+        )
+        == "active"
+    )
+    # 间隔恰好 30 天 → 严格 >30 不算回流,落回 active
+    assert (
+        classify(
+            first_day=20260101, last_day=20260905, prev_day=20260806, as_of=20260905, hist_msgs=50
+        )
+        == "active"
+    )
+    # 回流必须当日互动:前次间隔 31 天(>30)但 last_day=as_of-1 → 不算回流,按 gap=1 落 active
+    assert classify(20260101, 20260904, 20260804, 20260905, 50) == "active"
+
+
+def test_active_cooling_dormant_boundaries():
+    f, m = 20260101, 100
+    assert classify(f, 20260830, None, 20260905, m) == "active"  # 6 天
+    assert classify(f, 20260829, None, 20260905, m) == "cooling"  # 7 天
+    assert classify(f, 20260807, None, 20260905, m) == "cooling"  # 29 天
+    assert classify(f, 20260806, None, 20260905, m) == "dormant"  # 30 天
+
+
+def test_dormant_requires_history():
+    # ≥30 天无互动但历史<5 条:不计入任何状态(None)
+    assert classify(20260101, 20260806, None, 20260905, 4) is None
+    assert classify(20260101, 20260806, None, 20260905, 5) == "dormant"
+
+
+def test_no_interaction_returns_none():
+    assert classify(None, None, None, 20260905, 0) is None
+
+
+def test_day_add():
+    assert day_add(20260831, 1) == 20260901
+    assert day_add(20260901, -7) == 20260825

yunke-server/tests/test_management_models.py

@@ -0,0 +1,139 @@
+"""管理域 6 张新表的模型冒烟测试(批次A Task1)。
+
+只验证:建表可用、主键/唯一约束生效、默认值正确。业务逻辑在
+first_reply_facts / employee_metrics 的测试里。
+"""
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.exc import IntegrityError
+
+from app.db.models import (
+    EmployeeCallDaily,
+    EmployeeMetricDaily,
+    EmployeeScoreConfig,
+    EmployeeScoreSnapshot,
+    FirstReplyFact,
+    ManagementFollowup,
+)
+
+
+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_msg_id="m1",
+                         day=20260901, fr_ms=600_000))
+    await s.commit()
+    val = (await s.execute(sa.select(FirstReplyFact.fr_ms))).scalar_one()
+    assert val == 600_000
+
+
+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, "reply_msg_id": "m1",
+              "day": 20260901, "fr_ms": 1}
+    s.add(FirstReplyFact(**kwargs))
+    await s.commit()
+    s.add(FirstReplyFact(**kwargs))
+    with pytest.raises(IntegrityError):
+        await s.commit()
+
+
+async def test_metric_daily_versioning_coexists(make_session):
+    s = make_session
+    for v in ("v1", "v2"):
+        s.add(EmployeeMetricDaily(day=20260901, channel="wechat",
+                                  user_id="u1", metric_version=v))
+    await s.commit()
+    n = (await s.execute(
+        sa.select(sa.func.count()).select_from(EmployeeMetricDaily))).scalar_one()
+    assert n == 2
+
+
+async def test_metric_daily_defaults_zero_not_none(make_session):
+    s = make_session
+    s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1"))
+    await s.commit()
+    row = (await s.execute(sa.select(EmployeeMetricDaily))).scalar_one()
+    assert row.msg_sent == 0 and row.cust_threads == 0 and row.fr_sum_ms == 0
+    assert row.metric_version == "v1"
+    assert row.data_complete is False or row.data_complete == 0  # SQLite 0/1
+
+
+async def test_metric_daily_pk_duplicate_rejected(make_session):
+    """同 (day, channel, user_id, metric_version) 复合主键防重。
+
+    Task 6 delete+insert 幂等重算的安全网:版本内重复落表直接 IntegrityError。
+    """
+    s = make_session
+    s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1"))
+    await s.commit()
+    s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1"))
+    with pytest.raises(IntegrityError):
+        await s.commit()
+
+
+async def test_call_daily_roundtrip(make_session):
+    s = make_session
+    s.add(EmployeeCallDaily(day=20260901, user_id="u1", call_count=2,
+                            call_seconds=164, answered_call_count=1,
+                            call_followup_count=1))
+    await s.commit()
+    row = (await s.execute(sa.select(EmployeeCallDaily))).scalar_one()
+    assert row.call_seconds == 164
+
+
+async def test_call_daily_pk_duplicate_rejected(make_session):
+    s = make_session
+    s.add(EmployeeCallDaily(day=20260901, user_id="u1"))
+    await s.commit()
+    s.add(EmployeeCallDaily(day=20260901, user_id="u1"))
+    with pytest.raises(IntegrityError):
+        await s.commit()
+
+
+async def test_score_config_version_unique(make_session):
+    s = make_session
+    s.add(EmployeeScoreConfig(version="v1.0", name="一期试算",
+                              config_json={"dimensions": {}}))
+    await s.commit()
+    s.add(EmployeeScoreConfig(version="v1.0", name="重复",
+                              config_json={}))
+    with pytest.raises(IntegrityError):
+        await s.commit()
+
+
+async def test_score_config_config_json_not_null(make_session):
+    """config_json NOT NULL——注解与 DDL 一致(修复:去掉 | None)。
+
+    注意 JSON 列语义:Python None 经 JSON 序列化变成字符串 'null' 而非
+    SQL NULL,不会违反约束;测 NOT NULL 须用 sa.null() 显式发 SQL NULL。
+    """
+    s = make_session
+    s.add(EmployeeScoreConfig(version="v1.0", name="空配置",
+                              config_json=sa.null()))
+    with pytest.raises(IntegrityError):
+        await s.commit()
+
+
+async def test_snapshot_batch_user_unique(make_session):
+    s = make_session
+    common = {"snapshot_batch_id": "b1", "period_type": "week",
+              "period_start": 20260901, "period_end": 20260907,
+              "score_config_version": "v1.0", "metric_version": "v1",
+              "data_as_of": 20260907}
+    s.add(EmployeeScoreSnapshot(user_id="u1", **common))
+    await s.commit()
+    s.add(EmployeeScoreSnapshot(user_id="u1", **common))
+    with pytest.raises(IntegrityError):
+        await s.commit()
+
+
+async def test_followup_defaults(make_session):
+    s = make_session
+    s.add(ManagementFollowup(employee_user_id="u1", source_type="manual"))
+    await s.commit()
+    row = (await s.execute(sa.select(ManagementFollowup))).scalar_one()
+    assert row.status == "open"
+    assert row.created_at is not None