diff --git a/backend/packages/app/src/windup_app/server/mq/i2v_admit.py b/backend/packages/app/src/windup_app/server/mq/i2v_admit.py
index 88889361..a76aa447 100644
--- a/backend/packages/app/src/windup_app/server/mq/i2v_admit.py
+++ b/backend/packages/app/src/windup_app/server/mq/i2v_admit.py
@@ -1,4 +1,4 @@
-"""全站 i2v 建单闸：按 API key 分车道，每把 key 独立在途名额与 429 冷却。"""
+"""全站 i2v 建单闸：按 credential 分车道，每把 key 独立在途名额与 429 冷却。"""
 
 from __future__ import annotations
 
@@ -7,19 +7,26 @@
 from windup_app.server.mq.catalog import MSG_TYPE_CHARACTER_ACTION, stream_for_msg_type
 from windup_app.server.mq.i2v_state import I2V_KEY_PREFIX, load_i2v_state
 from windup_framework.db.redis import get_redis
+from windup_framework.gateway.pool_registry import (
+    RoutableEdge,
+    get_pool_snapshot,
+    legacy_route_id_map,
+)
+from windup_framework.gateway.types import Scene
 from windup_framework.mq.delayed import schedule_delayed
 
-# 全站「已占名额」索引，不算容量。容量在每条车道的 SET 上。
 INFLIGHT_KEY = "windup:i2v:gate:inflight"
-LANE_INFLIGHT_PREFIX = "windup:i2v:gate:inflight:"
-LANE_COOLING_PREFIX = "windup:i2v:gate:cooling:"
-LANE_COOLDOWN_PREFIX = "windup:i2v:gate:cooldown:"
-LANE_SHOT_PREFIX = "windup:i2v:gate:shot:"
+# #842 遗留：windup:i2v:gate:inflight:{primary.key0}
+LEGACY_LANE_INFLIGHT_PREFIX = "windup:i2v:gate:inflight:"
+LEGACY_LANE_COOLING_PREFIX = "windup:i2v:gate:cooling:"
+LEGACY_LANE_COOLDOWN_PREFIX = "windup:i2v:gate:cooldown:"
+LEGACY_LANE_SHOT_PREFIX = "windup:i2v:gate:shot:"
 TASK_HASH_PREFIX = "windup:i2v:gate:task:"
 
 ADMIT_RETRY_S = 5.0
 _FALLBACK_RETRY_S = 1.0
 _TASK_HASH_TTL_S = 2 * 3600
+_ROUTE_GROUP = Scene.CHARACTER_ACTION.value
 
 _ACQUIRE_LUA = """
 local setkey = KEYS[1]
@@ -62,17 +69,14 @@ def inflight_max() -> int:
     return max(1, int(raw))
 
 
-def lane_ids() -> tuple[str, ...]:
-    """当前视频路由上的 key 车道。测试可替换。"""
-    from windup_framework.config.provider import settings as ai_settings
-    from windup_framework.gateway.routes import routes_from_settings
-    from windup_framework.gateway.types import Scene
+def _action_edges() -> tuple[RoutableEdge, ...]:
+    return get_pool_snapshot(_ROUTE_GROUP).edges_for(_ROUTE_GROUP)
 
-    routes = routes_from_settings(
-        ai_settings, route_group=Scene.CHARACTER_ACTION.value
-    )
-    ids = tuple(route.route_id for route in routes)
-    return ids or ("primary.key0",)
+
+def lane_ids() -> tuple[str, ...]:
+    """当前视频池 credential_id 列表。测试可 monkeypatch ``_action_edges``。"""
+    ids = tuple(edge.credential_id for edge in _action_edges())
+    return ids or ("primary:0000000000000000",)
 
 
 def _task_member(task_id: int) -> str:
@@ -83,20 +87,54 @@ def _task_hash(task_id: int) -> str:
     return f"{TASK_HASH_PREFIX}{task_id}"
 
 
-def _lane_inflight(lane: str) -> str:
-    return f"{LANE_INFLIGHT_PREFIX}{lane}"
+def _legacy_inflight(lane: str) -> str:
+    return f"{LEGACY_LANE_INFLIGHT_PREFIX}{lane}"
+
 
+def _legacy_cooling(lane: str) -> str:
+    return f"{LEGACY_LANE_COOLING_PREFIX}{lane}"
 
-def _lane_cooling(lane: str) -> str:
-    return f"{LANE_COOLING_PREFIX}{lane}"
 
+def _legacy_cooldown(lane: str) -> str:
+    return f"{LEGACY_LANE_COOLDOWN_PREFIX}{lane}"
 
-def _lane_cooldown(lane: str) -> str:
-    return f"{LANE_COOLDOWN_PREFIX}{lane}"
 
+def _legacy_shot(lane: str) -> str:
+    return f"{LEGACY_LANE_SHOT_PREFIX}{lane}"
 
-def _lane_shot(lane: str) -> str:
-    return f"{LANE_SHOT_PREFIX}{lane}"
+
+def _normalize_route_id(route_id: str) -> str:
+    if not route_id:
+        return route_id
+    for edge in _action_edges():
+        if route_id in {edge.credential_id, edge.legacy_route_id, edge.route_id}:
+            return edge.credential_id
+    from windup_framework.config.provider import settings as ai_settings
+
+    return legacy_route_id_map(ai_settings, route_group=_ROUTE_GROUP).get(
+        route_id, route_id
+    )
+
+
+def _edge_for_credential(cred: str) -> RoutableEdge | None:
+    resolved = _normalize_route_id(cred)
+    for edge in _action_edges():
+        if edge.credential_id == resolved:
+            return edge
+    return None
+
+
+def _migrate_legacy_inflight(task_id: int, raw_lane: str, cred: str) -> None:
+    """#842 车道键迁到 ``inflight:cred:{credential_id}``。"""
+    if not raw_lane or raw_lane == cred:
+        return
+    redis_client = get_redis()
+    member = _task_member(task_id)
+    old_key = _legacy_inflight(raw_lane)
+    new_key = f"windup:i2v:gate:inflight:cred:{cred}"
+    if redis_client.sismember(old_key, member):
+        redis_client.sadd(new_key, member)
+        redis_client.srem(old_key, member)
 
 
 def _bound_lane(task_id: int) -> str | None:
