bak

services/mcp-server/app/agent_manager_client.py

@@ -6,6 +6,7 @@ Adapts to the actual Agent Manager interface (based on agent-manager-interface-d
 """
 
 import os
+import uuid
 import httpx
 import structlog
 from typing import Dict, List, Optional, Any
@@ -17,6 +18,10 @@ logger = structlog.get_logger(__name__)
 # Get Agent Manager URL from environment variables
 AGENT_MANAGER_URL = os.getenv("AGENT_MANAGER_URL", "http://localhost:8000")
 
+# 服务身份令牌（契约 §2.1 Option A）：调 agent-manager 必带 Authorization: Bearer。
+# 值与 agent-manager 的 AGNET_RUNTIME_SERVICE_TOKEN 等值，由 k8s secret 注入。
+AGENT_MANAGER_SERVICE_TOKEN = os.getenv("AGENT_MANAGER_SERVICE_TOKEN", "")
+
 # LLM_BASE_URL - 所有 Agent（平台和自定义）都必须传递的固定参数
 LLM_BASE_URL = os.getenv("LLM_BASE_URL", "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io")
 
@@ -587,10 +592,14 @@ class AgentManagerClient:
     async def _get_client(self) -> httpx.AsyncClient:
         """Get or create HTTP client"""
         if self._client is None or self._client.is_closed:
+            headers = {"Content-Type": "application/json"}
+            # 契约 §2.1：服务令牌鉴权（agent-manager v2 强制校验）
+            if AGENT_MANAGER_SERVICE_TOKEN:
+                headers["Authorization"] = f"Bearer {AGENT_MANAGER_SERVICE_TOKEN}"
             self._client = httpx.AsyncClient(
                 base_url=self.base_url,
                 timeout=self.timeout,
-                headers={"Content-Type": "application/json"}
+                headers=headers
             )
         return self._client
     
@@ -629,7 +638,8 @@ class AgentManagerClient:
                 method=method,
                 url=path,
                 json=json,
-                params=params
+                params=params,
+                headers={"X-Correlation-Id": str(uuid.uuid4())}  # 契约 §2.1 必填，全链路追踪
             )
             
             if response.status_code >= 400:

services/mcp-server/app/resource_control.py

@@ -308,7 +308,12 @@ class ResourceController:
                     cost=cost,
                     eu_consumed=float(cost),  # EU = Cost（1 EU = 1 美元）
                     start_time=now,
+                    # 🔧 修复双扣：本记录是一次性执行（费用已在 execute 路径即时扣除），
+                    # 标记 end_time 已结束，避免被周期任务(record_type=vm_runtime AND end_time IS NULL)
+                    # 误判为"运行中 VM"而按时长再扣一次（且永不结束、持续幽灵计费）。
+                    end_time=now,
                     period_start=now,
+                    period_end=now,
                 )
                 db.add(billing_record)
             

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

@@ -403,6 +403,7 @@ async def agent_manager_callback(
     """
     from models import AgentBillingRecord
     from sqlalchemy import and_
+    from sqlalchemy.exc import IntegrityError
     from app.billing import calculate_eu, calculate_platform_agent_cost, deduct_balance
     
     try:
@@ -420,7 +421,22 @@ async def agent_manager_callback(
         if not user:
             logger.warning(f"用户不存在: {callback_data.userId}")
             raise HTTPException(status_code=404, detail=f"用户不存在: {callback_data.userId}")
-        
+
+        # 🔧 幂等：防止 Agent Manager 重发同一 requestId 重复建记录 + 重复扣费
+        # （对齐 LiteLLM 回调按 call_id 去重的做法；request_id 维度的 api_call 记录唯一）
+        if callback_data.requestId:
+            dup = await db.execute(
+                select(AgentBillingRecord).where(
+                    and_(
+                        AgentBillingRecord.record_type == "api_call",
+                        AgentBillingRecord.request_id == callback_data.requestId,
+                    )
+                )
+            )
+            if dup.scalar_one_or_none():
+                logger.info(f"Agent 回调已处理过(幂等跳过): request_id={callback_data.requestId}")
+                return {"success": True, "message": "Already processed", "request_id": callback_data.requestId}
+
         # 解析时间
         start_time = None
         end_time = None
@@ -527,7 +543,14 @@ async def agent_manager_callback(
             )
         
         print(f"\n🔥 准备 commit...", flush=True)
-        await db.commit()
+        try:
+            await db.commit()
+        except IntegrityError:
+            # 🔧 并发安全网：唯一索引 uq_agent_billing_api_call_request_id 命中，
+            # 说明同 request_id 已被另一并发回调写入并扣费 → 幂等跳过，绝不重复扣。
+            await db.rollback()
+            logger.info(f"Agent 回调并发重复(唯一索引拦截，幂等跳过): request_id={callback_data.requestId}")
+            return {"success": True, "message": "Already processed", "request_id": callback_data.requestId}
         print(f"🔥 commit 完成！\n", flush=True)
         await db.refresh(billing_record)
         

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

@@ -1064,18 +1064,35 @@ async def stop_platform_agent(
     billing_record = billing_result.scalar_one_or_none()
     
     if billing_record:
+        # 计费/扣款（与周期任务同一套：增量扣 + 模板费率）
+        from app.billing import calculate_platform_agent_cost, deduct_balance
+
         billing_record.end_time = datetime.utcnow()
         duration = (billing_record.end_time - billing_record.start_time).total_seconds()
         billing_record.duration_seconds = int(duration)
-        # 计算成本: 平台Agent固定费率 $0.10/小时
-        cost = 0.10 * (duration / 3600)
+        # 🔧 修复硬编码 0.10：用模板费率（与周期任务一致），未知模板回退 $0.10/h
+        cost = float(calculate_platform_agent_cost(
+            billing_record.agent_type or agent.template, int(duration)
+        ))
+        # 🔧 修复末段漏扣：周期任务已按累计成本"增量"扣过（已扣额=记录里旧 cost）。
+        # 停机补扣"末段增量 = 全程成本 − 已扣"，使停机时机器费收齐（之前此处完全不扣款）。
+        already_charged = float(billing_record.cost or 0)
+        increment = cost - already_charged
         billing_record.cost = cost
-        # EU = Cost（1 EU = 1 美元）
-        billing_record.eu_consumed = cost
-        
+        billing_record.eu_consumed = cost  # EU = Cost（1 EU = 1 美元）
+
+        if increment > 0:
+            success, message = await deduct_balance(
+                user_id, increment, db,
+                f"平台Agent停机结算(末段增量): {agent_name} (运行{int(duration)}秒)",
+                auto_commit=False  # 与配额释放/Agent 删除在同一事务提交
+            )
+            if not success:
+                logger.warning(f"平台Agent停机扣款失败: {message}, 用户: {user_id}, Agent: {agent_name}")
+
         logger.info(
-            f"结束平台Agent计费: {agent_name}, "
-            f"运行时长={duration}秒, cost=${cost:.4f}, EU={billing_record.eu_consumed}"
+            f"结束平台Agent计费: {agent_name}, 运行时长={duration}秒, "
+            f"cost=${cost:.4f}, 末段增量=${increment:.4f}, EU={billing_record.eu_consumed}"
         )
     else:
         logger.warning(f"未找到Agent {agent_name} 的计费记录")

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

@@ -3486,9 +3486,10 @@ async def delete_custom_agent(
             )
         )
     )
+    billing_record = billing_result.scalar_one_or_none()  # 🔧 修复：之前漏了取值，billing_record 未定义会 NameError
     # 从 namespace 提取完整的 agent 名称（namespace 格式如 "agent-123123123-674ddd"）
     agent_full_name = name
-    if billing_record.namespace and billing_record.namespace.startswith("agent-"):
+    if billing_record and billing_record.namespace and billing_record.namespace.startswith("agent-"):
         agent_full_name = billing_record.namespace[6:]  # 去掉 "agent-" 前缀
         logger.info(f"使用完整 Agent 名称删除: {agent_full_name} (原名: {name}, namespace: {billing_record.namespace})")
     
@@ -3513,26 +3514,34 @@ async def delete_custom_agent(
             detail=f"删除 Agent 失败: {str(e)}"
         )
     
-    # ✅ Step 2: Pod 删除成功后，更新计费记录
-    billing_record.end_time = datetime.utcnow()
-    duration = (billing_record.end_time - billing_record.start_time).total_seconds()
-    billing_record.duration_seconds = int(duration)
-    
-    # ✅ Step 3: 计算成本并扣款（自定义Agent按资源使用量计费）
-    cpu_cores = _parse_cpu_to_cores(billing_record.cpu_used) if billing_record.cpu_used else 0.1
-    memory_gb = _parse_memory_to_gb(billing_record.memory_used) if billing_record.memory_used else 0.125
-    cost = calculate_agent_cost_by_resources(cpu_cores, memory_gb, int(duration))
-    billing_record.cost = float(cost)
-    billing_record.eu_consumed = float(cost)  # EU = Cost（1 EU = 1 美元）
-    
-    # 扣除用户余额
-    success, message = await deduct_balance(
-        user_id, cost, db,
-        f"自定义Agent使用: {name} (运行{int(duration)}秒)",
-        auto_commit=False  # 🔧 修复：与billing_record和quota更新在同一事务中
-    )
-    if not success:
-        logger.warning(f"扣款失败: {message}, 用户: {user_id}, Agent: {name}")
+    # ✅ Step 2+3: 结算计费记录（仅当找到运行中记录时）
+    if billing_record:
+        billing_record.end_time = datetime.utcnow()
+        duration = (billing_record.end_time - billing_record.start_time).total_seconds()
+        billing_record.duration_seconds = int(duration)
+
+        # 自定义Agent按资源使用量计费，cost 为 start→stop 全程累计成本
+        cpu_cores = _parse_cpu_to_cores(billing_record.cpu_used) if billing_record.cpu_used else 0.1
+        memory_gb = _parse_memory_to_gb(billing_record.memory_used) if billing_record.memory_used else 0.125
+        full_cost = float(calculate_agent_cost_by_resources(cpu_cores, memory_gb, int(duration)))
+
+        # 🔧 修复双扣：周期任务已按累计成本"增量"扣过（已扣额=记录里旧的 cost）。
+        # 停机只补扣"末段增量 = 全程成本 − 已扣"，绝不再整笔重扣（否则用户被多扣近一倍）。
+        already_charged = float(billing_record.cost or 0)
+        increment = full_cost - already_charged
+        billing_record.cost = full_cost
+        billing_record.eu_consumed = full_cost  # EU = Cost（1 EU = 1 美元）
+
+        if increment > 0:
+            success, message = await deduct_balance(
+                user_id, increment, db,
+                f"自定义Agent停机结算(末段增量): {name} (运行{int(duration)}秒)",
+                auto_commit=False  # 与billing_record和quota更新在同一事务中
+            )
+            if not success:
+                logger.warning(f"扣款失败: {message}, 用户: {user_id}, Agent: {name}")
+    else:
+        logger.warning(f"未找到 Agent {name} 的运行中计费记录，跳过结算")
     
     # ✅ Step 4: 释放配额（使用行锁防止并发问题）
     try:
@@ -3597,13 +3606,20 @@ async def scale_custom_agent_api(
             )
         )
     )
