feat: 统一登录门户（按角色分流五套工作台）+ 补齐权限覆盖缺口

- tools/portal.py：客户/风控/运营/管理员/投顾五个工作台，走真实登录与真实接口
- 新增 6 个此前从未建过的权限码（promotion:* 4 个 / financial:nl2sql:read / probe:read），
  它们让推广材料与金融 NL2SQL 两条线对所有角色都是 403
- 投顾权限 10 → 25 项：补投资目标创建确认、agent:run、行情、知识检索、客户画像、推广材料
- 运营补 financial:nl2sql:read；create_test_user.py 不再硬编码角色 id（改按 role_code 查库）
- portal 的写请求补 Idempotency-Key 头（漏了会被 AGENT_INPUT_INVALID 拒）
- 新增 tools/check_permission_coverage.py 做权限对账

tools/create_test_user.py

@@ -45,15 +45,14 @@ from app.service.identity_service import IdentityService  # noqa: E402
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(errors="replace")  # type: ignore[union-attr]
 
-#: 库里现成的角色。新用户复用它们。
-#: 要引入**新角色**得同时定义它的权限集合（`sys_role_permission`）——
-#: `advisor` 就是由 `tools/grant_advisor_role.py` 建立的。
-ROLE_IDS: dict[str, int] = {
-    "customer": 9001,
-    "risk_operator": 9002,
-    "admin": 9003,
-    "advisor": 9004,
-}
+#: 可创建的角色。**不再硬编码角色 id** —— 角色 id 是环境数据
+#: （`operator` 就是自增出来的 `1986594485028610`，各环境不同），硬编码会让脚本换个
+#: 环境就失效。现在运行时按 `role_code` 查库。
+#: 新角色得先有权限集合（`sys_role_permission`）：`advisor` 由
+#: `tools/grant_advisor_role.py` 建，`operator` 由 `tools/grant_operator_role.py` 补。
+KNOWN_ROLES: tuple[str, ...] = (
+    "customer", "risk_operator", "admin", "advisor", "operator",
+)
 
 #: 角色 → `sys_user.user_type`。注意这是 `user_type`，与 `employee_role` 不是一回事。
 ROLE_USER_TYPE: dict[str, str] = {
@@ -61,6 +60,7 @@ ROLE_USER_TYPE: dict[str, str] = {
     "risk_operator": "employee",
     "admin": "employee",
     "advisor": "employee",
+    "operator": "employee",
 }
 
 #: 客户的开户状态。风控扫描等链路会读它，写成 `closed` 会让部分规则不成立。
@@ -107,18 +107,21 @@ async def upsert_user(
     *, user_id: int, username: str, role: str, password: str
 ) -> int:
     """建/更新账号并绑定角色，最后验证权限能解析出来。"""
-    role_id = ROLE_IDS[role]
     user_type = ROLE_USER_TYPE[role]
     now = datetime.now(UTC).replace(tzinfo=None)
     assigned_at = now - timedelta(seconds=ASSIGN_BACKDATE_SECONDS)
 
     async with SessionFactory() as session, session.begin():
-        role_exists = await session.scalar(
-            text("SELECT id FROM sys_role WHERE id = :role_id"), {"role_id": role_id}
+        role_id = await session.scalar(
+            text("SELECT id FROM sys_role WHERE role_code = :code"), {"code": role}
         )
-        if role_exists is None:
-            print(f"[失败] 角色 {role}（id={role_id}）不存在，先跑 tools/seed_test_rbac.py")
+        if role_id is None:
+            available = (
+                await session.scalars(text("SELECT role_code FROM sys_role ORDER BY id"))
+            ).all()
+            print(f"[失败] 库里没有角色 {role}。现有角色：{list(available)}")
             return 1
+        role_id = int(role_id)
 
         # 覆盖语义：同一个 id 重跑不会堆出第二行。
         await session.execute(
@@ -192,7 +195,7 @@ async def main() -> int:
     parser.add_argument("--list", action="store_true", help="列出所有账号与角色")
     parser.add_argument("--id", type=int, help="用户 id（9001-9003 已被演示账号占用）")
     parser.add_argument("--username", help="登录用户名")
-    parser.add_argument("--role", choices=sorted(ROLE_IDS), help="角色")
+    parser.add_argument("--role", choices=KNOWN_ROLES, help="角色（库里需已存在该角色）")
     parser.add_argument("--password", help="登录密码（仅限演示环境）")
     args = parser.parse_args()