feat(gateway): P0 凭证池稳定身份与统一快照

backend/packages/framework/src/windup_framework/gateway/__init__.py

@@ -2,17 +2,29 @@
 from windup_framework.gateway.context import bind_call_context, fresh_gateway_request
 from windup_framework.gateway.image import ImageGateway, build_image_gateway
 from windup_framework.gateway.models import AIGatewayAttempt, AIGatewayAttemptDetail
+from windup_framework.gateway.pool_ids import credential_id
+from windup_framework.gateway.pool_registry import (
+    PoolSnapshot,
+    RoutableEdge,
+    get_pool_snapshot,
+    invalidate_pool_cache,
+)
 from windup_framework.gateway.video import VideoGateway, build_video_gateway
 
 __all__ = [
     "AIGatewayAttempt",
     "AIGatewayAttemptDetail",
     "ChatGateway",
     "ImageGateway",
+    "PoolSnapshot",
+    "RoutableEdge",
     "VideoGateway",
     "bind_call_context",
-    "fresh_gateway_request",
     "build_chat_gateway",
     "build_image_gateway",
     "build_video_gateway",
+    "credential_id",
+    "fresh_gateway_request",
+    "get_pool_snapshot",
+    "invalidate_pool_cache",
 ]

backend/packages/framework/src/windup_framework/gateway/chat.py

@@ -23,7 +23,7 @@
     config_for_route,
     key_circuit_id,
     lookup_adapter,
-    routes_from_settings,
+    pool_routes,
 )
 from windup_framework.gateway.trace import AttemptDetail, AttemptTrace, emit
 from windup_framework.gateway.sequencer import AttemptSequencer
@@ -182,9 +182,11 @@ def __init__(self, adapter, circuit, settings, route_adapters=None) -> None:
         self._adapter = adapter
         self._circuit = circuit
         self._settings = settings
-        self._routes = routes_from_settings(settings, route_group=Scene.CHAT.value)
         self._route_adapters = dict(route_adapters or {})
 
+    def _pool_routes(self) -> tuple[GatewayRoute, ...]:
+        return pool_routes(self._settings, route_group=Scene.CHAT.value)
+
     def _adapter_for(self, route: GatewayRoute):
         return lookup_adapter(self._route_adapters, route, self._adapter)
 
@@ -236,6 +238,7 @@ async def ainvoke(self, messages: Any, **kwargs: Any) -> Any:
         last_http_status: int | None = None
         seq = AttemptSequencer()
         budget = AttemptBudget()
+        routes = self._pool_routes()
 
         def total_ms() -> int:
             return int((time.monotonic() - started) * 1000)
@@ -252,7 +255,7 @@ def fail(http_status: int | None) -> None:
                     scene=Scene.CHAT,
                     model=models[0],
                     family=Family.CHAT_COMPLETIONS.value,
-                    route=self._routes[0],
+                    route=routes[0],
                     attempt_index=seq.next_index(),
                     retry_count=0,
                     route_reason="skip_circuit_open",
@@ -269,15 +272,15 @@ def fail(http_status: int | None) -> None:
             )
             fail(None)
 
-        for route_index, route in enumerate(self._routes):
+        for route_index, route in enumerate(routes):
             if self._circuit.is_open("base_url:" + route.base_url_id):
-                if route_index + 1 < len(self._routes):
+                if route_index + 1 < len(routes):
                     fallback_used = True
                     route_reason_override = "base_url_unreached"
                     continue
                 fail(last_http_status)
             if self._circuit.is_open(key_circuit_id(route)):
-                if route_index + 1 < len(self._routes):
+                if route_index + 1 < len(routes):
                     fallback_used = True
                     route_reason_override = "key_rate_limit"
                     continue
@@ -385,7 +388,7 @@ def fail(http_status: int | None) -> None:
                         retry_count=retry_count,
                         has_job_id=False,
                     )