+    billing_record = billing_result.scalar_one_or_none()  # 🔧 修复：之前漏了取值，3623 用 .cpu_used 会 NameError
+    if not billing_record:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"未找到 Agent {name} 或该 Agent 不属于您"
+        )
+
     quota_result = await db.execute(
         select(TenantCustomAgentQuota)
         .where(TenantCustomAgentQuota.tenant_id == user_id)
         .with_for_update()  # ✅ 添加行锁，防止并发扩缩容导致配额计算错误
     )
     quota = quota_result.scalar_one_or_none()
-    
+
     if not quota:
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,

services/mcp-server/config.py

@@ -65,6 +65,9 @@ class Settings(BaseSettings):
     # AI Agent Manager API 设置（K8s Pod 管理）
     agent_manager_url: str = os.getenv("AGENT_MANAGER_URL", "http://localhost:8000")
     agent_manager_timeout: float = 30.0  # 秒
+    # 服务身份令牌：调 agent-manager 时带 Authorization: Bearer（契约 §2.1 Option A）
+    # 值需与 agent-manager 的 AGNET_RUNTIME_SERVICE_TOKEN 等值，由 secret 注入
+    agent_manager_service_token: str = os.getenv("AGENT_MANAGER_SERVICE_TOKEN", "")
     agent_default_cpu_request: str = "100m"
     agent_default_cpu_limit: str = "500m"
     agent_default_memory_request: str = "128Mi"

