feat(compute): 算力引擎对接改造 + 用户生命周期自动同步

- config: 置顶 load_dotenv(.env)，修复 .env 未生效；COMPUTE_ADMIN_TOKEN 改读 PINEAGENTS_INTERNAL_TOKEN(回退旧名)，新增独立 COMPUTE_RELAY_TOKEN - compute_client: 空令牌不下发 Bearer(避免 httpx 非法头)；proxy 空令牌返 503 中文诊断；新增 sync_user_enabled/sync_delete_user - relay: /v1 转发改走 COMPUTE_RELAY_TOKEN(真实消费令牌)，空令牌不下发 - rbac_admin: 平台用户创建→自动建引擎账号+签PAT；启用/禁用→同步引擎状态；删除→同步删引擎账号(最佳努力,失败静默)
2026-08-25 12:08:14 +08:00
parent f067c02ec2
commit 1ca2e38559
4 changed files with 65 additions and 5 deletions
@@ -10,10 +10,19 @@ from ..schemas.admin import SetUserRoleRequest, SetUserStatusRequest
 from ..schemas.admin import UserCreateRequest, RolePermissionRequest, UpdateUserRequest
 from ...rbac import require_permission, require_roles, write_audit
 from ...infrastructure.repositories import Database
+from ...services import compute_client

 router = APIRouter(prefix="/admin", tags=["admin"])


