更新agent manager数据接口

services/mcp-server/app/auth.py

@@ -31,12 +31,41 @@ def get_password_hash(password: str) -> str:
 
 
 def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
+    """
+    创建访问令牌 (Access Token)
+    
+    默认有效期：根据配置 jwt_expire_minutes（通常为 24 小时）
+    """
     to_encode = data.copy()
     now = datetime.utcnow()
     expire = now + (expires_delta or timedelta(minutes=settings.jwt_expire_minutes))
     to_encode.update({
         "exp": expire,
         "iat": now,  # 添加签发时间，用于登出验证
+        "type": "access",  # 标识 token 类型
+    })
+    encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.jwt_algorithm)
+    return encoded_jwt
+
+
+# Refresh Token 有效期：7 天
+REFRESH_TOKEN_EXPIRE_DAYS = 7
+
+
+def create_refresh_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
+    """
+    创建刷新令牌 (Refresh Token)
+    
+    默认有效期：7 天，比 Access Token 更长
+    Refresh Token 仅用于获取新的 Access Token
+    """
+    to_encode = data.copy()
+    now = datetime.utcnow()
+    expire = now + (expires_delta or timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS))
+    to_encode.update({
+        "exp": expire,
+        "iat": now,
+        "type": "refresh",  # 标识 token 类型
     })
     encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.jwt_algorithm)
     return encoded_jwt

services/mcp-server/app/billing.py

@@ -137,23 +137,33 @@ async def deduct_balance(
     user_id: str,
     amount: Decimal,
     db: AsyncSession,
-    description: str = "消费"
+    description: str = "消费",
+    auto_commit: bool = False
 ) -> Tuple[bool, str]:
     """
     扣除余额（优先扣除账户余额，不足时使用授信额度）
     
+    使用行锁保护并发扣款操作，防止超扣。
+    
     Args:
         user_id: 用户ID
         amount: 扣除金额
         db: 数据库会话
         description: 描述
+        auto_commit: 是否自动提交（默认False，由调用者管理事务）
         
     Returns:
         (是否成功, 消息)
+        
+    Note:
+        默认不会 commit，由调用者统一管理事务。
+        如需独立提交，请设置 auto_commit=True。
     """
-    # 从 Balance 表获取余额
+    # 使用 FOR UPDATE 锁定余额行，防止并发扣款
     balance_result = await db.execute(
-        select(Balance).where(Balance.user_id == user_id)
+        select(Balance)
+        .where(Balance.user_id == user_id)
+        .with_for_update()
     )
     balance_obj = balance_result.scalar_one_or_none()
     