services/mcp-server/migrations/029_add_api_call_request_id_unique.sql

@@ -0,0 +1,27 @@
+-- Migration 029: agent_billing_records.request_id 部分唯一索引（api_call 幂等）
+--
+-- 背景：
+--   /agent-callback（API 调用计费）此前没有 request_id 去重，Agent Manager 重发同一
+--   requestId 的回调会重复创建 api_call 记录并重复扣费。
+--   应用层已在 billing_webhook.py 加 SELECT 去重（处理顺序重试），本迁移再加 DB 级
+--   部分唯一索引作为并发安全网（防两个相同 requestId 的回调并发插入）。
+--
+-- ⚠️ 注意：本迁移会先删除已存在的重复 api_call 记录（同 request_id 仅保留最早一条），
+--          再建唯一索引。删除的是因历史 bug 产生的重复计费记录（不影响已扣的余额）。
+
+-- 1. 去重：同一 request_id 的 api_call 记录只保留最早一条（按 created_at，平局取最小 id）
+DELETE FROM agent_billing_records a
+USING agent_billing_records b
+WHERE a.record_type = 'api_call'
+  AND b.record_type = 'api_call'
+  AND a.request_id IS NOT NULL AND a.request_id <> ''
+  AND a.request_id = b.request_id
+  AND (a.created_at > b.created_at OR (a.created_at = b.created_at AND a.id > b.id));
+
+-- 2. 部分唯一索引：api_call 且 request_id 非空时，request_id 唯一
+CREATE UNIQUE INDEX IF NOT EXISTS uq_agent_billing_api_call_request_id
+ON agent_billing_records (request_id)
+WHERE record_type = 'api_call' AND request_id IS NOT NULL AND request_id <> '';
+
+COMMENT ON INDEX uq_agent_billing_api_call_request_id IS
+  'api_call 计费按 request_id 幂等，防止 Agent Manager 重发回调重复扣费';

services/mcp-server/migrations/run_029_add_api_call_request_id_unique.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""
+Migration 029: agent_billing_records.request_id 部分唯一索引（api_call 幂等）
+
+背景：
+  /agent-callback（API 调用计费）此前无 request_id 去重，Agent Manager 重发同一
+  requestId 会重复创建 api_call 记录并重复扣费。应用层已加 SELECT 去重，本迁移再加
+  DB 级部分唯一索引作为并发安全网。
+
+⚠️ 会先删除重复的 api_call 记录（同 request_id 仅保留最早一条），再建唯一索引。
+
+使用方法：
+    cd services/mcp-server
+    python migrations/run_029_add_api_call_request_id_unique.py
+"""
+
+import asyncio
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+
+
+INDEX_NAME = "uq_agent_billing_api_call_request_id"
+
+
+async def run_migration():
+    database_url = os.environ.get("DATABASE_URL")
+    if not database_url:
+        print("❌ 错误：未设置 DATABASE_URL 环境变量")
+        return False
+
+    if database_url.startswith("postgresql://"):
+        database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
+    if "sslmode=" in database_url:
+        import re
+        database_url = re.sub(r'[?&]sslmode=[^&]*', '', database_url)
+        database_url = database_url.rstrip('?&')
+
+    print("📦 连接数据库...")
+    engine = create_async_engine(database_url, echo=False)
+    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+    async with async_session() as session:
+        try:
+            # 1. 索引已存在则跳过
+            print("\n🔍 检查唯一索引是否已存在...")
+            result = await session.execute(text("""
+                SELECT 1 FROM pg_indexes
+                WHERE tablename = 'agent_billing_records' AND indexname = :name
+            """), {"name": INDEX_NAME})
+            if result.fetchone():
+                print(f"✅ 索引 {INDEX_NAME} 已存在，跳过迁移")
+                return True
+
+            # 2. 统计重复
+            print("\n📊 统计重复的 api_call 记录（同 request_id）...")
+            result = await session.execute(text("""
+                SELECT COALESCE(SUM(cnt - 1), 0) AS dup_rows
+                FROM (
+                    SELECT request_id, COUNT(*) AS cnt
+                    FROM agent_billing_records
+                    WHERE record_type = 'api_call'
+                      AND request_id IS NOT NULL AND request_id <> ''
+                    GROUP BY request_id
+                    HAVING COUNT(*) > 1
+                ) t
+            """))
+            dup_rows = result.scalar() or 0
+            print(f"   将删除的重复记录数: {dup_rows}")
+
+            # 3. 去重（同 request_id 仅保留最早一条）
+            print("\n🔄 去重 api_call 记录...")
+            result = await session.execute(text("""
+                DELETE FROM agent_billing_records a
+                USING agent_billing_records b
+                WHERE a.record_type = 'api_call'
+                  AND b.record_type = 'api_call'
+                  AND a.request_id IS NOT NULL AND a.request_id <> ''
+                  AND a.request_id = b.request_id
+                  AND (a.created_at > b.created_at
+                       OR (a.created_at = b.created_at AND a.id > b.id))
+            """))
+            print(f"   实际删除: {result.rowcount} 条")
+
+            # 4. 建部分唯一索引
+            print("\n🔄 创建部分唯一索引...")
+            await session.execute(text(f"""
+                CREATE UNIQUE INDEX IF NOT EXISTS {INDEX_NAME}
+                ON agent_billing_records (request_id)
+                WHERE record_type = 'api_call' AND request_id IS NOT NULL AND request_id <> ''
+            """))
+            print(f"✅ 创建唯一索引: {INDEX_NAME}")
+
+            await session.commit()
+            print("\n✅ 迁移 029 成功完成！api_call 现按 request_id DB 级幂等。")
+            return True
+
+        except Exception as e:
+            await session.rollback()
+            print(f"\n❌ 迁移失败: {e}")
+            import traceback
+            traceback.print_exc()
+            return False
+
+
+if __name__ == "__main__":
+    success = asyncio.run(run_migration())
+    sys.exit(0 if success else 1)