From 697ea5ac7bcb66e4741a138b31e89288eefd6349 Mon Sep 17 00:00:00 2001 From: chenchen Date: Wed, 24 Jun 2026 18:01:46 +0800 Subject: [PATCH] bak --- .claude/scheduled_tasks.lock | 1 - .claude/settings.local.json | 51 +++++++- ...gic-link邮箱登录-给mcp-server的对接需求.md | 19 +++ k8s/prod/mcp-server.yaml | 10 +- .../mcp-server/app/agent_manager_client.py | 14 ++- services/mcp-server/app/resource_control.py | 5 + .../mcp-server/app/routes/billing_webhook.py | 27 ++++- .../app/routes/platform_agent_quota.py | 31 +++-- services/mcp-server/app/routes/user.py | 60 ++++++---- services/mcp-server/config.py | 3 + .../029_add_api_call_request_id_unique.sql | 27 +++++ .../run_029_add_api_call_request_id_unique.py | 113 ++++++++++++++++++ 12 files changed, 324 insertions(+), 37 deletions(-) delete mode 100644 .claude/scheduled_tasks.lock create mode 100644 services/mcp-server/migrations/029_add_api_call_request_id_unique.sql create mode 100644 services/mcp-server/migrations/run_029_add_api_call_request_id_unique.py diff --git a/.claude/scheduled_tasks.lock b/.claude/scheduled_tasks.lock deleted file mode 100644 index 0a4a6d6..0000000 --- a/.claude/scheduled_tasks.lock +++ /dev/null @@ -1 +0,0 @@ -{"sessionId":"1de9258a-6dbf-4827-ab74-de844edd79c7","pid":30640,"acquiredAt":1777959383082} \ No newline at end of file diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 36d1958..751848e 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -14,7 +14,56 @@ "Bash(kubectl logs *)", "Bash(netstat -ano)", "Bash(awk '{print $5}')", - "Bash(xargs -r -I PID powershell.exe -Command \"Stop-Process -Id PID -Force -ErrorAction SilentlyContinue\")" + "Bash(xargs -r -I PID powershell.exe -Command \"Stop-Process -Id PID -Force -ErrorAction SilentlyContinue\")", + "Bash(kubectl get *)", + "Bash(python3 -c \"import sys, json; d=json.loads\\(sys.stdin.read\\(\\)\\); print\\('CORS_ORIGINS:', d.get\\('CORS_ORIGINS', '[unset → defaults to *]'\\)\\); print\\('APP_ENV:', d.get\\('APP_ENV', '?'\\)\\)\")", + "Bash(kubectl set *)", + "Bash(kubectl rollout *)", + "Bash(MSYS_NO_PATHCONV=1 kubectl exec -n taiji-ai deploy/mcp-server -- python -c \"from app.routes import heicode_proxy; print\\(sorted\\([r.path for r in heicode_proxy.router.routes if hasattr\\(r,'path'\\)]\\)\\)\")", + "Bash(MSYS_NO_PATHCONV=1 kubectl exec -n taiji-ai deploy/mcp-server -- python -c \"from app.routes import agnet_stub; print\\(sorted\\([r.path for r in agnet_stub.router.routes if hasattr\\(r,'path'\\)]\\)\\)\")", + "Bash(python3 -c \"import ast; ast.parse\\(open\\('services/mcp-server/app/routes/agnet_stub.py', encoding='utf-8'\\).read\\(\\)\\); print\\('OK'\\)\")", + "Bash(kubectl port-forward *)", + "Bash(kubectl exec -i -n taiji-ai deploy/mcp-server -- python -c ' *)", + "Bash(MSYS_NO_PATHCONV=1 kubectl exec -n taiji-ai deploy/mcp-server -- python -c \"from app.routes import heicode_tasks; print\\(sorted\\([r.path for r in heicode_tasks.router.routes if hasattr\\(r,'path'\\)]\\)\\)\")", + "Bash(python3 -c ' *)", + "Bash(kubectl exec -n taiji-ai deploy/mcp-server -- python -c ' *)", + "Bash(MSYS_NO_PATHCONV=1 kubectl exec -n taiji-ai deploy/mcp-server -- python -c \"from app.routes import heicode_events; print\\(sorted\\(set\\(r.path for r in heicode_events.router.routes if hasattr\\(r,'path'\\)\\)\\)\\)\")", + "Bash(curl -sk --max-time 30 -x http://127.0.0.1:6578 https://aks.taiji-ai.com/api/mcp/health -w \"\\\\nHTTP=%{http_code}\\\\n\")", + "Bash(curl -sk --max-time 15 -v -x http://127.0.0.1:6578 https://aks.taiji-ai.com/api/mcp/health)", + "Bash(curl -s --noproxy '*' --max-time 10 http://127.0.0.1:18080/health -w \"\\\\nHTTP=%{http_code}\\\\n\")", + "Bash(kill 1883)", + "Bash(pkill -f \"port-forward.*mcp-server\")", + "Bash(kill 2231)", + "Bash(kubectl exec -n taiji-ai deploy/mcp-server -c mcp-server -- python -c ' *)", + "Bash(curl -sk --noproxy '*' --max-time 8 -o /dev/null -w \"agent-manager.taijiagnet.com HTTP=%{http_code} TIME=%{time_total}\\\\n\" https://agent-manager.taijiagnet.com/)", + "Bash(curl -sk --noproxy '*' --max-time 8 -o /dev/null -w \" /health HTTP=%{http_code}\\\\n\" https://agent-manager.taijiagnet.com/health)", + "Bash(curl -sk --noproxy '*' --max-time 8 -o /dev/null -w \" /api/agnet/deployments HTTP=%{http_code}\\\\n\" https://agent-manager.taijiagnet.com/api/agnet/deployments)", + "Bash(nslookup agent-manager.taijiagnet.com)", + "Bash(kubectl exec *)", + "Bash(curl -sk --noproxy '*' --max-time 8 --resolve agent-manager.taijiagnet.com:80:20.212.121.126 -i http://agent-manager.taijiagnet.com/)", + "Bash(curl -sk --noproxy '*' --max-time 8 --resolve agent-manager.taijiagnet.com:80:20.212.121.126 -w '\\\\n[HTTP=%{http_code}]\\\\n' http://agent-manager.taijiagnet.com/templates)", + "Bash(curl -sk --noproxy '*' --max-time 5 --resolve agent-manager.taijiagnet.com:80:20.212.121.126 -o /tmp/body -w '%{http_code}|%{size_download}' http://agent-manager.taijiagnet.com__TRACKED_VAR__)", + "Read(//tmp/**)", + "Bash(curl -sk --noproxy '*' --max-time 8 --resolve agent-manager.taijiagnet.com:80:20.212.121.126 http://agent-manager.taijiagnet.com/openapi.json -o /tmp/oa.json)", + "Bash(python -c ' *)", + "Bash(curl -sk --noproxy '*' --max-time 8 --resolve agent-manager.taijiagnet.com:80:20.212.121.126 http://agent-manager.taijiagnet.com/openapi.json)", + "Bash(MSYS_NO_PATHCONV=1 kubectl exec -n taiji-ai deploy/mcp-server -c mcp-server -- ls //app/scripts/)", + "Bash(MSYS_NO_PATHCONV=1 kubectl exec -n taiji-ai deploy/mcp-server -c mcp-server -- python //app/scripts/cleanup_orphan_litellm_keys.py --dry-run)", + "Bash(MSYS_NO_PATHCONV=1 kubectl exec -n taiji-ai deploy/mcp-server -c mcp-server -- bash -c \"cd /app && python -m scripts.cleanup_orphan_litellm_keys --dry-run\")", + "Bash(MSYS_NO_PATHCONV=1 kubectl exec -n taiji-ai deploy/mcp-server -c mcp-server -- bash -c \"cd /app && python -m scripts.cleanup_orphan_litellm_keys\")", + "Bash(pkill -f \"port-forward\")", + "Bash(curl -s --noproxy '*' --max-time 5 http://127.0.0.1:18083/health)", + "Bash(MSYS_NO_PATHCONV=1 kubectl logs -n taiji-ai deploy/mcp-server --tail=200)", + "Bash(MSYS_NO_PATHCONV=1 kubectl cp \"C:/Users/陈晨/AppData/Local/Temp/check_smoke_user.py\" \"taiji-ai/$\\(kubectl get pod -n taiji-ai -l app=mcp-server -o jsonpath='{.items[0].metadata.name}'\\):/tmp/check_smoke_user.py\")", + "Bash(MSYS_NO_PATHCONV=1 kubectl exec -n taiji-ai deploy/mcp-server -c mcp-server -- python /tmp/check_smoke_user.py smoke_1778571031@example.com)", + "Bash(git add *)", + "Bash(git commit -m ' *)", + "Bash(git push *)", + "Bash(unset HTTP_PROXY HTTPS_PROXY http_proxy https_proxy)", + "Bash(curl -sk --noproxy '*' --max-time 8 __TRACKED_VAR__/api/user/heicode/models -o /dev/null -w 'HTTP=%{http_code} \\(no auth\\)\\\\n')", + "Bash(curl -sk --noproxy '*' --max-time 15 -X POST __TRACKED_VAR__/api/auth/login -H 'Content-Type: application/json' -d '{\"email\":\"smoke_1778571031@example.com\",\"password\":\"GoodPwd1234\"}' -w '\\\\nHTTP=%{http_code}\\\\n')", + "Bash(python -c \"import sys,json,base64; d=json.load\\(sys.stdin\\); print\\({k: base64.b64decode\\(v\\).decode\\(\\)[:8]+'...' for k,v in d.items\\(\\)}\\)\")", + "Bash(python -c \"import ast; ast.parse\\(open\\('app/magic_link.py',encoding='utf-8'\\).read\\(\\)\\); ast.parse\\(open\\('app/routes/auth.py',encoding='utf-8'\\).read\\(\\)\\); ast.parse\\(open\\('config.py',encoding='utf-8'\\).read\\(\\)\\); print\\('syntax OK'\\)\")" ] } } diff --git a/Docs/Heicode-magic-link邮箱登录-给mcp-server的对接需求.md b/Docs/Heicode-magic-link邮箱登录-给mcp-server的对接需求.md index 5d133c8..18d9918 100644 --- a/Docs/Heicode-magic-link邮箱登录-给mcp-server的对接需求.md +++ b/Docs/Heicode-magic-link邮箱登录-给mcp-server的对接需求.md @@ -668,3 +668,22 @@ HM 已迁到 AKS,生产域名: ## 24.4 状态 落地域名改 `code.heicode.cc` **已上生产**;`request/verify/landing` 契约不变;只待 HM 验 §24.3 的"成功 302 经新反代透传"。 + +--- + +# 25. HM owner @zsbgnw12 — 验 §24.3:**查出 bug 并已修**(2026-06-12) + +§24.2 的安全判断同意:`/request` 可经 APIM 直达,裸读 `X-Forwarded-Host` 拼落地域名会被伪造钓鱼;你们选 23.4 域名常量更稳,**采纳**。(HM 侧 §23.3 的 X-Forwarded 透传保留无害——你们用常量、不读它;若将来要 23.3 自动双域名,确实需"白名单版",再排期。) + +## 25.1 §24.3 验证结果:HM 反代之前**不能**原样透回 302(已修) +按你们提示验了 `code.heicode.cc/api/heicode-auth` 的成功 302 透传,**查出两个 bug**: +1. `HeicodeAuthProxy` 的 `http.Client` 用 Go 默认 `CheckRedirect`(**自动 follow**)→ 收到 `302 Location: heicode://...` 会去 follow `heicode://`(非 http scheme)→ 失败 → 反代回 `502 BadGateway`(深链根本到不了浏览器)。 +2. 反代只回传 `Content-Type`,**未回传 `Location`** → 即便不 follow,浏览器也拿不到深链。 + +→ 即:**§24.3 担心的"成功 302 经反代透回"在修之前是断的。** 已修复(HM PR #78): +- `CheckRedirect=ErrUseLastResponse`:不 follow,把上游 302 **原样**返回浏览器; +- 透回 `Location`(深链)+ `Set-Cookie`(会话)+ `Cache-Control` + `Content-Type`; +- 与 §11 4 条一致;非重定向(/me、/refresh、/login、verify 的 JSON)行为不变。单测覆盖:`heicode://` Location 原样透回 + JSON 透传。 + +## 25.2 待办(HM 侧) +PR #78 待评审合并 + **部署到 AKS 生产 HM** 后,`code.heicode.cc/api/heicode-auth` 上的 landing 成功 302→`heicode://` 才会真正透回浏览器。**部署前**,经 code.heicode.cc 点 magic-link 邮件链接的成功跳转会拿到 502(失败分支 HTML 不受影响)。我会尽快合并 + 部署,完成后在此 §25.3 标注。mcp 侧无需再改。 diff --git a/k8s/prod/mcp-server.yaml b/k8s/prod/mcp-server.yaml index a5355ff..a9b7868 100644 --- a/k8s/prod/mcp-server.yaml +++ b/k8s/prod/mcp-server.yaml @@ -31,7 +31,7 @@ spec: spec: containers: - name: mcp-server - image: taiji.azurecr.io/mcp-server:magic-link-20260609-arm64-v2 + image: taiji.azurecr.io/mcp-server:agnet-auth-20260623-arm64 imagePullPolicy: Always ports: - containerPort: 8000 @@ -111,7 +111,13 @@ spec: configMapKeyRef: name: taiji-config key: AGENT_K8S_NAMESPACE - + # 服务令牌：调 agent-manager 鉴权（契约 §2.1，= agent-manager 的 AGNET_RUNTIME_SERVICE_TOKEN） + - name: AGENT_MANAGER_SERVICE_TOKEN + valueFrom: + secretKeyRef: + name: taiji-secrets + key: agent-manager-service-token + # JWT配置 - name: SECRET_KEY valueFrom: diff --git a/services/mcp-server/app/agent_manager_client.py b/services/mcp-server/app/agent_manager_client.py index 905798a..f56757f 100644 --- a/services/mcp-server/app/agent_manager_client.py +++ b/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: diff --git a/services/mcp-server/app/resource_control.py b/services/mcp-server/app/resource_control.py index 2982a14..87dc597 100644 --- a/services/mcp-server/app/resource_control.py +++ b/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) diff --git a/services/mcp-server/app/routes/billing_webhook.py b/services/mcp-server/app/routes/billing_webhook.py index f41f486..5276a10 100644 --- a/services/mcp-server/app/routes/billing_webhook.py +++ b/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) diff --git a/services/mcp-server/app/routes/platform_agent_quota.py b/services/mcp-server/app/routes/platform_agent_quota.py index fbe4770..e9111b8 100644 --- a/services/mcp-server/app/routes/platform_agent_quota.py +++ b/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} 的计费记录") diff --git a/services/mcp-server/app/routes/user.py b/services/mcp-server/app/routes/user.py index 27bc1a1..cd18447 100644 --- a/services/mcp-server/app/routes/user.py +++ b/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, diff --git a/services/mcp-server/config.py b/services/mcp-server/config.py index 9f016df..1517daf 100644 --- a/services/mcp-server/config.py +++ b/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" diff --git a/services/mcp-server/migrations/029_add_api_call_request_id_unique.sql b/services/mcp-server/migrations/029_add_api_call_request_id_unique.sql new file mode 100644 index 0000000..a96dde0 --- /dev/null +++ b/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 重发回调重复扣费'; diff --git a/services/mcp-server/migrations/run_029_add_api_call_request_id_unique.py b/services/mcp-server/migrations/run_029_add_api_call_request_id_unique.py new file mode 100644 index 0000000..21ee415 --- /dev/null +++ b/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)