@@ -105,77 +143,109 @@ def _bound_lane(task_id: int) -> str | None:
     return lane or None
 
 
-def _lane_index(lane: str) -> int:
-    lanes = lane_ids()
-    try:
-        return lanes.index(lane)
-    except ValueError:
-        return 0
+def _bound_credential(task_id: int) -> str | None:
+    lane = _bound_lane(task_id)
+    if not lane:
+        return None
+    return _normalize_route_id(lane)
 
 
-def _lane_cooling_wait_s(lane: str) -> float:
-    ttl = get_redis().ttl(_lane_cooldown(lane))
-    if ttl is None or int(ttl) < 0:
-        return 0.0
-    return float(ttl)
+def _ttl_or_zero(*keys: str) -> float:
+    redis_client = get_redis()
+    for key in keys:
+        if not key:
+            continue
+        ttl = redis_client.ttl(key)
+        if ttl is not None and int(ttl) >= 0:
+            return float(ttl)
+    return 0.0
+
+
+def _lane_cooling_wait_s_for_cred(cred: str) -> float:
+    edge = _edge_for_credential(cred)
+    if edge is None:
+        return _ttl_or_zero(_legacy_cooldown(cred))
+    return _ttl_or_zero(
+        edge.redis_cooldown_key(),
+        _legacy_cooldown(edge.legacy_route_id),
+        _legacy_cooldown(cred),
+    )
 
 
-def _lane_is_hot(lane: str) -> bool:
-    """冷却倒计时还没走完，新任务不该再挤这条车道。"""
-    return _lane_cooling_wait_s(lane) > 0
+def _lane_is_hot(edge: RoutableEdge) -> bool:
+    return _lane_cooling_wait_s_for_cred(edge.credential_id) > 0
 
 
-def _bind(task_id: int, lane: str) -> None:
+def _bind(task_id: int, edge: RoutableEdge) -> None:
     redis_client = get_redis()
     key = _task_hash(task_id)
     redis_client.hset(
         key,
         mapping={
-            "route_id": lane,
-            "route_skip": str(_lane_index(lane)),
+            "route_id": edge.credential_id,
+            "route_skip": str(edge.candidate_index),
         },
     )
     redis_client.expire(key, _TASK_HASH_TTL_S)
     redis_client.sadd(INFLIGHT_KEY, _task_member(task_id))
 
 
-def _unbind_lane(task_id: int, lane: str | None) -> None:
-    if not lane:
+def _unbind_edge(task_id: int, cred: str | None) -> None:
+    if not cred:
         return
-    get_redis().srem(_lane_inflight(lane), _task_member(task_id))
+    redis_client = get_redis()
+    member = _task_member(task_id)
+    edge = _edge_for_credential(cred)
+    if edge is not None:
+        redis_client.srem(edge.redis_inflight_key(), member)
+        if edge.legacy_route_id:
+            redis_client.srem(_legacy_inflight(edge.legacy_route_id), member)
+    raw = _bound_lane(task_id) or cred
+    if raw != cred:
+        redis_client.srem(_legacy_inflight(raw), member)
 
 
-def _acquire_lane(lane: str, task_id: int) -> bool:
+def _acquire_edge(edge: RoutableEdge, task_id: int) -> bool:
     got = get_redis().eval(
         _ACQUIRE_LUA,
         1,
-        _lane_inflight(lane),
+        edge.redis_inflight_key(),
         _task_member(task_id),
         inflight_max(),
     )
     return int(got or 0) == 1
 
 
-def _lanes_by_load() -> list[str]:
+def _edges_by_load() -> list[RoutableEdge]:
     redis_client = get_redis()
     scored = [
-        (int(redis_client.scard(_lane_inflight(lane)) or 0), index, lane)
-        for index, lane in enumerate(lane_ids())
+        (
+            int(redis_client.scard(edge.redis_inflight_key()) or 0),
+            edge.candidate_index,
+            edge,
+        )
+        for edge in _action_edges()
     ]
     scored.sort()
-    return [lane for _load, _index, lane in scored]
+    return [edge for _load, _index, edge in scored]
 
 
 def try_acquire(task_id: int) -> bool:
-    """占一条 key 车道的在途坑。已占过则成功（延迟再入队幂等）。"""
-    bound = _bound_lane(task_id)
-    if bound and get_redis().sismember(_lane_inflight(bound), _task_member(task_id)):
-        return True
-    for lane in _lanes_by_load():
-        if _lane_is_hot(lane):
+    """占一条 credential 车道的在途坑。已占过则成功（延迟再入队幂等）。"""
+    bound_raw = _bound_lane(task_id)
+    if bound_raw:
+        cred = _normalize_route_id(bound_raw)
+        _migrate_legacy_inflight(task_id, bound_raw, cred)
+        edge = _edge_for_credential(cred)
+        if edge and get_redis().sismember(
+            edge.redis_inflight_key(), _task_member(task_id)
+        ):
+            return True
+    for edge in _edges_by_load():
+        if _lane_is_hot(edge):
             continue
-        if _acquire_lane(lane, task_id):
-            _bind(task_id, lane)
+        if _acquire_edge(edge, task_id):
+            _bind(task_id, edge)
             return True
     return False
 
@@ -184,11 +254,18 @@ def release(task_id: int) -> None:
     redis_client = get_redis()
     member = _task_member(task_id)
     redis_client.srem(INFLIGHT_KEY, member)
-    for lane in lane_ids():
-        redis_client.srem(_lane_inflight(lane), member)
-    bound = _bound_lane(task_id)
-    if bound:
-        redis_client.srem(_lane_inflight(bound), member)
+    bound_raw = _bound_lane(task_id)
+    cred = _normalize_route_id(bound_raw) if bound_raw else None
+    for edge in _action_edges():
+        redis_client.srem(edge.redis_inflight_key(), member)
+        if edge.legacy_route_id:
+            redis_client.srem(_legacy_inflight(edge.legacy_route_id), member)
+    if bound_raw:
+        redis_client.srem(_legacy_inflight(bound_raw), member)
+    if cred:
+        edge = _edge_for_credential(cred)
+        if edge is not None:
+            redis_client.srem(edge.redis_inflight_key(), member)
     redis_client.delete(_task_hash(task_id))
 
 
@@ -197,10 +274,14 @@ def has_claim(task_id: int) -> bool:
 
 
 def claimed_ids() -> tuple[int, ...]:
-    """全局索引 ∪ 各车道 SET。终态对账时两边都要扫，避免只清了一边。"""
+    """全局索引 ∪ 凭证 SET ∪ 遗留 #842 车道。终态对账时都要扫。"""
     redis_client = get_redis()
     ids: set[int] = set()
-    keys = (INFLIGHT_KEY, *(_lane_inflight(lane) for lane in lane_ids()))
+    keys = [INFLIGHT_KEY]
+    for edge in _action_edges():
+        keys.append(edge.redis_inflight_key())
+        if edge.legacy_route_id:
+            keys.append(_legacy_inflight(edge.legacy_route_id))
     for key in keys:
         for raw in redis_client.smembers(key) or ():
             text = raw.decode() if isinstance(raw, bytes) else str(raw)
@@ -211,13 +292,16 @@ def claimed_ids() -> tuple[int, ...]:
 
 def can_submit(task_id: int) -> bool:
     """该任务所在车道健康时可并行；冷却期内等到点且只能一枪。"""
-    lane = _bound_lane(task_id) or lane_ids()[0]
+    cred = _bound_credential(task_id) or lane_ids()[0]
+    edge = _edge_for_credential(cred)
+    if edge is None:
+        return True
     got = get_redis().eval(
         _SUBMIT_LUA,
         3,
-        _lane_cooling(lane),
-        _lane_cooldown(lane),
-        _lane_shot(lane),
+        edge.redis_cooling_key(),
+        edge.redis_cooldown_key(),
+        edge.redis_shot_key(),
         _task_member(task_id),
     )
     return int(got or 0) == 1
@@ -225,23 +309,30 @@ def can_submit(task_id: int) -> bool:
 
 def cooldown_remaining_s(task_id: int | None = None) -> float:
     if task_id is not None:
-        lane = _bound_lane(task_id)
-        if lane:
-            return _lane_cooling_wait_s(lane)
-    waits = [_lane_cooling_wait_s(lane) for lane in lane_ids()]
+        cred = _bound_credential(task_id)
+        if cred:
+            return _lane_cooling_wait_s_for_cred(cred)
+    waits = [_lane_cooling_wait_s_for_cred(cred) for cred in lane_ids()]
     return min(waits) if waits else 0.0
 
 
 def on_rate_limit(*, wait_s: float, fallback_key: bool, task_id: int) -> float:
-    """只冷却当前这条 key。换 key 时立刻改挂空闲车道，不必连坐其它 key。"""
+    """只冷却当前这把 credential。换 key 时立刻改挂空闲车道，不必连坐其它 key。"""
     wait = max(1.0, float(wait_s))
     redis_client = get_redis()