+async def _sync_compute(coro) -> None:
+    """算力引擎用户同步为附加能力：失败不阻断平台用户操作（引擎未配置/不可达时静默）。"""
+    try:
+        await coro
+    except Exception:  # noqa: BLE001
+        pass
+
+
@router.post("/users", summary="新增账号（按类型/角色）")
 async def create_user(
    req: UserCreateRequest,
@@ -24,6 +33,9 @@ async def create_user(
    from ...services.user_admin_service import UserAdminService

    user = await UserAdminService(db).create_user(req, actor)
+    # 平台用户创建 → 引擎同步建号 + 签发 PAT（算力引擎用户与平台一致）
+    await _sync_compute(compute_client.create_user(user["username"]))
+    await _sync_compute(compute_client.issue_pat(user["username"]))
    await write_audit(db, action="user.create", resource="user", resource_id=user["id"],
                detail=f"role={req.role}", user=actor, request=request)
    return await db.users.to_profile(user)
@@ -78,9 +90,14 @@ async def set_user_status(
    db: Database = Depends(get_db),
    actor: dict = Depends(require_permission("action:user.disable")),
 ):
-    if await db.users.get_by_id(user_id) is None:
+    user = await db.users.get_by_id(user_id)
+    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    updated = await db.users.set_status(user_id, req.status)
+    # 平台用户启用/禁用 → 引擎镜像同步
+    await _sync_compute(compute_client.sync_user_enabled(
+        user.get("username"), req.status == "active",
+    ))
    await write_audit(
        db, action="user.disable", resource="user", resource_id=user_id,
        detail=f"{actor['username']} -> status={req.status}",
@@ -118,9 +135,13 @@ async def delete_user(
    db: Database = Depends(get_db),
    actor: dict = Depends(require_permission("action:user.manage")),
 ):
+    user = await db.users.get_by_id(user_id)
    ok = await db.users.delete(user_id)
    if not ok:
        raise HTTPException(status_code=404, detail="User not found")
+    # 平台用户删除 → 引擎镜像同步删除
+    if user:
+        await _sync_compute(compute_client.sync_delete_user(user.get("username")))
    await write_audit(db, action="user.delete", resource="user", resource_id=user_id,
                detail="deleted", user=actor, request=request)
    return {"ok": True, "user_id": user_id}
@@ -30,7 +30,12 @@ _ENGINE_MODEL_IDS = [


 def _auth_headers() -> dict[str, str]:
-    return {"Authorization": f"Bearer {config.COMPUTE_ADMIN_TOKEN}"}
+    # /v1 转发用引擎能识别的真实消费令牌（PINEAGENTS_COMPUTE_RELAY_TOKEN），
+    # 与管理的 COMPUTE_ADMIN_TOKEN(内部合成 root) 区分；空令牌不下发，避免 `Bearer ` 非法被 httpx 拒绝。
+    token = config.COMPUTE_RELAY_TOKEN or config.COMPUTE_ADMIN_TOKEN
+    if not token:
+        return {}
+    return {"Authorization": f"Bearer {token}"}


@router.post("/chat/completions")
@@ -12,6 +12,15 @@ from __future__ import annotations
 import os
 from pathlib import Path

+# 启动第一步即加载项目根 .env（含 PINEAGENTS_INTERNAL_TOKEN / DATABASE_URL 等），
+# 否则后续 os.environ.get 读到空串（.env 后续才会被 app.park/app.training 各自加载）。
+# 必须在下方任何 os.environ.get 之前执行。
+try:  # python-dotenv 为可选依赖；缺失时仍可用系统环境变量
+    from dotenv import load_dotenv
+    load_dotenv(Path(__file__).resolve().parent.parent / ".env")
+except Exception:  # noqa: BLE001
+    pass
+
 # 项目根目录 = 本文件的上上级（app/config.py -> <根>）
 BASE_DIR = Path(__file__).resolve().parent.parent

@@ -100,8 +109,12 @@ JWT_ACCESS_EXPIRY_SECONDS = DEFAULT_TOKEN_EXPIRY_SECONDS
 COMPUTE_BASE_URL = os.environ.get(
    "PINEAGENTS_COMPUTE_BASE_URL", "http://127.0.0.1:3000",
 )
-# new-api 系统访问令牌（管理 API 程序化调用用；非空则 /v1 中继走引擎）
-COMPUTE_ADMIN_TOKEN = os.environ.get("PINEAGENTS_COMPUTE_ADMIN_TOKEN", "")
+# 受信任内部令牌：server-core 作为唯一入口，无权限限制访问引擎管理/root 端点。
+# 与引擎侧 PINEAGENTS_INTERNAL_TOKEN 保持一致；回退兼容旧名 PINEAGENTS_COMPUTE_ADMIN_TOKEN。
+COMPUTE_ADMIN_TOKEN = os.environ.get("PINEAGENTS_INTERNAL_TOKEN") or os.environ.get("PINEAGENTS_COMPUTE_ADMIN_TOKEN", "")
+# /v1 模型中继转发（relay.py）用的引擎消费令牌：需是引擎 /v1 能识别的真实用户/渠道令牌，
+# 与聚合 root 的 COMPUTE_ADMIN_TOKEN 区分开（后者仅用于管理/root 端点）。回退 COMPUTE_ADMIN_TOKEN。
+COMPUTE_RELAY_TOKEN = os.environ.get("PINEAGENTS_COMPUTE_RELAY_TOKEN", "")
 COMPUTE_TIMEOUT = float(os.environ.get("PINEAGENTS_COMPUTE_TIMEOUT", "10"))
 COMPUTE_RETRIES = int(os.environ.get("PINEAGENTS_COMPUTE_RETRIES", "2"))

@@ -29,7 +29,11 @@ def _base() -> str:


 def _admin_headers() -> dict[str, str]:
-    return {"Authorization": f"Bearer {config.COMPUTE_ADMIN_TOKEN}"}
+    """引擎管理员请求头。令牌未配置时不下发 Authorization，避免 `Bearer ` 空令牌被 httpx 判定非法。"""
+    token = config.COMPUTE_ADMIN_TOKEN
+    if not token:
+        return {}
+    return {"Authorization": f"Bearer {token}"}


 async def _request(*, method: str, path: str, headers: dict[str, str] | None = None,
@@ -106,6 +110,21 @@ async def get_logs(user_pat: str) -> dict:
    return await _request(method="GET", path="/api/log/self", headers=headers)


+async def sync_user_enabled(username: str, enabled: bool) -> dict:
+    """按 username 同步引擎用户启用/禁用（平台用户生命周期自动同步）。"""
+    return await _request(
+        method="PATCH", path="/api/user/status", headers=_admin_headers(),
+        json={"username": username, "status": 1 if enabled else 0},
+    )
+
+
+async def sync_delete_user(username: str) -> dict:
+    """按 username 删除引擎用户（平台用户删除时同步）。"""
+    return await _request(
+        method="DELETE", path=f"/api/user/by_username/{username}", headers=_admin_headers(),
+    )
+
+
 async def proxy(method: str, path: str, *, json_body: dict | None = None, params: dict | None = None) -> dict:
    """管理 API 透传：转发到引擎 ``/api<path>``，返回 ``{status, body}``。

@@ -113,6 +132,8 @@ async def proxy(method: str, path: str, *, json_body: dict | None = None, params
    （models/channels/groups/tokens/users/logs/redemption/ratio 等），不改写每个端点契约。
    保留引擎原始状态码与响应体（网关层不吞错误）。
    """
+    if not config.COMPUTE_ADMIN_TOKEN:
+        return {"status": 503, "body": '{"success": false, "message": "算力引擎内部令牌未配置（PINEAGENTS_INTERNAL_TOKEN）"}'}
    url = f"{_base()}/api{path}"
    try:
        async with httpx.AsyncClient(timeout=config.COMPUTE_TIMEOUT) as client: