fix(gateway): resolve medium-risk issues #5-#10

- #5: Gemini generate_stream iteration moved into thread pool via
  asyncio.to_thread, unblocking the event loop during streaming
- #6: providers return real input/output token split; fallback cost
  tracking prefers it and falls back to 60/40 estimate instead of
  the misleading 50/50 split (output priced 2-3x higher)
- #7: balance queries use Key pool primary key (plural env vars),
  DeepSeek balance no longer silently skipped
- #8: JWKS fetch distinguishes network failures (stale cache reuse
  with 60s retry backoff) from HTTP/key errors (fail-closed),
  preventing whole-site 401 on transient Supabase hiccups
- #9: import_concept registered in TIMEOUT_CONFIG/RATE_LIMITS plus
  startup validation warning for unregistered features
- #10: error_pattern cache key includes user_id and hashes full
  content, preventing cross-user result reuse

Author: Aparencia

server/ai-gateway/config/fallback.py

@@ -214,12 +214,22 @@ async def _do_call():
                 try:
                     tokens_used = result.get("tokens_used", 0)
                     model_name = result.get("model", model_name)
+                    # GW-2#6: 优先使用 provider 返回的真实 input/output 拆分
+                    #（OpenAI 兼容 usage 均有 prompt/completion tokens）；
+                    # 缺失时按经验比例估算（output 占比 60%）——原对半拆分
+                    # 系统性低估费用（output 单价普遍 2-3 倍于 input），
+                    # 导致超预算用户未被拦截
+                    input_tokens = result.get("input_tokens", 0)
+                    output_tokens = result.get("output_tokens", 0)
+                    if not input_tokens and not output_tokens and tokens_used:
+                        output_tokens = int(tokens_used * 0.6)
+                        input_tokens = tokens_used - output_tokens
                     await get_cost_tracker().record(
                         user_id=user_id,
                         feature=feature,
                         model=model_name,
-                        input_tokens=tokens_used // 2 if tokens_used else 0,
-                        output_tokens=tokens_used // 2 if tokens_used else 0,
+                        input_tokens=input_tokens,
+                        output_tokens=output_tokens,
                     )
                 except Exception as cost_err:
                     logger.debug("CostTracker 记录失败（可忽略）: %s", cost_err)

server/ai-gateway/config/limits.py

@@ -50,6 +50,8 @@
     "conflict_detect": 30,
     # E1: 概念预检（JSON Mode，1-2 个探测问题）
     "concept_precheck": 30,
+    # 知识入籍概念化（JSON Mode，切块文本→概念候选，与同批 JSON Mode 链一致）
+    "import_concept": 30,
 }
 
 # ============================================================
@@ -100,4 +102,6 @@
     "conflict_detect": 5,
     # E1: 概念预检（生成成本高，严格限频）
     "concept_precheck": 5,
+    # 知识入籍概念化（批量导入场景，适度限频）
+    "import_concept": 10,
 }

server/ai-gateway/middleware/auth.py

@@ -143,6 +143,14 @@ def _normalize_pem_key(raw: str) -> str:
 _jwks_cache: Optional[dict] = None
 _jwks_cache_time: float = 0.0
 _JWKS_CACHE_TTL: int = 3600  # 缓存有效期（秒），默认 1 小时
+# GW-2#8: 网络类失败时允许复用过期缓存的宽限窗口（秒）——密钥轮换的越权
+# 风险与网络抖动导致的可用性风险权衡：JWKS 1 小时才刷新一次，期间 Supabase
+# 任何一次网络故障若 fail-closed 会让全部请求 401（认证风暴）
+_JWKS_STALE_GRACE_SECONDS: int = 300
+# GW-2#8: 网络失败后的重试退避（秒），避免每次请求都打 JWKS 端点
+_JWKS_FAIL_RETRY_SECONDS: int = 60
+# 网络失败后下次刷新尝试的最早时间戳（0 表示无退避）
+_jwks_retry_after: float = 0.0
 # GW-M3: 防缓存击穿锁 + 共享连接池客户端（避免每次请求新建 TCP/TLS 连接）
 _jwks_lock: asyncio.Lock = asyncio.Lock()
 _jwks_client: Optional[httpx.AsyncClient] = None
