feat(failover): 支持 Provider 级别故障转移规则，默认全部转移策略

- 新增 failover_rules 配置：支持 success_failover_patterns（成功响应匹配时转移）
  和 error_stop_patterns（错误响应匹配时终止），支持按 status_code 过滤
- 修改默认转移策略：ErrorClassifier 不再返回 RAISE，所有错误默认继续转移
- TaskService 中客户端错误不再直接抛出，改为 break 继续尝试下一个候选
- 修复 proxy tunnel 连接/断连竞态：引入 per-node 锁和事件时间戳排序
- 优化 ProxyNode 状态判定：OFFLINE 统一由心跳超时判定，兼容多 worker 场景
- has_tunnel 改为纯检查方法，避免在 finally 块中误清理新注册连接
- Redis stream NOGROUP 异常自愈处理
- OAuthAccountDialog 输入框焦点样式补全

frontend/src/api/endpoints/providers.ts

@@ -1,6 +1,7 @@
 import client from '../client'
 import type {
   ClaudeCodeAdvancedConfig,
+  FailoverRulesConfig,
   PoolAdvancedConfig,
   ProviderWithEndpointsSummary,
   ProxyConfig,
@@ -48,6 +49,7 @@ export async function updateProvider(
     is_active: boolean
     claude_code_advanced: ClaudeCodeAdvancedConfig | null
     pool_advanced: PoolAdvancedConfig | null
+    failover_rules: FailoverRulesConfig | null
   }>
 ): Promise<ProviderWithEndpointsSummary> {
   const response = await client.patch(`/api/admin/providers/${providerId}`, data)
@@ -77,6 +79,7 @@ export async function createProvider(
     proxy?: ProxyConfig | null
     claude_code_advanced?: ClaudeCodeAdvancedConfig | null
     pool_advanced?: PoolAdvancedConfig | null
+    failover_rules?: FailoverRulesConfig | null
   }
 ): Promise<{ id: string; name: string; message?: string }> {
   const response = await client.post('/api/admin/providers/', data)

frontend/src/api/endpoints/types/provider.ts

@@ -471,6 +471,17 @@ export interface PoolAdvancedConfig {
   unschedulable_rules?: Array<Record<string, unknown>> | null
 }
 
+export interface FailoverRuleItem {
+  pattern: string
+  description?: string
+  status_codes?: number[]
+}
+
+export interface FailoverRulesConfig {
+  success_failover_patterns: FailoverRuleItem[]
+  error_stop_patterns: FailoverRuleItem[]
+}
+
 export interface ProviderWithEndpointsSummary {
   id: string
   name: string
@@ -506,6 +517,7 @@ export interface ProviderWithEndpointsSummary {
   endpoint_health_details: EndpointHealthDetail[]
   claude_code_advanced?: ClaudeCodeAdvancedConfig | null
   pool_advanced?: PoolAdvancedConfig | null
+  failover_rules?: FailoverRulesConfig | null
   ops_configured: boolean  // 是否配置了扩展操作（余额监控等）
   ops_architecture_id?: string  // 扩展操作使用的架构 ID（如 cubence, anyrouter）
   created_at: string

frontend/src/features/providers/components/FailoverRulesDialog.vue

@@ -0,0 +1,253 @@
+<template>
+  <Dialog
+    :model-value="open"
+    title="故障转移规则"
+    description="配置提供商级别的故障转移规则。默认所有错误都会触发转移，此处可自定义例外。"
+    :icon="GitBranch"
+    size="lg"
+    @update:model-value="handleClose"
+  >
+    <div class="space-y-5 max-h-[60vh] overflow-y-auto px-0.5 py-0.5 -mx-0.5">
+      <!-- 成功转移规则 -->
+      <div class="space-y-3">
+        <div class="flex items-center justify-between">
+          <div>
+            <h3 class="text-sm font-medium">
+              成功转移规则
+            </h3>
+            <p class="text-xs text-muted-foreground mt-0.5">
+              HTTP 200 但响应体匹配正则时，视为失败并触发转移
+            </p>
+          </div>
+          <Button
+            type="button"
+            variant="outline"
+            size="sm"
+            @click="addRule('success')"
+          >
+            <Plus class="w-4 h-4 mr-1" />
+            添加
+          </Button>
+        </div>
+
+        <div
+          v-if="successPatterns.length === 0"
+          class="text-xs text-muted-foreground px-3 py-4 border border-dashed rounded-lg text-center"
+        >
+          暂无规则
+        </div>
+
+        <div
+          v-for="(rule, index) in successPatterns"
+          :key="'s-' + index"
+          class="flex items-center gap-1"
+        >
+          <Input
+            v-model="rule.pattern"
+            placeholder="例如: relay:.*格式错误"
+            class="font-mono text-xs flex-1"
+          />
+          <Button
+            variant="ghost"
+            size="sm"
+            class="shrink-0 h-8 w-8 p-0 text-muted-foreground hover:text-destructive"
+            @click="removeRule('success', index)"
+          >
+            <Trash2 class="w-3.5 h-3.5" />
+          </Button>
+        </div>
+      </div>
+
+      <!-- 错误终止规则 -->
+      <div class="space-y-3">
+        <div class="flex items-center justify-between">
+          <div>
+            <h3 class="text-sm font-medium">
+              错误终止规则
+            </h3>
+            <p class="text-xs text-muted-foreground mt-0.5">
+              HTTP 非 200 且响应体匹配正则时，停止转移并直接返回错误。可选填状态码缩小匹配范围
+            </p>
+          </div>
+          <Button
+            type="button"
+            variant="outline"
+            size="sm"
+            @click="addRule('error')"
+          >
+            <Plus class="w-4 h-4 mr-1" />
+            添加
+          </Button>
+        </div>
+
+        <div
+          v-if="errorPatterns.length === 0"
+          class="text-xs text-muted-foreground px-3 py-4 border border-dashed rounded-lg text-center"
+        >
+          暂无规则
+        </div>
+
+        <div
+          v-for="(rule, index) in errorPatterns"
+          :key="'e-' + index"
+          class="flex items-center gap-1"
+        >
+          <Input
+            :model-value="formatStatusCodes(rule.status_codes)"
+            placeholder="状态码"
+            size="sm"
+            class="font-mono text-xs w-24 shrink-0"
+            @update:model-value="(v: string | number) => rule.status_codes = parseStatusCodes(String(v))"
+          />
+          <Input
+            v-model="rule.pattern"
+            placeholder="例如: content_policy_violation"
+            class="font-mono text-xs flex-1"
+          />
+          <Button
+            variant="ghost"
+            size="sm"
+            class="shrink-0 h-8 w-8 p-0 text-muted-foreground hover:text-destructive"
+            @click="removeRule('error', index)"
+          >
+            <Trash2 class="w-3.5 h-3.5" />
+          </Button>
+        </div>
+      </div>
+    </div>
+
+    <template #footer>
+      <Button
+        variant="outline"
+        :disabled="saving"
+        @click="handleClose"
+      >
+        取消
+      </Button>
+      <Button
+        :disabled="saving"
+        @click="handleSave"
+      >
+        {{ saving ? '保存中...' : '保存' }}
+      </Button>
+    </template>
+  </Dialog>
+</template>
+
+<script setup lang="ts">
+import { ref, watch } from 'vue'
+import {
+  Dialog,
+  Button,
+  Input,
+} from '@/components/ui'
+import { GitBranch, Plus, Trash2 } from 'lucide-vue-next'
+import { useToast } from '@/composables/useToast'
+import { updateProvider, type ProviderWithEndpointsSummary } from '@/api/endpoints'
+import { parseApiError } from '@/utils/errorParser'
+import type { FailoverRuleItem } from '@/api/endpoints/types'
+
+const props = defineProps<{
+  open: boolean
+  provider: ProviderWithEndpointsSummary | null
+}>()
+
+const emit = defineEmits<{
+  'update:open': [value: boolean]
+  'saved': []
+}>()
+
+const { success, error: showError } = useToast()
+const saving = ref(false)
+
+const successPatterns = ref<FailoverRuleItem[]>([])
+const errorPatterns = ref<FailoverRuleItem[]>([])
+
+watch(() => [props.open, props.provider], () => {
+  if (props.open && props.provider) {
+    const rules = props.provider.failover_rules
+    successPatterns.value = (rules?.success_failover_patterns || []).map(r => ({ ...r }))
+    errorPatterns.value = (rules?.error_stop_patterns || []).map(r => ({ ...r }))
+  }
+}, { immediate: true })
+
+function addRule(type: 'success' | 'error') {
+  const rule: FailoverRuleItem = { pattern: '', description: '' }
+  if (type === 'success') {
+    successPatterns.value.push(rule)
+  } else {
+    errorPatterns.value.push(rule)
+  }
+}
+
+function removeRule(type: 'success' | 'error', index: number) {
+  if (type === 'success') {
+    successPatterns.value.splice(index, 1)
+  } else {
+    errorPatterns.value.splice(index, 1)
+  }
+}
+
+function handleClose() {
+  emit('update:open', false)
+}
+
+function formatStatusCodes(codes: number[] | undefined): string {
+  if (!codes || codes.length === 0) return ''
+  return codes.join(',')
+}
+
+function parseStatusCodes(input: string): number[] | undefined {
+  const trimmed = input.trim()
+  if (!trimmed) return undefined
+  const codes = trimmed
+    .split(/[,\s]+/)
+    .map(s => parseInt(s.trim(), 10))
+    .filter(n => !isNaN(n) && n >= 100 && n <= 599)
+  return codes.length > 0 ? codes : undefined
+}
+
+async function handleSave() {
+  if (!props.provider) return
+
+  // Validate patterns
+  const allPatterns = [...successPatterns.value, ...errorPatterns.value]
+  for (const rule of allPatterns) {
+    if (!rule.pattern.trim()) {
+      showError('正则表达式不能为空', '验证失败')
+      return
+    }
+    try {
+      new RegExp(rule.pattern)
+    } catch {
+      showError(`无效的正则表达式: ${rule.pattern}`, '验证失败')
+      return
+    }
+  }
+
+  saving.value = true
+  try {
+    const filteredSuccess = successPatterns.value.filter(r => r.pattern.trim())
+    const filteredError = errorPatterns.value.filter(r => r.pattern.trim())
+
+    const hasRules = filteredSuccess.length > 0 || filteredError.length > 0
+
+    await updateProvider(props.provider.id, {
+      failover_rules: hasRules
+        ? {
+            success_failover_patterns: filteredSuccess,
+            error_stop_patterns: filteredError,
+          }
+        : null,
+    })
+
+    success('故障转移规则已保存')
+    emit('saved')
+    handleClose()
+  } catch (err: unknown) {
+    showError(parseApiError(err, '保存故障转移规则失败'), '保存失败')
+  } finally {
+    saving.value = false
+  }
+}
+</script>

frontend/src/features/providers/components/OAuthAccountDialog.vue

@@ -138,7 +138,7 @@
                       v-model="device.start_url"
                       type="text"
                       placeholder="https://your-org.awsapps.com/start"
-                      class="w-full h-8 px-2 text-xs rounded-md border border-border bg-background font-mono"
+                      class="w-full h-8 px-2 text-xs rounded-md border border-border bg-background font-mono focus:outline-none focus:ring-1 focus:ring-ring focus:relative focus:z-10"
                       spellcheck="false"
                     >
                   </div>
@@ -154,7 +154,7 @@
                         <ComboboxInput
                           :display-value="() => device.region"
                           placeholder="输入或选择 Region"
-                          class="w-full h-8 px-2 pr-7 text-xs rounded-md border border-border bg-background font-mono focus:outline-none focus:ring-1 focus:ring-ring"
+                          class="w-full h-8 px-2 pr-7 text-xs rounded-md border border-border bg-background font-mono focus:outline-none focus:ring-1 focus:ring-ring focus:relative focus:z-10"
                           spellcheck="false"
                           @input="(e: Event) => regionSearch = (e.target as HTMLInputElement).value"
                           @keydown.enter.prevent="onRegionEnter"
@@ -193,7 +193,7 @@
                       v-model="device.totp_secret"
                       type="text"
                       placeholder="Base32 secret, 如 JBSWY3DPEHPK3PXP"
-                      class="w-full h-8 px-2 text-xs rounded-md border border-border bg-background font-mono"
+                      class="w-full h-8 px-2 text-xs rounded-md border border-border bg-background font-mono focus:outline-none focus:ring-1 focus:ring-ring focus:relative focus:z-10"
                       spellcheck="false"
                     >
                   </div>

frontend/src/features/providers/components/ProviderDetailDrawer.vue

@@ -50,6 +50,16 @@
                       <Shuffle class="w-4 h-4" />
                     </Button>
                   </span>
+                  <span :title="hasFailoverRules ? '已配置故障转移规则（点击编辑）' : '配置故障转移规则'">
+                    <Button
+                      variant="ghost"
+                      size="icon"
+                      :class="hasFailoverRules ? 'text-orange-500 dark:text-orange-400' : ''"
+                      @click="failoverRulesDialogOpen = true"
+                    >
+                      <GitBranch class="w-4 h-4" />
+                    </Button>
+                  </span>
                   <Popover
                     :open="providerProxyPopoverOpen"
                     @update:open="handleProviderProxyPopoverToggle"
@@ -1029,6 +1039,14 @@
     :key-id="antigravityQuotaDialogKey.id"
     @update:open="antigravityQuotaDialogOpen = $event"
   />
+
+  <!-- 故障转移规则弹窗 -->
+  <FailoverRulesDialog
+    :open="failoverRulesDialogOpen"
+    :provider="provider ?? null"
+    @update:open="failoverRulesDialogOpen = $event"
+    @saved="loadProvider()"
+  />
 </template>
 
 <script setup lang="ts">
@@ -1050,6 +1068,7 @@ import {
   BarChart3,
   ShieldX,
   Globe,
+  GitBranch,
 } from 'lucide-vue-next'
 import { parseApiError } from '@/utils/errorParser'
 import { useEscapeKey } from '@/composables/useEscapeKey'
@@ -1084,6 +1103,7 @@ import EndpointFormDialog from '@/features/providers/components/EndpointFormDial
 import ProviderModelFormDialog from '@/features/providers/components/ProviderModelFormDialog.vue'
 import AlertDialog from '@/components/common/AlertDialog.vue'
 import AntigravityQuotaDialog from '@/features/providers/components/AntigravityQuotaDialog.vue'
+import FailoverRulesDialog from '@/features/providers/components/FailoverRulesDialog.vue'
 import ProxyNodeSelect from '@/features/providers/components/ProxyNodeSelect.vue'
 import { useProxyNodesStore } from '@/stores/proxy-nodes'
 import {
@@ -1189,6 +1209,15 @@ const refreshingQuota = ref(false)
 const antigravityQuotaDialogOpen = ref(false)
 const antigravityQuotaDialogKey = ref<EndpointAPIKey | null>(null)
 
+// 故障转移规则
+const failoverRulesDialogOpen = ref(false)
+const hasFailoverRules = computed(() => {
+  const rules = provider.value?.failover_rules
+  if (!rules) return false
+  return (rules.success_failover_patterns?.length || 0) > 0
+    || (rules.error_stop_patterns?.length || 0) > 0
+})
+
 // Provider 级别代理配置状态
 const proxyNodesStore = useProxyNodesStore()
 const providerProxyPopoverOpen = ref(false)

src/api/admin/providers/routes.py

@@ -82,6 +82,32 @@ def _merge_pool_advanced_config(
     return merged_config or None, config_changed
 
 
+def _merge_failover_rules_config(
+    *,
+    provider_config: dict[str, Any] | None,
+    failover_rules: dict[str, Any] | None,
+    failover_rules_in_payload: bool,
+) -> tuple[dict[str, Any] | None, bool]:
+    """合并 failover_rules 到 provider.config。"""
+    merged_config = dict(provider_config or {})
+    config_changed = False
+
+    if not failover_rules_in_payload:
+        return merged_config or None, config_changed
+
+    if failover_rules is None:
+        if "failover_rules" in merged_config:
+            merged_config.pop("failover_rules", None)
+            config_changed = True
+    else:
+        next_value = dict(failover_rules)
+        if merged_config.get("failover_rules") != next_value:
+            merged_config["failover_rules"] = next_value
+            config_changed = True
+
+    return merged_config or None, config_changed
+
+
 def _merge_claude_code_advanced_config(
     *,
     provider_type: str | None,
@@ -394,6 +420,15 @@ class AdminCreateProviderAdapter(AdminApiAdapter):
                 ),
                 pool_advanced_in_payload=validated_data.pool_advanced is not None,
             )
+            provider_config, _ = _merge_failover_rules_config(
+                provider_config=provider_config,
+                failover_rules=(
+                    validated_data.failover_rules.model_dump()
+                    if validated_data.failover_rules is not None
+                    else None
+                ),
+                failover_rules_in_payload=validated_data.failover_rules is not None,
+            )
 
             # 创建 Provider 对象
             provider = Provider(
@@ -509,6 +544,7 @@ class AdminUpdateProviderAdapter(AdminApiAdapter):
             config_in_payload = "config" in update_data
             claude_advanced_in_payload = "claude_code_advanced" in update_data
             pool_advanced_in_payload = "pool_advanced" in update_data
+            failover_rules_in_payload = "failover_rules" in update_data
             provider_config = (
                 dict(update_data.pop("config") or {})
                 if config_in_payload
@@ -518,6 +554,9 @@ class AdminUpdateProviderAdapter(AdminApiAdapter):
                 update_data.pop("claude_code_advanced") if claude_advanced_in_payload else None
             )
             pool_advanced = update_data.pop("pool_advanced") if pool_advanced_in_payload else None
+            failover_rules = (
+                update_data.pop("failover_rules") if failover_rules_in_payload else None
+            )
             target_provider_type = (
                 update_data.get("provider_type")
                 or getattr(provider, "provider_type", None)
@@ -535,6 +574,11 @@ class AdminUpdateProviderAdapter(AdminApiAdapter):
                 pool_advanced=pool_advanced,
                 pool_advanced_in_payload=pool_advanced_in_payload,
             )
+            provider_config, config_changed_by_failover = _merge_failover_rules_config(
+                provider_config=provider_config,
+                failover_rules=failover_rules,
+                failover_rules_in_payload=failover_rules_in_payload,
+            )
 
             config_touched = (
                 config_in_payload
@@ -542,6 +586,8 @@ class AdminUpdateProviderAdapter(AdminApiAdapter):
                 or config_changed_by_claude
                 or pool_advanced_in_payload
                 or config_changed_by_pool
+                or failover_rules_in_payload
+                or config_changed_by_failover
             )
             if config_touched:
                 update_data["config"] = provider_config

src/api/admin/providers/summary.py

@@ -18,7 +18,11 @@ from src.core.enums import ProviderBillingType
 from src.core.exceptions import InvalidRequestException, NotFoundException
 from src.core.logger import logger
 from src.database import get_db
-from src.models.admin_requests import ClaudeCodeAdvancedConfig, PoolAdvancedConfig
+from src.models.admin_requests import (
+    ClaudeCodeAdvancedConfig,
+    FailoverRulesConfig,
+    PoolAdvancedConfig,
+)
 from src.models.database import (
     Model,
     Provider,
@@ -282,6 +286,38 @@ def _extract_claude_code_advanced_from_config(
         return None
 
 
+def _extract_failover_rules_from_config(
+    provider_config: dict[str, Any] | None,
+    *,
+    provider_id: str,
+) -> FailoverRulesConfig | None:
+    """从 Provider.config 中安全提取故障转移规则配置。"""
+    raw = (provider_config or {}).get("failover_rules")
+    if raw is None:
+        return None
+
+    if isinstance(raw, FailoverRulesConfig):
+        return raw
+
+    if not isinstance(raw, dict):
+        logger.warning(
+            "Provider {} 的 failover_rules 类型无效: {}，已忽略",
+            provider_id,
+            type(raw).__name__,
+        )
+        return None
+
+    try:
+        return FailoverRulesConfig.model_validate(raw)
+    except Exception as exc:
+        logger.warning(
+            "Provider {} 的 failover_rules 配置无效，已忽略: {}",
+            provider_id,
+            str(exc),
+        )
+        return None
+
+
 def _build_provider_summary(db: Session, provider: Provider) -> ProviderWithEndpointsSummary:
     endpoints = db.query(ProviderEndpoint).filter(ProviderEndpoint.provider_id == provider.id).all()
 
@@ -402,6 +438,10 @@ def _build_provider_summary(db: Session, provider: Provider) -> ProviderWithEndp
         provider_config,
         provider_id=str(provider.id),
     )
+    failover_rules = _extract_failover_rules_from_config(
+        provider_config,
+        provider_id=str(provider.id),
+    )
 
     return ProviderWithEndpointsSummary(
         id=provider.id,
@@ -425,6 +465,7 @@ def _build_provider_summary(db: Session, provider: Provider) -> ProviderWithEndp
         request_timeout=provider.request_timeout,
         claude_code_advanced=claude_code_advanced,
         pool_advanced=pool_advanced,
+        failover_rules=failover_rules,
         total_endpoints=total_endpoints,
         active_endpoints=active_endpoints,
         total_keys=total_keys,

src/api/admin/proxy_tunnel.py

@@ -8,10 +8,12 @@ aether-proxy 通过此端点建立 tunnel 连接。
 from __future__ import annotations
 
 import asyncio
+from datetime import datetime, timezone
 
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect
 
 from src.core.logger import logger
+from src.services.proxy_node.health_scheduler import heartbeat_is_stale
 from src.services.proxy_node.tunnel_manager import (
     TunnelConnection,
     get_tunnel_manager,
@@ -20,6 +22,18 @@ from src.services.proxy_node.tunnel_protocol import Frame, MsgType
 
 router = APIRouter()
 
+# Per-node 锁: 防止并发的 connect/disconnect 写入 DB 时出现竞态（后断连覆盖先连接）
+_node_status_locks: dict[str, asyncio.Lock] = {}
+
+
+def _get_node_lock(node_id: str) -> asyncio.Lock:
+    lock = _node_status_locks.get(node_id)
+    if lock is None:
+        lock = asyncio.Lock()
+        _node_status_locks[node_id] = lock
+    return lock
+
+
 # 单帧最大 64 MB -- AI API 请求体可能包含多张 base64 图片，需要足够余量
 _MAX_FRAME_SIZE = 64 * 1024 * 1024
 
@@ -111,10 +125,17 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
 
     manager = get_tunnel_manager()
     conn = TunnelConnection(node_id, node_name, ws, max_streams=max_streams)
+    node_lock = _get_node_lock(node_id)
+
     manager.register(conn)
 
-    # 更新 DB: tunnel_connected = True
+    # 在 per-node 锁保护下更新 DB，防止并发的 connect/disconnect 写入竞态
-    await _update_tunnel_status(node_id, connected=True)
+    async with node_lock:
+        await _update_tunnel_status(
+            node_id,
+            connected=True,
+            observed_at=datetime.now(timezone.utc),
+        )
 
     # 启动服务端 ping 任务，防止中间代理因空闲超时关闭连接
     ping_task = asyncio.create_task(_ping_loop(conn))
@@ -156,11 +177,20 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
         logger.error("tunnel WebSocket error for node_id={}: {}", node_id, e)
     finally:
         ping_task.cancel()
-        manager.unregister(conn)
+        # 在 per-node 锁保护下执行 unregister + 连接池计数检查 + DB 更新，
-        if not manager.has_tunnel(node_id):
+        # 确保整个序列是原子的，避免"断连写 OFFLINE 覆盖新连接写 ONLINE"的竞态
-            await _update_tunnel_status(node_id, connected=False, detail=disconnect_reason)
+        async with node_lock:
-        else:
+            manager.unregister(conn)
-            logger.info("tunnel connection closed but pool still active: node_id={}", node_id)
+            if manager.connection_count(node_id) == 0:
+                await _update_tunnel_status(
+                    node_id,
+                    connected=False,
+                    detail=disconnect_reason,
+                    observed_at=datetime.now(timezone.utc),
+                )
+                # 不清理锁: asyncio.Lock 极轻量，清理可能导致并发新连接拿到不同锁实例
+            else:
+                logger.info("tunnel connection closed but pool still active: node_id={}", node_id)
 
 
 async def _ping_loop(conn: TunnelConnection) -> None:
@@ -180,13 +210,15 @@ async def _ping_loop(conn: TunnelConnection) -> None:
 
 
 async def _update_tunnel_status(
-    node_id: str, *, connected: bool, detail: str | None = None
+    node_id: str,
+    *,
+    connected: bool,
+    detail: str | None = None,
+    observed_at: datetime | None = None,
 ) -> None:
     """更新 ProxyNode 的 tunnel 连接状态并记录事件（在线程池中执行）"""
 
     def _sync_update() -> None:
-        from datetime import datetime, timezone
-
         from src.database import create_session
         from src.models.database import ProxyNode, ProxyNodeEvent, ProxyNodeStatus
 
@@ -194,20 +226,47 @@ async def _update_tunnel_status(
         try:
             node = db.query(ProxyNode).filter(ProxyNode.id == node_id).first()
             if node:
-                node.tunnel_connected = connected
+                event_time = observed_at or datetime.now(timezone.utc)
-                now = datetime.now(timezone.utc)
+                last_transition = node.tunnel_connected_at
+                if last_transition and last_transition.tzinfo is None:
+                    last_transition = last_transition.replace(tzinfo=timezone.utc)
+
+                # 忽略乱序的旧事件，避免快速重连时旧状态覆盖新状态
+                stale_event = bool(last_transition and event_time < last_transition)
+                if stale_event:
+                    detail_text = f"[stale_ignored] {detail}" if detail else "[stale_ignored]"
+                    db.add(
+                        ProxyNodeEvent(
+                            node_id=node_id,
+                            event_type="connected" if connected else "disconnected",
+                            detail=detail_text,
+                        )
+                    )
+                    db.commit()
+                    return
+
+                event_detail = detail
                 if connected:
-                    node.tunnel_connected_at = now
+                    node.tunnel_connected = True
+                    node.tunnel_connected_at = event_time
                     node.status = ProxyNodeStatus.ONLINE
                 else:
-                    node.tunnel_connected_at = now
+                    # 断连不立即强制 OFFLINE。若心跳仍新鲜，可能仍有其他连接存活
-                    node.status = ProxyNodeStatus.OFFLINE
+                    # （连接池或跨 worker），避免误判写回 OFFLINE。
+                    if heartbeat_is_stale(node, event_time):
+                        node.tunnel_connected = False
+                        node.tunnel_connected_at = event_time
+                        node.status = ProxyNodeStatus.OFFLINE
+                    else:
+                        event_detail = (
+                            f"[heartbeat_fresh] {detail}" if detail else "[heartbeat_fresh]"
+                        )
 
                 # 记录连接事件
                 event = ProxyNodeEvent(
                     node_id=node_id,
                     event_type="connected" if connected else "disconnected",
-                    detail=detail,
+                    detail=event_detail,
                 )
                 db.add(event)
                 db.commit()

src/models/admin_requests.py

@@ -80,6 +80,53 @@ class ProxyConfig(BaseModel):
         return self
 
 
+class FailoverRuleItem(BaseModel):
+    """故障转移规则条目"""
+
+    pattern: str = Field(..., min_length=1, max_length=500, description="正则表达式")
+    description: str = Field("", max_length=200, description="规则描述")
+    status_codes: list[int] | None = Field(
+        default=None,
+        description="HTTP 状态码列表（可选，为空时匹配所有状态码）",
+    )
+
+    @field_validator("pattern")
+    @classmethod
+    def validate_pattern(cls, v: str) -> str:
+        """验证正则表达式语法"""
+        import re as _re
+
+        try:
+            _re.compile(v)
+        except _re.error as e:
+            raise ValueError(f"无效的正则表达式: {e}")
+        return v
+
+    @field_validator("status_codes")
+    @classmethod
+    def validate_status_codes(cls, v: list[int] | None) -> list[int] | None:
+        """验证 HTTP 状态码"""
+        if v is None:
+            return v
+        for code in v:
+            if not (100 <= code <= 599):
+                raise ValueError(f"无效的 HTTP 状态码: {code}")
+        return v
+
+
+class FailoverRulesConfig(BaseModel):
+    """故障转移规则配置"""
+
+    success_failover_patterns: list[FailoverRuleItem] = Field(
+        default_factory=list,
+        description="成功响应转移规则: HTTP 200 但响应体匹配正则时触发转移",
+    )
+    error_stop_patterns: list[FailoverRuleItem] = Field(
+        default_factory=list,
+        description="错误终止规则: HTTP 非 200 且响应体匹配正则时停止转移",
+    )
+
+
 class PoolAdvancedConfig(BaseModel):
     """通用号池配置（适用于所有 Provider 类型）。"""
 
@@ -247,6 +294,7 @@ class CreateProviderRequest(BaseModel):
     claude_code_advanced: ClaudeCodeAdvancedConfig | None = Field(
         None, description="Claude Code 特有配置"
     )
+    failover_rules: FailoverRulesConfig | None = Field(None, description="故障转移规则配置")
     config: dict[str, Any] | None = Field(None, description="其他配置")
 
     @field_validator("provider_type")
@@ -356,6 +404,7 @@ class UpdateProviderRequest(BaseModel):
     claude_code_advanced: ClaudeCodeAdvancedConfig | None = Field(
         None, description="Claude Code 特有配置"
     )
+    failover_rules: FailoverRulesConfig | None = Field(None, description="故障转移规则配置")
     config: dict[str, Any] | None = None
 
     # 复用相同的验证器

src/models/endpoint_models.py

@@ -10,7 +10,12 @@ from typing import Any, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
-from src.models.admin_requests import ClaudeCodeAdvancedConfig, PoolAdvancedConfig, ProxyConfig
+from src.models.admin_requests import (
+    ClaudeCodeAdvancedConfig,
+    FailoverRulesConfig,
+    PoolAdvancedConfig,
+    ProxyConfig,
+)
 
 # ========== Header Rule 类型定义 ==========
 # 请求头规则支持三种操作：
@@ -938,6 +943,7 @@ class ProviderUpdateRequest(BaseModel):
         None, description="Claude Code 高级配置"
     )
     pool_advanced: PoolAdvancedConfig | None = Field(None, description="通用号池配置")
+    failover_rules: FailoverRulesConfig | None = Field(None, description="故障转移规则配置")
 
 
 class ProviderWithEndpointsSummary(BaseModel):
@@ -982,6 +988,7 @@ class ProviderWithEndpointsSummary(BaseModel):
         default=None, description="Claude Code 高级配置"
     )
     pool_advanced: PoolAdvancedConfig | None = Field(default=None, description="通用号池配置")
+    failover_rules: FailoverRulesConfig | None = Field(default=None, description="故障转移规则配置")
 
     # Endpoint 统计
     total_endpoints: int = Field(default=0, description="总 Endpoint 数量")

src/services/candidate/failover.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import re
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Any, AsyncIterator
 
-import httpx
 from sqlalchemy import update
 from sqlalchemy.orm import Session
 
@@ -195,8 +195,33 @@ class FailoverEngine:
                         attempt_result = await self._probe_stream_first_chunk(
                             attempt_result=attempt_result,
                             record_id=record_id,
+                            candidate=candidate,
                         )
 
+                    # Sync: check success_failover_patterns on response body
+                    if attempt_result.kind == AttemptKind.SYNC_RESPONSE:
+                        body = getattr(attempt_result, "response_body", None)
+                        if body:
+                            if isinstance(body, bytes):
+                                body_text = body.decode("utf-8", errors="replace")
+                            elif isinstance(body, (dict, list)):
+                                body_text = json.dumps(body, ensure_ascii=False)
+                            else:
+                                body_text = str(body)
+                            rule_action = self._check_provider_failover_rules(
+                                candidate, is_success=True, response_text=body_text
+                            )
+                            if rule_action == FailoverAction.CONTINUE:
+                                self._record_attempt_failure(
+                                    record_id,
+                                    Exception("success_failover_pattern matched"),
+                                    200,
+                                )
+                                raise StreamProbeError(
+                                    "Success failover pattern matched",
+                                    http_status=200,
+                                )
+
                     self._record_attempt_success(record_id, attempt_result)
 
                     # PRE_EXPAND: mark unused slots after request ends (success)
@@ -598,14 +623,6 @@ class FailoverEngine:
             value = int(retry_policy.max_retries or 1)
         return max(1, value)
 
-    def _should_stop_on_http_error(self, *, status_code: int, error_text: str) -> bool:
-        # follow CandidateService rules
-        if status_code in (401, 403, 429):
-            return False
-        if 400 <= status_code < 500:
-            return self._error_classifier.is_client_error(error_text)
-        return False
-
     async def _handle_error(
         self,
         error: Exception,
@@ -613,29 +630,139 @@ class FailoverEngine:
         candidate: ProviderCandidate,
         has_retry_left: bool,
     ) -> FailoverAction:
-        # Special: HTTP client errors should stop failover.
+        # 检查提供商级别的错误终止规则
-        if isinstance(error, httpx.HTTPStatusError):
+        error_text = self._extract_error_text(error)
-            status_code = int(getattr(error.response, "status_code", 0) or 0)
+        status_code = int(getattr(error, "status_code", 0) or 0) or int(
-            try:
+            getattr(error, "http_status", 0) or 0
-                error_text = error.response.text or ""
+        )
-            except Exception:
+        # ExecutionError wrapping: check cause for status_code
-                error_text = ""
+        if not status_code:
-            if self._should_stop_on_http_error(status_code=status_code, error_text=error_text):
+            cause = getattr(error, "cause", None)
-                return FailoverAction.STOP
+            if cause is not None:
+                status_code = int(getattr(cause, "status_code", 0) or 0) or int(
+                    getattr(cause, "http_status", 0) or 0
+                )
+        if error_text:
+            rule_action = self._check_provider_failover_rules(
+                candidate,
+                is_success=False,
+                response_text=error_text,
+                status_code=status_code or None,
+            )
+            if rule_action is not None:
+                return rule_action
 
-        # Default: reuse legacy ErrorClassifier decision and map to FailoverAction.
+        # 默认全部转移: ErrorClassifier 结果统一映射为 CONTINUE/RETRY，不再 STOP
         action = self._error_classifier.classify(error, has_retry_left=has_retry_left)
-        if action == ErrorAction.RAISE:
+        if action == ErrorAction.CONTINUE:
-            return FailoverAction.STOP
+            return FailoverAction.RETRY
-        if action == ErrorAction.BREAK:
+        return FailoverAction.CONTINUE
-            return FailoverAction.CONTINUE
+
-        return FailoverAction.RETRY
+    def _check_provider_failover_rules(
+        self,
+        candidate: ProviderCandidate,
+        *,
+        is_success: bool,
+        response_text: str,
+        status_code: int | None = None,
+    ) -> FailoverAction | None:
+        """检查提供商级别的故障转移规则。返回 None 表示无规则命中，使用默认行为。"""
+        config = getattr(candidate.provider, "config", None) or {}
+        rules = config.get("failover_rules")
+        if not rules or not isinstance(rules, dict):
+            return None
+
+        compiled = self._get_compiled_patterns(rules)
+
+        if is_success:
+            for regex, rule in compiled.get("success", []):
+                if regex.search(response_text):
+                    logger.info(
+                        "[FailoverEngine] 成功转移规则命中: pattern={}, provider={}",
+                        rule.get("pattern", ""),
+                        candidate.provider.name,
+                    )
+                    return FailoverAction.CONTINUE
+        else:
+            for regex, rule in compiled.get("error", []):
+                # 检查状态码过滤
+                rule_status_codes = rule.get("status_codes")
+                if rule_status_codes and status_code not in rule_status_codes:
+                    continue
+                if regex.search(response_text):
+                    logger.info(
+                        "[FailoverEngine] 错误终止规则命中: pattern={}, status_code={}, provider={}",
+                        rule.get("pattern", ""),
+                        status_code,
+                        candidate.provider.name,
+                    )
+                    return FailoverAction.STOP
+
+        return None
+
+    @staticmethod
+    def _get_compiled_patterns(
+        rules: dict[str, Any],
+    ) -> dict[str, list[tuple[re.Pattern[str], dict[str, Any]]]]:
+        """编译 failover_rules 中的正则模式。
+
+        编译结果缓存在 rules dict 的 _compiled 键上，避免每次请求都重复编译。
+        """
+        cached = rules.get("_compiled")
+        if cached is not None:
+            return cached
+
+        result: dict[str, list[tuple[re.Pattern[str], dict[str, Any]]]] = {
+            "success": [],
+            "error": [],
+        }
+        for rule in rules.get("success_failover_patterns", []):
+            pattern = rule.get("pattern", "")
+            if pattern:
+                try:
+                    result["success"].append((re.compile(pattern), rule))
+                except re.error:
+                    pass
+        for rule in rules.get("error_stop_patterns", []):
+            pattern = rule.get("pattern", "")
+            if pattern:
+                try:
+                    result["error"].append((re.compile(pattern), rule))
+                except re.error:
+                    pass
+        rules["_compiled"] = result
+        return result
+
+    @staticmethod
+    def _extract_error_text(error: Exception) -> str:
+        """从异常中提取错误响应文本。"""
+        # ExecutionError wrapping
+        cause = getattr(error, "cause", None)
+        if cause is not None:
+            error = cause
+
+        # httpx.HTTPStatusError
+        response = getattr(error, "response", None)
+        if response is not None:
+            try:
+                return response.text or ""
+            except Exception:
+                pass
+
+        # upstream_response / upstream_error attribute
+        for attr in ("upstream_response", "upstream_error", "error_message"):
+            val = getattr(error, attr, None)
+            if val:
+                return str(val)
+
+        return str(error)
 
     async def _probe_stream_first_chunk(
         self,
         *,
         attempt_result: AttemptResult,
         record_id: str | None,
+        candidate: ProviderCandidate | None = None,
     ) -> AttemptResult:
         """
         Probe first chunk for a streaming response.
@@ -672,6 +799,22 @@ class FailoverEngine:
                 original_exception=exc,
             ) from exc
 
+        # Check success_failover_patterns on first chunk
+        if candidate is not None and first_chunk:
+            chunk_text = (
+                first_chunk.decode("utf-8", errors="replace")
+                if isinstance(first_chunk, bytes)
+                else str(first_chunk)
+            )
+            rule_action = self._check_provider_failover_rules(
+                candidate, is_success=True, response_text=chunk_text
+            )
+            if rule_action == FailoverAction.CONTINUE:
+                raise StreamProbeError(
+                    "Success failover pattern matched in first chunk",
+                    http_status=attempt_result.http_status,
+                )
+
         wrapped = self._wrap_stream_with_finalizer(
             first_chunk=first_chunk,
             original_iterator=original_iterator,

src/services/orchestration/error_classifier.py

@@ -384,6 +384,8 @@ class ErrorClassifier:
         """
         分类错误，返回处理动作
 
+        默认全部转移策略: 不再返回 RAISE，所有错误都允许故障转移
+
         Args:
             error: 异常对象
             has_retry_left: 当前候选是否还有重试次数
@@ -404,11 +406,8 @@ class ErrorClassifier:
         if isinstance(error, self.RETRIABLE_ERRORS):
             return ErrorAction.CONTINUE if has_retry_left else ErrorAction.BREAK
 
-        if isinstance(error, self.NON_RETRIABLE_ERRORS):
+        # 所有其他错误: 不再 RAISE，改为 BREAK（跳到下一个候选继续转移）
-            return ErrorAction.RAISE
+        return ErrorAction.BREAK
-
-        # 未知错误，直接抛出
-        return ErrorAction.RAISE
 
     async def handle_rate_limit(
         self,

src/services/proxy_node/health_scheduler.py

@@ -1,10 +1,9 @@
 """
 ProxyNode 心跳检测调度器
 
-定期检查 proxy_nodes 的 tunnel 连接状态，更新节点状态：
+定期检查 proxy_nodes 的连接健康状态，更新节点状态：
-- tunnel 实际连接中  -> ONLINE
+- 本地 TunnelManager 观测到连接 -> ONLINE（自愈）
-- tunnel 未连接      -> OFFLINE
+- 心跳超时（跨 worker 共享信号） -> OFFLINE
-以 TunnelManager 内存中的实际连接状态为准。
 """
 
 from __future__ import annotations
@@ -21,6 +20,28 @@ from src.services.system.scheduler import get_scheduler
 _EVENT_RETENTION_DAYS = 30
 # 每隔多少次心跳检测执行一次事件清理（15s * 240 = 1h）
 _EVENT_CLEANUP_INTERVAL = 240
+# 心跳超时判定：max(90s, heartbeat_interval * 3)
+HEARTBEAT_STALE_MIN_SECONDS = 90
+HEARTBEAT_STALE_MULTIPLIER = 3
+
+
+def heartbeat_is_stale(node: object, now: datetime) -> bool:
+    """根据 last_heartbeat_at 判定节点心跳是否超时。
+
+    接受任意具有 last_heartbeat_at / heartbeat_interval 属性的对象，
+    兼容 ProxyNode ORM 实例和在 asyncio.to_thread 中使用的场景。
+    """
+    last_heartbeat = getattr(node, "last_heartbeat_at", None)
+    if not last_heartbeat:
+        return True
+
+    # 兼容 DB 中可能出现的 naive datetime
+    if last_heartbeat.tzinfo is None:
+        last_heartbeat = last_heartbeat.replace(tzinfo=timezone.utc)
+
+    interval = max(int(getattr(node, "heartbeat_interval", None) or 30), 5)
+    stale_seconds = max(HEARTBEAT_STALE_MIN_SECONDS, interval * HEARTBEAT_STALE_MULTIPLIER)
+    return (now - last_heartbeat).total_seconds() > stale_seconds
 
 
 class ProxyNodeHealthScheduler:
@@ -83,27 +104,32 @@ class ProxyNodeHealthScheduler:
 
             changed = 0
             for node in nodes:
-                # 以 TunnelManager 内存中的实际连接状态为准，
+                # 注意：TunnelManager 仅是当前 worker 的进程内状态，跨 worker 不共享。
-                # 而非仅依赖 DB 的 tunnel_connected 字段。
+                # 因此“本地无 tunnel”不能直接判定 OFFLINE（可能连接在其他 worker）。
-                # 服务端重启后 DB 可能残留 tunnel_connected=True，
+                # OFFLINE 统一由心跳超时判定，避免多进程误判。
-                # 但 TunnelManager 内存中已无连接。
+                actually_connected_local = manager.has_tunnel(node.id)
-                actually_connected = manager.has_tunnel(node.id)
 
-                # 同步修正 DB 中不一致的 tunnel_connected 字段
+                if actually_connected_local:
-                if node.tunnel_connected != actually_connected:
+                    if not node.tunnel_connected:
-                    node.tunnel_connected = actually_connected
+                        node.tunnel_connected = True
-                    if not actually_connected:
                         node.tunnel_connected_at = now
-                    changed += 1
+                        changed += 1
+                    if node.status != ProxyNodeStatus.ONLINE:
+                        node.status = ProxyNodeStatus.ONLINE
+                        node.updated_at = now
+                        changed += 1
+                    continue
 
-                new_status = (
+                # 本地无 tunnel：仅在心跳超时时标记 OFFLINE
-                    ProxyNodeStatus.ONLINE if actually_connected else ProxyNodeStatus.OFFLINE
+                if heartbeat_is_stale(node, now):
-                )
+                    if node.tunnel_connected:
-
+                        node.tunnel_connected = False
-                if node.status != new_status:
+                        node.tunnel_connected_at = now
-                    node.status = new_status
+                        changed += 1
-                    node.updated_at = now
+                    if node.status != ProxyNodeStatus.OFFLINE:
-                    changed += 1
+                        node.status = ProxyNodeStatus.OFFLINE
+                        node.updated_at = now
+                        changed += 1
 
             if changed:
                 db.commit()

src/services/proxy_node/service.py

@@ -339,10 +339,11 @@ class ProxyNodeService:
 
         now = datetime.now(timezone.utc)
         # 心跳通过 tunnel 连接传输，能收到心跳说明 tunnel 一定连通。
-        # 如果状态不是 ONLINE（例如 _update_tunnel_status 执行失败），修正状态。
+        # 如果状态不是 ONLINE 或 tunnel_connected 不一致（例如并发写入覆盖），修正状态。
-        if node.status != ProxyNodeStatus.ONLINE:
+        if node.status != ProxyNodeStatus.ONLINE or not node.tunnel_connected:
             node.status = ProxyNodeStatus.ONLINE
             node.tunnel_connected = True
+            node.tunnel_connected_at = now
             node.updated_at = now
         node.last_heartbeat_at = now
         if heartbeat_interval is not None:
@@ -567,7 +568,23 @@ class ProxyNodeService:
                     "exit_ip": None,
                     "error": "tunnel 未连接",
                 }
-            return await _test_tunnel_connectivity(node.id)
+            result = await _test_tunnel_connectivity(node.id)
+
+            # 连通性测试成功但 DB 状态不一致时，修正为 ONLINE
+            if result.get("success") and (
+                node.status != ProxyNodeStatus.ONLINE or not node.tunnel_connected
+            ):
+                node.status = ProxyNodeStatus.ONLINE
+                node.tunnel_connected = True
+                node.tunnel_connected_at = datetime.now(timezone.utc)
+                node.updated_at = node.tunnel_connected_at
+                db.commit()
+
+                from .resolver import invalidate_proxy_node_cache
+
+                invalidate_proxy_node_cache(node.id)
+
+            return result
 
         # 手动节点：通过代理 URL 测试
         try:

src/services/proxy_node/tunnel_manager.py

@@ -278,8 +278,15 @@ class TunnelManager:
         return True
 
     def has_tunnel(self, node_id: str) -> bool:
-        conn = self.get_connection(node_id)
+        """检查指定 node 是否有存活的 tunnel 连接（纯检查，无副作用）
-        return conn is not None
+
+        与 get_connection 不同，此方法不会清理 dead 连接，
+        避免在 finally 块或 health_scheduler 中误清理刚注册的连接。
+        """
+        conns = self._connections.get(node_id)
+        if not conns:
+            return False
+        return any(c.is_alive for c in conns)
 
     def connection_count(self, node_id: str) -> int:
         """返回指定 node 当前存活的连接数"""

src/services/task/service.py

@@ -898,7 +898,7 @@ class TaskService:
                 error_message=str(exec_err),
                 extra_data=_proxy_extra,
             )
-            return "raise"
+            return "break"
 
         provider = candidate.provider
         endpoint = candidate.endpoint
@@ -1034,16 +1034,10 @@ class TaskService:
             embedded_status = cause.error_code or 200
             if error_classifier.is_client_error(error_message):
                 logger.warning(
-                    "  [{}] 嵌入式客户端错误，停止重试: {}",
+                    "  [{}] 嵌入式客户端错误，继续转移: {}",
                     request_id,
                     error_message[:200],
                 )
-                client_error = UpstreamClientException(
-                    message=error_message or "请求无效",
-                    provider_name=str(provider.name),
-                    status_code=embedded_status,
-                    upstream_error=error_message,
-                )
                 RequestCandidateService.mark_candidate_failed(
                     db=self.db,
                     candidate_id=candidate_record_id,
@@ -1054,14 +1048,7 @@ class TaskService:
                     concurrent_requests=captured_key_concurrent,
                     extra_data=_proxy_extra,
                 )
-                client_error.request_metadata = {
+                return "break"
-                    "provider": provider.name,
-                    "provider_id": str(provider.id),
-                    "provider_endpoint_id": str(endpoint.id),
-                    "provider_api_key_id": str(key.id),
-                    "api_format": str(api_format),
-                }
-                raise client_error
 
             logger.warning(
                 "  [{}] 嵌入式服务端错误，尝试重试: {}",
@@ -1124,7 +1111,7 @@ class TaskService:
 
             if isinstance(converted_error, UpstreamClientException):
                 logger.warning(
-                    "  [{}] 客户端请求错误，停止重试: {}",
+                    "  [{}] 客户端请求错误，继续转移: {}",
                     request_id,
                     str(converted_error.message),
                 )
@@ -1138,14 +1125,7 @@ class TaskService:
                     concurrent_requests=captured_key_concurrent,
                     extra_data=serializable_extra_data,
                 )
-                converted_error.request_metadata = {
+                return "break"
-                    "provider": provider.name,
-                    "provider_id": str(provider.id),
-                    "provider_endpoint_id": str(endpoint.id),
-                    "provider_api_key_id": str(key.id),
-                    "api_format": str(api_format),
-                }
-                raise converted_error
 
             RequestCandidateService.mark_candidate_failed(
                 db=self.db,
@@ -1207,7 +1187,7 @@ class TaskService:
             concurrent_requests=captured_key_concurrent,
             extra_data=_proxy_extra,
         )
-        return "raise"
+        return "break"
 
     async def submit_with_failover(
         self,

src/services/usage/consumer_streams.py

@@ -211,13 +211,19 @@ class UsageQueueConsumer:
         await self._process_messages(redis_client, messages)
 
     async def _read_new(self, redis_client: Any) -> None:
-        result = await redis_client.xreadgroup(
+        try:
-            groupname=self._stream_group,
+            result = await redis_client.xreadgroup(
-            consumername=self._consumer,
+                groupname=self._stream_group,
-            streams={self._stream_key: ">"},
+                consumername=self._consumer,
-            count=self._batch_size,
+                streams={self._stream_key: ">"},
-            block=self._block_ms,
+                count=self._batch_size,
-        )
+                block=self._block_ms,
+            )
+        except ResponseError as exc:
+            if "NOGROUP" in str(exc):
+                await ensure_usage_stream_group()
+                return
+            raise
         if not result:
             return
         for _stream, messages in result:

tests/services/test_failover_engine.py

@@ -26,8 +26,14 @@ def _make_candidate(
     skip_reason: str | None = None,
     needs_conversion: bool = False,
     provider_max_retries: int | None = None,
+    provider_config: dict[str, Any] | None = None,
 ) -> SimpleNamespace:
-    provider = SimpleNamespace(id=provider_id, name=provider_name, max_retries=provider_max_retries)
+    provider = SimpleNamespace(
+        id=provider_id,
+        name=provider_name,
+        max_retries=provider_max_retries,
+        config=provider_config,
+    )
     endpoint = SimpleNamespace(id=endpoint_id)
     key = SimpleNamespace(id=key_id, name=key_name, auth_type=auth_type, priority=priority)
     return SimpleNamespace(
@@ -156,7 +162,9 @@ async def test_failover_engine_retry_same_candidate_when_classifier_says_continu
 
 
 @pytest.mark.asyncio
-async def test_failover_engine_stop_when_classifier_raises() -> None:
+async def test_failover_engine_continues_when_classifier_raises() -> None:
+    """After the 'default failover' change, RAISE no longer stops failover.
+    All candidates should be attempted."""
     db = MagicMock()
     engine = FailoverEngine(db, error_classifier=_StubErrorClassifier(action=ErrorAction.RAISE))
 
@@ -172,10 +180,9 @@ async def test_failover_engine_stop_when_classifier_raises() -> None:
     )
 
     assert result.success is False
-    assert result.error_type == "RuntimeError"
+    assert result.error_type == "AllCandidatesFailed"
-    assert result.attempt_count == 1
+    # both candidates should be tried
-    # should not try candidate 2
+    assert attempt.await_count == 2
-    assert attempt.await_count == 1
 
 
 async def _stream_two_chunks() -> AsyncIterator[bytes]:
@@ -308,3 +315,144 @@ async def test_failover_engine_pre_expand_marks_unused_slots_on_success(
         if call.kwargs.get("status") == "unused"
     }
     assert unused_record_ids == {"r01", "r10"}
+
+
+# ========== error_stop_patterns with status_codes ==========
+
+
+class _HttpError(Exception):
+    """Stub exception with status_code and response text."""
+
+    def __init__(self, status_code: int, text: str) -> None:
+        super().__init__(text)
+        self.status_code = status_code
+
+
+@pytest.mark.asyncio
+async def test_error_stop_pattern_with_matching_status_code_stops_failover() -> None:
+    """When status_codes is set and matches, failover should stop."""
+    db = MagicMock()
+    engine = FailoverEngine(db, error_classifier=_StubErrorClassifier(action=ErrorAction.BREAK))
+
+    config = {
+        "failover_rules": {
+            "error_stop_patterns": [
+                {"pattern": "content_policy", "status_codes": [403]},
+            ],
+        },
+    }
+    candidates = [
+        _make_candidate(provider_id="p1", provider_config=config),
+        _make_candidate(provider_id="p2"),
+    ]
+
+    attempt = AsyncMock(
+        side_effect=[
+            _HttpError(403, "content_policy_violation"),
+            AttemptResult(
+                kind=AttemptKind.SYNC_RESPONSE,
+                http_status=200,
+                http_headers={},
+                response_body={"ok": True},
+            ),
+        ]
+    )
+
+    result = await engine.execute(
+        candidates=candidates,
+        attempt_func=attempt,
+        retry_policy=RetryPolicy(mode=RetryMode.DISABLED, max_retries=1),
+        skip_policy=SkipPolicy(),
+        request_id=None,
+    )
+
+    # Should stop at first candidate, not try second
+    assert result.success is False
+    assert attempt.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_error_stop_pattern_with_non_matching_status_code_continues() -> None:
+    """When status_codes is set but doesn't match, the rule is skipped and failover continues."""
+    db = MagicMock()
+    engine = FailoverEngine(db, error_classifier=_StubErrorClassifier(action=ErrorAction.BREAK))
+
+    config = {
+        "failover_rules": {
+            "error_stop_patterns": [
+                {"pattern": "content_policy", "status_codes": [403]},
+            ],
+        },
+    }
+    candidates = [
+        _make_candidate(provider_id="p1", provider_config=config),
+        _make_candidate(provider_id="p2"),
+    ]
+
+    attempt = AsyncMock(
+        side_effect=[
+            _HttpError(500, "content_policy_violation"),
+            AttemptResult(
+                kind=AttemptKind.SYNC_RESPONSE,
+                http_status=200,
+                http_headers={},
+                response_body={"ok": True},
+            ),
+        ]
+    )
+
+    result = await engine.execute(
+        candidates=candidates,
+        attempt_func=attempt,
+        retry_policy=RetryPolicy(mode=RetryMode.DISABLED, max_retries=1),
+        skip_policy=SkipPolicy(),
+        request_id=None,
+    )
+
+    # status_code 500 doesn't match [403], so rule is skipped; failover continues to p2
+    assert result.success is True
+    assert result.provider_id == "p2"
+    assert attempt.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_error_stop_pattern_without_status_codes_matches_any() -> None:
+    """When status_codes is not set, the rule matches any status code (existing behavior)."""
+    db = MagicMock()
+    engine = FailoverEngine(db, error_classifier=_StubErrorClassifier(action=ErrorAction.BREAK))
+
+    config = {
+        "failover_rules": {
+            "error_stop_patterns": [
+                {"pattern": "content_policy"},
+            ],
+        },
+    }
+    candidates = [
+        _make_candidate(provider_id="p1", provider_config=config),
+        _make_candidate(provider_id="p2"),
+    ]
+
+    attempt = AsyncMock(
+        side_effect=[
+            _HttpError(500, "content_policy_violation"),
+            AttemptResult(
+                kind=AttemptKind.SYNC_RESPONSE,
+                http_status=200,
+                http_headers={},
+                response_body={"ok": True},
+            ),
+        ]
+    )
+
+    result = await engine.execute(
+        candidates=candidates,
+        attempt_func=attempt,
+        retry_policy=RetryPolicy(mode=RetryMode.DISABLED, max_retries=1),
+        skip_policy=SkipPolicy(),
+        request_id=None,
+    )
+
+    # No status_codes filter, pattern matches -> stop
+    assert result.success is False
+    assert attempt.await_count == 1