-                    has_next_route = route_index + 1 < len(self._routes)
+                    has_next_route = route_index + 1 < len(routes)
                     if step is NextStep.FAIL:
                         tier_step = budget.tier_b_escalation(
                             error_type,
@@ -453,7 +456,7 @@ def fail(http_status: int | None) -> None:
                         break
                     if step is NextStep.FALLBACK_KEY:
                         if has_next_route:
-                            nxt = self._routes[route_index + 1]
+                            nxt = routes[route_index + 1]
                             time.sleep(
                                 rate_limit_wait_s(
                                     retry_count=retry_count,
@@ -505,7 +508,8 @@ async def astream(self, messages: Any, **kwargs: Any):
                 f"chat gateway failed request_id={request_id} http_status=None"
             )
         last_http_status: int | None = None
-        for route_index, route in enumerate(self._routes):
+        routes = self._pool_routes()
+        for route_index, route in enumerate(routes):
             if self._circuit.is_open("base_url:" + route.base_url_id):
                 continue
             if self._circuit.is_open(key_circuit_id(route)):
@@ -534,7 +538,7 @@ async def astream(self, messages: Any, **kwargs: Any):
                 break
             else:
                 return
-            if route_index + 1 >= len(self._routes):
+            if route_index + 1 >= len(routes):
                 break
         raise RuntimeError(
             f"chat gateway failed request_id={request_id} http_status={last_http_status}"
@@ -554,7 +558,7 @@ def build_chat_gateway(config=None, *, adapter=None, circuit=None, **client_kwar
     cfg: AIProviderSettings = config or default_settings
     route_adapters = None
     if adapter is None:
-        routes = routes_from_settings(cfg, route_group=Scene.CHAT.value)
+        routes = pool_routes(cfg, route_group=Scene.CHAT.value)
         route_adapters = {
             route.route_id: LangChainChatAdapter(config_for_route(cfg, route), **client_kwargs)
             for route in routes

backend/packages/framework/src/windup_framework/gateway/image.py

@@ -17,7 +17,7 @@
     config_for_route,
     key_circuit_id,
     lookup_adapter,
-    routes_from_settings,
+    pool_routes,
 )
 from windup_framework.gateway.trace import (
     AttemptDetail,
@@ -43,9 +43,11 @@ def __init__(self, registry, adapter, circuit, settings, route_adapters=None) ->
         self._adapter = adapter
         self._circuit = circuit
         self._settings = settings
-        self._routes = routes_from_settings(settings, route_group=Scene.CHARACTER_IMAGE.value)
         self._route_adapters = dict(route_adapters or {})
 
+    def _pool_routes(self) -> tuple[GatewayRoute, ...]:
+        return pool_routes(self._settings, route_group=Scene.CHARACTER_IMAGE.value)
+
     def _adapter_for(self, route: GatewayRoute):
         return lookup_adapter(self._route_adapters, route, self._adapter)
 
@@ -58,7 +60,7 @@ def gen_image(self, prompt: str, refs: list[bytes]) -> bytes:
         fallback_used = False
         fallback_reason: str | None = None
         route_reason_override: str | None = None
-        routes = self._routes
+        routes = self._pool_routes()
         seq = AttemptSequencer()
         budget = AttemptBudget()
 
@@ -410,7 +412,7 @@ def build_image_gateway(config=None, *, adapter=None, circuit=None) -> ImageGate
     if adapter is None:
         from windup_framework.providers.sufy import SufyImageProvider
 
-        routes = routes_from_settings(cfg, route_group=Scene.CHARACTER_IMAGE.value)
+        routes = pool_routes(cfg, route_group=Scene.CHARACTER_IMAGE.value)
         route_adapters = {
             route.route_id: SufyImageProvider(config=config_for_route(cfg, route))
             for route in routes

backend/packages/framework/src/windup_framework/gateway/pool_ids.py

@@ -0,0 +1,17 @@
+"""凭证池稳定身份：与 CSV 下标、列表顺序无关。"""
+
+from __future__ import annotations
+
+import hashlib
+
+
+def credential_id(endpoint_id: str, api_key: str) -> str:
+    """``{endpoint_id}:{sha256(api_key)[:16]}`` — key 材料不变则 id 不变。"""
+    endpoint = endpoint_id.strip() or "primary"
+    digest = hashlib.sha256(api_key.encode()).hexdigest()[:16]
+    return f"{endpoint}:{digest}"
+
+
+def default_account_id(credential: str) -> str:
+    """未单独建账号时，每把 key 自己一个账号（#842 / P0 默认）。"""
+    return credential

backend/packages/framework/src/windup_framework/gateway/pool_registry.py

@@ -0,0 +1,185 @@
+"""凭证池成员 → 运行时路由边。
+
+Admit 与 Gateway 都通过 :func:`get_pool_snapshot` 读同一份快照。
+P0 只从 env 物化：无 DB、不引入独立网关进程。
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from dataclasses import dataclass
+
+from windup_framework.config.provider import AIProviderSettings, settings as default_settings
+from windup_framework.gateway.pool_ids import default_account_id
+from windup_framework.gateway.routes import GatewayRoute, routes_from_settings
+
+_CACHE: dict[str, tuple[float, "PoolSnapshot"]] = {}
+
+
+@dataclass(frozen=True)
+class RoutableEdge:
+    """可调度边。``route_id`` 与 ``credential_id`` 相同，禁止 ``primary.key{i}``。"""
+
+    route_id: str
+    credential_id: str
+    endpoint_id: str
+    account_id: str
+    route_group: str
+    candidate_index: int
+    provider_name: str
+    base_url: str
+    api_key: str
+    legacy_route_id: str
+    selectable: bool = True
+
+    def to_gateway_route(self) -> GatewayRoute:
+        return GatewayRoute(
+            route_id=self.route_id,
+            route_group=self.route_group,
+            candidate_index=self.candidate_index,
+            provider_name=self.provider_name,
+            base_url_id=self.endpoint_id,
+            base_url=self.base_url,
+            api_key_id=self.credential_id,
+            api_key=self.api_key,
+            legacy_route_id=self.legacy_route_id,
+        )
+
+    def redis_inflight_key(self) -> str:
+        return f"windup:i2v:gate:inflight:cred:{self.credential_id}"
+
+    def redis_cooling_key(self) -> str:
+        return f"windup:i2v:gate:cooling:cred:{self.credential_id}"
+
+    def redis_cooldown_key(self) -> str:
+        return f"windup:i2v:gate:cooldown:cred:{self.credential_id}"
+
+    def redis_shot_key(self) -> str:
+        return f"windup:i2v:gate:shot:cred:{self.credential_id}"
+
+
+@dataclass(frozen=True)
+class PoolSnapshot:
+    edges: tuple[RoutableEdge, ...]
+    source: str  # "settings" | "test"
+
+    def edges_for(
+        self, route_group: str, *, selectable_only: bool = True
+    ) -> tuple[RoutableEdge, ...]:
+        out = [
+            edge
+            for edge in self.edges
+            if edge.route_group == route_group and (not selectable_only or edge.selectable)
+        ]
+        return tuple(sorted(out, key=lambda edge: edge.candidate_index))
+
+    def gateway_routes(self, route_group: str) -> tuple[GatewayRoute, ...]:
+        return tuple(edge.to_gateway_route() for edge in self.edges_for(route_group))
+
+    def edge_by_credential(self, route_group: str, cred: str) -> RoutableEdge | None:
+        for edge in self.edges_for(route_group, selectable_only=False):
+            if edge.credential_id == cred or edge.legacy_route_id == cred:
+                return edge
+        return None
+
+
+def pool_cache_ttl_s() -> float:
+    raw = os.getenv("WINDUP_GATEWAY_POOL_TTL_S", "").strip()
+    if not raw:
+        return 30.0
+    return max(1.0, float(raw))
+
+
+def invalidate_pool_cache() -> None:
+    _CACHE.clear()
+
+
+def _settings_fingerprint(cfg: AIProviderSettings) -> str:
+    return "|".join(
+        (
+            cfg.provider,
+            cfg.route_primary_name,
+            cfg.effective_route_primary_base_url,
+            cfg.effective_route_primary_api_key,
+            cfg.route_primary_api_keys,
+            cfg.route_fallback_name,
+            cfg.route_fallback_base_url,
+            cfg.route_fallback_api_key,
+            cfg.route_fallback_api_keys,
+        )
+    )
+
+
+def legacy_route_id_map(
+    cfg: AIProviderSettings,
+    *,
+    route_group: str,
+) -> dict[str, str]:
+    """#842 ``primary.key{i}`` → 稳定 ``credential_id``（deploy 迁移一轮）。"""
+    return {
+        route.legacy_route_id: route.route_id
+        for route in snapshot_from_settings(cfg, route_group=route_group).gateway_routes(
+            route_group
+        )
+        if route.legacy_route_id
+    }
+
+
+def resolve_credential_id(
+    route_id: str,
+    *,
+    cfg: AIProviderSettings | None = None,
+    route_group: str,
+) -> str:
+    """task / i2v_state 里可能仍是旧 ``primary.key0``。"""
+    if not route_id:
+        return route_id
+    cfg = cfg or default_settings
+    snap = get_pool_snapshot(route_group, cfg=cfg)
+    edge = snap.edge_by_credential(route_group, route_id)
+    if edge is not None:
+        return edge.credential_id
+    return legacy_route_id_map(cfg, route_group=route_group).get(route_id, route_id)
+
+
+def snapshot_from_settings(
+    cfg: AIProviderSettings,
+    *,
+    route_group: str,
+) -> PoolSnapshot:
+    edges: list[RoutableEdge] = []
+    for route in routes_from_settings(cfg, route_group=route_group):
+        cred = route.route_id
+        edges.append(
+            RoutableEdge(
+                route_id=cred,
+                credential_id=cred,
+                endpoint_id=route.base_url_id,
+                account_id=default_account_id(cred),
+                route_group=route_group,
+                candidate_index=route.candidate_index,
+                provider_name=route.provider_name,
+                base_url=route.base_url,
+                api_key=route.api_key,
+                legacy_route_id=route.legacy_route_id,
+            )
+        )
+    return PoolSnapshot(edges=tuple(edges), source="settings")
+
+
+def get_pool_snapshot(
+    route_group: str,
+    *,
+    cfg: AIProviderSettings | None = None,
+) -> PoolSnapshot:
+    """进程内 TTL 缓存；Admit 与 Gateway 统一入口。"""
+    now = time.monotonic()
+    cfg = cfg or default_settings
+    cache_key = f"{route_group}:{_settings_fingerprint(cfg)}"
+    cached = _CACHE.get(cache_key)
+    if cached is not None and now - cached[0] < pool_cache_ttl_s():
+        return cached[1]
+    snap = snapshot_from_settings(cfg, route_group=route_group)
+    _CACHE[cache_key] = (now, snap)
+    return snap