refactor(failover): 用 provider failover_rules 替代硬编码 ErrorClassifier 判断

移除 submit_with_failover 中基于 ErrorClassifier 的客户端错误硬编码逻辑,
改为读取 provider.config.failover_rules 进行规则匹配:
- error_stop_patterns: 错误响应命中时终止 failover
- success_failover_patterns: 2xx 响应命中时继续尝试下一个候选
同步更新相关注释、异常描述及测试用例

src/api/handlers/base/video_handler_base.py

@@ -305,7 +305,7 @@ class VideoHandlerBase(ABC):
 
         返回：
         - 成功：SubmitOutcome
-        - 上游客户端错误：直接返回脱敏后的 JSONResponse（保留 API 格式差异）
+        - 命中上游终止规则：直接返回脱敏后的 JSONResponse（保留 API 格式差异）
 
         失败时：
         - 无可用候选 / 全部失败：抛 HTTPException(503)

src/services/candidate/submit.py

@@ -20,7 +20,7 @@ class ExtractExternalTaskIdFunc(Protocol):
 
 
 class UpstreamClientRequestError(RuntimeError):
-    """可判定为客户端请求问题（不应 failover）的上游错误。"""
+    """命中上游终止规则（不应继续 failover）的错误。"""
 
     def __init__(
         self,
@@ -30,7 +30,7 @@ class UpstreamClientRequestError(RuntimeError):
     ) -> None:
         self.response = response
         self.candidate_keys = candidate_keys
-        super().__init__(f"Upstream client error: HTTP {response.status_code}")
+        super().__init__(f"Upstream stop rule matched: HTTP {response.status_code}")
 
 
 class AllCandidatesFailedError(RuntimeError):

src/services/task/service.py

@@ -1211,7 +1211,8 @@ class TaskService:
         Behavior notes:
         - sequentially try candidates (no per-candidate retries at submit stage)
         - record RequestCandidate audit rows
