登录接口配套：加人工具 + 给组员的转交文档（docs/29）

- tools/create_test_user.py：一条命令建"可登录的测试账号"（用户 + 角色 + bcrypt 密码），
  并用 IdentityService.resolve 打印**真实解析结果**。sys_user / sys_user_role 没有 ORM
  模型、全靠裸 SQL，手写容易漏必填字段；更要紧的是 assigned_at 那个静默陷阱（见下）。
  重复执行同一 --id 是覆盖语义，改角色也用它。
- tools/set_user_password.py：hash_password 改从 auth_service 取，消除第二份实现。
- app/service/auth_service.py：新增 hash_password，与 verify_password 放在一起，
  让"写密码"和"校验密码"永远同一套算法。
- docs/29-Agent组员登录接口使用说明.md：给组员的转交文档（接口契约、加人步骤、
  前端接入示例、常见问题、当前边界）。

文档里专门写清三条最容易踩的：

1. 登录用 username 而不是用户 id —— 演示账号是 cust_t / risk_t / admin_t，
   不是 9001/9002/9003。这条不写明，联调时一定有人按 id 试。
2. sys_user_role.assigned_at 的 DATETIME(0) 毫秒舍入陷阱：落在未来会让账号
   "登录成功但 roles=()"，**不报错**。create_test_user 统一往前留 5 秒。
3. roles / data_scope 只用于前端分流界面，不是权限凭证 —— 鉴权每次请求查库解析，
   所以权限变更立即生效，前端也不该拿它们做安全判断。

另：create_test_user 的 ON DUPLICATE KEY UPDATE 用 MySQL 8.0.19+ 的 `AS new` 别名语法，
避开已弃用的 VALUES()（实测本机 8.0.27 会打弃用警告）。

验证：文档守卫 38 份无编号冲突 / ruff 干净 / mypy 183 文件 0 错 /
登录集成测试 10 passed。

app/service/auth_service.py

@@ -103,6 +103,16 @@ def issue_access_token(user_id: int) -> tuple[str, int]:
     return token, expires_in
 
 
+def hash_password(password: str) -> str:
+    """生成 bcrypt 哈希（成本因子用库默认值）。
+
+    与 `verify_password` 放在一起，是为了让"写密码"和"校验密码"永远用同一套算法 ——
+    两个工具脚本（`set_user_password.py` / `create_test_user.py`）都从这里取，
+    避免第三次复制粘贴出不一致的实现。
+    """
+    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
+
+
 def verify_password(password: str, stored_hash: str | None) -> bool:
     """常数时间的密码校验；任何异常都当校验失败。
 

docs/29-Agent组员登录接口使用说明.md

@@ -0,0 +1,248 @@
+# 登录接口使用说明（给 Agent 组员）
+
+> **面向**：需要联调登录 / 需要造测试账号的人
+> **接口**：`POST /api/v1/auth/tokens`（`docs/05` §19 的 **A034**）
+> **归属**：认证是**平台级能力**，由平台侧维护。业务分支**不要**自己加登录路由
+> （会绕开公共鉴权，违反 `AGENTS.md` 规则 7）。
+> **日期**：2026-09-11
+
+---
+
+## 1. 一句话
+
+用 **`username` + `password`** 换一个 **Bearer 令牌**，之后所有接口带
+`Authorization: Bearer <token>` 即可。**令牌里只有用户 id**，你的角色、权限、能看多少数据
+全部由服务端按库里的 RBAC 实时解析——所以业务代码**永远只从 `RequestContext` 取身份**，
+不要自己解析令牌，也不要硬编码 `user_id`。
+
+---
+
+## 2. ⚠️ 先记住：登录用的是 `username`，不是用户 id
+
+演示账号的用户名**不是** `9001/9002/9003`：
+
+| 用户 id | **username（登录用这个）** | 密码 | 角色 | 进去该看到什么 |
+|---|---|---|---|---|
+| 9001 | **`cust_t`** | `123456` | `customer` | 客户界面：自己的会话、适当性、知识问答 |
+| 9002 | **`risk_t`** | `666666` | `risk_operator` | 风控界面：预警队列、证据、日报 |
+| 9003 | **`admin_t`** | `88888888` | `admin` | 管理界面：配置发布、模型端点、审计 |
+
+> 密码由 `tools/set_user_password.py` 设置。**这三种弱口令仅用于演示**，
+> 上线前必须全部更换。
+
+---
+
+## 3. 接口契约
+
+### 请求
+
+```http
+POST /api/v1/auth/tokens
+Content-Type: application/json
+
+{"username": "cust_t", "password": "123456"}
+```
+
+- **不需要** `Authorization` 头（本来就没有令牌）。
+- 请求体**只接受这两个字段**（`extra="forbid"`）。多传 `roles`、`user_id` 之类会 **422** ——
+  这是有意的，身份只能由服务端解析，不能由调用方声明。
+- 建议带 `X-Trace-ID`（便于和服务端日志对齐；不带也能用）。
+
+### 成功响应（`200`，`docs/05` §3.3 统一信封）
+
+```json
+{
+  "data": {
+    "access_token": "eyJhbGciOiJSUzI1NiIs...",
+    "token_type": "Bearer",
+    "expires_in": 1800,
+    "user_id": "9001",
+    "roles": ["customer"],
+    "data_scope": "self"
+  },
+  "meta": {"trace_id": "..."}
+}
+```
+
+| 字段 | 用途 |
+|---|---|
+| `access_token` | 后续所有请求的 `Authorization: Bearer <它>` |
+| `expires_in` | 有效期（秒），当前 **1800**（30 分钟） |
+| `roles` | **前端据此决定进哪个界面** |
+| `data_scope` | 该身份能看的数据范围（`self` / `own_customers` / `all`） |
+
+> ⚠️ `roles` 与 `data_scope` 是**给前端做界面分流用的**，不是权限凭证。
+> 真正的鉴权每次请求都由服务端查库解析，所以**权限被改后立刻生效**，
+> 不用等令牌过期、也不要用它们在前端做安全判断。
+
+### 失败响应
+
+| 场景 | 状态码 | 错误码 |
+|---|---|---|
+| 用户名不存在 / 密码错 / 账号停用 / 从没设过密码 | **401** | `AUTHENTICATION_REQUIRED` |
+| 缺少字段、多传字段 | **422** | `AGENT_INPUT_INVALID` |
+| 登录尝试过于频繁（60 秒 10 次） | **429** | `RATE_LIMITED`（带 `Retry-After`） |
+
+**401 的消息对所有失败原因都一样**（"用户名或密码不正确"）。这是有意设计——否则这个接口
+就成了账号枚举器。**前端不要试图从 401 的 message 里区分原因**，统一提示"账号或密码错误"。
+
+```json
+{"error": {"code": "AUTHENTICATION_REQUIRED", "message": "用户名或密码不正确",
+           "retryable": false, "field_errors": []},
+ "meta": {"trace_id": "..."}}
+```
+
+---
+
+## 4. 怎么加一个测试人员
+
+### 一条命令
+
+```powershell
+# 先看现在有哪些账号
+python tools/create_test_user.py --list
+
+# 建一个客户账号
+python tools/create_test_user.py --id 9010 --username test_cust --role customer --password abc12345
+
+# 建一个风控专员账号
+python tools/create_test_user.py --id 9011 --username test_risk --role risk_operator --password abc12345
+```
+
+**角色只有三个可选值**：`customer` / `risk_operator` / `admin`（`sys_role` 里现成的三个）。
+要引入**新角色**得同时定义它的权限集合（`sys_role_permission`），那超出这个脚本的范围，
+找平台侧。
+
+### 脚本会做三件事，并**验证第四件**
+
+1. 写 `sys_user`（含 bcrypt 密码哈希）；
+2. 绑 `sys_user_role`；
+3. 打印**真实解析结果**：
+
+```
+[OK] id=9010 username=test_cust role=customer 已写入
+     解析结果：roles=('customer',) data_scope=self
+     权限 10 项
+[OK] 可以用它登录：{"username": "test_cust", "password": "<你刚设的>"}
+```
+
+4. 第 3 步的验证很关键——它调的是 `IdentityService.resolve`，也就是请求进来时走的**同一条
+   链路**。**不要只看"插入成功"**：`sys_user_role` 有个静默陷阱（下节）。
+
+### ⚠️ 一个静默陷阱（脚本已处理，但你手写 SQL 时要当心）
+
+`MySQL` 的 `DATETIME(0)` 会把微秒**四舍五入到秒**。如果 `sys_user_role.assigned_at`
+用"当前时间"写入、进位后落在**未来**，而授权校验是 `assigned_at <= now`，那么：
+
+> 账号建好了、密码也对、**但一个角色都拿不到** —— 表现为 `roles=()`，
+> 登录照样成功，**不报错**。
+
+`tools/create_test_user.py` 统一把 `assigned_at` **往前留 5 秒**避开这个窗口。
+若你手写 SQL 或另写脚本，请照做。`tools/seed_test_rbac.py` 里也有这条注释。
+
+### 重复执行同一个 `--id`
+
+是**覆盖**语义（更新用户名 / 密码 / 角色），不会堆出重复行。改角色也用它：
+
+```powershell
+python tools/create_test_user.py --id 9010 --username test_cust --role admin --password abc12345
+```
+
+### 单独改密码
+
+```powershell
+python tools/set_user_password.py --list                              # 看现状
+python tools/set_user_password.py --user 9010 --password newpass123   # 改一个
+python tools/set_user_password.py                                     # 按演示规则设置 9001/9002/9003
+```
+
+---
+
+## 5. 前端怎么接
+
+### 登录并保存令牌
+
+```javascript
+async function login(username, password) {
+  const res = await fetch("/api/v1/auth/tokens", {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ username, password }),
+  });
+  if (res.status === 401) throw new Error("账号或密码错误");   // 不要区分原因
+  if (res.status === 429) throw new Error("尝试过于频繁，请稍后再试");
+  if (!res.ok) throw new Error("登录失败");
+  const { data } = await res.json();
+  localStorage.setItem("token", data.access_token);            // 演示够用；
+  return data;                                                 // 生产建议 httpOnly Cookie
+}
+
+// 按角色分流界面
+const { roles } = await login("cust_t", "123456");
+if (roles.includes("admin"))        router.push("/admin");
+else if (roles.includes("risk_operator")) router.push("/risk");
+else                                router.push("/chat");
+```
+
+### 之后每个请求
+
+```javascript
+fetch("/api/v1/risk/alerts", {
+  headers: { Authorization: `Bearer ${localStorage.getItem("token")}` },
+});
+```
+
+### 401 的处理
+
+拿到 **401** 就清掉令牌、回到登录页——它涵盖"令牌缺失/无效/过期/被吊销/账号停用"，
+不用也无法区分。**不要**在 401 里重试登录，那会撞上 429。
+
+### SSE 接口注意
+
+`GET /api/v1/agent-runs/{run_id}/events` 这类 SSE 端点**必须**带 `Authorization` 头，
+所以**不能用原生 `EventSource`**（它设不了自定义头），要用支持自定义头的 `fetch` 流式实现
+或合规的 SSE 客户端（`docs/05` §3.2）。
+
+---
+
+## 6. 常见问题
+
+| 现象 | 原因 |
+|---|---|
+| **登录一直 401** | ① 用错用户名——是 `cust_t` 不是 `9001`；② 密码没设过（`tools/set_user_password.py --list` 会显示"占位符，无法登录"）；③ 账号 `status` 不是 `正常` |
+| **登录成功但接口全 403** | 角色的权限不够（不是登录问题）。看该接口要求的权限码，再核对 `sys_role_permission` |
+| **登录成功但 `roles` 是 `()`** | 角色没绑上，多数是 `assigned_at` 落在未来——用 `create_test_user.py` 重跑，别手写 SQL |
+| **429** | 60 秒内超过 10 次登录（含失败）。等 `Retry-After` 秒 |
+| **令牌用一会儿就 401** | 有效期 30 分钟。**当前还没有刷新接口**，过期就重新登录 |
+| **改了这个用户的角色，但界面没变** | 前端缓存了 `roles`。令牌里不含权限，重新调一次登录接口拿最新 `roles` 即可 |
+
+---
+
+## 7. 当前边界（做之前先看这里）
+
+**已有**：
+- 登录（账号密码换令牌）、限流、登录成功/失败审计（写在 `interaction_audit`）。
+
+**还没有**（别假设它们存在）：
+- **刷新令牌**、**注销/吊销**、**改密码接口**、**注册**、**找回密码**。
+  其中刷新与注销按 `docs/05` §11 归"统一身份认证模块"；
+  `app/core/security.py` 已留好 `RevocationStore` 协议，接上 Redis 即可，属于二期。
+- 令牌**过期只能重新登录**（30 分钟）。
+
+**审计**：每次登录成功与失败都会写一条 `interaction_audit`
+（`action_type` 为 `auth.login_succeeded` / `auth.login_failed`）。
+排查"谁登过"用 `GET /api/v1/admin/audit-records`。
+
+---
+
+## 8. 给平台侧的反馈
+
+这个接口是平台侧新加的，边界是"**只做登录**"。如果你在联调中发现：
+
+- 需要**刷新令牌**（不想每 30 分钟重登）；
+- 需要**注销**（切换账号、退出登录）；
+- 需要更多**角色**（`operator`、`advisor`、`super_admin` 在 `bootstrap.py` 的
+  `allowed_roles` 里已经被引用，但 `sys_role` 里还没有对应的角色行）；
+- 需要**改密码**；
+
+直接提，别自己在业务分支里加路由——认证是共用的，两边各加一套会打架。

tools/create_test_user.py

@@ -0,0 +1,213 @@
+"""添加一个可登录的测试账号（用户 + 角色 + 密码），并**验证它真的能拿到权限**。
+
+## 为什么需要它
+
+`sys_user` / `sys_user_role` / `sys_role` 这些表**没有 ORM 模型**（全项目用裸 SQL 访问，
+见 `tools/seed_test_rbac.py`），手写 INSERT 要凑齐 10 个字段、还要自己算密码哈希。
+更要紧的是这里有个**静默陷阱**：
+
+    MySQL 的 DATETIME(0) 会把微秒**四舍五入到秒**。若 `sys_user_role.assigned_at`
+    用"当前时间"写入，进位后可能落在未来，而授权校验是 `assigned_at <= now` ——
+    于是刚建好的账号**一个角色都拿不到**，表现为 `roles=()`，**不报错也不失败**。
+
+`seed_test_rbac.py` 已经踩过一次（它的注释里写着）。本脚本统一把 `assigned_at`
+往前留 5 秒，并在最后**用 `IdentityService.resolve` 打印真实解析结果**而不是
+"插入成功" —— 后者根本不能说明这个账号能用。
+
+## 用法
+
+    python tools/create_test_user.py --list
+    python tools/create_test_user.py --id 9010 --username test_cust \\
+        --role customer --password abc12345
+
+重复执行同一个 `--id` 是**覆盖**语义：更新用户名 / 密码 / 角色，不会产生重复行。
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import sys
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import text
+
+from app.core.contracts import RequestContext
+from app.infrastructure.db import SessionFactory
+from app.service.auth_service import hash_password
+from app.service.identity_service import IdentityService
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(errors="replace")  # type: ignore[union-attr]
+
+#: 库里现成的三个角色（`tools/seed_test_rbac.py` 建的）。新用户复用它们。
+#: 要引入**新角色**得同时定义它的权限集合（`sys_role_permission`），超出本脚本范围。
+ROLE_IDS: dict[str, int] = {
+    "customer": 9001,
+    "risk_operator": 9002,
+    "admin": 9003,
+}
+
+#: 角色 → `sys_user.user_type`。注意这是 `user_type`，与 `employee_role` 不是一回事。
+ROLE_USER_TYPE: dict[str, str] = {
+    "customer": "customer",
+    "risk_operator": "employee",
+    "admin": "employee",
+}
+
+#: 客户的开户状态。风控扫描等链路会读它，写成 `closed` 会让部分规则不成立。
+FUND_ACCOUNT_STATUS: dict[str, str] = {
+    "customer": "已开户",
+    "employee": "closed",
+}
+
+#: `assigned_at` 往前留的秒数，见模块文档里的静默陷阱。
+ASSIGN_BACKDATE_SECONDS = 5
+
+
+async def list_users() -> None:
+    """列出所有账号、它的角色与密码状态。"""
+    async with SessionFactory() as session:
+        rows = (
+            await session.execute(
+                text(
+                    """
+                    SELECT u.id, u.username, u.user_type, u.status,
+                           u.password_hash,
+                           GROUP_CONCAT(r.role_code ORDER BY r.role_code) AS roles
+                    FROM sys_user u
+                    LEFT JOIN sys_user_role ur ON ur.user_id = u.id
+                    LEFT JOIN sys_role r ON r.id = ur.role_id
+                    GROUP BY u.id, u.username, u.user_type, u.status, u.password_hash
+                    ORDER BY u.id
+                    """
+                )
+            )
+        ).mappings().all()
+
+    print(f"{'id':<8}{'username':<18}{'user_type':<12}{'status':<8}{'roles':<28}密码")
+    for row in rows:
+        has_password = str(row["password_hash"] or "").startswith(("$2a$", "$2b$", "$2y$"))
+        print(
+            f"{row['id']:<8}{str(row['username']):<18}{str(row['user_type']):<12}"
+            f"{str(row['status']):<8}{str(row['roles'] or '(无角色)') or '(无角色)':<28}"
+            f"{'已设' if has_password else '占位符，无法登录'}"
+        )
+
+
+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}
+        )
+        if role_exists is None:
+            print(f"[失败] 角色 {role}（id={role_id}）不存在，先跑 tools/seed_test_rbac.py")
+            return 1
+
+        # 覆盖语义：同一个 id 重跑不会堆出第二行。
+        await session.execute(
+            text(
+                """
+                INSERT INTO sys_user
+                    (id, user_no, username, password_hash, user_type,
+                     professional_investor_status, fund_account_status, status,
+                     created_at, updated_at)
+                VALUES
+                    (:id, :user_no, :username, :password_hash, :user_type,
+                     'none', :fund_status, '正常', :now, :now)
+                AS new
+                ON DUPLICATE KEY UPDATE
+                    username = new.username,
+                    password_hash = new.password_hash,
+                    user_type = new.user_type,
+                    fund_account_status = new.fund_account_status,
+                    status = '正常',
+                    updated_at = new.updated_at
+                """
+            ),
+            {
+                "id": user_id,
+                "user_no": f"T-{username.upper()[:20]}",
+                "username": username,
+                "password_hash": hash_password(password),
+                "user_type": user_type,
+                "fund_status": FUND_ACCOUNT_STATUS[user_type],
+                "now": now,
+            },
+        )
+        # 角色绑定先清后插：`sys_user_role` 没有唯一约束，直接插会堆重复行。
+        await session.execute(
+            text("DELETE FROM sys_user_role WHERE user_id = :user_id"), {"user_id": user_id}
+        )
+        await session.execute(
+            text(
+                "INSERT INTO sys_user_role (user_id, role_id, assigned_at)"
+                " VALUES (:user_id, :role_id, :assigned_at)"
+            ),
+            {"user_id": user_id, "role_id": role_id, "assigned_at": assigned_at},
+        )
+
+    print(f"[OK] id={user_id} username={username} role={role} 已写入")
+    return await verify(user_id, username, role)
+
+
+async def verify(user_id: int, username: str, role: str) -> int:
+    """用真实链路解析身份 —— 这是唯一能证明"这个账号能用的"方式。"""
+    context = await IdentityService().resolve(
+        RequestContext(user_id=str(user_id), trace_id="create-test-user")
+    )
+    print(f"     解析结果：roles={context.roles} data_scope={context.data_scope}")
+    print(f"     权限 {len(context.permissions)} 项")
+    if role not in context.roles:
+        print(
+            "[失败] 角色没有解析出来。最可能的原因是 assigned_at 落在了未来"
+            "（DATETIME(0) 的毫秒舍入），请重跑本脚本。"
+        )
+        return 1
+    print(f"[OK] 可以用它登录：{{\"username\": \"{username}\", \"password\": \"<你刚设的>\"}}")
+    return 0
+
+
+async def main() -> int:
+    parser = argparse.ArgumentParser(description="添加可登录的测试账号")
+    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("--password", help="登录密码（仅限演示环境）")
+    args = parser.parse_args()
+
+    if args.list:
+        await list_users()
+        return 0
+
+    missing = [
+        name
+        for name, value in (
+            ("--id", args.id), ("--username", args.username),
+            ("--role", args.role), ("--password", args.password),
+        )
+        if value is None
+    ]
+    if missing:
+        print(f"[失败] 缺少参数：{' '.join(missing)}（或直接用 --list 看现有账号）")
+        return 1
+
+    return await upsert_user(
+        user_id=args.id,
+        username=args.username,
+        role=args.role,
+        password=args.password,
+    )
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

tools/set_user_password.py

@@ -29,10 +29,10 @@ import asyncio
 import sys
 from datetime import UTC, datetime
 
-import bcrypt
 from sqlalchemy import text
 
 from app.infrastructure.db import SessionFactory
+from app.service.auth_service import hash_password
 
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(errors="replace")  # type: ignore[union-attr]
@@ -49,10 +49,6 @@ DEMO_PASSWORDS: dict[str, str] = {
 BCRYPT_PREFIXES = ("$2a$", "$2b$", "$2y$")
 
 
-def hash_password(password: str) -> str:
-    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
-
-
 def _is_real_hash(value: str | None) -> bool:
     return bool(value) and str(value).startswith(BCRYPT_PREFIXES)