fix(gateway): AI网关高危修复——并发信号量/预算记账/流式限流/SSE资源释放/事件循环阻塞/请求体限制/降级链容错

- GW-H1: 启用 Phase3 并发信号量（ai_semaphore/ai_heavy_semaphore 统一入口保护，视频/多模态走 heavy 上限）
- GW-H2: 预算记账 user_id 从 request.state 取真实用户，BudgetMiddleware 日限额恢复生效
- GW-H3: 流式通配路径 /{feature}/stream 复用中间件 Redis 限流逻辑，堵住免限流漏洞
- GW-H4: SSE 所有退出路径（超时/断连/异常）关闭上游生成器 aclose，消除连接泄漏与幽灵计费；streaming.py 补 is_disconnected 检测
- GW-H5: gemini generate_vision_multi 同步阻塞改专用线程池执行，事件循环不再冻结
- GW-H6: Provider SDK 专用线程池隔离慢调用 + google-genai HttpOptions 120s 客户端超时
- GW-H7: 多模态/ASR/视觉端点纳入输入校验（64MB/32MB/32MB 前缀上限），base64/列表字段加 max_length
- GW-H8: 视频上传线程内流式计数复制（chunked 绕过 content-length 时由计数兜底），文件句柄 with 管理
- GW-H9: ffprobe/ffmpeg 改 create_subprocess_exec 异步执行，超时强杀防僵尸进程
- GW-H10: FallbackProvider.generate/generate_stream 补 **kwargs，云端全挂时降级内容正常返回

测试: python -m pytest tests/ -q → 193 passed

Author: Aparencia

server/ai-gateway/chains/video_analyze_chain.py

@@ -13,6 +13,7 @@
 使用 ffmpeg 均匀抽取关键帧后转为多图分析。
 """
 
+import asyncio
 import base64
 import logging
 import tempfile
@@ -30,8 +31,32 @@
 # 降级抽帧配置：均匀抽取的帧数上限
 _FALLBACK_FRAME_COUNT = 10
 
+# GW-H9: 子进程超时（秒）——ffprobe 时长探测与 ffmpeg 单帧抽取
+_PROBE_TIMEOUT = 30.0
+_FFMPEG_FRAME_TIMEOUT = 15.0
 
-def _extract_frames_ffmpeg(video_path: str, frame_count: int) -> list[str]:
+
+async def _run_subprocess(cmd: list[str], timeout: float) -> tuple[int, bytes, bytes]:
+    """异步执行子进程并等待完成（带超时，超时后强杀防僵尸进程）。
+
+    GW-H9: 原 subprocess.run 同步阻塞事件循环（单次最长 30s），
+    并发视频分析会相互拖垮；改为 create_subprocess_exec 非阻塞执行。
+    """
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+    except asyncio.TimeoutError:
+        proc.kill()
+        await proc.wait()
+        raise TimeoutError(f"子进程超时（{timeout:.0f}s）: {cmd[0]}")
+    return proc.returncode or 0, stdout, stderr
+
+
+async def _extract_frames_ffmpeg(video_path: str, frame_count: int) -> list[str]:
     """
     使用 ffmpeg 从视频中均匀抽取关键帧（返回 base64 列表）
 