-        - stop on "client error" (raise UpstreamClientRequestError)
+        - stop only when provider failover_rules.error_stop_patterns matches
+        - continue on provider failover_rules.success_failover_patterns matches (2xx body)
         - if all failed, raise AllCandidatesFailedError
         """
         from datetime import datetime, timezone
@@ -1230,16 +1231,38 @@ class TaskService:
                 return "request_failed"
             return _SENSITIVE_PATTERN.sub("[REDACTED]", message)[:max_length]
 
-        def _should_stop_on_http_error(
+        def _extract_response_text(response: httpx.Response) -> str:
-            *, status_code: int, error_text: str, classifier: ErrorClassifier
+            try:
-        ) -> bool:
+                return response.text or ""
-            # Keep rules aligned with previous behavior:
+            except Exception:
-            # - 401/403/429 should not hard-stop the traversal at submit stage
+                return ""
-            if status_code in (401, 403, 429):
+
-                return False
+        def _match_provider_failover_rule(
-            if 400 <= status_code < 500:
+            candidate: Any,
-                return classifier.is_client_error(error_text)
+            *,
-            return False
+            is_success: bool,
+            response_text: str,
+            status_code: int | None = None,
+        ) -> str | None:
+            from src.services.candidate.failover import FailoverEngine
+
+            provider_config = getattr(candidate.provider, "config", None) or {}
+            rules = provider_config.get("failover_rules")
+            if not rules or not isinstance(rules, dict):
+                return None
+
+            compiled = FailoverEngine._get_compiled_patterns(rules)
+            key = "success" if is_success else "error"
+
+            for regex, rule in compiled.get(key, []):
+                if not is_success:
+                    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):
+                    return rule.get("pattern", "")
+
+            return None
 
         # IMPORTANT:
         # This method awaits upstream HTTP calls. If we have an open DB transaction before awaiting,
@@ -1268,7 +1291,6 @@ class TaskService:
                 scheduling_mode=scheduling_mode,
             )
             resolver = CandidateResolver(db=self.db, cache_scheduler=cache_scheduler)
-            error_classifier = ErrorClassifier(db=self.db, cache_scheduler=cache_scheduler)
 
             candidates, _global_model_id = await resolver.fetch_candidates(
                 api_format=api_format,
@@ -1463,10 +1485,7 @@ class TaskService:
 
                 if response.status_code >= 400:
                     finished_at = datetime.now(timezone.utc)
-                    try:
+                    error_text = _extract_response_text(response)
-                        error_text = response.text or ""
-                    except Exception:
-                        error_text = ""
                     error_msg = _sanitize(error_text)
                     candidate_info.update(
                         {
@@ -1488,11 +1507,20 @@ class TaskService:
                             )
                         )
 
-                    if _should_stop_on_http_error(
+                    stop_pattern = _match_provider_failover_rule(
+                        cand,
+                        is_success=False,
+                        response_text=error_text,
                         status_code=response.status_code,
-                        error_text=error_text,
+                    )
-                        classifier=error_classifier,
+                    if stop_pattern:
-                    ):
+                        logger.info(
+                            "[TaskService] 错误终止规则命中: pattern={}, status_code={}, provider={}",
+                            stop_pattern,
+                            response.status_code,
+                            cand.provider.name,
+                        )
+                        candidate_info["stop_rule_pattern"] = stop_pattern
                         try:
                             self.db.commit()
                         except Exception:
@@ -1503,6 +1531,44 @@ class TaskService:
                         )
                     continue
 
+                success_text = _extract_response_text(response)
+                success_continue_pattern = _match_provider_failover_rule(
+                    cand,
+                    is_success=True,
+                    response_text=success_text,
+                    status_code=response.status_code,
+                )
+                if success_continue_pattern:
+                    logger.info(
+                        "[TaskService] 成功转移规则命中: pattern={}, status_code={}, provider={}",
+                        success_continue_pattern,
+                        response.status_code,
+                        cand.provider.name,
+                    )
+                    finished_at = datetime.now(timezone.utc)
+                    failover_reason = f"success_failover_rule_matched:{success_continue_pattern}"
+                    candidate_info.update(
+                        {
+                            "attempt_status": "success_failover",
+                            "status_code": response.status_code,
+                            "error_message": failover_reason,
+                            "success_rule_pattern": success_continue_pattern,
+                        }
+                    )
+                    if record_id:
+                        self.db.execute(
+                            update(RequestCandidate)
+                            .where(RequestCandidate.id == record_id)
+                            .values(
+                                status="failed",
+                                status_code=response.status_code,
+                                error_type="success_failover_pattern",
+                                error_message=failover_reason,
+                                finished_at=finished_at,
+                            )
+                        )
+                    continue
+
                 # Parse JSON
                 payload: dict[str, Any] | None = None
                 try:

tests/services/test_candidate_service_submit.py

@@ -14,6 +14,7 @@ def _make_candidate(
     *,
     provider_id: str = "p1",
     provider_name: str = "prov",
+    provider_config: dict[str, Any] | None = None,
     endpoint_id: str = "e1",
     key_id: str = "k1",
     key_name: str = "key",
@@ -24,7 +25,7 @@ def _make_candidate(
     skip_reason: str | None = None,
     needs_conversion: bool = False,
 ) -> SimpleNamespace:
-    provider = SimpleNamespace(id=provider_id, name=provider_name)
+    provider = SimpleNamespace(id=provider_id, name=provider_name, config=provider_config or {})
     endpoint = SimpleNamespace(id=endpoint_id)
     key = SimpleNamespace(id=key_id, name=key_name, auth_type=auth_type, priority=priority)
     return SimpleNamespace(
@@ -39,7 +40,7 @@ def _make_candidate(
 
 
 @pytest.mark.asyncio
-async def test_submit_with_failover_skips_http_500_then_succeeds(
+async def test_submit_with_failover_continues_on_http_400_without_stop_rule_then_succeeds(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     db = MagicMock()
@@ -65,13 +66,8 @@ async def test_submit_with_failover_skips_http_500_then_succeeds(
             )
         ),
     )
-    monkeypatch.setattr(
-        "src.services.orchestration.error_classifier.ErrorClassifier.is_client_error",
-        lambda _self, _text: False,
-    )
-
     responses = [
-        httpx.Response(500, text='{"error": {"message": "server"}}'),
+        httpx.Response(400, json={"error": {"type": "invalid_request_error", "message": "bad"}}),
         httpx.Response(200, json={"id": "task-123"}),
     ]
     submit = AsyncMock(side_effect=responses)
@@ -97,7 +93,9 @@ async def test_submit_with_failover_skips_http_500_then_succeeds(
 
 
 @pytest.mark.asyncio
-async def test_submit_with_failover_stops_on_client_error(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_submit_with_failover_stops_on_provider_error_stop_rule(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
     db = MagicMock()
     svc = TaskService(db)
 
@@ -111,11 +109,22 @@ async def test_submit_with_failover_stops_on_client_error(monkeypatch: pytest.Mo
     )
     monkeypatch.setattr(
         "src.services.candidate.resolver.CandidateResolver.fetch_candidates",
-        AsyncMock(return_value=([_make_candidate()], "gm1")),
+        AsyncMock(
-    )
+            return_value=(
-    monkeypatch.setattr(
+                [
-        "src.services.orchestration.error_classifier.ErrorClassifier.is_client_error",
+                    _make_candidate(
-        lambda _self, _text: True,
+                        provider_config={
+                            "failover_rules": {
+                                "error_stop_patterns": [
+                                    {"pattern": "invalid_request_error", "status_codes": [400]}
+                                ]
+                            }
+                        }
+                    )
+                ],
+                "gm1",
+            )
+        ),
     )
 
     response = httpx.Response(
@@ -140,6 +149,68 @@ async def test_submit_with_failover_stops_on_client_error(monkeypatch: pytest.Mo
         )
 
 
+@pytest.mark.asyncio
+async def test_submit_with_failover_continues_on_provider_success_failover_rule(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    db = MagicMock()
+    svc = TaskService(db)
+
+    monkeypatch.setattr(
+        "src.services.system.config.SystemConfigService.get_config",
+        lambda *_args, **_kwargs: "provider",
+    )
+    monkeypatch.setattr(
+        "src.services.scheduling.aware_scheduler.get_cache_aware_scheduler",
+        AsyncMock(return_value=None),
+    )
+    monkeypatch.setattr(
+        "src.services.candidate.resolver.CandidateResolver.fetch_candidates",
+        AsyncMock(
+            return_value=(
+                [
+                    _make_candidate(
+                        provider_id="p1",
+                        endpoint_id="e1",
+                        key_id="k1",
+                        provider_config={
+                            "failover_rules": {
+                                "success_failover_patterns": [{"pattern": "fallback_me"}]
+                            }
+                        },
+                    ),
+                    _make_candidate(provider_id="p2", endpoint_id="e2", key_id="k2"),
+                ],
+                "gm1",
+            )
+        ),
+    )
+
+    responses = [
+        httpx.Response(200, json={"id": "task-should-not-be-used", "message": "fallback_me"}),
+        httpx.Response(200, json={"id": "task-123"}),
+    ]
+    submit = AsyncMock(side_effect=responses)
+
+    outcome = await svc.submit_with_failover(
+        api_format="openai:video",
+        model_name="sora",
+        affinity_key="a1",
+        user_api_key=MagicMock(),
+        request_id=None,
+        task_type="video",
+        submit_func=submit,
+        extract_external_task_id=lambda payload: payload.get("id"),
+        supported_auth_types={"api_key"},
+        allow_format_conversion=False,
+        max_candidates=10,
+    )
+
+    assert outcome.external_task_id == "task-123"
+    assert outcome.candidate.provider.id == "p2"
+    assert submit.await_count == 2
+
+
 @pytest.mark.asyncio
 async def test_submit_with_failover_no_eligible_candidates_due_to_auth_type(
     monkeypatch: pytest.MonkeyPatch,
@@ -159,10 +230,6 @@ async def test_submit_with_failover_no_eligible_candidates_due_to_auth_type(
         "src.services.candidate.resolver.CandidateResolver.fetch_candidates",
         AsyncMock(return_value=([_make_candidate(auth_type="vertex_ai")], "gm1")),
     )
-    monkeypatch.setattr(
-        "src.services.orchestration.error_classifier.ErrorClassifier.is_client_error",
-        lambda _self, _text: False,
-    )
 
     with pytest.raises(AllCandidatesFailedError) as excinfo:
         await svc.submit_with_failover(
@@ -209,11 +276,6 @@ async def test_submit_with_failover_filters_missing_billing_rule(
             )
         ),
     )
-    monkeypatch.setattr(
-        "src.services.orchestration.error_classifier.ErrorClassifier.is_client_error",
-        lambda _self, _text: False,
-    )
-
     # enable require_rule
     old = config.billing_require_rule
     try: