feat(claude-code): 增加 TLS 指纹伪装、Cache TTL 统一、CLI 限制与流超时冷却

- 新增 curl_cffi Transport，支持真实浏览器 TLS 指纹伪装 (Chrome/Node.js)
- 增加 Cache TTL Override 功能，强制统一 cache_control 类型防止行为指纹差异
- 增加 CLI-only 客户端限制，支持仅允许 Claude Code CLI 访问
- 池健康策略增加 stream timeout 计数与自动冷却机制
- OAuth 账号 Region 选择改为从 AWS API 动态获取，支持搜索和自定义输入
- 前端 PoolConfigDialog 增加对应配置 UI

src/api/admin/system.py

@@ -2658,3 +2658,67 @@ class AdminPurgeStatsAdapter(AdminApiAdapter):
         return {
             "message": "聚合统计数据已清空",
         }
+
+
+# ---------------------------------------------------------------------------
+# AWS Regions (从 AWS Regional Table API 获取，Redis 缓存 24h)
+# ---------------------------------------------------------------------------
+
+_AWS_REGIONS_CACHE_KEY = "aws_regions"
+_AWS_REGIONS_CACHE_TTL = 86400  # 24h
+_AWS_REGIONAL_TABLE_URL = "https://api.regional-table.region-services.aws.a2z.com"
+
+# 内存级 fallback（进程生命周期内有效，Redis 不可用时兜底）
+_aws_regions_mem_cache: list[str] | None = None
+
+
+async def _fetch_aws_regions() -> list[str]:
+    """从 AWS Regional Table API 提取去重排序的 region 列表"""
+    import httpx as _httpx
+
+    from src.clients.http_client import HTTPClientPool
+
+    client = await HTTPClientPool.get_default_client_async()
+    resp = await client.get(
+        _AWS_REGIONAL_TABLE_URL,
+        timeout=_httpx.Timeout(connect=10, read=15),
+    )
+    resp.raise_for_status()
+    data = resp.json()
+    regions: set[str] = set()
+    for item in data.get("prices", []):
+        region = item.get("attributes", {}).get("aws:region", "")
+        if region:
+            regions.add(region)
+    return sorted(regions)
+
+
+@router.get("/aws-regions")
+async def get_aws_regions() -> Any:
+    """获取 AWS 全部可用 Region 列表（缓存 24h）"""
+    global _aws_regions_mem_cache
+
+    # 1. 尝试 Redis 缓存
+    from src.core.cache_service import CacheService
+
+    cached = await CacheService.get(_AWS_REGIONS_CACHE_KEY)
+    if cached and isinstance(cached, list):
+        return {"regions": cached}
+
+    # 2. 尝试内存 fallback
+    if _aws_regions_mem_cache:
+        return {"regions": _aws_regions_mem_cache}
+
+    # 3. 远程获取
+    try:
+        regions = await _fetch_aws_regions()
+    except Exception as e:
+        logger.warning("获取 AWS Regions 失败: {}", e)
+        # 返回最基础的 fallback
+        return {"regions": ["us-east-1", "us-east-2", "us-west-1", "us-west-2", "eu-north-1"]}
+
+    # 写入缓存
+    _aws_regions_mem_cache = regions
+    await CacheService.set(_AWS_REGIONS_CACHE_KEY, regions, ttl_seconds=_AWS_REGIONS_CACHE_TTL)
+
+    return {"regions": regions}

src/api/handlers/base/cli_adapter_base.py

@@ -73,6 +73,15 @@ class CliAdapterBase(HandlerAdapterBase):
         original_headers = context.original_headers
         query_params = context.query_params
 
+        # Store original headers for downstream envelope checks (e.g. CLI-only restriction).
+        # Only relevant for Claude Code CLI format; skip for others to avoid unnecessary coupling.
+        if self.FORMAT_ID == "claude:cli":
+            from src.services.provider.adapters.claude_code.client_restriction import (
+                set_original_request_headers,
+            )
+
+            set_original_request_headers(original_headers)
+
         original_request_body = context.ensure_json_body()
 
         # 合并 path_params 到请求体（如 Gemini API 的 model 在 URL 路径中）

src/api/handlers/base/cli_stream_mixin.py

@@ -805,6 +805,34 @@ class CliStreamMixin:
             prefetched_chunks,
         )
 