@@ -170,6 +180,7 @@ async def deduct_balance(
         # 如果余额记录不存在，创建一个新的（初始余额为0）
         balance_obj = Balance(user_id=user_id, eu_balance=0.0)
         db.add(balance_obj)
+        await db.flush()  # 确保记录创建后再继续
     
     balance = Decimal(str(balance_obj.eu_balance))
     credit_limit = Decimal(str(user.credit_limit))
@@ -178,16 +189,13 @@ async def deduct_balance(
     if available < amount:
         return False, f"余额不足，当前可用额度: {available}, 需要: {amount}"
     
-    # 优先扣除账户余额
-    if balance >= amount:
-        balance_obj.eu_balance = float(balance - amount)
-    else:
-        # 余额不足，使用授信额度
-        balance_obj.eu_balance = 0.0
-        # 注意：授信额度是额度上限，不是实际金额，这里简化处理
-        # 实际应该有单独的授信使用记录表
+    # 优先扣除账户余额，允许透支到授信额度
+    new_balance = balance - amount
+    balance_obj.eu_balance = float(new_balance)
     
-    await db.commit()
+    # 可选：自动提交
+    if auto_commit:
+        await db.commit()
     
     return True, f"成功扣除 {amount} 元"
 
@@ -196,19 +204,27 @@ async def add_balance(
     user_id: str,
     amount: Decimal,
     db: AsyncSession,
-    description: str = "充值"
+    description: str = "充值",
+    auto_commit: bool = False
 ) -> Tuple[bool, str]:
     """
     增加余额
     
+    使用行锁保护并发充值操作，确保余额准确。
+    
     Args:
         user_id: 用户ID
         amount: 充值金额
         db: 数据库会话
         description: 描述
+        auto_commit: 是否自动提交（默认False，由调用者管理事务）
         
     Returns:
         (是否成功, 消息)
+        
+    Note:
+        默认不会 commit，由调用者统一管理事务。
+        如需独立提交，请设置 auto_commit=True。
     """
     # 检查用户是否存在
     user_result = await db.execute(
@@ -219,9 +235,11 @@ async def add_balance(
     if not user:
         return False, "用户不存在"
     
-    # 从 Balance 表获取或创建余额记录
+    # 使用 FOR UPDATE 锁定余额行，防止并发更新
     balance_result = await db.execute(
-        select(Balance).where(Balance.user_id == user_id)
+        select(Balance)
+        .where(Balance.user_id == user_id)
+        .with_for_update()
     )
     balance_obj = balance_result.scalar_one_or_none()
     
@@ -229,13 +247,17 @@ async def add_balance(
         # 如果余额记录不存在，创建一个新的
         balance_obj = Balance(user_id=user_id, eu_balance=0.0)
         db.add(balance_obj)
+        await db.flush()  # 确保记录创建后再继续
     
     old_balance = Decimal(str(balance_obj.eu_balance))
-    balance_obj.eu_balance = float(old_balance + amount)
+    new_balance = old_balance + amount
+    balance_obj.eu_balance = float(new_balance)
     
-    await db.commit()
+    # 可选：自动提交
+    if auto_commit:
+        await db.commit()
     
-    return True, f"成功充值 {amount} 元，当前余额: {balance_obj.eu_balance}"
+    return True, f"成功充值 {amount} 元，当前余额: {new_balance}"
 
 
 # ============= 计费记录创建 =============

services/mcp-server/app/db_utils.py

@@ -0,0 +1,417 @@
+"""
+数据库工具模块
+提供事务重试机制、行锁保护等并发安全工具
+"""
+
+import asyncio
+import functools
+from typing import TypeVar, Callable, Any, Optional
+from decimal import Decimal
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.exc import OperationalError, DBAPIError
+from sqlalchemy.orm.exc import StaleDataError
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+T = TypeVar('T')
+
+# 数据库重试配置
+DEFAULT_MAX_RETRIES = 3
+DEFAULT_RETRY_DELAY = 0.1  # 秒
+DEFAULT_RETRY_BACKOFF = 2.0  # 指数退避倍数
+
+
+class DatabaseRetryError(Exception):
+    """数据库重试失败异常"""
+    def __init__(self, message: str, original_error: Exception = None):
+        super().__init__(message)
+        self.original_error = original_error
+
+
+class OptimisticLockError(Exception):
+    """乐观锁冲突异常"""
+    pass
+
+
+async def with_retry(
+    func: Callable[..., T],
+    *args,
+    max_retries: int = DEFAULT_MAX_RETRIES,
+    retry_delay: float = DEFAULT_RETRY_DELAY,
+    backoff: float = DEFAULT_RETRY_BACKOFF,
+    **kwargs
+) -> T:
+    """
+    带重试的异步函数执行器
+    
+    Args:
+        func: 要执行的异步函数
+        *args: 函数参数
+        max_retries: 最大重试次数
+        retry_delay: 初始重试延迟（秒）
+        backoff: 退避倍数
+        **kwargs: 函数关键字参数
+        
+    Returns:
+        函数执行结果
+        
+    Raises:
+        DatabaseRetryError: 重试次数用尽后仍然失败
+    """
+    last_error = None
+    current_delay = retry_delay
+    
+    for attempt in range(max_retries + 1):
+        try:
+            return await func(*args, **kwargs)
+        except (OperationalError, DBAPIError, StaleDataError) as e:
+            last_error = e
+            if attempt < max_retries:
+                logger.warning(
+                    "database_operation_retry",
+                    attempt=attempt + 1,
+                    max_retries=max_retries,
+                    error=str(e),
+                    delay=current_delay
+                )
+                await asyncio.sleep(current_delay)
+                current_delay *= backoff
+            else:
+                logger.error(
+                    "database_operation_failed",
+                    attempts=max_retries + 1,
+                    error=str(e)
+                )
+    
+    raise DatabaseRetryError(
+        f"数据库操作在 {max_retries + 1} 次尝试后失败",
+        original_error=last_error
+    )
+
+
+def retry_on_conflict(
+    max_retries: int = DEFAULT_MAX_RETRIES,
+    retry_delay: float = DEFAULT_RETRY_DELAY,
+    backoff: float = DEFAULT_RETRY_BACKOFF
+):
+    """
+    装饰器：自动重试数据库冲突操作
+    
+    用法：
+        @retry_on_conflict(max_retries=3)
+        async def my_db_operation(db: AsyncSession):
+            ...
+    """
+    def decorator(func: Callable[..., T]) -> Callable[..., T]:
+        @functools.wraps(func)
+        async def wrapper(*args, **kwargs) -> T:
+            return await with_retry(
+                func, *args,
+                max_retries=max_retries,
+                retry_delay=retry_delay,
+                backoff=backoff,
+                **kwargs
+            )
+        return wrapper
+    return decorator
+
+
+# ==================== 余额操作工具 ====================
+
+async def atomic_balance_update(
+    db: AsyncSession,
+    user_id: str,
+    amount: Decimal,
+    operation: str = "deduct",  # "deduct" 或 "add"
+    check_sufficient: bool = True
+) -> tuple[bool, Decimal, str]:
+    """
+    原子性余额更新（使用行锁）
+    
+    Args:
+        db: 数据库会话
+        user_id: 用户ID
+        amount: 金额（正数）
+        operation: 操作类型 "deduct" 扣减 / "add" 增加
+        check_sufficient: 扣减时是否检查余额充足
+        
+    Returns:
+        (是否成功, 新余额, 消息)
+        
+    Note:
+        此函数不会 commit，调用者负责事务管理
+    """
+    from models import Balance, User
+    
+    # 使用 FOR UPDATE 锁定余额行
+    balance_result = await db.execute(
+        select(Balance)
+        .where(Balance.user_id == user_id)
+        .with_for_update()
+    )
+    balance_obj = balance_result.scalar_one_or_none()
+    
+    # 获取用户授信额度
+    user_result = await db.execute(
+        select(User.credit_limit).where(User.id == user_id)
+    )
+    credit_row = user_result.first()
+    credit_limit = Decimal(str(credit_row[0])) if credit_row and credit_row[0] else Decimal(0)
+    
+    if balance_obj is None:
+        # 创建余额记录
+        balance_obj = Balance(user_id=user_id, eu_balance=0)
+        db.add(balance_obj)
+        await db.flush()  # 确保记录创建
+    
+    current_balance = Decimal(str(balance_obj.eu_balance))
+    
+    if operation == "deduct":
+        available = current_balance + credit_limit
+        if check_sufficient and available < amount:
+            return False, current_balance, f"余额不足，可用: {available}, 需要: {amount}"
+        
+        # 优先扣除余额
+        if current_balance >= amount:
+            new_balance = current_balance - amount
+        else:
+            # 余额不足，使用授信额度（余额可为负）
+            new_balance = current_balance - amount
+        
+        balance_obj.eu_balance = float(new_balance)
+        return True, new_balance, f"成功扣除 {amount}"
+        
+    elif operation == "add":
+        new_balance = current_balance + amount
+        balance_obj.eu_balance = float(new_balance)
+        return True, new_balance, f"成功充值 {amount}，当前余额: {new_balance}"
+    
+    else:
+        raise ValueError(f"未知操作类型: {operation}")
+
+
+async def atomic_eu_consume(
+    db: AsyncSession,
+    user_id: str,
+    eu_amount: Decimal
+) -> tuple[bool, str]:
+    """
+    原子性 EU 消耗（同时更新余额和用户消耗统计）
+    
+    Args:
+        db: 数据库会话
+        user_id: 用户ID
+        eu_amount: EU 消耗量
+        
+    Returns:
+        (是否成功, 消息)
+    """
+    from models import User
+    
+    # 扣减余额
+    success, new_balance, msg = await atomic_balance_update(
+        db, user_id, eu_amount, operation="deduct", check_sufficient=False
+    )
+    
+    if not success:
+        return False, msg
+    
+    # 更新用户 total_eu_consumed（使用原子更新）
+    await db.execute(
+        update(User)
+        .where(User.id == user_id)
+        .values(total_eu_consumed=User.total_eu_consumed + float(eu_amount))
+    )
+    
+    # 余额警告（不阻止操作）
+    if new_balance < 0:
+        logger.warning(
+            "user_balance_negative",
+            user_id=user_id,
+            balance=float(new_balance),
+            eu_consumed=float(eu_amount)
+        )
+    
+    return True, msg
+
+
+# ==================== 配额操作工具 ====================
+
+async def atomic_quota_update(
+    db: AsyncSession,
+    model_class,
+    conditions: dict,
+    field_name: str,
+    delta: int,
+    check_limit: bool = False,
+    limit_field: str = None,
+    max_value: int = None
+) -> tuple[bool, int, str]:
+    """
+    原子性配额更新（使用行锁和原子操作）
+    
+    Args:
+        db: 数据库会话
+        model_class: 模型类
+        conditions: 查询条件字典
+        field_name: 要更新的字段名
+        delta: 变化量（正数增加，负数减少）
+        check_limit: 是否检查上限
+        limit_field: 上限字段名
+        max_value: 固定上限值（与 limit_field 二选一）
+        
+    Returns:
+        (是否成功, 新值, 消息)
+    """
+    from sqlalchemy import and_
+    
+    # 构建查询条件
+    where_clauses = [getattr(model_class, k) == v for k, v in conditions.items()]
+    
+    # 使用 FOR UPDATE 锁定行
+    query = select(model_class).where(and_(*where_clauses)).with_for_update()
+    result = await db.execute(query)
+    record = result.scalar_one_or_none()
+    
+    if not record:
+        return False, 0, "记录不存在"
+    
+    current_value = getattr(record, field_name) or 0
+    new_value = current_value + delta
+    
+    # 检查下限
+    if new_value < 0:
+        return False, current_value, f"配额不足，当前: {current_value}, 变化: {delta}"
+    
+    # 检查上限
+    if check_limit:
+        if limit_field:
+            limit_value = getattr(record, limit_field) or 0
+        elif max_value is not None:
+            limit_value = max_value
+        else:
+            limit_value = float('inf')
+        
+        if new_value > limit_value:
+            return False, current_value, f"超出配额限制，上限: {limit_value}, 请求: {new_value}"
+    
+    # 使用原子更新
+    field = getattr(model_class, field_name)
+    await db.execute(
+        update(model_class)
+        .where(and_(*where_clauses))
+        .values({field_name: field + delta})
+    )
+    
+    return True, new_value, f"配额更新成功: {current_value} -> {new_value}"
+
+
+async def atomic_increment(
+    db: AsyncSession,
+    model_class,
+    conditions: dict,
+    updates: dict[str, Any]
+) -> bool:
+    """
+    原子性字段增量更新
+    
+    Args:
+        db: 数据库会话
+        model_class: 模型类
+        conditions: 查询条件
+        updates: 更新字典 {字段名: 增量值}
+        
+    Returns:
+        是否成功
+    """
+    from sqlalchemy import and_
+    
+    where_clauses = [getattr(model_class, k) == v for k, v in conditions.items()]
+    
+    # 构建原子更新表达式
+    update_values = {}
+    for field_name, delta in updates.items():
+        field = getattr(model_class, field_name)
+        update_values[field_name] = field + delta
+    
+    result = await db.execute(
+        update(model_class)
+        .where(and_(*where_clauses))
+        .values(**update_values)
+    )
+    
+    return result.rowcount > 0
+
+
+# ==================== 幂等性工具 ====================
+
+async def ensure_idempotent(
+    db: AsyncSession,
+    model_class,
+    unique_field: str,
+    unique_value: str
+) -> bool:
+    """
+    检查操作是否已执行（幂等性检查）
+    
+    Args:
+        db: 数据库会话
+        model_class: 模型类
+        unique_field: 唯一字段名
+        unique_value: 唯一字段值
+        
+    Returns:
+        True 如果记录已存在（操作已执行），False 如果不存在
+    """
+    result = await db.execute(
+        select(model_class.id)
+        .where(getattr(model_class, unique_field) == unique_value)
+        .limit(1)
+    )
+    return result.scalar_one_or_none() is not None
+
+
+async def get_or_create_with_lock(
+    db: AsyncSession,
+    model_class,
+    defaults: dict,
+    **lookup_kwargs
+) -> tuple[Any, bool]:
+    """
+    获取或创建记录（带锁保护）
+    
+    Args:
+        db: 数据库会话
+        model_class: 模型类
+        defaults: 创建时的默认值
+        **lookup_kwargs: 查找条件
+        
+    Returns:
+        (记录对象, 是否新创建)
+    """
+    from sqlalchemy import and_
+    
+    where_clauses = [getattr(model_class, k) == v for k, v in lookup_kwargs.items()]
+    
+    # 尝试获取（带锁）
+    result = await db.execute(
+        select(model_class)
+        .where(and_(*where_clauses))
+        .with_for_update()
+    )
+    instance = result.scalar_one_or_none()
+    
+    if instance:
+        return instance, False
+    
+    # 创建新记录
+    create_kwargs = {**lookup_kwargs, **defaults}
+    instance = model_class(**create_kwargs)
+    db.add(instance)
+    await db.flush()
+    
+    return instance, True
+

services/mcp-server/app/email_verification.py

@@ -15,16 +15,19 @@ from app.state import get_state
 logger = structlog.get_logger(__name__)
 
 # 邮箱配置 - 从环境变量读取
-SMTP_SERVER = os.getenv("SMTP_SERVER", "smtp.office365.com")
-SMTP_PORT = int(os.getenv("SMTP_PORT", "587"))
+SMTP_SERVER = os.getenv("SMTP_SERVER", "smtpdm-ap-southeast-1.aliyun.com")
+SMTP_PORT = int(os.getenv("SMTP_PORT", "465"))
 SMTP_EMAIL = os.getenv("SMTP_EMAIL", "supportagnet@taijiaicloud.com")
-# 注意：生产环境应该通过环境变量设置SMTP_PASSWORD，不要硬编码密码
-# 这里使用默认值仅用于开发测试，生产环境必须通过环境变量配置
-SMTP_PASSWORD = os.getenv("SMTP_PASSWORD", "l5YYL7TOK2WvRKtf")
+# 注意：生产环境必须通过环境变量设置SMTP_PASSWORD
+# 如果未设置，邮件发送功能将不可用（测试模式下可通过日志或Redis获取验证码）
+SMTP_PASSWORD = os.getenv("SMTP_PASSWORD")
+# 是否使用SSL（端口465使用SSL，端口587使用STARTTLS）
+SMTP_USE_SSL = os.getenv("SMTP_USE_SSL", "true").lower() == "true"
 
 # 验证码配置
 VERIFICATION_CODE_LENGTH = 6
 VERIFICATION_CODE_EXPIRE_SECONDS = 600  # 10分钟
+VERIFICATION_CODE_RATE_LIMIT_SECONDS = 60  # 发送频率限制：60秒内只能发送一次
 
 
 def _send_email_sync(msg: MIMEMultipart) -> None:
@@ -32,9 +35,14 @@ def _send_email_sync(msg: MIMEMultipart) -> None:
     if not SMTP_PASSWORD:
         raise ValueError("SMTP_PASSWORD环境变量未设置，无法发送邮件")
     
-    server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)
-    try:
+    # 根据端口选择连接方式：465用SSL，587用STARTTLS
+    if SMTP_PORT == 465 or SMTP_USE_SSL:
+        server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT)
+    else:
+        server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)
         server.starttls()
+    
+    try:
         server.login(SMTP_EMAIL, SMTP_PASSWORD)
         server.send_message(msg)
         logger.info("邮件发送成功", to=msg['To'])
@@ -117,6 +125,64 @@ Taiji AI-PAD 团队
         return False
 
 
+async def check_rate_limit(email: str) -> tuple[bool, int]:
+    """
+    检查验证码发送频率限制
+    
+    Args:
+        email: 邮箱地址
+        
+    Returns:
+        (是否可以发送, 剩余等待秒数)
+    """
+    try:
+        state = get_state()
+        if not state.redis_client:
+            logger.warning("Redis未连接，跳过频率限制检查")
+            return True, 0
+        
+        rate_limit_key = f"verification_rate_limit:{email}"
+        
+        # 检查是否存在频率限制
+        ttl = await state.redis_client.ttl(rate_limit_key)
+        if ttl > 0:
+            logger.warning("验证码发送频率限制", email=email, remaining_seconds=ttl)
+            return False, ttl
+        
+        return True, 0
+    except Exception as e:
+        logger.error("检查频率限制失败", email=email, error=str(e))
+        # 出错时允许发送，避免阻塞用户
+        return True, 0
+
+
+async def set_rate_limit(email: str) -> bool:
+    """
+    设置验证码发送频率限制
+    
+    Args:
+        email: 邮箱地址
+        
+    Returns:
+        是否设置成功
+    """
+    try:
+        state = get_state()
+        if not state.redis_client:
+            return False
+        
+        rate_limit_key = f"verification_rate_limit:{email}"
+        await state.redis_client.setex(
+            rate_limit_key,
+            VERIFICATION_CODE_RATE_LIMIT_SECONDS,
+            "1"
+        )
+        return True
+    except Exception as e:
+        logger.error("设置频率限制失败", email=email, error=str(e))
+        return False
+
+
 async def store_verification_code(email: str, code: str) -> bool:
     """
     存储验证码到Redis
@@ -258,12 +324,17 @@ async def send_and_store_verification_code(email: str) -> Optional[str]:
                 verification_code=code,
                 hint="验证码已存储到Redis，可通过Redis获取或查看日志（仅测试环境）"
             )
+            # 设置发送频率限制
+            await set_rate_limit(email)
             return code
         else:
             # 生产模式：邮件发送失败则不返回验证码
             logger.error("邮件发送失败，验证码已存储但未发送", email=email)
             return None
     
+    # 发送成功后设置频率限制
+    await set_rate_limit(email)
+    
     logger.info("验证码已发送并存储", email=email)
     return code
 

services/mcp-server/app/quota_manager.py

@@ -156,6 +156,8 @@ async def create_quota_alert(
     """
     创建配额预警记录
     
+    使用行锁保护并发创建/更新预警记录。
+    
     Args:
         user_id: 用户ID
         channel_id: 渠道ID
@@ -168,15 +170,17 @@ async def create_quota_alert(
     Returns:
         QuotaAlert记录
     """
-    # 检查是否已有相同的活跃预警
+    # 检查是否已有相同的活跃预警（使用行锁防止并发创建重复预警）
     result = await db.execute(
-        select(QuotaAlert).where(
+        select(QuotaAlert)
+        .where(
             and_(
                 QuotaAlert.user_id == user_id,
                 QuotaAlert.alert_type == alert_type,
                 QuotaAlert.status == "active",
             )
         )
+        .with_for_update()  # 行锁
     )
     existing = result.scalar_one_or_none()
     

services/mcp-server/app/resource_control.py

@@ -6,13 +6,14 @@
 from datetime import datetime, timedelta
 from decimal import Decimal
 from typing import Dict, Optional, Tuple
-from sqlalchemy import select, func, and_
+from sqlalchemy import select, func, and_, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from fastapi import HTTPException, status
 import structlog
 
 from models import User, Channel, ResourceUsage, BillingRecord
 from app.quota_manager import check_user_balance_quota, check_channel_quota
+from app.db_utils import with_retry
 
 logger = structlog.get_logger()
 
@@ -208,6 +209,8 @@ class ResourceController:
         """
         记录资源消耗
         
+        使用行锁和重试机制防止并发更新丢失。
+        
         Args:
             user_id: 用户ID
             resource_type: 资源类型
@@ -219,39 +222,52 @@ class ResourceController:
             network_io: 网络IO（KB）
             db: 数据库会话
         """
-        try:
+        async def _do_record():
             now = datetime.utcnow()
             
             # 记录到ResourceUsage表（每小时聚合）
             period_start = now.replace(minute=0, second=0, microsecond=0)
             period_end = period_start + timedelta(hours=1)
             
-            # 检查是否已有该小时的记录
+            # 计算增量值
+            cpu_increment = cpu_usage * (execution_time_ms / 1000.0)
+            memory_increment = memory_usage * (execution_time_ms / 1000.0)
+            network_increment = int(network_io * 1024)
+            
+            # 使用行锁保护并发更新（FOR UPDATE）
             result = await db.execute(
-                select(ResourceUsage).where(
+                select(ResourceUsage)
+                .where(
                     and_(
                         ResourceUsage.user_id == user_id,
                         ResourceUsage.period_start == period_start,
                         ResourceUsage.granularity == "hourly"
                     )
                 )
+                .with_for_update()  # 行锁
             )
             usage = result.scalar_one_or_none()
             
             if usage:
-                # 更新现有记录
-                usage.cpu_seconds += cpu_usage * (execution_time_ms / 1000.0)
-                usage.memory_mb_seconds += memory_usage * (execution_time_ms / 1000.0)
-                usage.network_bytes += int(network_io * 1024)
-                usage.api_calls += 1
+                # 使用原子更新语句，而不是 ORM 属性修改
+                await db.execute(
+                    update(ResourceUsage)
+                    .where(ResourceUsage.id == usage.id)
+                    .values(
+                        cpu_seconds=ResourceUsage.cpu_seconds + cpu_increment,
+                        memory_mb_seconds=ResourceUsage.memory_mb_seconds + memory_increment,
+                        network_bytes=ResourceUsage.network_bytes + network_increment,
+                        api_calls=ResourceUsage.api_calls + 1
+                    )
+                )
             else:
                 # 创建新记录
                 usage = ResourceUsage(
                     user_id=user_id,
                     agent_id=resource_id if resource_type == "agent" else None,
-                    cpu_seconds=cpu_usage * (execution_time_ms / 1000.0),
-                    memory_mb_seconds=memory_usage * (execution_time_ms / 1000.0),
-                    network_bytes=int(network_io * 1024),
+                    cpu_seconds=cpu_increment,
+                    memory_mb_seconds=memory_increment,
+                    network_bytes=network_increment,
                     storage_bytes=0,
                     api_calls=1,
                     period_start=period_start,
@@ -285,6 +301,10 @@ class ResourceController:
                 cost=float(cost),
                 execution_time_ms=execution_time_ms
             )
+        
+        try:
+            # 使用重试机制处理数据库冲突
+            await with_retry(_do_record, max_retries=3)
             
         except Exception as e:
             logger.error(

services/mcp-server/app/routes/agents.py

@@ -81,8 +81,23 @@ def _build_agent_card(agent: Agent) -> AgentCard:
     )
 
 
-async def _get_balance(db: AsyncSession, user_id: uuid.UUID) -> Balance:
-    result = await db.execute(select(Balance).where(Balance.user_id == user_id))
+async def _get_balance(db: AsyncSession, user_id: uuid.UUID, for_update: bool = False) -> Balance:
+    """
+    获取或创建用户余额记录
+    
+    Args:
+        db: 数据库会话
+        user_id: 用户ID
+        for_update: 是否使用行锁（用于更新操作）
+        
+    Returns:
+        Balance 对象
+    """
+    query = select(Balance).where(Balance.user_id == user_id)
+    if for_update:
+        query = query.with_for_update()  # 行锁保护并发更新
+    
+    result = await db.execute(query)
     balance = result.scalar_one_or_none()
     if balance is None:
         balance = Balance(user_id=user_id, eu_balance=0.0)
@@ -852,9 +867,9 @@ async def execute_agent(
         )
         db.add(billing)
 
-        balance = await _get_balance(db, agent.owner_id)
+        # 使用行锁保护余额更新，防止并发扣款
+        balance = await _get_balance(db, agent.owner_id, for_update=True)
         balance.eu_balance = (balance.eu_balance or 0) - execution.eu_consumed
-        db.add(balance)
 
         # ========== 记录资源消耗 ==========
         # 记录到资源监控系统

services/mcp-server/app/routes/auth.py

@@ -6,6 +6,7 @@ from datetime import timedelta
 from fastapi import APIRouter, Depends, HTTPException, status, Query
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.exc import IntegrityError
 import secrets
 import hashlib
 from typing import Optional
@@ -18,6 +19,7 @@ from models import (
 from app.auth import (
     authenticate_user,
     create_access_token,
+    create_refresh_token,
     get_password_hash,
     verify_password,
     require_auth,
@@ -31,7 +33,7 @@ from app.schemas import (
     RegenerateAPIKeyResponse,
     UserCreate,
 )
-from app.email_verification import verify_code, send_and_store_verification_code
+from app.email_verification import verify_code, send_and_store_verification_code, check_rate_limit
 from app.agent_manager_client import get_agent_manager_client, AgentManagerError
 from app.litellm_client import get_litellm_client, LiteLLMClientError
 from config import settings
@@ -84,19 +86,19 @@ async def login(req: LoginRequest, db: AsyncSession = Depends(get_db)):
         await db.commit()
         
         # 创建JWT token
-        token = create_access_token(
-            data={
-                "sub": str(entity.id),
-                "email": entity.email,
-                "role": "channel_admin",
-                "channelId": str(entity.id),
-            }
-        )
+        token_data = {
+            "sub": str(entity.id),
+            "email": entity.email,
+            "role": "channel_admin",
+            "channelId": str(entity.id),
+        }
+        access_token = create_access_token(data=token_data)
+        refresh_token = create_refresh_token(data=token_data)
         
         return SuccessResponse(
             data={
-                "token": token,
-                "refreshToken": token,  # 简化处理，实际应该生成单独的refresh token
+                "token": access_token,
+                "refreshToken": refresh_token,
                 "user": {
                     "id": str(entity.id),
                     "name": entity.name,
@@ -159,19 +161,19 @@ async def login(req: LoginRequest, db: AsyncSession = Depends(get_db)):
         await db.commit()
         
         # 创建JWT token
-        token = create_access_token(
-            data={
-                "sub": str(user.id),
-                "email": user.email,
-                "role": user_role,
-                "channelId": str(user.channel_id) if user.channel_id else None,
-            }
-        )
+        token_data = {
+            "sub": str(user.id),
+            "email": user.email,
+            "role": user_role,
+            "channelId": str(user.channel_id) if user.channel_id else None,
+        }
+        access_token = create_access_token(data=token_data)
+        refresh_token = create_refresh_token(data=token_data)
         
         return SuccessResponse(
             data={
-                "token": token,
-                "refreshToken": token,
+                "token": access_token,
+                "refreshToken": refresh_token,
                 "user": {
                     "id": str(user.id),
                     "name": user.name or user.full_name,
@@ -248,14 +250,21 @@ async def logout(
 
 
 @router.post("/refresh", response_model=SuccessResponse)
-async def refresh_token(
+async def refresh_token_endpoint(
     principal: dict = Depends(require_auth),
     db: AsyncSession = Depends(get_db)
 ):
     """
     刷新访问令牌
+    
+    使用 Refresh Token 获取新的 Access Token
     """
     user_id = principal.get("user_id")
+    claims = principal.get("claims", {})
+    
+    # 验证是否为 refresh token（可选，如果前端确保传入的是 refresh token）
+    token_type = claims.get("type")
+    # 为了向后兼容，不强制要求 type 为 refresh
     
     # 查询用户信息
     result = await db.execute(select(User).where(User.id == user_id))
@@ -267,20 +276,20 @@ async def refresh_token(
             detail="用户不存在"
         )
     
-    # 生成新token
-    token = create_access_token(
-        data={
-            "sub": str(user.id),
-            "email": user.email,
-            "role": user.role,
-            "channelId": str(user.channel_id) if user.channel_id else None,
-        }
-    )
+    # 生成新的 access token 和 refresh token
+    token_data = {
+        "sub": str(user.id),
+        "email": user.email,
+        "role": user.role,
+        "channelId": str(user.channel_id) if user.channel_id else None,
+    }
+    new_access_token = create_access_token(data=token_data)
+    new_refresh_token = create_refresh_token(data=token_data)
     
     return SuccessResponse(
         data={
-            "token": token,
-            "refreshToken": token,
+            "token": new_access_token,
+            "refreshToken": new_refresh_token,
         }
     )
 
@@ -431,8 +440,18 @@ async def send_verification_code_endpoint(
     发送邮箱验证码
     
     在用户注册前，先调用此接口发送验证码到邮箱
+    
+    频率限制：同一邮箱60秒内只能发送一次
     """
-    # 检查邮箱是否已存在
+    # 1. 检查发送频率限制
+    can_send, remaining_seconds = await check_rate_limit(email)
+    if not can_send:
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail=f"请等待{remaining_seconds}秒后再重新发送验证码"
+        )
+    
+    # 2. 检查邮箱是否已存在
     result = await db.execute(select(User).where(User.email == email))
     existing_user = result.scalar_one_or_none()
     
@@ -442,7 +461,7 @@ async def send_verification_code_endpoint(
             detail="该邮箱已被注册"
         )
     
-    # 发送验证码
+    # 3. 发送验证码
     code = await send_and_store_verification_code(email)
     if not code:
         raise HTTPException(
@@ -468,15 +487,7 @@ async def register(req: UserCreate, db: AsyncSession = Depends(get_db)):
     - 供应商所有模型
     - 余额：20元
     """
-    # 验证邮箱验证码
-    is_valid = await verify_code(req.email, req.verification_code)
-    if not is_valid:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="验证码错误或已过期"
-        )
-    
-    # 检查邮箱是否已存在
+    # 1. 先检查邮箱是否已存在（不消耗验证码）
     result = await db.execute(select(User).where(User.email == req.email))
     existing_user = result.scalar_one_or_none()
     
@@ -486,15 +497,22 @@ async def register(req: UserCreate, db: AsyncSession = Depends(get_db)):
             detail="该邮箱已被注册"
         )
     
-    # 检查用户名是否已存在
-    if req.username:
-        result = await db.execute(select(User).where(User.username == req.username))
-        existing_username = result.scalar_one_or_none()
-        if existing_username:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="该用户名已被使用"
-            )
+    # 2. 检查用户名是否已存在（username 是必填字段）
+    result = await db.execute(select(User).where(User.username == req.username))
+    existing_username = result.scalar_one_or_none()
+    if existing_username:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="该用户名已被使用"
+        )
+    
+    # 3. 最后验证邮箱验证码（验证成功后会消耗验证码）
+    is_valid = await verify_code(req.email, req.verification_code)
+    if not is_valid:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="验证码错误或已过期"
+        )
     
     # taiji 渠道 ID
     TAIJI_CHANNEL_ID = uuid.UUID("b415e70b-8d37-481c-b229-bc3b7871607b")
@@ -533,8 +551,28 @@ async def register(req: UserCreate, db: AsyncSession = Depends(get_db)):
         total_eu_consumed=0,
     )
     
-    db.add(new_user)
-    await db.flush()  # 获取 user.id
+    try:
+        db.add(new_user)
+        await db.flush()  # 获取 user.id，这里会触发唯一约束检查
+    except IntegrityError as e:
+        await db.rollback()
+        error_str = str(e.orig) if e.orig else str(e)
+        if "email" in error_str.lower() or "users_email_key" in error_str.lower():
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="该邮箱已被注册"
+            )
+        elif "username" in error_str.lower() or "users_username_key" in error_str.lower():
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="该用户名已被使用"
+            )
+        else:
+            logger.error(f"注册时数据库约束冲突: {e}")
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="注册信息与现有用户冲突，请更换邮箱或用户名"
+            )
     
     user_id = new_user.id
     
@@ -673,20 +711,22 @@ async def register(req: UserCreate, db: AsyncSession = Depends(get_db)):
         )
     
     # 创建JWT token，自动登录
-    token = create_access_token(
-        data={
-            "sub": str(new_user.id),
-            "email": new_user.email,
-            "role": new_user.role,
-            "user_id": str(new_user.id),
-            "channelId": str(TAIJI_CHANNEL_ID),
-        }
-    )
+    token_data = {
+        "sub": str(new_user.id),
+        "email": new_user.email,
+        "role": new_user.role,
+        "user_id": str(new_user.id),
+        "channelId": str(TAIJI_CHANNEL_ID),
+    }
+    
+    # 创建 Access Token（短期有效）和 Refresh Token（长期有效）
+    access_token = create_access_token(data=token_data)
+    refresh_token = create_refresh_token(data=token_data)
     
     return SuccessResponse(
         data={
-            "token": token,
-            "refreshToken": token,
+            "token": access_token,
+            "refreshToken": refresh_token,
             "user": {
                 "id": str(new_user.id),
                 "name": new_user.name,

services/mcp-server/app/routes/billing_webhook.py

@@ -10,12 +10,13 @@ from decimal import Decimal
 
 from fastapi import APIRouter, Depends, HTTPException, Request
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import select
+from sqlalchemy import select, update
 from pydantic import BaseModel, Field
 
 from database import get_db
 from models import ModelBillingRecord, TenantModelKey, Balance, User
 from app.schemas import AgentManagerCallbackData, AgentManagerCallbackResponse
+from app.db_utils import ensure_idempotent, atomic_eu_consume, with_retry
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/v1/billing", tags=["计费Webhook"])
@@ -220,46 +221,38 @@ async def process_single_callback(callback_data: LiteLLMCallbackData, db: AsyncS
     except Exception as e:
         logger.warning(f"时间解析失败: {e}")
     
-    # 创建计费记录
-    record = ModelBillingRecord(
-        tenant_id=tenant_id,
-        channel_id=channel_id,
-        litellm_call_id=call_id,
-        api_key=api_key[:8] + "..." + api_key[-4:] if api_key and len(api_key) > 12 else api_key,
-        team_id=team_id,
-        model_name=callback_data.model,
-        input_tokens=prompt_tokens,
-        output_tokens=completion_tokens,
-        total_tokens=total_tokens,
-        total_cost=Decimal(callback_data.response_cost or 0),
-        eu_consumed=eu_consumed,
-        status=callback_data.status,
-        start_time=start_time,
-        end_time=end_time,
-        response_time_ms=int((callback_data.response_time or 0) * 1000),
-        raw_callback_data=callback_data.model_dump() if hasattr(callback_data, 'model_dump') else callback_data.dict()
-    )
-    
-    # ✅ 幂等性检查 - 防止重复处理
-    if call_id:
-        existing = await db.execute(
-            select(ModelBillingRecord).where(
-                ModelBillingRecord.litellm_call_id == call_id
-            )
-        )
-        if existing.scalar_one_or_none():
-            logger.info(f"LiteLLM回调已处理过: {call_id}")
-            return {"message": "Already processed", "call_id": call_id}
-    
     # ✅ 原子操作：同时创建记录和更新余额
     if not tenant_id:
         logger.error(f"无法解析租户ID，跳过计费: call_id={call_id}")
         raise HTTPException(status_code=400, detail="无法解析租户ID")
     
     try:
-        # 查询或创建用户余额记录
+        # 创建计费记录
+        record = ModelBillingRecord(
+            tenant_id=tenant_id,
+            channel_id=channel_id,
+            litellm_call_id=call_id,
+            api_key=api_key[:8] + "..." + api_key[-4:] if api_key and len(api_key) > 12 else api_key,
+            team_id=team_id,
+            model_name=callback_data.model,
+            input_tokens=prompt_tokens,
+            output_tokens=completion_tokens,
+            total_tokens=total_tokens,
+            total_cost=Decimal(callback_data.response_cost or 0),
+            eu_consumed=eu_consumed,
+            status=callback_data.status,
+            start_time=start_time,
+            end_time=end_time,
+            response_time_ms=int((callback_data.response_time or 0) * 1000),
+            raw_callback_data=callback_data.model_dump() if hasattr(callback_data, 'model_dump') else callback_data.dict()
+        )
+        db.add(record)
+        
+        # 使用行锁保护余额更新，防止并发扣款
         balance_result = await db.execute(
-            select(Balance).where(Balance.user_id == tenant_id)
+            select(Balance)
+            .where(Balance.user_id == tenant_id)
+            .with_for_update()  # 行锁
         )
         balance = balance_result.scalar_one_or_none()
         
@@ -267,6 +260,7 @@ async def process_single_callback(callback_data: LiteLLMCallbackData, db: AsyncS
             # 如果余额记录不存在，创建一个新的（初始余额为0）
             balance = Balance(user_id=tenant_id, eu_balance=0)
             db.add(balance)
+            await db.flush()  # 确保记录创建
             logger.warning(f"用户 {tenant_id} 余额记录不存在，已创建初始余额为0")
         
         # 扣减EU余额（使用Decimal精确计算）
@@ -274,20 +268,12 @@ async def process_single_callback(callback_data: LiteLLMCallbackData, db: AsyncS
         new_balance = old_balance - Decimal(str(eu_consumed))
         balance.eu_balance = float(new_balance)
         
-        # 创建计费记录
-        db.add(record)
-        
-        # 同时更新用户表的total_eu_consumed字段
-        user_result = await db.execute(
-            select(User).where(User.id == tenant_id)
+        # 使用原子操作更新用户表的 total_eu_consumed 字段
+        await db.execute(
+            update(User)
+            .where(User.id == tenant_id)
+            .values(total_eu_consumed=User.total_eu_consumed + float(eu_consumed))
         )
-        user = user_result.scalar_one_or_none()
-        if user:
-            user.total_eu_consumed = float(Decimal(str(user.total_eu_consumed or 0)) + Decimal(str(eu_consumed)))
-            logger.info(
-                f"✅ 更新用户EU消耗: user_id={tenant_id}, "
-                f"total_eu_consumed={user.total_eu_consumed}"
-            )
         
         logger.info(
             f"✅ 扣减用户余额: user_id={tenant_id}, "

services/mcp-server/app/routes/channel.py

@@ -5,7 +5,7 @@
 from datetime import datetime, timedelta
 from typing import List, Optional, Dict, Any
 from fastapi import APIRouter, Depends, HTTPException, Query, status
-from sqlalchemy import select, func, and_, desc, or_
+from sqlalchemy import select, func, and_, desc, or_, update
 from sqlalchemy.ext.asyncio import AsyncSession
 import uuid
 import structlog
@@ -349,15 +349,17 @@ async def allocate_tenant_resources(
         # agentId 在这里是模板名称（如 echo_agent）
         template_name = agent_alloc.agentId
         
-        # 获取渠道的平台 Agent 配额
+        # 获取渠道的平台 Agent 配额（使用行锁防止并发更新）
         channel_quota_result = await db.execute(
-            select(PlatformAgentQuota).where(
+            select(PlatformAgentQuota)
+            .where(
                 and_(
                     PlatformAgentQuota.target_id == channel_id,
                     PlatformAgentQuota.target_type == "channel",
                     PlatformAgentQuota.template_name == template_name
                 )
             )
+            .with_for_update()  # 行锁
         )
         channel_quota = channel_quota_result.scalar_one_or_none()
         
@@ -378,15 +380,17 @@ async def allocate_tenant_resources(
             )
             other_quota = other_tenants_quota_result.scalar() or 0
             
-            # 获取当前租户已有配额
+            # 获取当前租户已有配额（使用行锁防止并发更新）
             tenant_quota_result = await db.execute(
-                select(PlatformAgentQuota).where(
+                select(PlatformAgentQuota)
+                .where(
                     and_(
                         PlatformAgentQuota.target_id == tenant_id,
                         PlatformAgentQuota.target_type == "tenant",
                         PlatformAgentQuota.template_name == template_name
                     )
                 )
+                .with_for_update()  # 行锁
             )
             tenant_quota = tenant_quota_result.scalar_one_or_none()
             current_tenant_quota = tenant_quota.pod_quota if tenant_quota else 0
@@ -816,9 +820,11 @@ async def recharge_tenant(
             detail="租户不存在或不属于该渠道"
         )
     
-    # 从 Balance 表获取或创建余额记录
+    # 使用行锁保护余额更新，防止并发充值问题
     balance_result = await db.execute(
-        select(Balance).where(Balance.user_id == tenant_id)
+        select(Balance)
+        .where(Balance.user_id == tenant_id)
+        .with_for_update()  # 行锁
     )
     balance_obj = balance_result.scalar_one_or_none()
     
@@ -826,11 +832,12 @@ async def recharge_tenant(
         # 如果余额记录不存在，创建一个新的
         balance_obj = Balance(user_id=tenant_id, eu_balance=0.0)
         db.add(balance_obj)
+        await db.flush()  # 确保记录创建
     
     # 更新余额（使用 Balance 表）
     old_balance = float(balance_obj.eu_balance)
-    balance_obj.eu_balance = old_balance + req.amount
-    new_balance = balance_obj.eu_balance
+    new_balance = old_balance + req.amount
+    balance_obj.eu_balance = new_balance
     
     # 创建充值记录
     recharge = RechargeRecord(
@@ -3152,15 +3159,17 @@ async def allocate_platform_agent_to_tenant(
             detail=f"平台 Agent 模板 '{req.templateName}' 不存在"
         )
     
-    # 获取渠道的配额
+    # 获取渠道的配额（使用行锁防止并发更新）
     channel_quota_result = await db.execute(
-        select(PlatformAgentQuota).where(
+        select(PlatformAgentQuota)
+        .where(
             and_(
                 PlatformAgentQuota.target_id == channel_id,
                 PlatformAgentQuota.target_type == "channel",
                 PlatformAgentQuota.template_name == req.templateName
             )
         )
+        .with_for_update()  # 行锁
     )
     channel_quota = channel_quota_result.scalar_one_or_none()
     
@@ -3194,15 +3203,17 @@ async def allocate_platform_agent_to_tenant(
             detail=f"配额超出渠道剩余配额。渠道剩余: {remaining}，请求: {req.podQuota}"
         )
     
-    # 查找或创建租户配额记录
+    # 查找或创建租户配额记录（使用行锁防止并发更新）
     tenant_quota_result = await db.execute(
-        select(PlatformAgentQuota).where(
+        select(PlatformAgentQuota)
+        .where(
             and_(
                 PlatformAgentQuota.target_id == tenant_id,
                 PlatformAgentQuota.target_type == "tenant",
                 PlatformAgentQuota.template_name == req.templateName
             )
         )
+        .with_for_update()  # 行锁
     )
     tenant_quota = tenant_quota_result.scalar_one_or_none()
     

services/mcp-server/app/routes/platform_agent_quota.py

@@ -431,30 +431,34 @@ async def allocate_platform_agent_to_tenant(
     if not tenant or tenant.channel_id != channel_uuid:
         raise HTTPException(status_code=404, detail="租户不存在或不属于该渠道")
     
-    # 检查渠道配额
+    # 检查渠道配额（使用行锁防止并发更新）
     channel_quota_result = await db.execute(
-        select(PlatformAgentQuota).where(
+        select(PlatformAgentQuota)
+        .where(
             and_(
                 PlatformAgentQuota.target_id == channel_uuid,
                 PlatformAgentQuota.target_type == "channel",
                 PlatformAgentQuota.template_name == request.templateName
             )
         )
+        .with_for_update()  # 行锁
     )
     channel_quota = channel_quota_result.scalar_one_or_none()
     
     if not channel_quota:
         raise HTTPException(status_code=400, detail=f"渠道没有 {request.templateName} 的配额")
     
-    # 检查租户是否已有该模板的配额
+    # 检查租户是否已有该模板的配额（使用行锁防止并发更新）
     existing_quota_result = await db.execute(
-        select(PlatformAgentQuota).where(
+        select(PlatformAgentQuota)
+        .where(
             and_(
                 PlatformAgentQuota.target_id == tenant_uuid,
                 PlatformAgentQuota.target_type == "tenant",
                 PlatformAgentQuota.template_name == request.templateName
             )
         )
+        .with_for_update()  # 行锁
     )
     existing_quota = existing_quota_result.scalar_one_or_none()
     
@@ -1055,15 +1059,17 @@ async def stop_platform_agent(
     else:
         logger.warning(f"未找到Agent {agent_name} 的计费记录")
     
-    # 更新租户配额
+    # 更新租户配额（使用行锁防止并发更新）
     tenant_quota_result = await db.execute(
-        select(PlatformAgentQuota).where(
+        select(PlatformAgentQuota)
+        .where(
             and_(
                 PlatformAgentQuota.target_id == uuid.UUID(user_id),
                 PlatformAgentQuota.target_type == "tenant",
                 PlatformAgentQuota.template_name == agent.template
             )
         )
+        .with_for_update()  # 行锁
     )
     tenant_quota = tenant_quota_result.scalar_one_or_none()
     if tenant_quota and tenant_quota.pod_used > 0:

services/mcp-server/app/routes/user.py

@@ -1231,15 +1231,17 @@ async def deploy_agent(
     user_id = principal.get("user_id")
     channel_id = principal.get("channel_id")
     
-    # 查询平台Agent配额记录（agentId 是配额ID，来自 /platform-agents/available）
+    # 查询平台Agent配额记录（使用行锁防止并发部署超限）
     quota_result = await db.execute(
-        select(PlatformAgentQuota).where(
+        select(PlatformAgentQuota)
+        .where(
             and_(
                 PlatformAgentQuota.id == req.agentId,
                 PlatformAgentQuota.target_type == "tenant",
                 PlatformAgentQuota.target_id == user_id
             )
         )
+        .with_for_update()  # 行锁
     )
     quota = quota_result.scalar_one_or_none()
     
@@ -2175,15 +2177,17 @@ async def deploy_platform_agent(
     user_id = principal.get("user_id")
     channel_id = principal.get("channel_id")
     
-    # 检查用户是否有该 Agent 类型的配额
+    # 检查用户是否有该 Agent 类型的配额（使用行锁防止并发部署超限）
     quota_result = await db.execute(
-        select(PlatformAgentQuota).where(
+        select(PlatformAgentQuota)
+        .where(
             and_(
                 PlatformAgentQuota.target_type == "tenant",
                 PlatformAgentQuota.target_id == user_id,
                 PlatformAgentQuota.template_name == req.agentType
             )
         )
+        .with_for_update()  # 行锁
     )
     quota = quota_result.scalar_one_or_none()
     
@@ -2355,15 +2359,17 @@ async def use_platform_agent(
     user_id = principal.get("user_id")
     channel_id = principal.get("channel_id")
     
-    # 检查用户是否有该 Agent 类型的配额
+    # 检查用户是否有该 Agent 类型的配额（使用行锁防止并发部署超限）
     quota_result = await db.execute(
-        select(PlatformAgentQuota).where(
+        select(PlatformAgentQuota)
+        .where(
             and_(
                 PlatformAgentQuota.target_type == "tenant",
                 PlatformAgentQuota.target_id == user_id,
                 PlatformAgentQuota.template_name == req.agentType
             )
         )
+        .with_for_update()  # 行锁
     )
     quota = quota_result.scalar_one_or_none()
     
@@ -2496,15 +2502,17 @@ async def stop_platform_agent(
             billing_record.duration_seconds = int(duration)
             billing_record.eu_consumed = _calculate_eu(int(duration))
             
-            # 释放配额
+            # 释放配额（使用行锁防止并发更新配额）
             quota_result = await db.execute(
-                select(PlatformAgentQuota).where(
+                select(PlatformAgentQuota)
+                .where(
                     and_(
                         PlatformAgentQuota.target_type == "tenant",
                         PlatformAgentQuota.target_id == user_id,
                         PlatformAgentQuota.template_name == billing_record.agent_type
                     )
                 )
+                .with_for_update()  # 行锁
             )
             quota = quota_result.scalar_one_or_none()
             if quota and quota.pod_used > 0:

services/mcp-server/migrations/014_add_username_unique_constraint.sql

@@ -0,0 +1,34 @@
+-- Migration: 014_add_username_unique_constraint
+-- Description: 为 users 表的 username 字段添加唯一约束，防止并发注册竞态条件
+-- Date: 2026-01-12
+
+-- 先检查是否已存在该约束
+DO $$
+BEGIN
+    -- 检查是否存在重复的 username（需要先处理）
+    IF EXISTS (
+        SELECT username, COUNT(*) 
+        FROM users 
+        WHERE username IS NOT NULL 
+        GROUP BY username 
+        HAVING COUNT(*) > 1
+    ) THEN
+        RAISE NOTICE '发现重复的 username，请先手动处理重复数据';
+        -- 可以选择自动处理：为重复的 username 添加后缀
+        -- UPDATE users SET username = username || '_' || id::text WHERE ... 
+    END IF;
+    
+    -- 检查约束是否已存在
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_constraint 
+        WHERE conname = 'users_username_key' 
+        AND conrelid = 'users'::regclass
+    ) THEN
+        -- 添加唯一约束（允许 NULL 值，只对非 NULL 值检查唯一性）
+        ALTER TABLE users ADD CONSTRAINT users_username_key UNIQUE (username);
+        RAISE NOTICE '成功添加 username 唯一约束';
+    ELSE
+        RAISE NOTICE 'username 唯一约束已存在，跳过';
+    END IF;
+END $$;
+

services/mcp-server/migrations/run_014_migration.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""
+运行 014 迁移：添加 username 唯一约束
+"""
+import asyncio
+import os
+import sys
+
+# 添加项目根目录到 path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from sqlalchemy import text
+from database import engine
+
+
+async def run_migration():
+    """执行迁移"""
+    migration_sql = """
+    DO $$
+    BEGIN
+        -- 检查是否存在重复的 username（需要先处理）
+        IF EXISTS (
+            SELECT username, COUNT(*) 
+            FROM users 
+            WHERE username IS NOT NULL 
+            GROUP BY username 
+            HAVING COUNT(*) > 1
+        ) THEN
+            RAISE NOTICE '发现重复的 username，请先手动处理重复数据';
+        END IF;
+        
+        -- 检查约束是否已存在
+        IF NOT EXISTS (
+            SELECT 1 FROM pg_constraint 
+            WHERE conname = 'users_username_key' 
+            AND conrelid = 'users'::regclass
+        ) THEN
+            -- 添加唯一约束（允许 NULL 值，只对非 NULL 值检查唯一性）
+            ALTER TABLE users ADD CONSTRAINT users_username_key UNIQUE (username);
+            RAISE NOTICE '成功添加 username 唯一约束';
+        ELSE
+            RAISE NOTICE 'username 唯一约束已存在，跳过';
+        END IF;
+    END $$;
+    """
+    
+    print("=" * 60)
+    print("开始执行迁移: 014_add_username_unique_constraint")
+    print("=" * 60)
+    
+    async with engine.begin() as conn:
+        try:
+            # 先检查是否有重复数据
+            result = await conn.execute(text("""
+                SELECT username, COUNT(*) as cnt
+                FROM users 
+                WHERE username IS NOT NULL 
+                GROUP BY username 
+                HAVING COUNT(*) > 1
+            """))
+            duplicates = result.fetchall()
+            
+            if duplicates:
+                print("\n⚠️  发现重复的 username:")
+                for row in duplicates:
+                    print(f"   - {row[0]}: {row[1]} 条记录")
+                print("\n请先手动处理重复数据，然后重新运行迁移")
+                return False
+            
+            # 执行迁移
+            await conn.execute(text(migration_sql))
+            print("\n✅ 迁移执行成功！")
+            
+            # 验证约束是否创建成功
+            result = await conn.execute(text("""
+                SELECT 1 FROM pg_constraint 
+                WHERE conname = 'users_username_key' 
+                AND conrelid = 'users'::regclass
+            """))
+            if result.fetchone():
+                print("✅ 验证: users_username_key 约束已存在")
+            else:
+                print("❌ 验证失败: 约束未创建")
+                return False
+                
+            return True
+            
+        except Exception as e:
+            print(f"\n❌ 迁移失败: {e}")
+            raise
+
+
+if __name__ == "__main__":
+    success = asyncio.run(run_migration())
+    sys.exit(0 if success else 1)
+

services/mcp-server/models.py

@@ -102,7 +102,7 @@ class User(BaseModel, Base):
     permissions = Column(JSON, default=list)  # 用户权限列表，如 ["use:platform_agents", "read:billing"]
     
     # 兼容旧字段
-    username = Column(String(50))
+    username = Column(String(50), unique=True, nullable=True)  # 添加唯一约束防止并发竞态
     hashed_password = Column(String(255))
     full_name = Column(String(100))
     is_active = Column(Boolean, default=True)