From ade5e0c08512918be8f628fbf630308eb02a81f9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com>
Date: Sat, 12 Sep 2026 11:46:29 +0800
Subject: [PATCH] =?UTF-8?q?=E5=9B=9E=E5=A4=8D=20ZSY=20=E7=9A=84=E5=BA=95?=
 =?UTF-8?q?=E5=BA=A7=E6=89=A9=E5=B1=95=E7=A1=AE=E8=AE=A4=20v2=EF=BC=9A?=
 =?UTF-8?q?=E9=80=90=E8=A1=8C=E6=A0=B8=E9=AA=8C=E4=B8=A4=E5=8F=A5=E7=A1=AE?=
 =?UTF-8?q?=E8=AE=A4=EF=BC=9B=E6=8C=87=E5=87=BA=20docs/05=20=E7=AB=AF?=
 =?UTF-8?q?=E7=82=B9=E7=BC=96=E5=8F=B7=E6=92=9E=E8=BD=A6=E3=80=81=E4=B8=89?=
 =?UTF-8?q?=E4=B8=AA=E6=96=B0=E6=9D=83=E9=99=90=E6=9C=AA=E9=9A=8F=E4=BB=A3?=
 =?UTF-8?q?=E7=A0=81=E5=90=88=E5=B9=B6=EF=BC=9B=E8=A1=A5=E6=9D=83=E9=99=90?=
 =?UTF-8?q?=E7=A7=8D=E5=AD=90=E8=84=9A=E6=9C=AC?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 docs/34-ZSY底座扩展确认-回复v2.md             | 249 ++++++++++++++++++
 ...ant_customer_service_phase2_permissions.py | 210 +++++++++++++++
 2 files changed, 459 insertions(+)
 create mode 100644 docs/34-ZSY底座扩展确认-回复v2.md
 create mode 100644 tools/grant_customer_service_phase2_permissions.py

diff --git a/docs/34-ZSY底座扩展确认-回复v2.md b/docs/34-ZSY底座扩展确认-回复v2.md
new file mode 100644
index 0000000..19ded09
--- /dev/null
+++ b/docs/34-ZSY底座扩展确认-回复v2.md
@@ -0,0 +1,249 @@
+# 回复 ZSY：底座扩展确认（v2）
+
+> **致**：ZSY
+> **被回复**：`致qyqy_底座扩展确认沟通文案_v2.md`
+> **我方基线**：`qyqy_develop`（本地，领先 `origin/qyqy_develop` 6 个提交）
+> **被评审的分支**：`origin/ZSY_develop` = `9aaacc2`（PR **#7**）
+> **结论**：**你要补的两句确认，我独立核验后全部成立 —— 可以照批**；
+> 但合并前有 **3 项必须处理**（§2），其中 2 项是**功能会直接不可用**，不是风格问题。
+
+---
+
+## 0. 先说核验方法与结论
+
+我把 `origin/ZSY_develop`（`9aaacc2`）取到本地逐行读了，不是照抄你的描述：
+
+| 你的主张 | 我实测到的 | 结论 |
+|---|---|---|
+| `security.py:87-91` 是那段 `sub` 校验 | 第 87 行 `subject = claims["sub"]`，88–91 行正是 `isascii`/`isdecimal`/`len ≤ 20`/`≤ 2⁶⁴−1` | **一致** |
+| 访客 `sub` 是纯数字，`[1, 9×10¹⁸]` | 第 43 行 `str(uuid4().int % 9_000_000_000_000_000_000 + 1)`，19 位上限 | **一致** |
+| 访客分支复用同一套 `decode` 与校验 | 访客判断在第 92 行，**在校验之后**；`auth.py` 只跳过 `IdentityService.resolve()`（第 52 行），没另写入口 | **一致** |
+| `contracts.py:88` 默认 `True` | 第 88 行 `recalls_customer_memory: bool = True` | **一致** |
+| `base.py:138` 短路 | 第 138 行 `if not self.definition.recalls_customer_memory or "visitor" in context.roles:` | **一致** |
+| `customer_service.py:197` 显式 `False` | 第 197 行 `recalls_customer_memory=False` | **一致** |
+
+**`docs/33` §1.2 提的两个确认，答复如下：**
+
+1. **`sub` 是纯数字 —— 满足，这一条关掉了。**
+2. **`visitor` 分支复用了同一个 `jwt.decode` 与同一段 `sub` 校验 —— 满足，这一条也关掉了。**
+
+顺带把 72 处 `int(context.user_id)` 的结论说明白：**入口校验保证所有通过鉴权的 `sub` 都是纯十进制数**，
+所以那 69 处未防御的 `int()` 不会因访客而抛 `ValueError`，**不需要**改成统一转换函数，
+那轮跨模块收敛**不必做**。
+
+**你主动提的两个「要不要改」，我的答复都是「不用改」**：
+
+- **访客 `sub` 不必换号段**。数值撞上真实用户 ID 时，`auth.py:52` 已按 `visitor` 角色跳过
+  `IdentityService`，拿不到任何 RBAC；`data_scope="public"` 的 26 个消费点我已逐点核过，
+  对未知值一致 fail closed。而 `visitor: true` 需要私钥签名，普通用户令牌伪造不出来 —— 碰撞无实际后果。
+- **「访客不召回」保留在 `base.py` 的角色判断里，不要下放给声明位**。角色判断是**底座级安全兜底**：
+  声明位漏写一个 Agent 就会静默召回访客记忆，而角色判断不会。加上 `recall_memory` 里那道
+  `memory.customer_id != context.user_id` 的越界检查，是三层，比一层好。
+
+---
+
+## 1. 照批的部分
+
+| 项 | 依据 |
+|---|---|
+| §5 删掉的两个模块 | 我按你在 PR 描述里的口径实测：`git grep -n "knowledge_tool_service"`、`git grep -n "milvus_knowledge_adapter"`（限 `app/ tests/ tools/`）**输出均为空**。**照批** |
+| `9aaacc2` 的提交拆分 | 实测 9 文件 +189/−483，与 `docs/33` §6 记录一致。**做法请保持** |
+| C3 画像 ORM 清理 | 与我方独立核实结论一致（`EXTRA=''`、`GENERATION_EXPRESSION=''`）。`docs/00` 那处偏差已按 `docs/33` §3 登记进 `docs/08` 的「已知文档偏差」，**你不用再管** |
+| `docs/05` §8.5 | **对**：没挤掉 §8.1–§8.4，`AGENTS.md`/`docs/09`/`docs/14` 的既有引用继续成立 |
+| `docs/05` §9.7 | **不撞号**：我方 §9 只到 §9.6 |
+| 元数据不可外部注入 | 比你自己注释的更强：`build_outbox_metadata()` 用 `model_copy(update=...)` **服务端覆盖**客户端提交的 `chitchat_streak`/`clarification_round`/`session_context`，不是「不接收」而是「不采信」。加分项 |
+
+---
+
+## 2. 合并前必须处理的三项
+
+### 2.1 ⚠️ `docs/05` §19 端点编号撞车：`A034`、`A035` 各被占用两次
+
+`e85989b` 你合进了含我方登录 + RBAC 只读接口的 `qyqy_develop`，那批在 §19 已占：
+
+```
+A034 | POST /api/v1/auth/tokens
+A035 | GET  /api/v1/admin/roles
+A036 | GET  /api/v1/admin/roles/{role_code}
+A037 | GET  /api/v1/admin/roles/{role_code}/permissions
+A038 | GET  /api/v1/admin/users/{user_id}/roles
+```
+
+你在 `9aaacc2` 里又插了两行同名编号：
+
+```
+A034 | GET  /api/v1/admin/customer-profile-candidates        ← 与上面重复
+A035 | POST /api/v1/admin/customer-profile-candidates/{id}/reviews   ← 与上面重复
+```
+
+**实测**：`9aaacc2:docs/05` 里 `A034` 出现 2 次、`A035` 出现 2 次。
+
+**请改成 `A039`/`A040`**（`M003`/`M004` 不冲突，保留）。这与你 §8.5 的处置逻辑是同一件事 ——
+编号被复用比编号不够更麻烦，我们刚因两家同时占 `docs/21`、`docs/22` 让过两次号。
+
+> ⚠️ 顺带告知一个**守卫盲区**：`tools/check_authoritative_docs.py` 只校验 **`docs/` 文件名编号唯一性**
+> 与权威性声明，**不校验 §19 的端点编号**。所以合并后文档守卫会**通过**，这处重复会静默遗留 ——
+> 这也是为什么我在评审里手工核了一遍。
+
+### 2.2 ⚠️ 三个新权限不在任何环境脚本里：接口会全 403
+
+你新增的接口要三个新权限，但我实测：
+
+- **我方库**（`127.0.0.1:3306/jr`）`sys_permission` 共 38 条，
+  `memory:candidate:confirm`、`memory:candidate:review`、`handover:read` **都不存在**
+  （已有的只有 `handover:create`=9008、`memory:read:self`=9009、`memory:read:customer`=9010）；
+- 你的分支里这三个字符串**只出现在业务代码、`docs/05`、和测试里**，
+  `tools/` 与 `alembic/` 下**没有任何种子或迁移**建立它们。
+
+**后果**（合并后立即生效）：
+
+| 接口 | 需要的权限 | 现状 |
+|---|---|---|
+| `GET /api/v1/users/me/memory-candidates` | `memory:read:self` | ✅ 已有，能通 |
+| `POST /api/v1/users/me/memory-candidates/{id}/decisions` | `memory:candidate:confirm` | ❌ **403** |
+| `GET/POST /api/v1/admin/customer-profile-candidates*` | `memory:candidate:review` | ❌ **403** |
+| `GET /api/v1/admin/customer-service/handover-tickets*` | `handover:read` | ❌ **403** |
+
+fail closed 是对的（不会误放行），但**功能等于没上**。你的集成测试能过，说明**你的环境里已经手工插过** ——
+这正是 `AGENTS.md` §E 那条「`config_release` / `sys_role` / `sys_permission` / `sys_user` 是环境数据、
+不随代码合并」的**第 4 次踩坑**（前面已有三条线各踩一次）。
+
+**分工建议**：这一份由**我方**出 `tools/grant_customer_service_phase2_permissions.py`
+（id 从 **9036** 起，避开我方 9001–9010 与投顾 9020–9035），**两边各跑一次**。
+你若已在自己环境手工插过，请把实际 id 报给我，我按你的号段对齐，避免两边不一致。
+
+### 2.3 `milvus-lite` 放进了主 `dependencies`，与确认口径相反
+
+`docs/33` §4 的口径是：
+
+> 依赖新增 `milvus-lite` → **请放进 `pyproject.toml` 的 `optional-dependencies`**，
+> **不要进主 `dependencies`**。
+
+实测 `9aaacc2` 的实际改动是：
+
+```
+pyproject.toml  → dependencies 段新增 "milvus-lite>=3.2,<4"
+requirements.txt → 新增 "milvus-lite>=3.2,<4"
+```
+
+而且**你自己的接入说明**（`docs/客服Agent接入底座扩展说明_v1.md` §四 B 类表格）写的也是
+「（**可选**本地开发依赖）」—— 所以这是**放错了段**，不是理念分歧：挪到
+`[project.optional-dependencies]` 即可（`requirements.txt` 同理只留在本地开发那份里）。
+
+理由不变：它只是本地开发用，进主依赖会让生产环境多背一个包，而 `MILVUS_LOCAL_URI` 那个坑正是它引入的。
+
+---
+
+## 3. 品牌名：新增一个事实，请连同这条一起上报
+
+`docs/33` §2.4 我的口径是「需项目方拍板」。**现在给你一个补充事实，它对你有利**：
+
+`app/core/customer_service_rules.py` 第 116、129 行**早就是「奶龙基金」**：
+
+```
+116: "奶龙基金不会通过电话、短信或聊天索要您的验证码、密码，也不会要求您把钱转到指定账户。"
+129: "这件事需要人工为您办理。奶龙基金智能助手不能代办交易、修改资料、销户或受理投诉赔偿。"
+```
+
+也就是说，改之前我方底座的现状是**自相矛盾**的：转人工/防诈骗话术说「奶龙基金」，
+而 `COMPANY` 是「南方科技」。**你这一改实际是消除了一处内部不一致**，不只是换品牌。
+
+所以请在上报时把这层理由写进去（比单说「改个常量」更容易通过）。
+
+两个附带提醒：
+
+- **建议把品牌名单独一个提交**，便于项目方拍板后单独保留或回滚，不要混在功能提交里；
+- 我方 `tools/chat_console.py` 的标题栏现在也还是「南方科技」（第 48、81 行）。
+  若最终定「奶龙」，**我方自己也要改**，这条算在我账上，不占你时间。
+
+---
+
+## 4. 合并预演结果：干净，可以合
+
+我按 `docs/32-平台侧交接与联调准备.md` §5 的口径做了预演：
+
+| 检查 | 结果 |
+|---|---|
+| `git merge-tree --write-tree qyqy_develop 9aaacc2` | **无冲突**（返回单一 tree，无 conflict 段） |
+| 与你这 89 个文件的交集 | **我方领先的 6 个提交与 89 个文件零重叠**（`Compare-Object` 结果为空） |
+| PR #7 元数据（Gitea API 实测） | `state=open`、`mergeable=True`、`base=qyqy_develop`、`head=ZSY_develop`、**89 files / +7635 / −73** |
+| 迁移脚本 | **无新增**；`app/model` 的改动经我逐行核对只是别名与映射整理（`FinKnowledgeMeta` 别名、`ProfileSnapshot` 转出、`current_customer_id` 普通列、去重复声明），**未改任何表结构** → 不需要新迁移 |
+| PR 标题 | 「客服 Agent 接入底座：访客身份、画像候选、转人工工单（含底座合规清理）」—— **范围写得准**，评审时以它为准 |
+
+**合并顺序建议**：我方本地 `qyqy_develop` 领先远端 6 个提交、且与你的改动零重叠，
+所以 **先推我这 6 个、再合 PR #7**（两个方向都无冲突，但这样 PR 的 diff 更干净）。
+
+**合并后我方会跑**（`docs/32` §5）：`alembic upgrade head` → `python tools/audit_schema.py`
+→ `python tools/check_authoritative_docs.py` → `ruff` / `mypy` / `pytest` 四项门禁 → 真机联调；
+外加 §2.2 的权限种子脚本。
+
+---
+
+## 5. 结论
+
+- **你要补的两句确认：核验通过**，§1.2 与 §2.2 两条全部关掉，**不用再改代码**；
+- **你主动提的两个「要不要改」：都不用改**（访客号段、角色兜底）；
+- **合并前请改两处**：§2.1 的 `A034`/`A035` → `A039`/`A040`；§2.3 的 `milvus-lite` 挪到
+  `optional-dependencies`；
+- **由我方补一件**：§2.2 的权限种子脚本（等你报 id 号段后我对齐）；
+- **品牌名**：按你原计划上报项目方，请带上 §3 的补充事实，并单独一个提交。
+
+这四项处理完，PR #7 我这边没有别的反对意见，可以合。
+
+辛苦了。
+
+---
+
+## 6. 附：可直接复制发送的微信版
+
+> 给不读长文档的场景用；内容与上文一致，逐条对应。
+
+```
+ZSY 你好，你的 v2 我逐行核过了（把 origin/ZSY_develop 取到本地读的，不是照抄描述）。
+
+你补的两句确认：都成立，可以关掉。
+1）sub 是纯数字 —— security.py:87-91 那段校验在访客分支之前，你第 43 行的生成方式
+   落在 [1, 9e18]，19 位，全部满足；
+2）默认值 True —— contracts.py:88 实测就是 True，base.py:138 的短路也读到了。
+另外你主动提的两个"要不要改"，我的答复都是不用改：访客 sub 不用换号段（撞上真实 ID 也
+拿不到权限，auth.py:52 已跳过身份解析）；"访客不召回"也别下放到声明位，放 base.py 的
+角色判断是底座级兜底，比声明位可靠。
+
+顺带一个好消息：入口校验保证了所有 sub 都是纯数字，所以我上次担心的 69 处没防御的
+int(context.user_id) 不会被访客打炸，那轮跨模块收敛不用做了。
+
+但合并前有三件事要处理，两件是功能会直接不可用：
+
+1）docs/05 §19 端点编号撞车：A034/A035 各被占了两次（我方登录 + RBAC 只读已占 A034-A038，
+   你又插了画像候选那两行）。麻烦改成 A039/A040，M003/M004 不冲突、保留。
+   提醒一下：tools/check_authoritative_docs.py 只查文档文件名编号，不查 §19 端点号，
+   所以这个重复合并后守卫会放过去，我是手工核出来的。
+
+2）三个新权限不在任何脚本里：memory:candidate:confirm、memory:candidate:review、
+   handover:read。我实测我方库 sys_permission 38 条里都没有，你分支的 tools/ 和 alembic/
+   里也没有种子或迁移。这样合过来以后，候选确认、候选审核、转人工工单这 5 个接口会全 403
+   （fail closed 是对的，但等于没上）。你的集成测试能过，说明你环境里已经手工插过了 ——
+   这正是"环境数据不随代码合并"的第 4 次踩坑。
+   这份我出了：tools/grant_customer_service_phase2_permissions.py，id 从 9036 起
+   （避开种子 9001-9019 和投顾 9020-9035），customer 角色给 confirm、admin 给 review +
+   handover:read。你如果已经插过，把你用的 id 号段报我，我按你的对齐。
+   注意它只增不删，跑过 seed_test_rbac.py（DELETE 9001-9099）之后要重跑。
+
+3）milvus-lite 加到主 dependencies 了（pyproject.toml 和 requirements.txt 都是）。
+   按确认口径应该放 optional-dependencies，你自己的接入说明里也写的"（可选本地开发依赖）"，
+   所以是放错段了，挪一下就行。
+
+还有品牌名那条，给你一个对你有利的补充事实：app/core/customer_service_rules.py 第 116、
+129 行早就是"奶龙基金"了（转人工和防诈骗话术），而 COMPANY 是"南方科技" —— 所以改之前
+底座自己就是矛盾的，你这一改实际是消除内部不一致，不只是换品牌。上报时把这层理由写进去
+更容易过。建议单独一个提交，便于项目方拍板后单独回滚。我方 tools/chat_console.py 也还是
+南方科技，定了奶龙我们自己改。
+
+合并预演我做了：merge-tree 无冲突，你这 89 个文件与我方本地领先的 6 个提交零重叠，
+没有新迁移、app/model 的改动只是别名和映射整理、表结构没变。
+建议顺序是先推我这 6 个、再合 PR #7（都无冲突，但这样 PR 的 diff 更干净）。
+
+那三件处理完，PR #7 我这边没有别的反对意见。
+
+辛苦了。
+```
diff --git a/tools/grant_customer_service_phase2_permissions.py b/tools/grant_customer_service_phase2_permissions.py
new file mode 100644
index 0000000..22144fd
--- /dev/null
+++ b/tools/grant_customer_service_phase2_permissions.py
@@ -0,0 +1,210 @@
+"""补客服二期（画像候选 + 转人工工单）需要的三个权限码。
+
+## 为什么需要它
+
+客服二期这条线新增了 6 个接口，其中 5 个要三个**平台此前不存在的权限码**：
+
+| 接口 | 权限码 |
+|---|---|
+| `POST /api/v1/users/me/memory-candidates/{id}/decisions` | `memory:candidate:confirm` |
+| `GET /api/v1/admin/customer-profile-candidates` | `memory:candidate:review` |
+| `POST /api/v1/admin/customer-profile-candidates/{id}/reviews` | `memory:candidate:review` |
+| `GET /api/v1/admin/customer-service/handover-tickets` | `handover:read` |
+| `GET /api/v1/admin/customer-service/handover-tickets/{ticket_no}` | `handover:read` |
+
+这三个码在 **没有种子脚本、也没有迁移** 的情况下被业务代码直接引用，
+所以它们属于**环境数据**：代码合过来以后，如果库里没有这三行，上面 5 个接口一律
+`403 缺少操作权限` —— 报错看起来像"权限配错了"，实际是**权限码根本不存在**。
+（`GET /api/v1/users/me/memory-candidates` 用的是既有的 `memory:read:self`，不受影响。）
+
+这与 `AGENTS.md` §E 那条环境口径是同一类问题：
+`config_release`、`sys_role`/`sys_permission`/`sys_user`、Milvus 集合 schema、
+`advisor_product_*` **都不随代码合并**，换环境必须重放。
+
+## 授给谁 —— 依据是代码里的真实校验
+
+`app/service/authorization_service.py:11-17` 的语义是：
+
+```python
+allowed = permission in context.permissions
+if admin:
+    allowed = allowed and bool({"admin", "super_admin"}.intersection(context.roles))
+```
+
+| 权限码 | 校验方式 | 授给 | 理由 |
+|---|---|---|---|
+| `memory:candidate:confirm` | `require(..., USER_CONFIRM_PERMISSION)`，**非 admin** | `customer` | 客户确认**自己**的候选；查询按 `customer_id` 过滤，不会越权 |
+| `memory:candidate:review` | `require(..., ADMIN_REVIEW_PERMISSION, admin=True)` | `admin` | 需同时满足权限与管理员角色 |
+| `handover:read` | `require(..., "handover:read", admin=True)` | `admin` | 同上 |
+
+我方库里没有 `super_admin` 角色，`admin=True` 那一支由 `admin` 满足。
+
+## ⚠️ 与 `seed_test_rbac.py` 的冲突（与投顾那次同源）
+
+`seed_test_rbac.py` 是 **DELETE 重建**语义，它的
+`DELETE FROM sys_permission WHERE id BETWEEN 9001 AND 9099` 会**清掉本脚本建的权限**。
+所以本脚本**只增不删**，用于局部补齐；若要长期固化，请把这三个权限并进
+`seed_test_rbac.py` 的 `PERMISSIONS` 常量（投顾那 16 个也是同一个待办）。
+
+权限 id 从 **9036** 起：避开种子占用的 9001-9019，也避开投顾的 9020-9035。
+
+用法：
+
+    python tools/grant_customer_service_phase2_permissions.py --dry-run   # 只打印将写入什么
+    python tools/grant_customer_service_phase2_permissions.py
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import sys
+from datetime import UTC, datetime
+
+from sqlalchemy import text
+
+from app.infrastructure.db import SessionFactory
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(errors="replace")  # type: ignore[union-attr]
+
+#: (id, permission_code, resource, action, data_scope)
+PHASE2_PERMISSIONS: tuple[tuple[int, str, str, str, str], ...] = (
+    (9036, "memory:candidate:confirm", "memory:candidate", "confirm", "self"),
+    (9037, "memory:candidate:review", "memory:candidate", "review", "all"),
+    (9038, "handover:read", "handover", "read", "all"),
+)
+
+#: 角色 id（`seed_test_rbac.py` 建的）。
+CUSTOMER_ROLE_ID = 9001
+ADMIN_ROLE_ID = 9003
+
+#: 每个角色该拿到哪些码。
+GRANTS: tuple[tuple[int, str, tuple[str, ...]], ...] = (
+    (CUSTOMER_ROLE_ID, "customer", ("memory:candidate:confirm",)),
+    (ADMIN_ROLE_ID, "admin", ("memory:candidate:confirm", "memory:candidate:review",
+                              "handover:read")),
+)
+
+
+async def apply(*, dry_run: bool) -> int:
+    now = datetime.now(UTC).replace(tzinfo=None)
+    async with SessionFactory() as session, session.begin():
+        existing = dict(
+            (await session.execute(
+                text("SELECT permission_code, id FROM sys_permission")
+            )).all()
+        )
+        to_create = [p for p in PHASE2_PERMISSIONS if p[1] not in existing]
+        print(f"权限：库里已有 {len(existing)} 个，本次新增 {len(to_create)} 个")
+        for _, code, resource, action, scope in to_create:
+            print(f"  + {code:<32} {resource}:{action} scope={scope}")
+        if not to_create:
+            print("  （三个权限都已在库中）")
+
+        role_ids: dict[str, int] = {}
+        for role_id, role_code, _codes in GRANTS:
+            found = await session.scalar(
+                text("SELECT id FROM sys_role WHERE role_code = :code"),
+                {"code": role_code},
+            )
+            print(f"角色 {role_code:<10} {'已存在' if found else '缺失（请先跑 seed_test_rbac.py）'}")
+            if found is not None:
+                role_ids[role_code] = int(found)
+
+        if dry_run:
+            print("\n[dry-run] 未写入任何数据。")
+            return 0
+
+        for perm_id, code, resource, action, scope in to_create:
+            await session.execute(
+                text(
+                    """
+                    INSERT INTO sys_permission
+                        (id, permission_code, resource, action, data_scope, created_at, updated_at)
+                    VALUES
+                        (:id, :code, :resource, :action, :scope, :now, :now)
+                    """
+                ),
+                {"id": perm_id, "code": code, "resource": resource,
+                 "action": action, "scope": scope, "now": now},
+            )
+
+        # 重新取一遍完整映射，避免依赖本次插入的 id。
+        permission_ids = dict(
+            (await session.execute(
+                text("SELECT permission_code, id FROM sys_permission")
+            )).all()
+        )
+
+        for role_id, role_code, codes in GRANTS:
+            if role_code not in role_ids:
+                print(f"跳过 {role_code}：角色不存在")
+                continue
+            have = set(
+                (await session.scalars(
+                    text("SELECT permission_id FROM sys_role_permission WHERE role_id = :r"),
+                    {"r": role_ids[role_code]},
+                )).all()
+            )
+            added = 0
+            for code in codes:
+                perm_id = permission_ids.get(code)
+                if perm_id is None or int(perm_id) in have:
+                    continue
+                await session.execute(
+                    text(
+                        "INSERT INTO sys_role_permission (role_id, permission_id, created_at)"
+                        " VALUES (:r, :p, :now)"
+                    ),
+                    {"r": role_ids[role_code], "p": int(perm_id), "now": now},
+                )
+                added += 1
+            print(f"授权：{role_code:<10} 新增 {added} 项（共 {len(codes)} 项）")
+
+    await verify()
+    return 0
+
+
+async def verify() -> None:
+    """按权限码实测一遍，而不是只看插了几行。"""
+    async with SessionFactory() as session:
+        rows = (
+            await session.execute(
+                text(
+                    """
+                    SELECT p.permission_code, r.role_code
+                    FROM sys_permission p
+                    JOIN sys_role_permission rp ON rp.permission_id = p.id
+                    JOIN sys_role r ON r.id = rp.role_id
+                    WHERE p.permission_code IN
+                        ('memory:candidate:confirm', 'memory:candidate:review', 'handover:read')
+                    ORDER BY p.permission_code, r.role_code
+                    """
+                )
+            )
+        ).all()
+    print("\n实测绑定关系：")
+    for code, role in rows:
+        print(f"  {str(code):<32} → {role}")
+    missing = {
+        "memory:candidate:confirm", "memory:candidate:review", "handover:read",
+    } - {str(code) for code, _ in rows}
+    if missing:
+        print(f"[失败] 仍未绑定：{sorted(missing)}")
+        raise SystemExit(1)
+    print(
+        "\n客服二期接口现在应当可用了。若之后跑过 `seed_test_rbac.py`（DELETE 重建 9001-9099），"
+        "必须重跑本脚本 —— 或先把这三个权限并进那个种子的 PERMISSIONS 常量。"
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="补客服二期的画像候选与转人工工单权限")
+    parser.add_argument("--dry-run", action="store_true", help="只打印将写入什么")
+    args = parser.parse_args()
+    return asyncio.run(apply(dry_run=args.dry_run))
+
+
+if __name__ == "__main__":
+    sys.exit(main())