+    @staticmethod
+    def _fire_stream_timeout_policy(ctx: StreamContext) -> None:
+        """Fire-and-forget: record stream timeout for pool health policy."""
+        if not ctx.provider_id or not ctx.key_id:
+            return
+        try:
+            from src.services.provider.adapters.claude_code.context import (
+                get_claude_code_request_context,
+            )
+
+            cc_ctx = get_claude_code_request_context()
+            pool_cfg = cc_ctx.pool_config if cc_ctx else None
+            if not pool_cfg:
+                return
+
+            from src.services.provider.pool.health_policy import apply_stream_timeout_policy
+
+            task = asyncio.create_task(
+                apply_stream_timeout_policy(
+                    provider_id=ctx.provider_id,
+                    key_id=ctx.key_id,
+                    config=pool_cfg,
+                )
+            )
+            task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+        except Exception as exc:
+            logger.debug("Stream timeout policy trigger failed: {}", exc)
+
     async def _create_response_stream(
         self,
         ctx: StreamContext,
@@ -900,6 +928,9 @@ class CliStreamMixin:
                                 f"elapsed={elapsed:.1f}s, "
                                 f"chunk_count={ctx.chunk_count}, data_count=0"
                             )
+
+                            self._fire_stream_timeout_policy(ctx)
+
                             error_event = {
                                 "type": "error",
                                 "error": {

src/clients/curl_cffi_transport.py

@@ -0,0 +1,233 @@
+"""curl_cffi-based httpx AsyncTransport for TLS fingerprint impersonation.
+
+When ``curl_cffi`` is installed, this transport can replace the default httpx
+transport to send upstream requests with a browser-grade TLS fingerprint
+(JA3/JA4), making the traffic indistinguishable from a real browser or
+Node.js client.
+
+The transport is used exclusively when ``tls_profile == "claude_code_nodejs"``
+and ``curl_cffi`` is available. Otherwise, the system falls back to the
+default httpx SSL context (best-effort cipher ordering only).
+
+Design notes:
+- curl_cffi AsyncSession instances are **reused** per (impersonate, proxy) pair
+  to avoid rebuilding the TLS session on every request.
+- Streaming is supported via ``aiter_content()`` on the curl_cffi response.
+- The transport implements ``httpx.AsyncBaseTransport`` so it plugs into
+  the existing ``HTTPClientPool`` without changing callers.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import httpx
+
+from src.core.logger import logger
+
+# ---------------------------------------------------------------------------
+# Availability check
+# ---------------------------------------------------------------------------
+
+try:
+    from curl_cffi.requests import AsyncSession  # type: ignore[import-untyped]
+
+    CURL_CFFI_AVAILABLE = True
+except ImportError:
+    CURL_CFFI_AVAILABLE = False
+
+# ---------------------------------------------------------------------------
+# Default impersonate profile
+# ---------------------------------------------------------------------------
+# "chrome120" closely matches the TLS fingerprint of Node.js 20.x on Linux
+# (which Claude Code CLI uses). If the upstream introduces fingerprint
+# rotation, this can be made configurable per-profile.
+DEFAULT_IMPERSONATE = "chrome120"
+
+
+# ---------------------------------------------------------------------------
+# Session pool (module-level, async-safe)
+# ---------------------------------------------------------------------------
+_session_pool: dict[str, AsyncSession] = {}
+_pool_lock = asyncio.Lock()
+
+
+def _session_key(impersonate: str, proxy: str | None) -> str:
+    return f"{impersonate}::{proxy or '__direct__'}"
+
+
+async def _get_or_create_session(
+    impersonate: str = DEFAULT_IMPERSONATE,
+    proxy: str | None = None,
+) -> AsyncSession:
+    """Get or create a cached curl_cffi AsyncSession."""
+    key = _session_key(impersonate, proxy)
+    async with _pool_lock:
+        session = _session_pool.get(key)
+        if session is not None:
+            return session
+
+        kwargs: dict[str, Any] = {
+            "impersonate": impersonate,
+            "verify": True,
+        }
+        if proxy:
+            kwargs["proxy"] = proxy
+
+        session = AsyncSession(**kwargs)
+        _session_pool[key] = session
+        logger.info(
+            "curl_cffi session created: impersonate={}, proxy={}",
+            impersonate,
+            proxy or "direct",
+        )
+        return session
+
+
+async def close_all_sessions() -> None:
+    """Close all cached curl_cffi sessions (called at shutdown)."""
+    async with _pool_lock:
+        sessions = list(_session_pool.values())
+        _session_pool.clear()
+    for s in sessions:
+        try:
+            await s.close()
+        except Exception:
+            pass
+
+
+# ---------------------------------------------------------------------------
+# Exception mapping (curl_cffi -> httpx)
+# ---------------------------------------------------------------------------
+
+
+def _map_curl_exception(exc: Exception) -> httpx.HTTPError:
+    """Map curl_cffi exceptions to the closest httpx equivalents.
+
+    This lets the upstream failover / error_classifier distinguish between
+    transient timeouts (retryable) and hard connection failures.
+    """
+    if CURL_CFFI_AVAILABLE:
+        from curl_cffi.requests.exceptions import ConnectionError as CurlConnectionError
+        from curl_cffi.requests.exceptions import ProxyError as CurlProxyError
+        from curl_cffi.requests.exceptions import Timeout as CurlTimeout
+
+        if isinstance(exc, CurlTimeout):
+            return httpx.ReadTimeout(f"curl_cffi timeout: {exc}")
+        if isinstance(exc, CurlProxyError):
+            return httpx.ProxyError(f"curl_cffi proxy error: {exc}")
+        if isinstance(exc, CurlConnectionError):
+            return httpx.ConnectError(f"curl_cffi connection error: {exc}")
+    return httpx.ConnectError(f"curl_cffi request failed: {exc}")
+
+
+# ---------------------------------------------------------------------------
+# httpx AsyncTransport implementation
+# ---------------------------------------------------------------------------
+
+
+class CurlCffiStream(httpx.AsyncByteStream):
+    """Async byte stream backed by curl_cffi response content iterator."""
+
+    def __init__(self, curl_response: Any) -> None:
+        self._response = curl_response
+        self._consumed = False
+
+    async def __aiter__(self) -> Any:  # type: ignore[override]
+        if self._consumed:
+            return
+        try:
+            async for chunk in self._response.aiter_content():
+                yield chunk
+        finally:
+            self._consumed = True
+
+    async def aclose(self) -> None:
+        self._consumed = True
+        close_fn = getattr(self._response, "aclose", None)
+        if close_fn and callable(close_fn):
+            try:
+                await close_fn()
+            except Exception:
+                pass
+
+
+class CurlCffiTransport(httpx.AsyncBaseTransport):
+    """httpx-compatible async transport using curl_cffi for TLS impersonation.
+
+    Usage::
+
+        transport = CurlCffiTransport(proxy="http://proxy:8080")
+        client = httpx.AsyncClient(transport=transport)
+        resp = await client.post(url, json=payload, headers=headers)
+    """
+
+    def __init__(
+        self,
+        impersonate: str = DEFAULT_IMPERSONATE,
+        proxy: str | None = None,
+    ) -> None:
+        self._impersonate = impersonate
+        self._proxy = proxy
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        session = await _get_or_create_session(self._impersonate, self._proxy)
+
+        # Build headers dict (skip host header, curl_cffi handles it).
+        headers: dict[str, str] = {}
+        for key, value in request.headers.raw:
+            k = key.decode("latin-1").lower()
+            if k in ("host", "content-length", "transfer-encoding"):
+                continue
+            headers[key.decode("latin-1")] = value.decode("latin-1")
+
+        body = request.content if request.content else None
+        method = request.method.upper()
+        url = str(request.url)
+
+        # Determine timeout from request extensions.
+        timeout = 60.0
+        if hasattr(request, "extensions") and isinstance(request.extensions, dict):
+            raw_timeout = request.extensions.get("timeout")
+            if isinstance(raw_timeout, dict):
+                # httpx timeout pool format: {"connect": ..., "read": ..., "write": ..., "pool": ...}
+                read_timeout = raw_timeout.get("read")
+                if isinstance(read_timeout, (int, float)) and read_timeout > 0:
+                    timeout = float(read_timeout)
+            elif isinstance(raw_timeout, (int, float)) and raw_timeout > 0:
+                timeout = float(raw_timeout)
+
+        try:
+            # Use stream=True for all requests so we can support streaming responses.
+            curl_resp = await session.request(
+                method,
+                url,
+                headers=headers,
+                data=body,
+                timeout=timeout,
+                stream=True,
+            )
+        except Exception as exc:
+            raise _map_curl_exception(exc) from exc
+
+        # Build response headers.
+        resp_headers_list: list[tuple[bytes, bytes]] = []
+        if hasattr(curl_resp, "headers") and curl_resp.headers:
+            for k, v in curl_resp.headers.multi_items():
+                resp_headers_list.append((k.encode("latin-1"), v.encode("latin-1")))
+
+        return httpx.Response(
+            status_code=curl_resp.status_code,
+            headers=resp_headers_list,
+            stream=CurlCffiStream(curl_resp),
+            request=request,
+        )
+
+
+__all__ = [
+    "CURL_CFFI_AVAILABLE",
+    "CurlCffiTransport",
+    "close_all_sessions",
+    "DEFAULT_IMPERSONATE",
+]

src/clients/http_client.py

@@ -242,6 +242,44 @@ class HTTPClientPool:
             # 淘汰旧客户端（如果超过上限）
             await cls._evict_lru_proxy_client()
 
+            # 添加代理配置
+            proxy_url = build_proxy_url(proxy_config) if proxy_config else None
+
+            # curl_cffi Transport: real TLS fingerprint impersonation.
+            # When tls_profile requires fingerprint impersonation and curl_cffi
+            # is available, use CurlCffiTransport instead of the default httpx
+            # transport. This gives us a genuine browser/Node.js TLS handshake.
+            if tls_profile_key == "claude_code_nodejs":
+                from src.clients.curl_cffi_transport import (
+                    CURL_CFFI_AVAILABLE,
+                    CurlCffiTransport,
+                )
+
+                if CURL_CFFI_AVAILABLE:
+                    transport = CurlCffiTransport(proxy=proxy_url)
+                    client = httpx.AsyncClient(
+                        transport=transport,
+                        follow_redirects=True,
+                        timeout=httpx.Timeout(
+                            connect=config.http_connect_timeout,
+                            read=config.http_read_timeout,
+                            write=config.http_write_timeout,
+                            pool=config.http_pool_timeout,
+                        ),
+                    )
+                    cls._proxy_clients[cache_key] = (client, time.time())
+                    logger.info(
+                        "创建 curl_cffi TLS 指纹客户端: profile={}, proxy={}",
+                        tls_profile_key,
+                        proxy_url or "direct",
+                    )
+                    return client
+                else:
+                    logger.warning(
+                        "curl_cffi 不可用，回退到 best-effort TLS 配置 (profile={})",
+                        tls_profile_key,
+                    )
+
             # 创建新客户端（使用默认超时，请求时可覆盖）
             client_config: dict[str, Any] = {
                 "http2": False,
@@ -260,8 +298,6 @@ class HTTPClientPool:
                 ),
             }
 
-            # 添加代理配置
-            proxy_url = build_proxy_url(proxy_config) if proxy_config else None
             proxy_param = make_proxy_param(proxy_url)
             if proxy_param:
                 client_config["proxy"] = proxy_param
@@ -315,6 +351,17 @@ class HTTPClientPool:
                 logger.warning("关闭 tunnel 客户端失败: {}", e)
 
         cls._tunnel_clients.clear()
+
+        # 关闭 curl_cffi session 缓存
+        try:
+            from src.clients.curl_cffi_transport import CURL_CFFI_AVAILABLE, close_all_sessions
+
+            if CURL_CFFI_AVAILABLE:
+                await close_all_sessions()
+                logger.debug("curl_cffi sessions 已关闭")
+        except Exception as e:
+            logger.debug("关闭 curl_cffi sessions 失败: {}", e)
+
         logger.info("所有HTTP客户端已关闭")
 
     @classmethod

src/models/admin_requests.py

@@ -147,6 +147,18 @@ class ClaudeCodeAdvancedConfig(BaseModel):
     session_id_masking_enabled: bool = Field(
         False, description="是否启用会话 ID 伪装（固定 metadata.user_id 中 session 片段）"
     )
+    cache_ttl_override_enabled: bool = Field(
+        False, description="是否启用 Cache TTL 强制替换（统一所有请求的 cache_control 类型）"
+    )
+    cache_ttl_override_target: str = Field(
+        "ephemeral",
+        description="Cache TTL 目标类型: ephemeral (5min) 或 1h",
+        pattern="^(ephemeral|1h)$",
+    )
+    cli_only_enabled: bool = Field(
+        False,
+        description="是否仅允许 Claude Code CLI 客户端访问（非 CLI 流量返回 403）",
+    )
 
     @model_validator(mode="after")
     def normalize_session_control(self) -> "ClaudeCodeAdvancedConfig":

src/services/provider/adapters/claude_code/client_restriction.py

@@ -0,0 +1,96 @@
+"""Claude Code CLI client restriction.
+
+When cli_only_enabled is True, only requests from genuine Claude Code CLI
+clients are allowed. Non-CLI traffic receives a 403 response.
+"""
+
+from __future__ import annotations
+
+import contextvars
+from typing import Any
+
+from src.core.logger import logger
+
+# Contextvar to carry the original request headers into the envelope layer.
+_original_request_headers: contextvars.ContextVar[dict[str, str] | None] = contextvars.ContextVar(
+    "claude_code_original_request_headers",
+    default=None,
+)
+
+
+def set_original_request_headers(headers: dict[str, str] | None) -> None:
+    _original_request_headers.set(headers)
+
+
+def get_original_request_headers() -> dict[str, str] | None:
+    return _original_request_headers.get()
+
+
+# Known Claude Code CLI User-Agent patterns.
+_CLI_USER_AGENT_PATTERNS = (
+    "claude-code",
+    "claudecode",
+    "claude_code",
+)
+
+# Known originator / x-app values indicating CLI usage.
+_CLI_APP_VALUES = {"cli"}
+
+
+def is_claude_code_client(headers: dict[str, Any]) -> bool:
+    """Detect whether the request originates from a Claude Code CLI client.
+
+    Detection signals (any match is sufficient):
+    1. User-Agent contains a known Claude Code CLI pattern
+    2. x-app header equals "cli"
+    """
+    # Normalize header keys to lowercase for case-insensitive matching.
+    lower_headers = {k.lower(): v for k, v in headers.items()}
+
+    # Check User-Agent
+    ua = str(lower_headers.get("user-agent", "")).lower()
+    for pattern in _CLI_USER_AGENT_PATTERNS:
+        if pattern in ua:
+            return True
+
+    # Check x-app header
+    x_app = str(lower_headers.get("x-app", "")).strip().lower()
+    if x_app in _CLI_APP_VALUES:
+        return True
+
+    return False
+
+
+def enforce_cli_only(cli_only_enabled: bool) -> None:
+    """Enforce CLI-only restriction if enabled.
+
+    Reads original request headers from contextvar, checks whether the
+    client is a Claude Code CLI, and raises HTTPException(403) if not.
+    """
+    if not cli_only_enabled:
+        return
+
+    headers = get_original_request_headers()
+    if headers is None:
+        # No headers available; skip enforcement (should not happen in normal flow).
+        logger.debug("CLI-only check skipped: no request headers in context")
+        return
+
+    if is_claude_code_client(headers):
+        return
+
+    from fastapi import HTTPException
+
+    logger.info("CLI-only restriction: rejected non-CLI client")
+    raise HTTPException(
+        status_code=403,
+        detail="This endpoint only accepts requests from Claude Code CLI clients.",
+    )
+
+
+__all__ = [
+    "enforce_cli_only",
+    "get_original_request_headers",
+    "is_claude_code_client",
+    "set_original_request_headers",
+]

src/services/provider/adapters/claude_code/context.py

@@ -24,6 +24,9 @@ class ClaudeCodeRequestContext:
     session_idle_timeout_minutes: int = 5
     enable_tls_fingerprint: bool = False
     session_id_masking_enabled: bool = False
+    cache_ttl_override_enabled: bool = False
+    cache_ttl_override_target: str = "ephemeral"
+    cli_only_enabled: bool = False
     # Account Pool fields
     provider_id: str | None = None
     pool_config: PoolConfig | None = None
@@ -88,6 +91,15 @@ def build_claude_code_request_context(
     session_id_masking_enabled = (
         bool(advanced_config.session_id_masking_enabled) if advanced_config else False
     )
+    cache_ttl_override_enabled = (
+        bool(advanced_config.cache_ttl_override_enabled) if advanced_config else False
+    )
+    cache_ttl_override_target = (
+        str(advanced_config.cache_ttl_override_target or "ephemeral")
+        if advanced_config
+        else "ephemeral"
+    )
+    cli_only_enabled = bool(advanced_config.cli_only_enabled) if advanced_config else False
 
     # Parse pool config (None = non-pool provider, keep as None for semantic consistency)
     pool_cfg = parse_pool_config(provider_config_dict)
@@ -100,6 +112,9 @@ def build_claude_code_request_context(
         session_idle_timeout_minutes=idle_timeout_minutes,
         enable_tls_fingerprint=enable_tls_fingerprint,
         session_id_masking_enabled=session_id_masking_enabled,
+        cache_ttl_override_enabled=cache_ttl_override_enabled,
+        cache_ttl_override_target=cache_ttl_override_target,
+        cli_only_enabled=cli_only_enabled,
         provider_id=str(provider_id or "").strip() or None,
         pool_config=pool_cfg,
     )

src/services/provider/adapters/claude_code/envelope.py

@@ -218,6 +218,59 @@ def _apply_session_id_masking(request_body: dict[str, Any], *, scope_key: str) -
     )
 
 
+# -- Cache TTL Override -------------------------------------------------------
+
+_VALID_CACHE_TTL_TARGETS = {"ephemeral", "1h"}
+
+
+def _override_cache_control_in_blocks(blocks: list[Any], target: str) -> int:
+    """Override cache_control in a list of content blocks. Returns count of overrides."""
+    count = 0
+    for block in blocks:
+        if not isinstance(block, dict):
+            continue
+        cc = block.get("cache_control")
+        if isinstance(cc, dict):
+            if cc.get("type") != target:
+                cc["type"] = target
+                count += 1
+    return count
+
+
+def _apply_cache_ttl_override(request_body: dict[str, Any], target: str) -> None:
+    """Force all cache_control entries to use a unified TTL type.
+
+    Prevents multi-user behavioral fingerprinting when sharing an OAuth account.
+    """
+    if target not in _VALID_CACHE_TTL_TARGETS:
+        return
+
+    overridden = 0
+
+    # system prompt (can be string or list of blocks)
+    system = request_body.get("system")
+    if isinstance(system, list):
+        overridden += _override_cache_control_in_blocks(system, target)
+
+    # messages
+    messages = request_body.get("messages")
+    if isinstance(messages, list):
+        for msg in messages:
+            if not isinstance(msg, dict):
+                continue
+            content = msg.get("content")
+            if isinstance(content, list):
+                overridden += _override_cache_control_in_blocks(content, target)
+
+    # tools
+    tools = request_body.get("tools")
+    if isinstance(tools, list):
+        overridden += _override_cache_control_in_blocks(tools, target)
+
+    if overridden:
+        logger.debug("Cache TTL override: {} block(s) -> {}", overridden, target)
+
+
 def _register_or_reject_session(
     *,
     scope_key: str,
@@ -446,6 +499,15 @@ class ClaudeCodeEnvelope:
         ctx = get_claude_code_request_context()
         if ctx is None:
             ctx = ClaudeCodeRequestContext()
+
+        # CLI-only restriction: reject non-CLI clients early.
+        if ctx.cli_only_enabled:
+            from src.services.provider.adapters.claude_code.client_restriction import (
+                enforce_cli_only,
+            )
+
+            enforce_cli_only(ctx.cli_only_enabled)
+
         # Extract session_uuid from metadata.user_id for pool sticky session.
         session_uuid: str | None = None
         user_id = _get_metadata_user_id(request_body)
@@ -456,6 +518,10 @@ class ClaudeCodeEnvelope:
 
         _sanitize_thinking_blocks(request_body)
 
+        # Cache TTL override: unify cache_control types to prevent behavioral fingerprinting.
+        if ctx.cache_ttl_override_enabled:
+            _apply_cache_ttl_override(request_body, ctx.cache_ttl_override_target)
+
         _enforce_session_controls(
             request_body,
             ctx,

src/services/provider/pool/config.py

@@ -51,6 +51,11 @@ class PoolConfig:
     # -- Temporary Unschedulable Rules ----------------------------------------
     unschedulable_rules: list[UnschedulableRule] = field(default_factory=list)
 
+    # -- Stream Timeout Auto-Pause --------------------------------------------
+    stream_timeout_threshold: int = 3  # N timeouts within window trigger cooldown
+    stream_timeout_window_seconds: int = 1800  # 30 min counting window
+    stream_timeout_cooldown_seconds: int = 300  # 5 min cooldown
+
     # -- Pluggable Strategies -------------------------------------------------
     strategies: tuple[str, ...] = ()
 
@@ -127,6 +132,9 @@ def parse_pool_config(provider_config: Any) -> PoolConfig | None:
         proactive_refresh_seconds=_int_or("proactive_refresh_seconds", 180),
         health_policy_enabled=_bool_or("health_policy_enabled", True),
         unschedulable_rules=rules,
+        stream_timeout_threshold=_int_or("stream_timeout_threshold", 3),
+        stream_timeout_window_seconds=_int_or("stream_timeout_window_seconds", 1800),
+        stream_timeout_cooldown_seconds=_int_or("stream_timeout_cooldown_seconds", 300),
         strategies=_parse_strategies(raw_advanced.get("strategies")),
     )
 

src/services/provider/pool/health_policy.py

@@ -202,3 +202,56 @@ async def _apply(
                     rule.duration_minutes,
                 )
                 return
+
+
+async def apply_stream_timeout_policy(
+    *,
+    provider_id: str,
+    key_id: str,
+    config: PoolConfig,
+) -> None:
+    """Record a stream timeout event and apply cooldown if threshold is reached.
+
+    Called when an upstream stream response times out (no data within the
+    configured interval). Increments a per-key counter in Redis and sets
+    a cooldown if the count reaches the configured threshold.
+    """
+    if not config.health_policy_enabled:
+        return
+
+    try:
+        count = await redis_ops.incr_stream_timeout_count(
+            provider_id,
+            key_id,
+            config.stream_timeout_window_seconds,
+        )
+        if count >= config.stream_timeout_threshold:
+            ttl = config.stream_timeout_cooldown_seconds
+            await redis_ops.set_cooldown(
+                provider_id,
+                key_id,
+                f"stream_timeout_x{count}",
+                ttl=ttl,
+            )
+            logger.warning(
+                "Pool[{}]: key {} stream timeout count {} >= threshold {}, cooldown {}s",
+                provider_id[:8],
+                key_id[:8],
+                count,
+                config.stream_timeout_threshold,
+                ttl,
+            )
+        else:
+            logger.info(
+                "Pool[{}]: key {} stream timeout count {}/{}",
+                provider_id[:8],
+                key_id[:8],
+                count,
+                config.stream_timeout_threshold,
+            )
+    except Exception as exc:
+        logger.warning(
+            "Pool stream timeout policy failed for key {}: {}",
+            key_id[:8],
+            str(exc),
+        )

src/services/provider/pool/redis_ops.py

@@ -468,3 +468,45 @@ async def batch_get_cooldown_ttls(provider_id: str, key_ids: list[str]) -> dict[
         return out
     except Exception:
         return {k: None for k in key_ids}
+
+
+# ---------------------------------------------------------------------------
+# Stream timeout counter
+# ---------------------------------------------------------------------------
+
+_STREAM_TIMEOUT_KEY_FMT = f"{PREFIX}:{{}}:stream_timeout:{{}}"
+
+
+def _stream_timeout_key(provider_id: str, key_id: str) -> str:
+    return _STREAM_TIMEOUT_KEY_FMT.format(provider_id, key_id)
+
+
+async def incr_stream_timeout_count(
+    provider_id: str,
+    key_id: str,
+    window_seconds: int,
+) -> int:
+    """Increment stream timeout counter and return count within the window.
+
+    Uses a ZSET with timestamps as scores. Old entries beyond the window
+    are pruned on each call. Returns the count of timeouts in the window.
+    """
+    redis = await _get_redis()
+    if redis is None:
+        return 0
+    try:
+        now = time.time()
+        window_start = now - window_seconds
+        key = _stream_timeout_key(provider_id, key_id)
+        member = f"{uuid.uuid4().hex}"
+        pipe = redis.pipeline()
+        pipe.zremrangebyscore(key, "-inf", window_start)
+        pipe.zadd(key, {member: now})
+        pipe.zcard(key)
+        pipe.expire(key, window_seconds + 60)
+        results = await pipe.execute()
+        count = int(results[2]) if results[2] else 0
+        return count
+    except Exception:
+        logger.debug("Pool: stream timeout INCR failed for key {}", key_id[:8])
+        return 0