@@ -181,11 +189,21 @@ async def _fetch_jwks(jwks_url: str) -> dict:
     GW-M3: 网络失败时 fail-closed（拒绝验证）而非返回过期缓存——
     密钥轮换后过期缓存会让旧 token 在验证窗口内继续有效（越权窗口）。
     asyncio.Lock 防止冷启动多请求并发刷新（缓存击穿）。
+    GW-2#8: 区分两类失败——HTTP 状态错误（密钥轮换/端点失效）保持
+    fail-closed；网络类失败（超时/DNS/TLS/连接）短暂复用过期缓存
+    （最多 5 分钟宽限）并退避 60s 重试，避免 Supabase 短暂抖动全站 401。
     """
-    global _jwks_cache, _jwks_cache_time
+    global _jwks_cache, _jwks_cache_time, _jwks_retry_after
     now = time.time()
     if _jwks_cache is not None and (now - _jwks_cache_time) < _JWKS_CACHE_TTL:
         return _jwks_cache
+    # 缓存过期但在宽限窗口内且未到重试时间：降级复用（网络抖动自愈窗口）
+    if (
+        _jwks_cache is not None
+        and (now - _jwks_cache_time) < _JWKS_CACHE_TTL + _JWKS_STALE_GRACE_SECONDS
+        and now < _jwks_retry_after
+    ):
+        return _jwks_cache
     async with _jwks_lock:
         # 双重检查：等待锁期间其他协程可能已刷新缓存
         now = time.time()
@@ -196,10 +214,23 @@ async def _fetch_jwks(jwks_url: str) -> dict:
             resp.raise_for_status()
             _jwks_cache = resp.json()
             _jwks_cache_time = time.time()
+            _jwks_retry_after = 0.0
             logger.info("JWKS 获取成功，缓存已更新: %s", jwks_url)
             return _jwks_cache
+        except httpx.HTTPStatusError as e:
+            # HTTP 状态错误（401/404/5xx）：密钥轮换或端点失效，fail-closed
+            logger.error("JWKS 获取失败: %s, status=%s", jwks_url, e.response.status_code)
+            raise
         except Exception as e:
-            logger.error("JWKS 获取失败: %s, error=%s", jwks_url, str(e))
+            # 网络类失败（超时/DNS/TLS/连接中断）：短暂复用过期缓存降级
+            logger.error("JWKS 获取失败(网络): %s, error=%s", jwks_url, str(e))
+            if _jwks_cache is not None:
+                _jwks_retry_after = time.time() + _JWKS_FAIL_RETRY_SECONDS
+                logger.warning(
+                    "JWKS 网络失败，降级复用过期缓存（%ds 内重试）",
+                    _JWKS_FAIL_RETRY_SECONDS,
+                )
+                return _jwks_cache
             raise
 
 

server/ai-gateway/middleware/rate_limit.py

@@ -18,7 +18,7 @@
 from starlette.middleware.base import BaseHTTPMiddleware
 
 from cache.redis_cache import get_cache
-from config import RATE_LIMITS
+from config import RATE_LIMITS, TIMEOUT_CONFIG
 
 logger = logging.getLogger(__name__)
 
@@ -108,6 +108,19 @@
 # 仅受各自功能级上限约束
 GLOBAL_EXEMPT_FEATURES: frozenset[str] = frozenset({"transcribe", "chat"})
 
+# GW-2#9: 启动校验——PATH_TO_FEATURE 登记的 feature 必须同时在
+# TIMEOUT_CONFIG 与 RATE_LIMITS 登记，否则静默兜底（300s 超时/默认 10 次限流）
+# 会掩盖配置缺失（import_concept 曾因此落入 300s 超时预算）
+_MISSING_CONFIG = sorted(
+    f for f in set(PATH_TO_FEATURE.values())
+    if f not in RATE_LIMITS or f not in TIMEOUT_CONFIG
+)
+if _MISSING_CONFIG:
+    logger.warning(
+        "以下 feature 缺少 TIMEOUT_CONFIG/RATE_LIMITS 登记（将使用兜底值）: %s",
+        _MISSING_CONFIG,
+    )
+
 
 class RateLimitMiddleware(BaseHTTPMiddleware):
     """频率限制中间件 — 基于 Redis 滑动窗口"""

server/ai-gateway/providers/deepseek_provider.py

@@ -97,8 +97,14 @@ async def generate(
             # 提取结果
             content = response.choices[0].message.content or ""
             tokens_used = 0
+            input_tokens = 0
+            output_tokens = 0
             if response.usage:
                 tokens_used = response.usage.total_tokens or 0
+                # GW-2#6: 提取真实 input/output 拆分供成本记账（OpenAI 兼容
+                # usage 字段），fallback 链不再对半估算
+                input_tokens = getattr(response.usage, "prompt_tokens", 0) or 0
+                output_tokens = getattr(response.usage, "completion_tokens", 0) or 0
 
             # DeepSeek 缓存命中信息（如果有的话）
             cache_hit_tokens = 0
@@ -132,6 +138,8 @@ async def generate(
             return {
                 "content": content,
                 "tokens_used": tokens_used,
+                "input_tokens": input_tokens,
+                "output_tokens": output_tokens,
                 "effective_tokens": effective_tokens,
                 "cache_hit": cache_hit,
                 "cache_hit_tokens": cache_hit_tokens,

server/ai-gateway/providers/gemini_provider.py

@@ -91,11 +91,17 @@ async def generate(
             latency_ms = int((time.monotonic() - start_time) * 1000)
             content = response.text or ""
             tokens_used = 0
+            input_tokens = 0
+            output_tokens = 0
             if response.usage_metadata:
                 tokens_used = response.usage_metadata.total_token_count or 0
+                # GW-2#6: Gemini usage_metadata 的拆分字段供成本记账
+                input_tokens = getattr(response.usage_metadata, "prompt_token_count", 0) or 0
+                output_tokens = getattr(response.usage_metadata, "candidates_token_count", 0) or 0
             logger.info("GeminiProvider.generate 成功: model=%s, tokens=%d, latency=%dms",
                         model, tokens_used, latency_ms)
             return {"content": content, "tokens_used": tokens_used,
+                    "input_tokens": input_tokens, "output_tokens": output_tokens,
                     "model": model, "latency_ms": latency_ms}
         except Exception as e:
             logger.error("GeminiProvider.generate 失败: %s", str(e))
@@ -294,7 +300,20 @@ async def generate_stream(
                 contents=prompt,
                 config=config,
             )
-            for chunk in response:
+
+            # GW-2#5: 同步 SDK 生成器的迭代（含底层阻塞网络 IO）直接在事件循环
+            # 线程执行会卡死整个网关（流式期间全站延迟升高）。改为逐块
+            # asyncio.to_thread 搬进线程池，chunk 间让出控制权
+            def _next_chunk(iterator):
+                try:
+                    return next(iterator)
+                except StopIteration:
+                    return None
+
+            while True:
+                chunk = await asyncio.to_thread(_next_chunk, response)
+                if chunk is None:
+                    break
                 if chunk.text:
                     yield chunk.text
                 # 让出事件循环控制权

server/ai-gateway/providers/glm_provider.py

@@ -102,8 +102,14 @@ async def generate(
             # 提取结果
             content = response.choices[0].message.content or ""
             tokens_used = 0
+            input_tokens = 0
+            output_tokens = 0
             if response.usage:
                 tokens_used = response.usage.total_tokens or 0
+                # GW-2#6: 提取真实 input/output 拆分供成本记账（OpenAI 兼容
+                # usage 字段），fallback 链不再对半估算
+                input_tokens = getattr(response.usage, "prompt_tokens", 0) or 0
+                output_tokens = getattr(response.usage, "completion_tokens", 0) or 0
 
             logger.info(
                 "GLMProvider 调用成功: model=%s, tokens=%d, latency=%dms",
@@ -113,6 +119,8 @@ async def generate(
             return {
                 "content": content,
                 "tokens_used": tokens_used,
+                "input_tokens": input_tokens,
+                "output_tokens": output_tokens,
                 "model": model,
                 "latency_ms": latency_ms,
             }

server/ai-gateway/providers/qwen_provider.py

@@ -87,8 +87,14 @@ async def generate(
             # 提取结果
             content = response.choices[0].message.content or ""
             tokens_used = 0
+            input_tokens = 0
+            output_tokens = 0
             if response.usage:
                 tokens_used = response.usage.total_tokens or 0
+                # GW-2#6: 提取真实 input/output 拆分供成本记账（OpenAI 兼容
+                # usage 字段），fallback 链不再对半估算
+                input_tokens = getattr(response.usage, "prompt_tokens", 0) or 0
+                output_tokens = getattr(response.usage, "completion_tokens", 0) or 0
 
             logger.info(
                 "QwenProvider 调用成功: model=%s, tokens=%d, latency=%dms",
@@ -98,6 +104,8 @@ async def generate(
             return {
                 "content": content,
                 "tokens_used": tokens_used,
+                "input_tokens": input_tokens,
+                "output_tokens": output_tokens,
                 "model": model,
                 "latency_ms": latency_ms,
             }

server/ai-gateway/routers/balance.py

@@ -25,6 +25,7 @@
 
 from cache.redis_cache import get_cache
 from config import AI_PROVIDERS, is_valid_api_key, ALIYUN_ACCESS_KEY_ID, ALIYUN_ACCESS_KEY_SECRET
+from config.key_pool import get_primary_key
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/v1/ai", tags=["余额查询"])
@@ -301,7 +302,10 @@ async def _query_all_balances(cache, start_time: float) -> dict:
 
     for provider_key, (display_name, query_fn) in _PROVIDER_QUERIES.items():
         cfg = AI_PROVIDERS.get(provider_key, {})
-        api_key = cfg.get("api_key", "")
+        # GW-2#7: 优先从 Key 池取主 Key（复数环境变量 DEEPSEEK_API_KEYS 的
+        # 首 Key）——原实现只读 AI_PROVIDERS 的单数变量 DEEPSEEK_API_KEY，
+        # 部署仅配置复数变量时余额查询被静默跳过（余额面板缺 DeepSeek 条目）
+        api_key = get_primary_key(provider_key) or cfg.get("api_key", "")
 
         # 跳过未配置的 Provider
         if not is_valid_api_key(api_key):

server/ai-gateway/routers/error_pattern.py

@@ -111,11 +111,14 @@ async def error_pattern(request: Request, body: ErrorPatternRequest) -> ErrorPat
     logger.info("错误模式分析请求: user=%s, errors_count=%d", user_id, len(body.goldenErrors))
 
     # 生成 AI 响应缓存键（基于输入错误记录）
+    # GW-2#10: 缓存键加入 user_id 隔离——错误模式分析结果基于用户私有
+    # 学习数据，原键不含用户维度，内容相同的不同用户会串用彼此的分析结论
+    #（隐私泄露 + 正确性错误）；同时不再截断 50 字符（sha256 本身压缩长度）
     errors_str = "|".join([
-        f"{e.flashcardId}:{e.correctAnswer[:50]}:{e.userAnswer[:50]}"
+        f"{e.flashcardId}:{e.correctAnswer}:{e.userAnswer}"
         for e in body.goldenErrors
     ])
-    cache_key = hashlib.sha256(errors_str.encode()).hexdigest()
+    cache_key = hashlib.sha256(f"{user_id}:{errors_str}".encode()).hexdigest()
 
     # 检查 Redis AI 响应缓存
     cache = get_cache()