-    lane = _bound_lane(task_id) or lane_ids()[0]
-    redis_client.set(_lane_cooling(lane), "1")
-    redis_client.set(_lane_cooldown(lane), "1", ex=int(wait))
-    redis_client.delete(_lane_shot(lane))
+    cred = _bound_credential(task_id) or lane_ids()[0]
+    edge = _edge_for_credential(cred)
+    if edge is not None:
+        redis_client.set(edge.redis_cooling_key(), "1")
+        redis_client.set(edge.redis_cooldown_key(), "1", ex=int(wait))
+        redis_client.delete(edge.redis_shot_key())
+    bound_raw = _bound_lane(task_id)
+    if bound_raw and bound_raw != cred:
+        redis_client.set(_legacy_cooling(bound_raw), "1")
+        redis_client.set(_legacy_cooldown(bound_raw), "1", ex=int(wait))
+        redis_client.delete(_legacy_shot(bound_raw))
     if fallback_key:
-        _unbind_lane(task_id, lane)
+        _unbind_edge(task_id, cred)
         skip, _retry = retry_state(task_id)
         redis_client.hset(
             _task_hash(task_id),
@@ -257,17 +348,34 @@ def on_rate_limit(*, wait_s: float, fallback_key: bool, task_id: int) -> float:
 
 def clear_cooling(task_id: int | None = None) -> None:
     redis_client = get_redis()
-    lanes = []
+    edges: list[RoutableEdge] = []
     if task_id is not None:
-        bound = _bound_lane(task_id)
-        if bound:
-            lanes = [bound]
-    if not lanes:
-        lanes = list(lane_ids())
-    for lane in lanes:
+        cred = _bound_credential(task_id)
+        if cred:
+            edge = _edge_for_credential(cred)
+            if edge is not None:
+                edges = [edge]
+            bound_raw = _bound_lane(task_id)
+            if bound_raw and bound_raw != cred:
+                redis_client.delete(
+                    _legacy_cooling(bound_raw),
+                    _legacy_cooldown(bound_raw),
+                    _legacy_shot(bound_raw),
+                )
+    if not edges:
+        edges = list(_action_edges())
+    for edge in edges:
         redis_client.delete(
-            _lane_cooling(lane), _lane_cooldown(lane), _lane_shot(lane)
+            edge.redis_cooling_key(),
+            edge.redis_cooldown_key(),
+            edge.redis_shot_key(),
         )
+        if edge.legacy_route_id:
+            redis_client.delete(
+                _legacy_cooling(edge.legacy_route_id),
+                _legacy_cooldown(edge.legacy_route_id),
+                _legacy_shot(edge.legacy_route_id),
+            )
 
 
 def retry_state(task_id: int) -> tuple[int, int]:
@@ -309,7 +417,8 @@ def schedule_retry(task_id: int, delay_s: float) -> None:
 def rebuild() -> None:
     """进程重启后按已建单的 i2v 状态把在途集合补回去。"""
     redis_client = get_redis()
-    fallback = lane_ids()[0]
+    edges = _action_edges()
+    fallback = edges[0].credential_id if edges else lane_ids()[0]
     for key in redis_client.scan_iter(match=f"{I2V_KEY_PREFIX}*"):
         name = key.decode() if isinstance(key, bytes) else str(key)
         suffix = name[len(I2V_KEY_PREFIX) :]
@@ -319,7 +428,23 @@ def rebuild() -> None:
         if not state or not state.get("job_id"):
             continue
         task_id = int(suffix)
-        lane = str(state.get("route_id") or "") or fallback
+        raw_lane = str(state.get("route_id") or "") or fallback
+        cred = _normalize_route_id(raw_lane)
+        edge = _edge_for_credential(cred)
+        lane_cred = edge.credential_id if edge else cred
         redis_client.sadd(INFLIGHT_KEY, suffix)
-        redis_client.sadd(_lane_inflight(lane), suffix)
-        _bind(task_id, lane)
+        inflight_key = (
+            edge.redis_inflight_key()
+            if edge
+            else f"windup:i2v:gate:inflight:cred:{lane_cred}"
+        )
+        redis_client.sadd(inflight_key, suffix)
+        _migrate_legacy_inflight(task_id, raw_lane, lane_cred)
+        if edge is not None:
+            _bind(task_id, edge)
+        else:
+            redis_client.hset(
+                _task_hash(task_id),
+                mapping={"route_id": lane_cred, "route_skip": "0"},
+            )
+            redis_client.expire(_task_hash(task_id), _TASK_HASH_TTL_S)
diff --git a/backend/packages/framework/src/windup_framework/gateway/__init__.py b/backend/packages/framework/src/windup_framework/gateway/__init__.py
index 0d4ffbaa..5c9cbcf4 100644
--- a/backend/packages/framework/src/windup_framework/gateway/__init__.py
+++ b/backend/packages/framework/src/windup_framework/gateway/__init__.py
@@ -2,6 +2,13 @@
 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__ = [
@@ -9,10 +16,15 @@
     "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",
 ]
diff --git a/backend/packages/framework/src/windup_framework/gateway/chat.py b/backend/packages/framework/src/windup_framework/gateway/chat.py
index 84cb97a1..5dd866b0 100644
--- a/backend/packages/framework/src/windup_framework/gateway/chat.py
+++ b/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
diff --git a/backend/packages/framework/src/windup_framework/gateway/image.py b/backend/packages/framework/src/windup_framework/gateway/image.py
index 21cd574d..c4bb5c90 100644
--- a/backend/packages/framework/src/windup_framework/gateway/image.py
+++ b/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
diff --git a/backend/packages/framework/src/windup_framework/gateway/pool_ids.py b/backend/packages/framework/src/windup_framework/gateway/pool_ids.py
new file mode 100644
index 00000000..a254aad1
--- /dev/null
+++ b/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
diff --git a/backend/packages/framework/src/windup_framework/gateway/pool_registry.py b/backend/packages/framework/src/windup_framework/gateway/pool_registry.py
new file mode 100644
index 00000000..9713d01a
--- /dev/null
+++ b/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
diff --git a/backend/packages/framework/src/windup_framework/gateway/routes.py b/backend/packages/framework/src/windup_framework/gateway/routes.py
index 89db9aae..a718c746 100644
--- a/backend/packages/framework/src/windup_framework/gateway/routes.py
+++ b/backend/packages/framework/src/windup_framework/gateway/routes.py
@@ -4,6 +4,7 @@
 from urllib.parse import urlparse
 
 from windup_framework.config.provider import AIProviderSettings
+from windup_framework.gateway.pool_ids import credential_id
 
 
 @dataclass(frozen=True)
@@ -16,6 +17,7 @@ class GatewayRoute:
     base_url: str
     api_key_id: str | None
     api_key: str
+    legacy_route_id: str = ""
 
     @property
     def host(self) -> str | None:
@@ -46,7 +48,8 @@ def _expand_url(
 ) -> list[GatewayRoute]:
     routes: list[GatewayRoute] = []
     for i, api_key in enumerate(_unique_keys(first_key, extra_keys)):
-        api_key_id = f"{base_url_id}.key{i}"
+        legacy_route_id = f"{base_url_id}.key{i}"
+        api_key_id = credential_id(base_url_id, api_key)
         routes.append(
             GatewayRoute(
                 route_id=api_key_id,
@@ -57,11 +60,19 @@ def _expand_url(
                 base_url=base_url,
                 api_key_id=api_key_id,
                 api_key=api_key,
+                legacy_route_id=legacy_route_id,
             )
         )
     return routes
 
 
+def pool_routes(cfg: AIProviderSettings, *, route_group: str) -> tuple[GatewayRoute, ...]:
+    """凭证池物化路由；Admit 与 Gateway 统一读 :func:`get_pool_snapshot`。"""
+    from windup_framework.gateway.pool_registry import get_pool_snapshot
+
+    return get_pool_snapshot(route_group, cfg=cfg).gateway_routes(route_group)
+
+
 def routes_from_settings(cfg: AIProviderSettings, *, route_group: str) -> tuple[GatewayRoute, ...]:
     primary_name = cfg.route_primary_name.strip() or "primary"
     routes = _expand_url(
@@ -97,6 +108,7 @@ def lookup_adapter(route_adapters: dict, route: GatewayRoute, default):
     return (
         route_adapters.get(route.route_id)
         or route_adapters.get(route.api_key_id)
+        or route_adapters.get(route.legacy_route_id)
         or route_adapters.get(route.base_url_id)
         or default
     )
diff --git a/backend/packages/framework/src/windup_framework/gateway/video.py b/backend/packages/framework/src/windup_framework/gateway/video.py
index d06c8982..29a0e3e4 100644
--- a/backend/packages/framework/src/windup_framework/gateway/video.py
+++ b/backend/packages/framework/src/windup_framework/gateway/video.py
@@ -14,13 +14,14 @@
 from windup_framework.gateway.context import current_call_context
 from windup_framework.gateway.image import _CIRCUIT
 from windup_framework.gateway.policy import decide, rate_limit_wait_s
+from windup_framework.gateway.pool_registry import resolve_credential_id
 from windup_framework.gateway.registry import ModelRegistry, RegistryError
 from windup_framework.gateway.routes import (
     GatewayRoute,
     config_for_route,
     key_circuit_id,
     lookup_adapter,
-    routes_from_settings,
+    pool_routes,
 )
 from windup_framework.gateway.sequencer import AttemptSequencer
 from windup_framework.gateway.trace import (
@@ -53,9 +54,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_ACTION.value)
         self._route_adapters = dict(route_adapters or {})
 
+    def _pool_routes(self) -> tuple[GatewayRoute, ...]:
+        return pool_routes(self._settings, route_group=Scene.CHARACTER_ACTION.value)
+
     def _adapter_for(self, route: GatewayRoute):
         return lookup_adapter(self._route_adapters, route, self._adapter)
 
@@ -82,12 +85,19 @@ def poll_i2v(
         """
         adapter = self._adapter
         if route_id:
-            for route in self._routes:
-                if route.route_id == route_id:
+            resolved = resolve_credential_id(
+                route_id,
+                cfg=self._settings,
+                route_group=Scene.CHARACTER_ACTION.value,
+            )
+            for route in self._pool_routes():
+                if route.route_id in {route_id, resolved}:
                     adapter = self._adapter_for(route)
                     break
             else:
-                mapped = self._route_adapters.get(route_id)
+                mapped = self._route_adapters.get(route_id) or self._route_adapters.get(
+                    resolved
+                )
                 if mapped is not None:
                     adapter = mapped
         if hasattr(adapter, "inspect_job"):
@@ -119,7 +129,7 @@ def i2v(
         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()
 
@@ -608,7 +618,7 @@ def build_video_gateway(
     if adapter is None:
         from windup_framework.providers.sufy import SufyVideoProvider
 
-        routes = routes_from_settings(cfg, route_group=Scene.CHARACTER_ACTION.value)
+        routes = pool_routes(cfg, route_group=Scene.CHARACTER_ACTION.value)
         route_adapters = {
             route.route_id: SufyVideoProvider(
                 config=config_for_route(cfg, route), uploader=uploader
diff --git a/backend/tests/test_gateway_chat.py b/backend/tests/test_gateway_chat.py
index 765a0255..b1e46e73 100644
--- a/backend/tests/test_gateway_chat.py
+++ b/backend/tests/test_gateway_chat.py
@@ -15,7 +15,7 @@
     LangChainChatAdapter,
 )
 from windup_framework.gateway.circuit import CircuitBreaker
-from windup_framework.gateway.routes import key_circuit_id, routes_from_settings
+from windup_framework.gateway.routes import key_circuit_id, pool_routes
 from windup_framework.gateway.types import Scene
 from windup_framework.providers.chat import create_chat_model
 
@@ -194,7 +194,7 @@ def test_chat_skips_open_key_circuit_to_next_key():
     key_a = FakeChatAdapter({"gpt-4o-mini": [OK]})
     key_b = FakeChatAdapter({"gpt-4o-mini": [OK]})
     cfg = _primary_cfg(route_primary_api_keys="key-b")
-    routes = routes_from_settings(cfg, route_group=Scene.CHAT.value)
+    routes = pool_routes(cfg, route_group=Scene.CHAT.value)
     circuit = CircuitBreaker()
     circuit.open(key_circuit_id(routes[0]))
     gw = ChatGateway(
diff --git a/backend/tests/test_gateway_image.py b/backend/tests/test_gateway_image.py
index 69ef4a65..99f5c3c3 100644
--- a/backend/tests/test_gateway_image.py
+++ b/backend/tests/test_gateway_image.py
@@ -182,7 +182,9 @@ def test_429_switches_key_on_same_base_url_before_model(monkeypatch, caplog):
     assert line["route_reason"] == "key_rate_limit"
     assert line["route_layer"] == "key"
     assert line["base_url_id"] == "primary"
-    assert line["api_key_id"].endswith("key1")
+    from windup_framework.gateway.pool_ids import credential_id
+
+    assert line["api_key_id"] == credential_id("primary", "key-b")
 
 
 def test_429_exhausted_keys_switches_backup_entry(monkeypatch, caplog):
diff --git a/backend/tests/test_gateway_pool_ids.py b/backend/tests/test_gateway_pool_ids.py
new file mode 100644
index 00000000..0bb0c18e
--- /dev/null
+++ b/backend/tests/test_gateway_pool_ids.py
@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from windup_framework.config.provider import AIProviderSettings
+from windup_framework.gateway.pool_ids import credential_id, default_account_id
+from windup_framework.gateway.pool_registry import (
+    invalidate_pool_cache,
+    snapshot_from_settings,
+)
+from windup_framework.gateway.routes import pool_routes, routes_from_settings
+
+
+def test_credential_id_is_stable_for_same_key():
+    a = credential_id("primary", "sk-secret")
+    b = credential_id("primary", "sk-secret")
+    assert a == b
+    assert a.startswith("primary:")
+    assert ".key" not in a
+
+
+def test_credential_id_differs_for_different_keys():
+    assert credential_id("primary", "sk-a") != credential_id("primary", "sk-b")
+
+
+def test_default_account_id_matches_credential():
+    cred = credential_id("primary", "sk-x")
+    assert default_account_id(cred) == cred
+
+
+def _cfg(*keys: str, extra: str = "") -> AIProviderSettings:
+    first, *rest = keys
+    return AIProviderSettings(
+        route_primary_name="primary",
+        route_primary_base_url="https://api.qnaigc.com/v1",
+        route_primary_api_key=first,
+        route_primary_api_keys=",".join(rest) if rest else extra,
+        route_fallback_name="",
+        route_fallback_base_url="",
+        route_fallback_api_key="",
+        route_fallback_api_keys="",
+    )
+
+
+def test_routes_use_stable_id_not_csv_index():
+    routes = routes_from_settings(_cfg("key-a", "key-b"), route_group="character_action")
+    assert routes[0].route_id == credential_id("primary", "key-a")
+    assert routes[1].route_id == credential_id("primary", "key-b")
+    assert routes[0].legacy_route_id == "primary.key0"
+    assert routes[1].legacy_route_id == "primary.key1"
+    assert routes[0].api_key_id == routes[0].route_id
+
+
+def test_insert_or_delete_key_keeps_physical_credential_id():
+    before = {r.api_key: r.route_id for r in routes_from_settings(
+        _cfg("key-a", "key-b", "key-c"), route_group="character_image"
+    )}
+    inserted = {r.api_key: r.route_id for r in routes_from_settings(
+        _cfg("key-new", "key-a", "key-b", "key-c"), route_group="character_image"
+    )}
+    deleted = {r.api_key: r.route_id for r in routes_from_settings(
+        _cfg("key-a", "key-c"), route_group="character_image"
+    )}
+    assert inserted["key-a"] == before["key-a"]
+    assert inserted["key-b"] == before["key-b"]
+    assert inserted["key-c"] == before["key-c"]
+    assert deleted["key-a"] == before["key-a"]
+    assert deleted["key-c"] == before["key-c"]
+    assert inserted["key-new"] != before["key-a"]
+
+
+def test_pool_routes_match_snapshot():
+    invalidate_pool_cache()
+    cfg = _cfg("key-a", "key-b")
+    snap = snapshot_from_settings(cfg, route_group="character_action")
+    routes = pool_routes(cfg, route_group="character_action")
+    assert snap.source == "settings"
+    assert [r.route_id for r in routes] == [e.credential_id for e in snap.edges]
+    assert all(".key" not in e.credential_id for e in snap.edges)
diff --git a/backend/tests/test_gateway_route_config.py b/backend/tests/test_gateway_route_config.py
index e4117335..678fb122 100644
--- a/backend/tests/test_gateway_route_config.py
+++ b/backend/tests/test_gateway_route_config.py
@@ -56,6 +56,10 @@ def test_routes_expand_extra_keys_on_same_base_url_before_fallback_url():
         ("backup", "key-c", "https://backup.example.com/v1"),
     ]
     assert routes[0].api_key_id != routes[1].api_key_id
+    assert routes[0].legacy_route_id == "primary.key0"
+    assert routes[1].legacy_route_id == "primary.key1"
+    assert routes[2].legacy_route_id == "backup.key0"
+    assert routes[0].route_id != routes[0].legacy_route_id
     assert routes[0].candidate_index == 0
     assert routes[1].candidate_index == 1
     assert routes[2].candidate_index == 2
diff --git a/backend/tests/test_i2v_admit.py b/backend/tests/test_i2v_admit.py
index 65382aef..ad911e16 100644
--- a/backend/tests/test_i2v_admit.py
+++ b/backend/tests/test_i2v_admit.py
@@ -8,13 +8,32 @@
 
 from windup_app.server.mq import i2v_admit as admit
 from windup_common.enums.model import ModelErrorType
+from windup_framework.config.provider import AIProviderSettings
 from windup_framework.gateway.context import bind_call_context
 from windup_framework.gateway.errors import RateLimitBackoff
+from windup_framework.gateway.pool_ids import credential_id
+from windup_framework.gateway.pool_registry import RoutableEdge, snapshot_from_settings
 from windup_framework.gateway.types import AdapterResult
 
 from test_gateway_video import FakeVideoAdapter, _video_gw
 
 
+def _test_edge(cred_id: str, *, index: int = 0, legacy: str = "") -> RoutableEdge:
+    return RoutableEdge(
+        route_id=cred_id,
+        credential_id=cred_id,
+        endpoint_id="primary",
+        account_id=cred_id,
+        route_group="character_action",
+        candidate_index=index,
+        provider_name="openai-compatible",
+        base_url="https://example.com/v1",
+        api_key="sk-test",
+        legacy_route_id=legacy or f"primary.key{index}",
+        selectable=True,
+    )
+
+
 class _MemRedis:
     def __init__(self) -> None:
         self.sets: dict[str, set[str]] = {}
@@ -142,18 +161,19 @@ def eval(self, script, nkeys, *args):
 def _patch_redis(
     monkeypatch,
     mem: _MemRedis | None = None,
-    lanes: tuple[str, ...] = ("primary.key0",),
+    cred_ids: tuple[str, ...] = ("cred-a",),
 ) -> _MemRedis:
     mem = mem or _MemRedis()
+    edges = tuple(_test_edge(cred, index=i) for i, cred in enumerate(cred_ids))
     monkeypatch.setattr("windup_app.server.mq.i2v_admit.get_redis", lambda: mem)
-    monkeypatch.setattr("windup_app.server.mq.i2v_admit.lane_ids", lambda: lanes)
+    monkeypatch.setattr("windup_app.server.mq.i2v_admit._action_edges", lambda: edges)
     return mem
 
 
-def test_claimed_ids_unions_global_and_lane_sets(monkeypatch):
-    mem = _patch_redis(monkeypatch, lanes=("primary.key0", "primary.key1"))
+def test_claimed_ids_unions_global_cred_and_legacy_sets(monkeypatch):
+    mem = _patch_redis(monkeypatch, cred_ids=("cred-a", "cred-b"))
     mem.sadd("windup:i2v:gate:inflight", "632")
-    mem.sadd("windup:i2v:gate:inflight:primary.key0", "632")
+    mem.sadd("windup:i2v:gate:inflight:cred:cred-a", "632")
     mem.sadd("windup:i2v:gate:inflight:primary.key1", "710")
     assert admit.claimed_ids() == (632, 710)
 
@@ -192,7 +212,7 @@ def test_fallback_key_advances_route_skip(monkeypatch):
 
 
 def test_two_keys_spread_load_and_cap_each(monkeypatch):
-    _patch_redis(monkeypatch, lanes=("primary.key0", "primary.key1"))
+    _patch_redis(monkeypatch, cred_ids=("cred-a", "cred-b"))
     assert admit.try_acquire(1)
     assert admit.try_acquire(2)
     skip1, _ = admit.retry_state(1)
@@ -204,7 +224,7 @@ def test_two_keys_spread_load_and_cap_each(monkeypatch):
 
 
 def test_429_on_one_key_does_not_cool_the_other(monkeypatch):
-    _patch_redis(monkeypatch, lanes=("primary.key0", "primary.key1"))
+    _patch_redis(monkeypatch, cred_ids=("cred-a", "cred-b"))
     assert admit.try_acquire(1)
     assert admit.try_acquire(2)
     wait = admit.on_rate_limit(wait_s=8, fallback_key=False, task_id=1)
@@ -214,7 +234,7 @@ def test_429_on_one_key_does_not_cool_the_other(monkeypatch):
 
 
 def test_fallback_moves_claim_to_idle_key(monkeypatch):
-    _patch_redis(monkeypatch, lanes=("primary.key0", "primary.key1"))
+    _patch_redis(monkeypatch, cred_ids=("cred-a", "cred-b"))
     assert admit.try_acquire(1)
     skip_before, _ = admit.retry_state(1)
     wait = admit.on_rate_limit(wait_s=16, fallback_key=True, task_id=1)
@@ -225,6 +245,84 @@ def test_fallback_moves_claim_to_idle_key(monkeypatch):
     assert admit.can_submit(1)
 
 
+def test_insert_key_at_head_keeps_inflight_on_physical_credential(monkeypatch):
+    """配置头部插 key 后，已占坑任务仍绑原 credential，不戴到新钥匙上。"""
+    mem = _MemRedis()
+    original = (
+        _test_edge("cred-a", index=0, legacy="primary.key0"),
+        _test_edge("cred-b", index=1, legacy="primary.key1"),
+    )
+    monkeypatch.setattr("windup_app.server.mq.i2v_admit.get_redis", lambda: mem)
+    monkeypatch.setattr("windup_app.server.mq.i2v_admit._action_edges", lambda: original)
+    assert admit.try_acquire(1)
+    assert admit.try_acquire(2)
+    bound_one = admit.retry_state(1)
+    assert "1" in mem.sets["windup:i2v:gate:inflight:cred:cred-a"]
+    assert "2" in mem.sets["windup:i2v:gate:inflight:cred:cred-b"]
+
+    inserted = (
+        _test_edge("cred-new", index=0, legacy="primary.key0"),
+        _test_edge("cred-a", index=1, legacy="primary.key1"),
+        _test_edge("cred-b", index=2, legacy="primary.key2"),
+    )
+    monkeypatch.setattr("windup_app.server.mq.i2v_admit._action_edges", lambda: inserted)
+    assert admit.has_claim(1)
+    assert admit.try_acquire(1)
+    assert "1" in mem.sets["windup:i2v:gate:inflight:cred:cred-a"]
+    assert "1" not in mem.sets.get("windup:i2v:gate:inflight:cred:cred-new", set())
+    assert admit.retry_state(1) == bound_one
+
+
+def test_env_head_insert_keeps_claim_on_same_physical_key(monkeypatch):
+    mem = _MemRedis()
+    monkeypatch.setattr("windup_app.server.mq.i2v_admit.get_redis", lambda: mem)
+    first = snapshot_from_settings(
+        AIProviderSettings(
+            route_primary_name="primary",
+            route_primary_base_url="https://api.qnaigc.com/v1",
+            route_primary_api_key="key-a",
+            route_primary_api_keys="key-b",
+        ),
+        route_group="character_action",
+    )
+    monkeypatch.setattr(
+        "windup_app.server.mq.i2v_admit._action_edges", lambda: first.edges
+    )
+    assert admit.try_acquire(1)
+    cred_a = credential_id("primary", "key-a")
+    assert "1" in mem.sets[f"windup:i2v:gate:inflight:cred:{cred_a}"]
+
+    inserted = snapshot_from_settings(
+        AIProviderSettings(
+            route_primary_name="primary",
+            route_primary_base_url="https://api.qnaigc.com/v1",
+            route_primary_api_key="key-new",
+            route_primary_api_keys="key-a,key-b",
+        ),
+        route_group="character_action",
+    )
+    monkeypatch.setattr(
+        "windup_app.server.mq.i2v_admit._action_edges", lambda: inserted.edges
+    )
+    assert inserted.edges[0].credential_id != cred_a
+    assert admit.try_acquire(1)
+    assert "1" in mem.sets[f"windup:i2v:gate:inflight:cred:{cred_a}"]
+    cred_new = credential_id("primary", "key-new")
+    assert "1" not in mem.sets.get(
+        f"windup:i2v:gate:inflight:cred:{cred_new}", set()
+    )
+
+
+def test_legacy_primary_key_claim_migrates_to_credential(monkeypatch):
+    mem = _patch_redis(monkeypatch, cred_ids=("cred-a",))
+    mem.hashes["windup:i2v:gate:task:9"] = {"route_id": "primary.key0"}
+    mem.sadd("windup:i2v:gate:inflight", "9")
+    mem.sadd("windup:i2v:gate:inflight:primary.key0", "9")
+    assert admit.try_acquire(9)
+    assert "9" in mem.sets["windup:i2v:gate:inflight:cred:cred-a"]
+    assert "9" not in mem.sets.get("windup:i2v:gate:inflight:primary.key0", set())
+
+
 def test_rebuild_restores_job_holders(monkeypatch):
     mem = _patch_redis(monkeypatch)
     mem.hashes["windup:i2v:11"] = {"job_id": "j1"}