@@ -44,8 +69,6 @@ def _extract_frames_ffmpeg(video_path: str, frame_count: int) -> list[str]:
     Raises:
         RuntimeError: ffmpeg 不可用或抽帧失败
     """
-    import subprocess
-
     try:
         # 获取视频时长
         probe_cmd = [
@@ -54,10 +77,10 @@ def _extract_frames_ffmpeg(video_path: str, frame_count: int) -> list[str]:
             "-of", "default=noprint_wrappers=1:nokey=1",
             video_path,
         ]
-        result = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=30)
-        if result.returncode != 0:
-            raise RuntimeError(f"ffprobe 获取时长失败: {result.stderr}")
-        duration = float(result.stdout.strip())
+        code, stdout, stderr = await _run_subprocess(probe_cmd, _PROBE_TIMEOUT)
+        if code != 0:
+            raise RuntimeError(f"ffprobe 获取时长失败: {stderr.decode(errors='replace')}")
+        duration = float(stdout.decode(errors="replace").strip())
 
         # 均匀间隔抽帧
         interval = max(duration / (frame_count + 1), 1.0)
@@ -75,7 +98,11 @@ def _extract_frames_ffmpeg(video_path: str, frame_count: int) -> list[str]:
                     str(out_path),
                     "-y",
                 ]
-                subprocess.run(cmd, capture_output=True, timeout=15, check=True)
+                code, _, stderr = await _run_subprocess(cmd, _FFMPEG_FRAME_TIMEOUT)
+                if code != 0:
+                    raise RuntimeError(
+                        f"ffmpeg 抽帧失败: {stderr.decode(errors='replace')}"
+                    )
                 if out_path.exists():
                     frames.append(base64.b64encode(out_path.read_bytes()).decode())
 
@@ -87,6 +114,8 @@ def _extract_frames_ffmpeg(video_path: str, frame_count: int) -> list[str]:
 
     except FileNotFoundError as e:
         raise RuntimeError("ffmpeg 未安装，无法进行视频抽帧降级") from e
+    except asyncio.TimeoutError as e:
+        raise RuntimeError(f"视频抽帧超时: {e}") from e
     except Exception as e:
         raise RuntimeError(f"视频抽帧失败: {e}") from e
 
@@ -171,7 +200,7 @@ async def run(
             )
 
         try:
-            frames = _extract_frames_ffmpeg(str(video_input), _FALLBACK_FRAME_COUNT)
+            frames = await _extract_frames_ffmpeg(str(video_input), _FALLBACK_FRAME_COUNT)
         except RuntimeError as e:
             raise RuntimeError(f"视频分析失败：Provider 不支持视频且抽帧不可用: {e}") from e
 

server/ai-gateway/config/fallback.py

@@ -22,6 +22,31 @@
 
 _FIRST_TOKEN_RETRY_TIMEOUT = 5.0  # 首 token 探测重试的超时预算（秒）
 
+# 高耗时功能（走独立并发上限 ai_heavy_semaphore）
+_HEAVY_CONCURRENCY_FEATURES: frozenset[str] = frozenset(
+    {"video_analyze", "multimodal_analyze"}
+)
+
+
+def _get_concurrency_semaphore(app, feature: str):
+    """获取并发信号量；未初始化（如测试环境）时返回 None 表示不限流。
+
+    @ai-context: Phase3 并发控制——主信号量限制全部 AI 调用，
+    heavy 信号量进一步限制高耗时功能（视频/多模态分析）。
+    """
+    if feature in _HEAVY_CONCURRENCY_FEATURES:
+        return getattr(app.state, "ai_heavy_semaphore", None)
+    return getattr(app.state, "ai_semaphore", None)
+
+
+async def _run_under_semaphore(app, feature: str, coro_factory: Callable[[], Awaitable[Any]]) -> Any:
+    """在信号量保护下执行 AI 调用（信号量缺失时直通）。"""
+    sem = _get_concurrency_semaphore(app, feature)
+    if sem is None:
+        return await coro_factory()
+    async with sem:
+        return await coro_factory()
+
 # ============================================================
 # Provider Fallback 链：主 Provider 失败时依次尝试的备选
 # ============================================================
@@ -109,6 +134,7 @@ async def call_with_fallback(
     app,
     feature: str,
     fn: Callable[..., Awaitable[dict[str, Any]]],
+    user_id: str = "system",
 ) -> tuple[dict[str, Any], str]:
     """
     使用 Provider fallback 链执行 AI 调用（非流式路径）。
@@ -122,6 +148,7 @@ async def call_with_fallback(
         app:     FastAPI 应用实例（通过 app.state.providers 获取 Provider）
         feature: 功能标识，对应 PROVIDER_FALLBACK_CHAIN 的 key
         fn:      异步可调用对象，签名为 async fn(provider, model_name) -> dict
+        user_id: 记账归属用户（预算控制按此维度计数）
 
     Returns:
         tuple: (result_dict, provider_key)
@@ -168,18 +195,22 @@ async def _run_fallback_chain():
             )
 
             try:
-                _FEATURE_CONTEXT.set(feature)
-                result = await fn(provider, model_name)
+                async def _do_call():
+                    _FEATURE_CONTEXT.set(feature)
+                    return await fn(provider, model_name)
+
+                # Phase3: 并发信号量保护（视频/多模态走 heavy 上限）
+                result = await _run_under_semaphore(app, feature, _do_call)
                 # 调用成功：通知熔断器重置失败计数，恢复 Provider 健康状态
                 cb = get_circuit(provider_key)
                 if cb:
                     await cb.on_success()
-                # 记录成本
+                # 记录成本（按真实用户记账，预算中间件按 user_id 维度查询）
                 try:
                     tokens_used = result.get("tokens_used", 0)
                     model_name = result.get("model", model_name)
                     await get_cost_tracker().record(
-                        user_id="system",
+                        user_id=user_id,
                         feature=feature,
                         model=model_name,
                         input_tokens=tokens_used // 2 if tokens_used else 0,
@@ -235,7 +266,10 @@ async def call_with_fallback_for_request(
     Raises:
         RuntimeError: 所有 Provider 均不可用
     """
-    result, provider_key = await call_with_fallback(app, feature, fn)
+    # GW-H2: 预算记账必须归属真实用户，预算中间件（BudgetMiddleware）
+    # 按 request.state.user_id 查询日用量，写死 system 会导致限额永不触发
+    user_id = getattr(request.state, "user_id", "system")
+    result, provider_key = await call_with_fallback(app, feature, fn, user_id=user_id)
     return result, provider_key, False
 
 
@@ -313,11 +347,20 @@ async def _run_stream_fallback_chain():
                     probe_timeout = min(_FIRST_TOKEN_PROBE_TIMEOUT, remaining)
                     first_chunk = await asyncio.wait_for(agen.__anext__(), timeout=probe_timeout)
 
-                    # 首 token 成功：包装生成器，先吐首 token 再吐剩余
-                    async def _wrapped_gen(a=agen, first=first_chunk):
-                        yield first
-                        async for c in a:
-                            yield c
+                    # 首 token 成功：包装生成器，先吐首 token 再吐剩余；
+                    # 并发信号量在生成器整个生命周期内持有，防止流式长连接堆积
+                    sem = _get_concurrency_semaphore(app, feature)
+                    if sem is None:
+                        async def _wrapped_gen(a=agen, first=first_chunk):
+                            yield first
+                            async for c in a:
+                                yield c
+                    else:
+                        async def _wrapped_gen(a=agen, first=first_chunk, s=sem):
+                            async with s:
+                                yield first
+                                async for c in a:
+                                    yield c
 
                     # 通知熔断器成功
                     cb = get_circuit(provider_key)

server/ai-gateway/middleware/input_validation.py

@@ -16,11 +16,16 @@
 
 logger = logging.getLogger(__name__)
 
-# 需要输入验证的路径前缀
-VALIDATED_PATHS = ("/api/v1/ai/",)
+# 需要输入验证的路径前缀 → 各前缀允许的最大请求体字节数
+# GW-H7: 多模态/ASR/视觉端点原不在校验范围，base64 大载荷可致内存 DoS；
+# 各前缀按业务实际载荷设定独立上限（文本端点 1MB，多模态 64MB，音频/视觉 32MB）
+VALIDATED_PATHS: dict[str, int] = {
+    "/api/v1/ai/": 1 * 1024 * 1024,          # 1MB
+    "/api/v1/multimodal/": 64 * 1024 * 1024,  # 64MB（多图联合分析，100 帧 × ≤10MB 编码）
+    "/api/v1/asr/": 32 * 1024 * 1024,         # 32MB（音频转写）
+    "/api/v1/vision/": 32 * 1024 * 1024,      # 32MB（视觉提取）
+}
 
-# 限制常量
-MAX_CONTENT_LENGTH = 1 * 1024 * 1024  # 1MB
 MAX_TEXT_FIELD_LENGTH = 50000  # 字符
 
 
@@ -29,27 +34,33 @@ class InputValidationMiddleware(BaseHTTPMiddleware):
 
     async def dispatch(self, request: Request, call_next):
         # 仅对 AI 功能 API 进行输入验证
-        if not request.url.path.startswith(VALIDATED_PATHS):
+        max_content_length = None
+        for prefix, limit in VALIDATED_PATHS.items():
+            if request.url.path.startswith(prefix):
+                max_content_length = limit
+                break
+        if max_content_length is None:
             return await call_next(request)
 
         # 仅检查有请求体的方法
         if request.method in ("POST", "PUT", "PATCH"):
-            # 检查 Content-Length 头
+            # 检查 Content-Length 头（chunked 编码无此头时跳过，由 JSON 解析兜底）
             content_length = request.headers.get("content-length")
             if content_length:
                 try:
-                    if int(content_length) > MAX_CONTENT_LENGTH:
+                    if int(content_length) > max_content_length:
                         return JSONResponse(
-                            status_code=422,
+                            status_code=413,
                             content={
-                                "detail": f"request body exceeds {MAX_CONTENT_LENGTH} bytes (max 1MB)",
+                                "detail": f"request body exceeds {max_content_length} bytes",
                             },
                         )
                 except (ValueError, TypeError):
                     # 畸形 Content-Length 头，忽略该检查（后续 JSON 解析会进一步校验）
                     pass
 
-            # 解析 JSON 请求体并检查文本字段
+            # 解析 JSON 请求体并检查文本字段（body 全量读入内存前先受
+            # Content-Length 限制；chunked 大载荷由路由层 Pydantic max_length 兜底）
             content_type = request.headers.get("content-type", "")
             if "application/json" in content_type:
                 try: