refactor: 统一任务框架 Phase 3 - 用 TaskService/FailoverEngine 替代 FallbackOrchestrator

核心重构：
- 移除 FallbackOrchestrator，用 TaskService + FailoverEngine 替代
- TaskService 作为统一入口，支持 SYNC/ASYNC 两种任务模式
- FailoverEngine 实现候选遍历、重试、故障转移逻辑
- 新增 AttemptFunc/AttemptResult 协议，统一尝试结果表示

功能改进：
- 流式响应首字节探测（30s 超时，空流触发故障转移）
- 流式取消归因优化（区分客户端断连 vs 服务端中断）
- 新增 OpenAI Sora 视频取消路由 POST /v1/videos/{task_id}/cancel
- OpenAI 流式请求自动添加 stream_options.include_usage

代码规范：
- 修复 loguru 日志格式（%s → {}）
- 新增 FORMAT_CONVERSION_ENABLED 环境变量说明

测试覆盖：
- test_failover_engine.py: FailoverEngine 单元测试
- test_task_service_async_execute.py: TaskService ASYNC 模式测试
- test_video_cancel_e2e.py: 视频取消端到端测试

src/services/billing/dimension_collector_service.py

@@ -188,7 +188,7 @@ class DimensionCollectorRuntime:
             except (ValueError, UnsafeExpressionError, ExpressionEvaluationError, Exception) as exc:
                 # 注意：这里选择“不中断，尝试下一优先级”
                 logger.debug(
-                    "Dimension collector failed (dim=%s, id=%s): %s",
+                    "Dimension collector failed (dim={}, id={}): {}",
                     dim_name,
                     getattr(c, "id", None),
                     str(exc),
@@ -261,7 +261,7 @@ class DimensionCollectorRuntime:
                 except UnsafeExpressionError:
                     # 配置错误：按无依赖处理，避免阻塞
                     logger.error(
-                        "Invalid computed transform_expression (dim=%s, id=%s)",
+                        "Invalid computed transform_expression (dim={}, id={})",
                         dim_name,
                         getattr(c, "id", None),
                     )
@@ -293,7 +293,7 @@ class DimensionCollectorRuntime:
         if len(ordered) != len(computed_only):
             # 有环依赖：保护性降级（按名称补齐），避免阻塞整条计费链路
             remaining = sorted(list(computed_only - set(ordered)))
-            logger.error("Computed dimension cycle detected: %s", remaining)
+            logger.error("Computed dimension cycle detected: {}", remaining)
             ordered.extend(remaining)
 
         return ordered

src/services/billing/service.py

@@ -125,7 +125,7 @@ class BillingService:
             return CostResult(cost=cost, status="legacy", snapshot=snapshot)
 
         logger.warning(
-            "No billing rule for task (task_type=%s, model=%s, provider_id=%s)",
+            "No billing rule for task (task_type={}, model={}, provider_id={})",
             task_type,
             model,
             provider_id,

src/services/cache/aware_scheduler.py

@@ -621,7 +621,7 @@ class CacheAwareScheduler:
         target_format = normalize_endpoint_signature(api_format)
 
         logger.debug(
-            "[Scheduler] list_all_candidates: model=%s, api_format=%s",
+            "[Scheduler] list_all_candidates: model={}, api_format={}",
             model_name,
             target_format,
         )
@@ -636,7 +636,7 @@ class CacheAwareScheduler:
             raise ModelNotSupportedException(model=model_name)
 
         logger.debug(
-            "[Scheduler] GlobalModel resolved: id=%s, name=%s",
+            "[Scheduler] GlobalModel resolved: id={}, name={}",
             global_model.id,
             global_model.name,
         )
@@ -691,12 +691,12 @@ class CacheAwareScheduler:
         self._release_db_connection_before_await(db)
 
         logger.debug(
-            "[Scheduler] Found %d active providers",
+            "[Scheduler] Found {} active providers",
             len(providers),
         )
         for p in providers:
             logger.debug(
-                "[Scheduler] Provider: id=%s, name=%s, is_active=%s, endpoints=%d, models=%d",
+                "[Scheduler] Provider: id={}, name={}, is_active={}, endpoints={}, models={}",
                 p.id[:8] if p.id else "N/A",
                 p.name,
                 p.is_active,
@@ -724,10 +724,14 @@ class CacheAwareScheduler:
         from src.config.settings import config
         from src.services.system.config import SystemConfigService
 
-        # 全局格式转换开关：优先使用数据库配置，回退到环境变量
+        # 格式转换总开关（环境变量）：关闭时禁止任何跨格式候选进入队列
+        master_conversion_enabled = bool(config.format_conversion_enabled)
+
+        # 全局覆盖开关（数据库）：开启时强制允许所有提供商的格式转换（跳过端点格式接受策略）
         global_conversion_enabled = SystemConfigService.is_format_conversion_enabled(db)
-        # 如果环境变量明确禁用，则禁用（环境变量可作为强制禁用开关）
-        if not config.format_conversion_enabled:
+
+        # 如果环境变量明确禁用，则全局覆盖也视为关闭（并最终禁止跨格式转换）
+        if not master_conversion_enabled:
             global_conversion_enabled = False
         candidates = await self._build_candidates(
             db=db,
@@ -740,6 +744,7 @@ class CacheAwareScheduler:
             is_stream=is_stream,
             capability_requirements=capability_requirements,
             global_conversion_enabled=global_conversion_enabled,
+            master_conversion_enabled=master_conversion_enabled,
         )
 
         # 3. 应用优先级模式排序
@@ -1064,6 +1069,7 @@ class CacheAwareScheduler:
         is_stream: bool = False,
         capability_requirements: dict[str, bool] | None = None,
         global_conversion_enabled: bool = False,
+        master_conversion_enabled: bool = True,
     ) -> list[ProviderCandidate]:
         """
         构建候选列表
@@ -1080,7 +1086,8 @@ class CacheAwareScheduler:
             max_candidates: 最大候选数
             is_stream: 是否是流式请求，如果为 True 则过滤不支持流式的 Provider
             capability_requirements: 能力需求（可选）
-            global_conversion_enabled: 全局格式转换开关
+            global_conversion_enabled: 全局覆盖开关（DB），开启时跳过端点格式接受策略检查
+            master_conversion_enabled: 总开关（ENV），关闭时禁止任何跨格式转换
 
         Returns:
             候选列表
@@ -1099,7 +1106,7 @@ class CacheAwareScheduler:
 
         for provider in providers:
             logger.debug(
-                "[Scheduler] Checking provider: %s, endpoints=%d",
+                "[Scheduler] Checking provider: {}, endpoints={}",
                 provider.name,
                 len(provider.endpoints) if provider.endpoints else 0,
             )
@@ -1155,7 +1162,7 @@ class CacheAwareScheduler:
 
             for endpoint in endpoints:
                 logger.debug(
-                    "[Scheduler] Checking endpoint: family=%s, kind=%s, is_active=%s, base_url=%s",
+                    "[Scheduler] Checking endpoint: family={}, kind={}, is_active={}, base_url={}",
                     getattr(endpoint, "api_family", None),
                     getattr(endpoint, "endpoint_kind", None),
                     getattr(endpoint, "is_active", None),
@@ -1170,15 +1177,14 @@ class CacheAwareScheduler:
                     str(getattr(endpoint, "endpoint_kind", "")).strip().lower(),
                 )
 
-                # 计算格式转换的有效开关状态（三层优先级）
-                # 全局 ON → 强制允许（跳过端点检查）
-                # 全局 OFF → 提供商 ON → 强制允许（跳过端点检查）
-                # 全局 OFF → 提供商 OFF → 看端点配置
+                # 计算格式转换开关状态（三层优先级）
+                #
+                # 1) 总开关（ENV）关闭 -> 禁止任何跨格式转换
+                # 2) 全局覆盖（DB）开启 -> 强制允许（跳过端点检查）
+                # 3) 提供商覆盖（Provider.enable_format_conversion）开启 -> 强制允许（跳过端点检查）
+                # 4) 否则 -> 由端点配置 format_acceptance_config 决定是否允许
                 provider_allows_conversion = getattr(provider, "enable_format_conversion", True)
-                effective_conversion_enabled = (
-                    global_conversion_enabled or provider_allows_conversion
-                )
-                # 如果全局或提供商开关为 ON，跳过端点配置检查
+                effective_conversion_enabled = bool(master_conversion_enabled)
                 skip_endpoint_check = global_conversion_enabled or provider_allows_conversion
 
                 is_compatible, needs_conversion, _compat_reason = is_format_compatible(
@@ -1190,11 +1196,12 @@ class CacheAwareScheduler:
                     skip_endpoint_check=skip_endpoint_check,
                 )
                 logger.debug(
-                    "[Scheduler] Format compatibility: client=%s, endpoint=%s, compatible=%s, "
-                    "global=%s, provider=%s, skip_endpoint=%s, reason=%s",
+                    "[Scheduler] Format compatibility: client={}, endpoint={}, compatible={}, "
+                    "master={}, global={}, provider={}, skip_endpoint={}, reason={}",
                     client_format_str,
                     endpoint_format_str,
                     is_compatible,
+                    master_conversion_enabled,
                     global_conversion_enabled,
                     provider_allows_conversion,
                     skip_endpoint_check,
@@ -1217,7 +1224,7 @@ class CacheAwareScheduler:
                     model_support_cache[endpoint_format_str]
                 )
                 logger.debug(
-                    "[Scheduler] Model support: provider=%s, model=%s, supports=%s, reason=%s",
+                    "[Scheduler] Model support: provider={}, model={}, supports={}, reason={}",
                     provider.name,
                     model_name,
                     supports_model,

src/services/candidate/failover.py

@@ -1,28 +1,54 @@
 from __future__ import annotations
 
-from typing import Any, Protocol
+import asyncio
+import re
+from collections.abc import Awaitable, Callable
+from datetime import datetime, timezone
+from typing import Any, AsyncIterator
 
+import httpx
+from sqlalchemy import update
 from sqlalchemy.orm import Session
 
+from src.core.logger import logger
+from src.models.database import RequestCandidate
 from src.services.cache.aware_scheduler import ProviderCandidate
+from src.services.orchestration.error_classifier import ErrorAction, ErrorClassifier
+from src.services.request.candidate import RequestCandidateService
+from src.services.task.exceptions import StreamProbeError
+from src.services.task.protocol import AttemptFunc, AttemptKind, AttemptResult
+from src.services.task.schema import ExecutionResult
 
-from .policy import RetryPolicy, SkipPolicy
-from .schema import CandidateKey, CandidateResult
+from .policy import FailoverAction, RetryMode, RetryPolicy, SkipPolicy
+from .recorder import CandidateRecorder
+from .schema import CandidateKey
 
-
-class AttemptFunc(Protocol):
-    async def __call__(self, candidate: ProviderCandidate) -> Any: ...
+_SENSITIVE_PATTERN = re.compile(
+    r"(api[_-]?key|token|bearer|authorization)[=:\s]+\S+",
+    re.IGNORECASE,
+)
 
 
 class FailoverEngine:
     """
     FailoverEngine executes candidate attempts under policies.
 
-    Phase2 scaffolding: implementation will gradually replace legacy orchestrators.
+    Phase3 core: unified failover loop used by TaskService.
     """
 
-    def __init__(self, db: Session):
+    # Hard constraint: streaming first chunk probe timeout
+    STREAM_FIRST_CHUNK_TIMEOUT_SECONDS: int = 30
+
+    def __init__(
+        self,
+        db: Session,
+        *,
+        error_classifier: ErrorClassifier | None = None,
+        recorder: CandidateRecorder | None = None,
+    ) -> None:
         self.db = db
+        self._error_classifier = error_classifier or ErrorClassifier(db=db)
+        self._recorder = recorder or CandidateRecorder(db)
 
     async def execute(
         self,
@@ -32,26 +58,688 @@ class FailoverEngine:
         retry_policy: RetryPolicy,
         skip_policy: SkipPolicy,
         request_id: str | None = None,
+        user_id: str | None = None,
+        api_key_id: str | None = None,
+        candidate_record_map: dict[tuple[int, int], str] | None = None,
         max_candidates: int | None = None,
-    ) -> CandidateResult:
-        # NOTE: intentionally minimal for now; legacy orchestrators still in use.
-        # This will be implemented when migrating video/chat flows to CandidateService.
-        _ = (retry_policy, skip_policy, request_id, max_candidates)
-        candidate_keys: list[CandidateKey] = []
+        max_attempts: int | None = None,
+        execution_error_handler: (
+            Callable[
+                ...,
+                Awaitable[tuple[FailoverAction, int | None]],
+            ]
+            | None
+        ) = None,
+    ) -> ExecutionResult:
+        """
+        Execute candidate traversal + retry + failover.
+
+        Notes:
+        - For PRE_EXPAND: `candidate_record_map` should be provided (created by CandidateResolver).
+        - For ON_DEMAND/DISABLED: records are created when used (and on skip, best-effort).
+        """
+        candidate_keys_fallback: list[CandidateKey] = []
+
+        if max_candidates is not None and max_candidates > 0:
+            candidates = candidates[:max_candidates]
+
+        attempt_count = 0
+        last_status_code: int | None = None
+
+        # For logging / dispatcher parity only; callers may pass an exact value.
+        if max_attempts is None:
+            computed = 0
+            for cand in candidates:
+                should_skip, _ = self._should_skip(cand, skip_policy)
+                if should_skip:
+                    continue
+                computed += self._get_max_retries(cand, retry_policy)
+            max_attempts = computed
+
+        for candidate_index, candidate in enumerate(candidates):
+            should_skip, skip_reason = self._should_skip(candidate, skip_policy)
+            if should_skip:
+                # PRE_EXPAND: mark all retry slots skipped.
+                if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
+                    self._mark_candidate_skipped(
+                        candidate_record_map=candidate_record_map,
+                        candidate_index=candidate_index,
+                        candidate=candidate,
+                        retry_policy=retry_policy,
+                        skip_reason=skip_reason,
+                    )
+                else:
+                    # ON_DEMAND/DISABLED: create a skipped record for audit (best-effort).
+                    if request_id:
+                        await self._create_skipped_record(
+                            request_id=request_id,
+                            candidate=candidate,
+                            candidate_index=candidate_index,
+                            user_id=user_id,
+                            api_key_id=api_key_id,
+                            skip_reason=skip_reason,
+                        )
+                candidate_keys_fallback.append(
+                    self._make_candidate_key(
+                        candidate=candidate,
+                        candidate_index=candidate_index,
+                        retry_index=0,
+                        status="skipped",
+                        skip_reason=skip_reason,
+                    )
+                )
+                continue
+
+            max_retries = self._get_max_retries(candidate, retry_policy)
+            retry_index = 0
+            while retry_index < max_retries:
+                attempt_count += 1
+
+                # Resolve/create record_id
+                record_id = None
+                if candidate_record_map:
+                    record_id = candidate_record_map.get((candidate_index, retry_index))
+                    if record_id is None:
+                        # Rectify may extend retries beyond pre-created range; reuse retry 0 record.
+                        record_id = candidate_record_map.get((candidate_index, 0))
+                if record_id is None and request_id and retry_policy.mode != RetryMode.PRE_EXPAND:
+                    record_id = await self._ensure_record_exists(
+                        request_id=request_id,
+                        candidate=candidate,
+                        candidate_index=candidate_index,
+                        retry_index=retry_index,
+                        user_id=user_id,
+                        api_key_id=api_key_id,
+                    )
+
+                # Attach per-attempt context onto candidate for attempt_func (keeps AttemptFunc signature stable).
+                try:
+                    setattr(candidate, "_utf_candidate_index", candidate_index)
+                    setattr(candidate, "_utf_retry_index", retry_index)
+                    setattr(candidate, "_utf_candidate_record_id", record_id)
+                    setattr(candidate, "_utf_attempt_count", attempt_count)
+                    setattr(candidate, "_utf_max_attempts", max_attempts)
+                except Exception:
+                    # Best-effort only; attempt_func may not rely on these attributes.
+                    pass
+
+                # Mark pending
+                now = datetime.now(timezone.utc)
+                if record_id:
+                    self._update_record(
+                        record_id,
+                        status="pending",
+                        started_at=now,
+                    )
+
+                # Commit BEFORE await (avoid holding DB connections during slow upstream calls)
+                self._commit_before_await()
+
+                try:
+                    attempt_result = await attempt_func(candidate)
+                    last_status_code = int(getattr(attempt_result, "http_status", 0) or 0)
+
+                    # Stream: probe first chunk, failover only before first chunk
+                    if attempt_result.kind == AttemptKind.STREAM:
+                        attempt_result = await self._probe_stream_first_chunk(
+                            attempt_result=attempt_result,
+                            record_id=record_id,
+                        )
+
+                    # Mark success-like status
+                    if record_id:
+                        if attempt_result.kind == AttemptKind.STREAM:
+                            # For streaming, mark "streaming" (final status is recorded elsewhere).
+                            self._update_record(
+                                record_id,
+                                status="streaming",
+                                status_code=attempt_result.http_status,
+                            )
+                        else:
+                            self._update_record(
+                                record_id,
+                                status="success",
+                                status_code=attempt_result.http_status,
+                                finished_at=datetime.now(timezone.utc),
+                            )
+                        self.db.commit()
+
+                    # PRE_EXPAND: mark unused slots after request ends (success)
+                    if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
+                        self._mark_remaining_slots_unused(
+                            candidate_record_map=candidate_record_map,
+                            candidates=candidates,
+                            success_candidate_idx=candidate_index,
+                            success_retry_idx=retry_index,
+                            retry_policy=retry_policy,
+                        )
+
+                    return ExecutionResult(
+                        success=True,
+                        attempt_result=attempt_result,
+                        candidate=candidate,
+                        candidate_index=candidate_index,
+                        retry_index=retry_index,
+                        provider_id=str(candidate.provider.id),
+                        provider_name=str(candidate.provider.name),
+                        endpoint_id=str(candidate.endpoint.id),
+                        key_id=str(candidate.key.id),
+                        candidate_keys=self._get_candidate_keys(
+                            request_id=request_id,
+                            fallback=candidate_keys_fallback,
+                            candidates=candidates,
+                        ),
+                        attempt_count=attempt_count,
+                        request_candidate_id=record_id,
+                    )
+
+                except StreamProbeError as exc:
+                    # Probe failed (before first chunk) => eligible for failover
+                    last_status_code = exc.http_status
+                    if record_id:
+                        self._update_record(
+                            record_id,
+                            status="failed",
+                            status_code=exc.http_status,
+                            error_type=type(exc).__name__,
+                            error_message=self._sanitize(str(exc)),
+                            finished_at=datetime.now(timezone.utc),
+                        )
+                        self.db.commit()
+                    action = FailoverAction.CONTINUE
+
+                except Exception as exc:
+                    has_retry_left = retry_index + 1 < max_retries
+
+                    # If caller provides an execution_error_handler, prefer it for RequestExecutor's ExecutionError.
+                    handler_used = False
+                    if execution_error_handler is not None:
+                        try:
+                            from src.services.request.executor import (
+                                ExecutionError as _ExecutionError,
+                            )
+
+                            if isinstance(exc, _ExecutionError):
+                                handler_used = True
+                                action, new_max_retries = await execution_error_handler(
+                                    exec_err=exc,
+                                    candidate=candidate,
+                                    candidate_index=candidate_index,
+                                    retry_index=retry_index,
+                                    max_retries_for_candidate=max_retries,
+                                    record_id=record_id,
+                                    attempt_count=attempt_count,
+                                    max_attempts=max_attempts,
+                                )
+                                if new_max_retries is not None:
+                                    max_retries = max(max_retries, int(new_max_retries))
+                        except Exception:
+                            # Fall back to internal handler below.
+                            handler_used = False
+
+                    if not handler_used:
+                        action = await self._handle_error(
+                            exc,
+                            candidate=candidate,
+                            has_retry_left=has_retry_left,
+                        )
+
+                        last_status_code = int(getattr(exc, "status_code", 0) or 0) or int(
+                            getattr(exc, "http_status", 0) or 0
+                        )
+
+                        if record_id:
+                            self._update_record(
+                                record_id,
+                                status="failed",
+                                status_code=last_status_code or None,
+                                error_type=type(exc).__name__,
+                                error_message=self._sanitize(str(exc)),
+                                finished_at=datetime.now(timezone.utc),
+                            )
+                            self.db.commit()
+
+                        if action == FailoverAction.STOP:
+                            # PRE_EXPAND: STOP ends the request => mark remaining slots unused.
+                            if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
+                                self._mark_remaining_slots_unused(
+                                    candidate_record_map=candidate_record_map,
+                                    candidates=candidates,
+                                    success_candidate_idx=candidate_index,
+                                    success_retry_idx=retry_index,
+                                    retry_policy=retry_policy,
+                                )
+                            return ExecutionResult(
+                                success=False,
+                                error_type=type(exc).__name__,
+                                error_message=self._sanitize(str(exc)),
+                                last_status_code=last_status_code or None,
+                                candidate_keys=self._get_candidate_keys(
+                                    request_id=request_id,
+                                    fallback=candidate_keys_fallback,
+                                    candidates=candidates,
+                                ),
+                                attempt_count=attempt_count,
+                            )
+
+                # action switch: continue/ retry
+                if action == FailoverAction.CONTINUE:
+                    # PRE_EXPAND: if we break early, mark remaining retries of this candidate unused.
+                    if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
+                        self._mark_candidate_remaining_retries_unused(
+                            candidate_record_map=candidate_record_map,
+                            candidate_idx=candidate_index,
+                            from_retry_idx=retry_index + 1,
+                            retry_policy=retry_policy,
+                        )
+                    break
+                if action == FailoverAction.RETRY:
+                    retry_index += 1
+                    continue
+
+                # Safety: unknown action -> stop retrying this candidate.
+                break
+
+        # exhausted: PRE_EXPAND should not leave 'available' records behind
+        if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
+            self._mark_all_remaining_available_unused(candidate_record_map)
+
+        return ExecutionResult(
+            success=False,
+            error_type="AllCandidatesFailed",
+            error_message="All candidates exhausted",
+            last_status_code=last_status_code,
+            candidate_keys=self._get_candidate_keys(
+                request_id=request_id,
+                fallback=candidate_keys_fallback,
+                candidates=candidates,
+            ),
+            attempt_count=attempt_count,
+        )
+
+    def _sanitize(self, message: str, max_length: int = 200) -> str:
+        if not message:
+            return "request_failed"
+        return _SENSITIVE_PATTERN.sub("[REDACTED]", message)[:max_length]
+
+    def _make_candidate_key(
+        self,
+        *,
+        candidate: ProviderCandidate,
+        candidate_index: int,
+        retry_index: int,
+        status: str,
+        skip_reason: str | None = None,
+        error_type: str | None = None,
+        error_message: str | None = None,
+        status_code: int | None = None,
+    ) -> CandidateKey:
+        return CandidateKey(
+            candidate_index=candidate_index,
+            retry_index=retry_index,
+            provider_id=str(candidate.provider.id),
+            provider_name=str(candidate.provider.name),
+            endpoint_id=str(candidate.endpoint.id),
+            key_id=str(candidate.key.id),
+            key_name=str(getattr(candidate.key, "name", "") or ""),
+            auth_type=str(getattr(candidate.key, "auth_type", "") or ""),
+            priority=int(getattr(candidate.key, "priority", 0) or 0),
+            is_cached=bool(getattr(candidate, "is_cached", False)),
+            status=status,
+            skip_reason=skip_reason,
+            error_type=error_type,
+            error_message=error_message,
+            status_code=status_code,
+        )
+
+    def _get_candidate_keys(
+        self,
+        *,
+        request_id: str | None,
+        fallback: list[CandidateKey],
+        candidates: list[ProviderCandidate],
+    ) -> list[CandidateKey]:
+        if request_id:
+            try:
+                return self._recorder.get_candidate_keys(request_id)
+            except Exception as exc:
+                # 降级到 fallback 但记录 warning（影响审计追踪可见性）
+                logger.warning(
+                    "[FailoverEngine] get_candidate_keys failed, using fallback: {}",
+                    self._sanitize(str(exc)),
+                )
+        if fallback:
+            return fallback
+        # fallback snapshot (no DB audit)
+        result: list[CandidateKey] = []
         for idx, cand in enumerate(candidates):
-            candidate_keys.append(
-                CandidateKey(
+            result.append(
+                self._make_candidate_key(
+                    candidate=cand,
                     candidate_index=idx,
-                    provider_id=str(cand.provider.id),
-                    provider_name=str(cand.provider.name),
-                    endpoint_id=str(cand.endpoint.id),
-                    key_id=str(cand.key.id),
-                    key_name=str(getattr(cand.key, "name", "") or ""),
-                    auth_type=str(getattr(cand.key, "auth_type", "") or ""),
-                    priority=int(getattr(cand.key, "priority", 0) or 0),
-                    is_cached=bool(getattr(cand, "is_cached", False)),
+                    retry_index=0,
                     status="available",
                 )
             )
+        return result
 
-        raise NotImplementedError("FailoverEngine.execute is not implemented yet")
+    def _commit_before_await(self) -> None:
+        if self.db.in_transaction():
+            try:
+                self.db.commit()
+            except Exception:
+                self.db.rollback()
+                raise
+
+    def _update_record(self, record_id: str, /, **values: Any) -> None:
+        self.db.execute(
+            update(RequestCandidate).where(RequestCandidate.id == record_id).values(**values)
+        )
+
+    async def _ensure_record_exists(
+        self,
+        *,
+        request_id: str,
+        candidate: ProviderCandidate,
+        candidate_index: int,
+        retry_index: int,
+        user_id: str | None,
+        api_key_id: str | None,
+    ) -> str:
+        # Create "available" record, then caller will mark pending.
+        row = RequestCandidateService.create_candidate(
+            db=self.db,
+            request_id=request_id,
+            candidate_index=candidate_index,
+            retry_index=retry_index,
+            user_id=user_id,
+            api_key_id=api_key_id,
+            provider_id=str(candidate.provider.id),
+            endpoint_id=str(candidate.endpoint.id),
+            key_id=str(candidate.key.id),
+            status="available",
+            is_cached=bool(getattr(candidate, "is_cached", False)),
+            extra_data={},
+        )
+        return str(row.id)
+
+    async def _create_skipped_record(
+        self,
+        *,
+        request_id: str,
+        candidate: ProviderCandidate,
+        candidate_index: int,
+        user_id: str | None,
+        api_key_id: str | None,
+        skip_reason: str | None,
+    ) -> str:
+        row = RequestCandidateService.create_candidate(
+            db=self.db,
+            request_id=request_id,
+            candidate_index=candidate_index,
+            retry_index=0,
+            user_id=user_id,
+            api_key_id=api_key_id,
+            provider_id=str(candidate.provider.id),
+            endpoint_id=str(candidate.endpoint.id),
+            key_id=str(candidate.key.id),
+            status="skipped",
+            skip_reason=skip_reason,
+            is_cached=bool(getattr(candidate, "is_cached", False)),
+            extra_data={},
+        )
+        # ensure visible for subsequent recorder reads
+        if self.db.in_transaction():
+            self.db.commit()
+        return str(row.id)
+
+    def _should_skip(
+        self, candidate: ProviderCandidate, skip_policy: SkipPolicy
+    ) -> tuple[bool, str | None]:
+        if bool(getattr(candidate, "is_skipped", False)):
+            return True, str(getattr(candidate, "skip_reason", None) or "scheduler_marked")
+
+        auth_type = str(getattr(getattr(candidate, "key", None), "auth_type", "") or "api_key")
+        if (
+            skip_policy.supported_auth_types is not None
+            and auth_type not in skip_policy.supported_auth_types
+        ):
+            return True, "unsupported_auth_type"
+
+        needs_conversion = bool(getattr(candidate, "needs_conversion", False))
+        if needs_conversion and not skip_policy.allow_format_conversion:
+            return True, "format_conversion_not_supported"
+
+        return False, None
+
+    def _get_max_retries(self, candidate: ProviderCandidate, retry_policy: RetryPolicy) -> int:
+        if retry_policy.mode == RetryMode.DISABLED:
+            return 1
+        if retry_policy.retry_on_cached_only and not bool(getattr(candidate, "is_cached", False)):
+            return 1
+        provider_max = getattr(getattr(candidate, "provider", None), "max_retries", None)
+        try:
+            value = int(provider_max or retry_policy.max_retries or 1)
+        except Exception:
+            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,
+        *,
+        candidate: ProviderCandidate,
+        has_retry_left: bool,
+    ) -> FailoverAction:
+        # Special: HTTP client errors should stop failover.
+        if isinstance(error, httpx.HTTPStatusError):
+            status_code = int(getattr(error.response, "status_code", 0) or 0)
+            try:
+                error_text = error.response.text or ""
+            except Exception:
+                error_text = ""
+            if self._should_stop_on_http_error(status_code=status_code, error_text=error_text):
+                return FailoverAction.STOP
+
+        # Default: reuse legacy ErrorClassifier decision and map to FailoverAction.
+        action = self._error_classifier.classify(error, has_retry_left=has_retry_left)
+        if action == ErrorAction.RAISE:
+            return FailoverAction.STOP
+        if action == ErrorAction.BREAK:
+            return FailoverAction.CONTINUE
+        return FailoverAction.RETRY
+
+    async def _probe_stream_first_chunk(
+        self,
+        *,
+        attempt_result: AttemptResult,
+        record_id: str | None,
+    ) -> AttemptResult:
+        """
+        Probe first chunk for a streaming response.
+
+        Strong constraints:
+        - Must have timeout.
+        - Empty stream before first chunk is treated as probe failure (eligible for failover).
+        """
+        assert attempt_result.kind == AttemptKind.STREAM
+        assert attempt_result.stream_iterator is not None
+
+        original_iterator = attempt_result.stream_iterator
+        try:
+            first_chunk = await asyncio.wait_for(
+                original_iterator.__anext__(),
+                timeout=self.STREAM_FIRST_CHUNK_TIMEOUT_SECONDS,
+            )
+        except asyncio.TimeoutError as exc:
+            raise StreamProbeError(
+                "Timeout waiting for first chunk",
+                http_status=attempt_result.http_status,
+                original_exception=exc,
+            ) from exc
+        except StopAsyncIteration as exc:
+            raise StreamProbeError(
+                "Empty stream: no data received before EOF",
+                http_status=attempt_result.http_status,
+                original_exception=exc,
+            ) from exc
+        except Exception as exc:
+            raise StreamProbeError(
+                f"Failed to read first chunk: {exc}",
+                http_status=attempt_result.http_status,
+                original_exception=exc,
+            ) from exc
+
+        wrapped = self._wrap_stream_with_finalizer(
+            first_chunk=first_chunk,
+            original_iterator=original_iterator,
+            record_id=record_id,
+        )
+        return AttemptResult(
+            kind=AttemptKind.STREAM,
+            http_status=attempt_result.http_status,
+            http_headers=attempt_result.http_headers,
+            stream_iterator=wrapped,
+            raw_response=attempt_result.raw_response,
+        )
+
+    def _wrap_stream_with_finalizer(
+        self,
+        *,
+        first_chunk: bytes,
+        original_iterator: AsyncIterator[bytes],
+        record_id: str | None,
+    ) -> AsyncIterator[bytes]:
+        async def _gen() -> AsyncIterator[bytes]:
+            yield first_chunk
+            try:
+                async for chunk in original_iterator:
+                    yield chunk
+            except Exception as exc:
+                # Best-effort: mark stream interrupted using a new session (stream may outlive request session).
+                if record_id:
+                    self._mark_record_stream_interrupted(record_id, exc)
+                raise
+
+        return _gen()
+
+    def _mark_record_stream_interrupted(self, record_id: str, exc: Exception) -> None:
+        try:
+            from src.database import create_session
+
+            with create_session() as db:
+                db.execute(
+                    update(RequestCandidate)
+                    .where(RequestCandidate.id == record_id)
+                    .values(
+                        status="stream_interrupted",
+                        error_type=type(exc).__name__,
+                        error_message=self._sanitize(str(exc)),
+                        finished_at=datetime.now(timezone.utc),
+                    )
+                )
+                db.commit()
+        except Exception as inner:
+            logger.debug(
+                "[FailoverEngine] Failed to mark stream_interrupted: {}",
+                self._sanitize(str(inner)),
+            )
+
+    def _mark_candidate_skipped(
+        self,
+        *,
+        candidate_record_map: dict[tuple[int, int], str],
+        candidate_index: int,
+        candidate: ProviderCandidate,
+        retry_policy: RetryPolicy,
+        skip_reason: str | None,
+    ) -> None:
+        max_retries = self._get_max_retries(candidate, retry_policy)
+        now = datetime.now(timezone.utc)
+        for retry_index in range(max_retries):
+            record_id = candidate_record_map.get((candidate_index, retry_index))
+            if record_id:
+                self._update_record(
+                    record_id,
+                    status="skipped",
+                    skip_reason=skip_reason,
+                    finished_at=now,
+                )
+        self.db.commit()
+
+    def _mark_remaining_slots_unused(
+        self,
+        *,
+        candidate_record_map: dict[tuple[int, int], str],
+        candidates: list[ProviderCandidate],
+        success_candidate_idx: int,
+        success_retry_idx: int,
+        retry_policy: RetryPolicy,
+    ) -> None:
+        now = datetime.now(timezone.utc)
+        for candidate_idx, cand in enumerate(candidates):
+            max_retries = self._get_max_retries(cand, retry_policy)
+            for retry_idx in range(max_retries):
+                if candidate_idx < success_candidate_idx:
+                    continue
+                if candidate_idx == success_candidate_idx and retry_idx <= success_retry_idx:
+                    continue
+                record_id = candidate_record_map.get((candidate_idx, retry_idx))
+                if record_id:
+                    self._update_record(
+                        record_id,
+                        status="unused",
+                        finished_at=now,
+                    )
+        self.db.commit()
+
+    def _mark_candidate_remaining_retries_unused(
+        self,
+        *,
+        candidate_record_map: dict[tuple[int, int], str],
+        candidate_idx: int,
+        from_retry_idx: int,
+        retry_policy: RetryPolicy,
+    ) -> None:
+        # Only meaningful for PRE_EXPAND.
+        # We don't have access to candidate object list here, so infer max_retries from map keys.
+        # Fallback to retry_policy.max_retries.
+        now = datetime.now(timezone.utc)
+        # try best-effort upper bound
+        upper = max(
+            (ri for (ci, ri) in candidate_record_map.keys() if ci == candidate_idx),
+            default=retry_policy.max_retries - 1,
+        )
+        for retry_idx in range(from_retry_idx, upper + 1):
+            record_id = candidate_record_map.get((candidate_idx, retry_idx))
+            if record_id:
+                self._update_record(record_id, status="unused", finished_at=now)
+        self.db.commit()
+
+    def _mark_all_remaining_available_unused(
+        self, candidate_record_map: dict[tuple[int, int], str]
+    ) -> None:
+        # As a safety net: do not leave available records behind in PRE_EXPAND mode.
+        try:
+            ids = list(candidate_record_map.values())
+            if not ids:
+                return
+            now = datetime.now(timezone.utc)
+            self.db.execute(
+                update(RequestCandidate)
+                .where(RequestCandidate.id.in_(ids))
+                .where(RequestCandidate.status == "available")
+                .values(status="unused", finished_at=now)
+            )
+            self.db.commit()
+        except Exception:
+            self.db.rollback()
+            raise

src/services/candidate/policy.py

@@ -12,6 +12,14 @@ class RetryMode(str, Enum):
     DISABLED = "disabled"  # no retry
 
 
+class FailoverAction(str, Enum):
+    """Failover decision after an attempt error."""
+
+    STOP = "stop"  # stop failover (client error / non-retriable)
+    CONTINUE = "continue"  # continue with next candidate
+    RETRY = "retry"  # retry current candidate
+
+
 @dataclass(frozen=True)
 class RetryPolicy:
     """Unified retry policy."""

src/services/candidate/service.py

@@ -1,24 +1,11 @@
 from __future__ import annotations
 
-import re
-from datetime import datetime, timezone
 from typing import Any
 
-import httpx
-from sqlalchemy import update
 from sqlalchemy.orm import Session
 
-from src.config.settings import config
-from src.core.exceptions import ProviderNotAvailableException
-from src.core.logger import logger
-from src.models.database import ApiKey, RequestCandidate
-from src.services.billing.rule_service import BillingRuleLookupResult, BillingRuleService
+from src.models.database import ApiKey
 from src.services.cache.aware_scheduler import ProviderCandidate, get_cache_aware_scheduler
-from src.services.candidate.submit import (
-    AllCandidatesFailedError,
-    SubmitOutcome,
-    UpstreamClientRequestError,
-)
 from src.services.orchestration.error_classifier import ErrorClassifier
 from src.services.system.config import SystemConfigService
 
@@ -26,17 +13,6 @@ from .recorder import CandidateRecorder
 from .resolver import CandidateResolver
 from .schema import CandidateKey
 
-_SENSITIVE_PATTERN = re.compile(
-    r"(api[_-]?key|token|bearer|authorization)[=:\s]+\S+",
-    re.IGNORECASE,
-)
-
-
-def _sanitize(message: str, max_length: int = 200) -> str:
-    if not message:
-        return "request_failed"
-    return _SENSITIVE_PATTERN.sub("[REDACTED]", message)[:max_length]
-
 
 class CandidateService:
     """
@@ -101,396 +77,6 @@ class CandidateService:
             preferred_key_ids=preferred_key_ids,
         )
 
-    def _should_stop_on_http_error(self, *, status_code: int, error_text: str) -> bool:
-        """
-        Decide whether an upstream HTTP error is a client error (no failover).
-
-        Rules:
-        - 401/403/429 are usually key/permission/ratelimit issues -> allow failover
-        - other 4xx: stop only if ErrorClassifier says it's a client error
-        """
-        if status_code in (401, 403, 429):
-            return False
-        if 400 <= status_code < 500:
-            assert self._error_classifier is not None
-            return self._error_classifier.is_client_error(error_text)
-        return False
-
-    async def submit_with_failover(
-        self,
-        *,
-        api_format: str,
-        model_name: str,
-        affinity_key: str,
-        user_api_key: ApiKey,
-        request_id: str | None,
-        task_type: str,
-        submit_func: Any,
-        extract_external_task_id: Any,
-        supported_auth_types: set[str] | None = None,
-        allow_format_conversion: bool = False,
-        capability_requirements: dict[str, bool] | None = None,
-        max_candidates: int | None = None,
-    ) -> SubmitOutcome:
-        """
-        Submit async task with failover, returning the selected candidate + external_task_id.
-
-        Phase2 submit entrypoint (replaces legacy submit orchestrator).
-        """
-        # IMPORTANT:
-        # This method awaits upstream HTTP calls. If we have an open DB transaction before awaiting,
-        # the connection can be held for a long time (pool exhaustion under concurrency).
-        #
-        # Also note SQLAlchemy's default expire_on_commit=True would expire ORM objects and may
-        # trigger unexpected lazy DB loads after we commit (potentially during the await).
-        # We disable it temporarily to keep candidate/provider/key objects in-memory.
-        original_expire_on_commit = getattr(self.db, "expire_on_commit", True)
-        self.db.expire_on_commit = False
-        await self._ensure_initialized()
-        assert self._resolver is not None
-        try:
-            candidates, _global_model_id = await self._resolver.fetch_candidates(
-                api_format=api_format,
-                model_name=model_name,
-                affinity_key=affinity_key,
-                user_api_key=user_api_key,
-                request_id=request_id,
-                is_stream=False,
-                capability_requirements=capability_requirements,
-            )
-
-            if not candidates:
-                raise ProviderNotAvailableException("No candidates available")
-
-            if max_candidates is not None and max_candidates > 0:
-                candidates = candidates[:max_candidates]
-
-            # Pre-create RequestCandidate records (no retry expand for async submit stage)
-            record_map: dict[tuple[int, int], str] = {}
-            if request_id:
-                try:
-                    record_map = self.create_candidate_records(
-                        candidates=candidates,
-                        request_id=request_id,
-                        user_api_key=user_api_key,
-                        required_capabilities=capability_requirements,
-                        expand_retries=False,
-                    )
-                except Exception as exc:
-                    logger.warning(
-                        "[CandidateService] Failed to create candidate records: %s",
-                        _sanitize(str(exc)),
-                    )
-                    record_map = {}
-
-            candidate_keys: list[dict[str, Any]] = []
-            eligible_count = 0
-            last_status_code: int | None = None
-
-            for idx, cand in enumerate(candidates):
-                now = datetime.now(timezone.utc)
-                auth_type = getattr(cand.key, "auth_type", "api_key") or "api_key"
-
-                candidate_info: dict[str, Any] = {
-                    "index": idx,
-                    "provider_id": cand.provider.id,
-                    "provider_name": cand.provider.name,
-                    "endpoint_id": cand.endpoint.id,
-                    "key_id": cand.key.id,
-                    "key_name": getattr(cand.key, "name", None),
-                    "auth_type": auth_type,
-                    "priority": getattr(cand.key, "priority", 0) or 0,
-                    "is_cached": bool(getattr(cand, "is_cached", False)),
-                }
-                candidate_keys.append(candidate_info)
-
-                record_id = record_map.get((idx, 0))
-
-                # Scheduler marked skip
-                if getattr(cand, "is_skipped", False):
-                    skip_reason = getattr(cand, "skip_reason", None) or "skipped"
-                    candidate_info.update({"skipped": True, "skip_reason": skip_reason})
-                    if record_id:
-                        # record is usually already skipped, but keep it consistent
-                        self.db.execute(
-                            update(RequestCandidate)
-                            .where(RequestCandidate.id == record_id)
-                            .values(status="skipped", skip_reason=skip_reason)
-                        )
-                    continue
-
-                # Format conversion checks
-                # 优先级：全局开关 ON 强制允许，全局开关 OFF 看提供商开关
-                needs_conversion = bool(getattr(cand, "needs_conversion", False))
-                if needs_conversion:
-                    # 1. Check handler-level switch (handler 不支持则直接跳过)
-                    if not allow_format_conversion:
-                        skip_reason = "format_conversion_not_supported"
-                        candidate_info.update({"skipped": True, "skip_reason": skip_reason})
-                        if record_id:
-                            self.db.execute(
-                                update(RequestCandidate)
-                                .where(RequestCandidate.id == record_id)
-                                .values(status="skipped", skip_reason=skip_reason)
-                            )
-                        continue
-
-                    # 2. Check global + provider switches
-                    # 全局 ON → 允许；全局 OFF → 看提供商
-                    from src.services.system.config import SystemConfigService
-
-                    global_enabled = SystemConfigService.is_format_conversion_enabled(self.db)
-                    provider_enabled = getattr(cand.provider, "enable_format_conversion", True)
-                    effective_enabled = global_enabled or provider_enabled
-
-                    if not effective_enabled:
-                        skip_reason = "format_conversion_disabled"
-                        candidate_info.update(
-                            {
-                                "skipped": True,
-                                "skip_reason": skip_reason,
-                                "global_conversion_enabled": global_enabled,
-                                "provider_conversion_enabled": provider_enabled,
-                            }
-                        )
-                        if record_id:
-                            self.db.execute(
-                                update(RequestCandidate)
-                                .where(RequestCandidate.id == record_id)
-                                .values(status="skipped", skip_reason=skip_reason)
-                            )
-                        continue
-
-                # auth_type filter
-                if supported_auth_types is not None and auth_type not in supported_auth_types:
-                    skip_reason = f"unsupported_auth_type:{auth_type}"
-                    candidate_info.update({"skipped": True, "skip_reason": skip_reason})
-                    if record_id:
-                        self.db.execute(
-                            update(RequestCandidate)
-                            .where(RequestCandidate.id == record_id)
-                            .values(status="skipped", skip_reason=skip_reason)
-                        )
-                    continue
-
-                # billing rule filter
-                rule_lookup: BillingRuleLookupResult | None = None
-                has_billing_rule = True
-                if config.billing_require_rule:
-                    rule_lookup = BillingRuleService.find_rule(
-                        self.db,
-                        provider_id=cand.provider.id,
-                        model_name=model_name,
-                        task_type=task_type,
-                    )
-                    has_billing_rule = rule_lookup is not None
-                    if not has_billing_rule:
-                        skip_reason = "billing_rule_missing"
-                        candidate_info.update(
-                            {"has_billing_rule": False, "skipped": True, "skip_reason": skip_reason}
-                        )
-                        if record_id:
-                            self.db.execute(
-                                update(RequestCandidate)
-                                .where(RequestCandidate.id == record_id)
-                                .values(status="skipped", skip_reason=skip_reason)
-                            )
-                        continue
-                candidate_info["has_billing_rule"] = has_billing_rule
-
-                eligible_count += 1
-
-                # Mark pending
-                if record_id:
-                    self.db.execute(
-                        update(RequestCandidate)
-                        .where(RequestCandidate.id == record_id)
-                        .values(status="pending", started_at=now)
-                    )
-
-                # Flush/commit BEFORE awaiting upstream submit to avoid holding DB connections
-                # during potentially slow network operations.
-                if self.db.in_transaction():
-                    try:
-                        self.db.commit()
-                    except Exception:
-                        self.db.rollback()
-                        raise
-
-                # Attempt submit (upstream HTTP)
-                try:
-                    response: httpx.Response = await submit_func(cand)
-                except Exception as exc:
-                    finished_at = datetime.now(timezone.utc)
-                    error_msg = _sanitize(str(exc))
-                    candidate_info.update(
-                        {
-                            "attempt_status": "exception",
-                            "error_type": type(exc).__name__,
-                            "error_message": error_msg,
-                        }
-                    )
-                    if record_id:
-                        self.db.execute(
-                            update(RequestCandidate)
-                            .where(RequestCandidate.id == record_id)
-                            .values(
-                                status="failed",
-                                error_type=type(exc).__name__,
-                                error_message=error_msg,
-                                finished_at=finished_at,
-                            )
-                        )
-                    continue
-
-                last_status_code = int(getattr(response, "status_code", 0) or 0)
-
-                if response.status_code >= 400:
-                    finished_at = datetime.now(timezone.utc)
-                    try:
-                        error_text = response.text or ""
-                    except Exception:
-                        error_text = ""
-                    error_msg = _sanitize(error_text)
-                    candidate_info.update(
-                        {
-                            "attempt_status": "http_error",
-                            "status_code": response.status_code,
-                            "error_message": error_msg,
-                        }
-                    )
-                    if record_id:
-                        self.db.execute(
-                            update(RequestCandidate)
-                            .where(RequestCandidate.id == record_id)
-                            .values(
-                                status="failed",
-                                status_code=response.status_code,
-                                error_type="http_error",
-                                error_message=error_msg,
-                                finished_at=finished_at,
-                            )
-                        )
-
-                    if self._should_stop_on_http_error(
-                        status_code=response.status_code, error_text=error_text
-                    ):
-                        try:
-                            self.db.commit()
-                        except Exception:
-                            self.db.rollback()
-                        raise UpstreamClientRequestError(
-                            response=response,
-                            candidate_keys=candidate_keys,
-                        )
-                    continue
-
-                # Parse JSON
-                payload: dict[str, Any] | None = None
-                try:
-                    data = response.json()
-                    if isinstance(data, dict):
-                        payload = data
-                except Exception as exc:
-                    finished_at = datetime.now(timezone.utc)
-                    error_msg = _sanitize(str(exc))
-                    candidate_info.update(
-                        {
-                            "attempt_status": "invalid_json",
-                            "error_type": type(exc).__name__,
-                            "error_message": error_msg,
-                        }
-                    )
-                    if record_id:
-                        self.db.execute(
-                            update(RequestCandidate)
-                            .where(RequestCandidate.id == record_id)
-                            .values(
-                                status="failed",
-                                status_code=response.status_code,
-                                error_type="invalid_json",
-                                error_message=error_msg,
-                                finished_at=finished_at,
-                            )
-                        )
-                    continue
-
-                external_task_id = extract_external_task_id(payload or {})
-                if not external_task_id:
-                    finished_at = datetime.now(timezone.utc)
-                    candidate_info.update(
-                        {
-                            "attempt_status": "empty_task_id",
-                            "error_message": "Upstream returned empty task id",
-                        }
-                    )
-                    if record_id:
-                        self.db.execute(
-                            update(RequestCandidate)
-                            .where(RequestCandidate.id == record_id)
-                            .values(
-                                status="failed",
-                                status_code=response.status_code,
-                                error_type="empty_task_id",
-                                error_message="Upstream returned empty task id",
-                                finished_at=finished_at,
-                            )
-                        )
-                    continue
-
-                # Success
-                finished_at = datetime.now(timezone.utc)
-                candidate_info.update({"attempt_status": "success", "selected": True})
-                if record_id:
-                    self.db.execute(
-                        update(RequestCandidate)
-                        .where(RequestCandidate.id == record_id)
-                        .values(
-                            status="success",
-                            status_code=response.status_code,
-                            finished_at=finished_at,
-                        )
-                    )
-                try:
-                    self.db.commit()
-                except Exception:
-                    self.db.rollback()
-
-                return SubmitOutcome(
-                    candidate=cand,
-                    candidate_keys=candidate_keys,
-                    external_task_id=str(external_task_id),
-                    rule_lookup=rule_lookup,
-                    upstream_payload=payload,
-                    upstream_headers=dict(response.headers),
-                    upstream_status_code=response.status_code,
-                )
-
-            # Persist candidate records before raising
-            try:
-                self.db.commit()
-            except Exception:
-                self.db.rollback()
-
-            if eligible_count == 0:
-                reason = "no_eligible_candidates"
-                if config.billing_require_rule:
-                    reason = "no_candidate_with_billing_rule"
-                raise AllCandidatesFailedError(
-                    reason=reason,
-                    candidate_keys=candidate_keys,
-                    last_status_code=last_status_code,
-                )
-
-            raise AllCandidatesFailedError(
-                reason="all_candidates_failed",
-                candidate_keys=candidate_keys,
-                last_status_code=last_status_code,
-            )
-        finally:
-            # Restore Session behavior for the rest of the request lifecycle.
-            self.db.expire_on_commit = original_expire_on_commit
-
     def create_candidate_records(
         self,
         *,

src/services/model/availability.py

@@ -170,7 +170,7 @@ class ModelAvailabilityQuery:
                 key_formats_norm = set(endpoint_formats)
             elif not isinstance(key_formats, list):
                 logger.warning(
-                    "[ModelAvailability] Key api_formats 类型异常, provider_id=%s, type=%s",
+                    "[ModelAvailability] Key api_formats 类型异常, provider_id={}, type={}",
                     provider_id,
                     type(key_formats).__name__,
                 )
@@ -237,7 +237,7 @@ class ModelAvailabilityQuery:
                 key_formats_norm = set(endpoint_formats)
             elif not isinstance(key_formats, list):
                 logger.warning(
-                    "[ModelAvailability] Key api_formats 类型异常, key_id=%s, type=%s",
+                    "[ModelAvailability] Key api_formats 类型异常, key_id={}, type={}",
                     key_id,
                     type(key_formats).__name__,
                 )

src/services/model/cost.py

@@ -430,7 +430,7 @@ class ModelCostService:
             price_per_request = self.get_request_price(provider, model)
             if price_per_request is None or price_per_request == 0.0:
                 logger.warning(
-                    "未找到模型价格配置: %s/%s，请在 GlobalModel 中配置价格",
+                    "未找到模型价格配置: {}/{}，请在 GlobalModel 中配置价格",
                     provider_name,
                     model,
                 )

src/services/orchestration/__init__.py

@@ -2,7 +2,6 @@
 Orchestration 模块
 
 提供请求编排相关的组件：
-- FallbackOrchestrator: 故障转移编排器，协调请求的完整生命周期
 - CandidateResolver: 候选解析器，负责获取和排序可用的 Provider 组合
 - RequestDispatcher: 请求分发器，负责执行单个候选请求
 - ErrorClassifier: 错误分类器，负责错误分类和处理策略
@@ -10,11 +9,9 @@ Orchestration 模块
 
 from .candidate_resolver import CandidateResolver
 from .error_classifier import ErrorAction, ErrorClassifier
-from .fallback_orchestrator import FallbackOrchestrator
 from .request_dispatcher import RequestDispatcher
 
 __all__ = [
-    "FallbackOrchestrator",
     "CandidateResolver",
     "RequestDispatcher",
     "ErrorClassifier",

src/services/orchestration/candidate_resolver.py

@@ -77,7 +77,7 @@ class CandidateResolver:
         global_model_id: str | None = None
 
         logger.debug(
-            "[CandidateResolver] fetch_candidates starting: model=%s, api_format=%s",
+            "[CandidateResolver] fetch_candidates starting: model={}, api_format={}",
             model_name,
             api_format,
         )
@@ -96,7 +96,7 @@ class CandidateResolver:
             )
 
             logger.debug(
-                "[CandidateResolver] list_all_candidates batch: offset=%d, returned=%d candidates",
+                "[CandidateResolver] list_all_candidates batch: offset={}, returned={} candidates",
                 provider_offset,
                 len(candidates),
             )
@@ -111,7 +111,7 @@ class CandidateResolver:
             provider_offset += provider_batch_size
 
             logger.debug(
-                "[CandidateResolver] fetch_candidates completed: total=%d candidates",
+                "[CandidateResolver] fetch_candidates completed: total={} candidates",
                 len(all_candidates),
             )
 

src/services/provider/transport.py

@@ -254,7 +254,7 @@ def get_vertex_ai_effective_format(
             return normalize_endpoint_signature(user_format_mapping[model])
         except Exception:
             logger.warning(
-                "Invalid vertex_ai model_format_mapping value for model '%s': %r",
+                "Invalid vertex_ai model_format_mapping value for model '{}': {!r}",
                 model,
                 user_format_mapping[model],
             )
@@ -266,7 +266,7 @@ def get_vertex_ai_effective_format(
                 return normalize_endpoint_signature(api_format)
             except Exception:
                 logger.warning(
-                    "Invalid vertex_ai model_format_mapping value for prefix '%s': %r",
+                    "Invalid vertex_ai model_format_mapping value for prefix '{}': {!r}",
                     prefix,
                     api_format,
                 )
@@ -282,7 +282,7 @@ def get_vertex_ai_effective_format(
         try:
             return normalize_endpoint_signature(user_default_format)
         except Exception:
-            logger.warning("Invalid vertex_ai default_format: %r", user_default_format)
+            logger.warning("Invalid vertex_ai default_format: {!r}", user_default_format)
 
     # 5. 内置默认格式
     return VERTEX_AI_DEFAULT_FORMAT

src/services/request/executor.py

@@ -95,8 +95,7 @@ class RequestExecutor:
             # 获取当前 RPM 计数用于计算负载
             # 注意：key 侧返回的是 RPM 计数（不会在请求结束时减少，靠 TTL 过期）
             try:
-                _, current_key_rpm = await self.concurrency_manager.get_current_concurrency(
-                    endpoint_id=endpoint.id,
+                current_key_rpm = await self.concurrency_manager.get_key_rpm_count(
                     key_id=key.id,
                 )
             except Exception as e:

src/services/request/result.py

@@ -8,7 +8,7 @@
 
 使用场景：
 - ProviderService 创建 RequestMetadata
-- FallbackOrchestrator 在异常时补充 RequestMetadata
+- TaskService 在异常时补充 RequestMetadata
 - ChatHandlerBase 使用 RequestResult 记录 Usage
 - ChatAdapterBase 使用 RequestResult 处理异常响应
 """

src/services/task/__init__.py

@@ -1,13 +1,32 @@
 """
-任务服务层（Phase2）
+任务服务层（Phase2/Phase3）
 
 统一任务框架相关的应用层入口：
-- 候选提交阶段：`services.candidate.CandidateService`
-- 终态结算：`services.task.application.TaskApplicationService`
+- 候选域能力：`services.candidate.CandidateService`（resolve/record 等）
+- 终态结算：`services.task.service.TaskService.finalize_video_task`
+- 统一门面：`services.task.service.TaskService`
 """
 
-from .application import TaskApplicationService
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:  # pragma: no cover
+    # NOTE: keep package import lightweight; avoid importing heavy modules on submodule imports
+    from .service import TaskService as TaskService
 
 __all__ = [
-    "TaskApplicationService",
+    "TaskService",
 ]
+
+
+def __getattr__(name: str) -> type:  # pragma: no cover
+    if name == "TaskService":
+        from .service import TaskService
+
+        return TaskService
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def __dir__() -> list[str]:  # pragma: no cover
+    return sorted(__all__)

src/services/task/application.py

@@ -1,312 +0,0 @@
-from __future__ import annotations
-
-from datetime import datetime, timezone
-from typing import Any
-
-from sqlalchemy.orm import Session
-
-from src.config.settings import config
-from src.core.logger import logger
-from src.models.database import ApiKey, Provider, Usage, User, VideoTask
-from src.services.billing.dimension_collector_service import DimensionCollectorService
-from src.services.billing.formula_engine import BillingIncompleteError, FormulaEngine
-from src.services.billing.rule_service import BillingRuleService
-from src.services.usage.service import UsageService
-
-
-class TaskApplicationService:
-    """
-    TaskApplicationService (Phase2)
-
-    当前仅先收敛"终态结算"入口，用于替代旧版 VideoTelemetry 直写 Usage 的流程。
-    后续将扩展 submit/cancel 并迁移候选编排逻辑。
-    """
-
-    def __init__(self, db: Session, *, redis_client: Any | None = None) -> None:
-        self.db = db
-        self.redis = redis_client
-
-    async def finalize_video_task(self, task: VideoTask) -> bool:
-        """
-        更新视频任务的计费信息（轮询完成后调用）。
-
-        异步任务的计费流程：
-        1. 提交成功时：Usage 已结算（billing_status='settled'，费用=0）
-        2. 轮询完成时：更新实际费用（成功则计费，失败则保持0）
-
-        返回 True 表示成功更新，False 表示无需更新（如已是最终状态）
-        """
-        request_id = getattr(task, "request_id", None) or task.id
-
-        existing = self.db.query(Usage).filter(Usage.request_id == request_id).first()
-        if not existing:
-            # Usage 不存在，尝试创建并结算（兜底逻辑）
-            logger.warning(
-                "Usage not found for video task, creating fallback: task_id=%s request_id=%s",
-                task.id,
-                request_id,
-            )
-            return await self._create_fallback_usage(task, request_id)
-
-        # 检查是否已有计费更新标记（避免重复计费）
-        metadata = existing.request_metadata or {}
-        if metadata.get("billing_updated_at"):
-            logger.debug(
-                "Video task billing already updated: task_id=%s request_id=%s",
-                task.id,
-                request_id,
-            )
-            return False
-
-        # 计算异步任务总耗时（ms）
-        response_time_ms: int | None = None
-        if task.submitted_at and task.completed_at:
-            delta = task.completed_at - task.submitted_at
-            response_time_ms = int(delta.total_seconds() * 1000)
-
-        # === 收集计费维度 ===
-        base_dimensions: dict[str, Any] = {
-            "duration_seconds": task.duration_seconds,
-            "resolution": task.resolution,
-            "aspect_ratio": task.aspect_ratio,
-            "size": task.size or "",
-            "retry_count": task.retry_count,
-        }
-
-        collector_metadata: dict[str, Any] = {
-            "task": {
-                "id": task.id,
-                "external_task_id": task.external_task_id,
-                "model": task.model,
-                "duration_seconds": task.duration_seconds,
-                "resolution": task.resolution,
-                "aspect_ratio": task.aspect_ratio,
-                "size": task.size,
-                "retry_count": task.retry_count,
-                "video_size_bytes": task.video_size_bytes,
-            },
-            "result": {
-                "video_url": task.video_url,
-                "video_urls": task.video_urls or [],
-            },
-        }
-
-        dims = DimensionCollectorService(self.db).collect_dimensions(
-            api_format=task.provider_api_format,
-            task_type="video",
-            request=task.original_request_body or {},
-            response=(
-                (task.request_metadata or {}).get("poll_raw_response")
-                if isinstance(task.request_metadata, dict)
-                else None
-            ),
-            metadata=collector_metadata,
-            base_dimensions=base_dimensions,
-        )
-
-        # === 计算成本（优先使用冻结的 billing_rule_snapshot）===
-        rule_snapshot = None
-        if isinstance(task.request_metadata, dict):
-            rule_snapshot = task.request_metadata.get("billing_rule_snapshot")
-
-        expression = None
-        variables: dict[str, Any] | None = None
-        dimension_mappings: dict[str, dict[str, Any]] | None = None
-        rule_id = None
-        rule_name = None
-        rule_scope = None
-
-        if isinstance(rule_snapshot, dict) and rule_snapshot.get("status") == "ok":
-            rule_id = rule_snapshot.get("rule_id")
-            rule_name = rule_snapshot.get("rule_name")
-            rule_scope = rule_snapshot.get("scope")
-            expression = rule_snapshot.get("expression")
-            variables = rule_snapshot.get("variables") or {}
-            dimension_mappings = rule_snapshot.get("dimension_mappings") or {}
-        else:
-            lookup = BillingRuleService.find_rule(
-                self.db,
-                provider_id=task.provider_id,
-                model_name=task.model,
-                task_type="video",
-            )
-            if lookup:
-                rule = lookup.rule
-                rule_id = rule.id
-                rule_name = rule.name
-                rule_scope = getattr(lookup, "scope", None)
-                expression = rule.expression
-                variables = rule.variables or {}
-                dimension_mappings = rule.dimension_mappings or {}
-
-        billing_snapshot: dict[str, Any] = {
-            "schema_version": "1.0",
-            "rule_id": str(rule_id) if rule_id else None,
-            "rule_name": str(rule_name) if rule_name else None,
-            "scope": str(rule_scope) if rule_scope else None,
-            "expression": str(expression) if expression else None,
-            "dimensions_used": dims,
-            "missing_required": [],
-            "cost": 0.0,
-            "status": "no_rule",
-            "calculated_at": datetime.now(timezone.utc).isoformat(),
-        }
-
-        cost = 0.0
-        # 只有任务成功时才计费
-        if task.status == "completed" and expression:
-            engine = FormulaEngine()
-            try:
-                result = engine.evaluate(
-                    expression=str(expression),
-                    variables=variables,
-                    dimensions=dims,
-                    dimension_mappings=dimension_mappings,
-                    strict_mode=config.billing_strict_mode,
-                )
-                billing_snapshot["status"] = result.status
-                billing_snapshot["missing_required"] = result.missing_required
-                if result.status == "complete":
-                    cost = float(result.cost)
-                billing_snapshot["cost"] = cost
-            except BillingIncompleteError as exc:
-                # strict_mode=true：标记任务失败并隐藏产物，避免"免费放行"
-                task.status = "failed"
-                task.error_code = "billing_incomplete"
-                task.error_message = f"Missing required dimensions: {exc.missing_required}"
-                task.video_url = None
-                task.video_urls = None
-                billing_snapshot["status"] = "incomplete"
-                billing_snapshot["missing_required"] = exc.missing_required
-                billing_snapshot["cost"] = 0.0
-            except Exception as exc:
-                billing_snapshot["status"] = "incomplete"
-                billing_snapshot["error"] = str(exc)
-                billing_snapshot["cost"] = 0.0
-
-        # 回写到 task.request_metadata 便于审计/重算
-        # 重新赋值整个字典，确保 SQLAlchemy 检测到变更
-        metadata = dict(task.request_metadata) if task.request_metadata else {}
-        metadata["billing_snapshot"] = billing_snapshot
-        task.request_metadata = metadata
-
-        # === 更新已结算的 Usage 计费信息 ===
-        updated = UsageService.update_settled_billing(
-            self.db,
-            request_id=request_id,
-            total_cost_usd=cost,
-            request_cost_usd=cost,
-            status="completed" if task.status == "completed" else "failed",
-            status_code=200 if task.status == "completed" else 500,
-            error_message=(
-                None
-                if task.status == "completed"
-                else (task.error_message or task.error_code or "video_task_failed")
-            ),
-            response_time_ms=response_time_ms,
-            billing_snapshot=billing_snapshot,
-            extra_metadata={
-                "dimensions": dims,
-                "raw_response_ref": {
-                    "video_task_id": task.id,
-                    "field": "video_tasks.request_metadata.poll_raw_response",
-                },
-            },
-        )
-
-        if updated:
-            logger.debug(
-                "Updated video task billing: task_id=%s request_id=%s cost=%.6f",
-                task.id,
-                request_id,
-                cost,
-            )
-        else:
-            logger.warning(
-                "Failed to update video task billing (may already be updated): "
-                "task_id=%s request_id=%s",
-                task.id,
-                request_id,
-            )
-
-        return updated
-
-    async def _create_fallback_usage(self, task: VideoTask, request_id: str) -> bool:
-        """
-        兜底逻辑：当 Usage 不存在时创建完整记录。
-        这种情况理论上不应发生（submit 阶段已创建），但保留以防万一。
-        """
-        user_obj = self.db.query(User).filter(User.id == task.user_id).first()
-        api_key_obj = (
-            self.db.query(ApiKey).filter(ApiKey.id == task.api_key_id).first()
-            if task.api_key_id
-            else None
-        )
-        provider_obj = (
-            self.db.query(Provider).filter(Provider.id == task.provider_id).first()
-            if task.provider_id
-            else None
-        )
-        provider_name = provider_obj.name if provider_obj else "unknown"
-
-        # 计算响应时间
-        response_time_ms: int | None = None
-        if task.submitted_at and task.completed_at:
-            delta = task.completed_at - task.submitted_at
-            response_time_ms = int(delta.total_seconds() * 1000)
-
-        try:
-            await UsageService.record_usage_with_custom_cost(
-                db=self.db,
-                user=user_obj,
-                api_key=api_key_obj,
-                provider=provider_name,
-                model=task.model,
-                request_type="video",
-                total_cost_usd=0.0,  # 兜底记录不计费
-                request_cost_usd=0.0,
-                input_tokens=0,
-                output_tokens=0,
-                cache_creation_input_tokens=0,
-                cache_read_input_tokens=0,
-                api_format=task.client_api_format,
-                endpoint_api_format=task.provider_api_format,
-                has_format_conversion=bool(task.format_converted),
-                is_stream=False,
-                response_time_ms=response_time_ms,
-                first_byte_time_ms=None,
-                status_code=200 if task.status == "completed" else 500,
-                error_message=(
-                    None
-                    if task.status == "completed"
-                    else (task.error_message or task.error_code or "video_task_failed")
-                ),
-                metadata={
-                    "fallback_created": True,
-                    "video_task_id": task.id,
-                },
-                request_headers=(
-                    (task.request_metadata or {}).get("request_headers")
-                    if isinstance(task.request_metadata, dict)
-                    else None
-                ),
-                request_body=task.original_request_body,
-                provider_request_headers=None,
-                response_headers=None,
-                client_response_headers=None,
-                response_body=None,
-                request_id=request_id,
-                provider_id=task.provider_id,
-                provider_endpoint_id=task.endpoint_id,
-                provider_api_key_id=task.key_id,
-                status="completed" if task.status == "completed" else "failed",
-                target_model=None,
-            )
-            return True
-        except Exception as exc:
-            logger.exception(
-                "Failed to create fallback usage for video task=%s: %s",
-                task.id,
-                str(exc),
-            )
-            return False

src/services/task/exceptions.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+
+class StreamProbeError(RuntimeError):
+    """Streaming probe failed before first chunk (eligible for failover)."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        http_status: int,
+        original_exception: Exception | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.http_status = http_status
+        self.original_exception = original_exception
+
+
+class TaskNotFoundError(LookupError):
+    """Task not found (by internal id or external id)."""
+
+    def __init__(self, task_id: str) -> None:
+        super().__init__(f"Task not found: {task_id}")
+        self.task_id = task_id

src/services/task/impl/video_poller.py

@@ -35,7 +35,7 @@ from src.core.crypto import crypto_service
 from src.core.logger import logger
 from src.database import create_session
 from src.models.database import ProviderAPIKey, ProviderEndpoint, VideoTask
-from src.services.task.application import TaskApplicationService
+from src.services.task.service import TaskService
 
 
 @dataclass(slots=True)
@@ -164,7 +164,7 @@ class VideoTaskPollerAdapter:
         try:
             upstream_key = crypto_service.decrypt(key.api_key)
         except Exception:
-            logger.warning("Failed to decrypt provider key for task %s", task.id)
+            logger.warning("Failed to decrypt provider key for task {}", task.id)
             return InternalVideoPollResult(
                 status=VideoStatus.FAILED,
                 error_code="decryption_error",
@@ -241,7 +241,7 @@ class VideoTaskPollerAdapter:
         with create_session() as db:
             task = db.get(VideoTask, task_id)
             if not task:
-                logger.warning("Task %s disappeared during poll update", task_id)
+                logger.warning("Task {} disappeared during poll update", task_id)
                 return
 
             if error_exception is not None and ctx is not None:
@@ -284,12 +284,10 @@ class VideoTaskPollerAdapter:
             # 终态结算
             if task.status in (VideoStatus.COMPLETED.value, VideoStatus.FAILED.value):
                 try:
-                    await TaskApplicationService(db, redis_client=redis_client).finalize_video_task(
-                        task
-                    )
+                    await TaskService(db, redis_client=redis_client).finalize_video_task(task)
                 except Exception as exc:
                     logger.exception(
-                        "Failed to record video usage for task=%s: %s",
+                        "Failed to record video usage for task={}: {}",
                         task.id,
                         sanitize_error_message(str(exc)),
                     )
@@ -300,7 +298,7 @@ class VideoTaskPollerAdapter:
         """处理轮询错误"""
         task.poll_count += 1
         error_msg = sanitize_error_message(str(exc))
-        logger.warning("Poll error for task %s: %s", task.id, error_msg)
+        logger.warning("Poll error for task {}: {}", task.id, error_msg)
         task.progress_message = f"Poll error: {error_msg}"
 
         status_code = exc.status_code if isinstance(exc, PollHTTPError) else None
@@ -337,7 +335,7 @@ class VideoTaskPollerAdapter:
         url = self._build_gemini_url(ctx.base_url, operation_name)
 
         logger.debug(
-            "[VideoPoller] Gemini poll: task=%s external_id=%s url=%s",
+            "[VideoPoller] Gemini poll: task={} external_id={} url={}",
             ctx.task_id,
             ctx.external_task_id,
             url,
@@ -347,7 +345,7 @@ class VideoTaskPollerAdapter:
         response = await client.get(url, headers=ctx.headers)
         if response.status_code >= 400:
             logger.warning(
-                "[VideoPoller] Gemini poll failed: task=%s status=%s response=%s",
+                "[VideoPoller] Gemini poll failed: task={} status={} response={}",
                 ctx.task_id,
                 response.status_code,
                 response.text[:500] if response.text else "(empty)",
@@ -394,7 +392,7 @@ class VideoTaskPollerAdapter:
         except Exception as exc:
             task.poll_count += 1
             error_msg = sanitize_error_message(str(exc))
-            logger.warning("Poll error for task %s: %s", task.id, error_msg)
+            logger.warning("Poll error for task {}: {}", task.id, error_msg)
             task.progress_message = f"Poll error: {error_msg}"
 
             status_code = exc.status_code if isinstance(exc, PollHTTPError) else None
@@ -427,12 +425,10 @@ class VideoTaskPollerAdapter:
         # 终态结算
         if task.status in (VideoStatus.COMPLETED.value, VideoStatus.FAILED.value):
             try:
-                await TaskApplicationService(db, redis_client=redis_client).finalize_video_task(
-                    task
-                )
+                await TaskService(db, redis_client=redis_client).finalize_video_task(task)
             except Exception as exc:
                 logger.exception(
-                    "Failed to record video usage for task=%s: %s",
+                    "Failed to record video usage for task={}: {}",
                     task.id,
                     sanitize_error_message(str(exc)),
                 )
@@ -470,7 +466,7 @@ class VideoTaskPollerAdapter:
         try:
             upstream_key = crypto_service.decrypt(key.api_key)
         except Exception:
-            logger.warning("Failed to decrypt provider key for task %s", task.id)
+            logger.warning("Failed to decrypt provider key for task {}", task.id)
             return InternalVideoPollResult(
                 status=VideoStatus.FAILED,
                 error_code="decryption_error",
@@ -539,7 +535,7 @@ class VideoTaskPollerAdapter:
         headers = self._build_headers(endpoint_sig, upstream_key, endpoint, auth_info)
 
         logger.debug(
-            "[VideoPoller] Gemini poll: task=%s external_id=%s url=%s",
+            "[VideoPoller] Gemini poll: task={} external_id={} url={}",
             task.id,
             task.external_task_id,
             url,
@@ -549,7 +545,7 @@ class VideoTaskPollerAdapter:
         response = await client.get(url, headers=headers)
         if response.status_code >= 400:
             logger.warning(
-                "[VideoPoller] Gemini poll failed: task=%s status=%s response=%s",
+                "[VideoPoller] Gemini poll failed: task={} status={} response={}",
                 task.id,
                 response.status_code,
                 response.text[:500] if response.text else "(empty)",

src/services/task/protocol.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, AsyncIterator, Protocol, runtime_checkable
+
+import httpx
+
+from src.services.cache.aware_scheduler import ProviderCandidate
+
+
+class AttemptKind(str, Enum):
+    """`attempt_func` return kind."""
+
+    SYNC_RESPONSE = "sync_response"
+    STREAM = "stream"
+    ASYNC_SUBMIT = "async_submit"
+
+
+@dataclass(slots=True)
+class AttemptResult:
+    """
+    Unified attempt result returned by `AttemptFunc`.
+
+    Notes:
+    - `http_status` / `http_headers` MUST be filled for all kinds (for audit/classification).
+    - The payload fields are filled depending on `kind`.
+    """
+
+    kind: AttemptKind
+
+    # HTTP meta (always filled)
+    http_status: int
+    http_headers: dict[str, str]
+
+    # SYNC_RESPONSE
+    response_body: Any = None
+
+    # STREAM
+    stream_iterator: AsyncIterator[bytes] | None = None
+
+    # ASYNC_SUBMIT
+    provider_task_id: str | None = None
+
+    # Raw response reference (optional, for audit/debugging)
+    raw_response: httpx.Response | None = None
+
+
+@runtime_checkable
+class AttemptFunc(Protocol):
+    async def __call__(self, candidate: ProviderCandidate) -> AttemptResult: ...

src/services/task/schema.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from src.services.cache.aware_scheduler import ProviderCandidate
+from src.services.candidate.schema import CandidateKey
+
+from .protocol import AttemptKind, AttemptResult
+
+
+@dataclass(slots=True)
+class ExecutionResult:
+    """FailoverEngine.execute() unified result."""
+
+    success: bool
+
+    # payload (filled based on AttemptKind)
+    attempt_result: AttemptResult | None = None
+
+    # selected candidate
+    candidate: ProviderCandidate | None = None
+    candidate_index: int = -1
+    retry_index: int = 0
+
+    provider_id: str | None = None
+    provider_name: str | None = None
+    endpoint_id: str | None = None
+    key_id: str | None = None
+
+    # audit
+    candidate_keys: list[CandidateKey] = field(default_factory=list)
+    attempt_count: int = 0
+    request_candidate_id: str | None = None
+
+    # failure
+    error_type: str | None = None
+    error_message: str | None = None
+    last_status_code: int | None = None
+
+    @property
+    def response(self) -> Any:
+        """Compatibility accessor: returns response body or stream iterator."""
+        if not self.attempt_result:
+            return None
+        if self.attempt_result.kind == AttemptKind.STREAM:
+            return self.attempt_result.stream_iterator
+        return self.attempt_result.response_body
+
+    @property
+    def provider_task_id(self) -> str | None:
+        if self.attempt_result and self.attempt_result.kind == AttemptKind.ASYNC_SUBMIT:
+            return self.attempt_result.provider_task_id
+        return None
+
+
+@dataclass(slots=True)
+class TaskStatusResult:
+    """Generic task status payload returned by TaskService.poll()."""
+
+    task_id: str
+    status: str
+
+    progress_percent: int | None = None
+    result_url: str | None = None
+    error_message: str | None = None
+
+    # optional metadata (best-effort)
+    provider_id: str | None = None
+    provider_name: str | None = None
+    endpoint_id: str | None = None
+    key_id: str | None = None

src/services/task/task_poller.py

@@ -133,7 +133,7 @@ class TaskPollerService:
                                 task_obj = self.adapter.get_task(task_db, task_id)
                                 if not task_obj:
                                     logger.warning(
-                                        "[%s] Task %s disappeared during poll",
+                                        "[{}] Task {} disappeared during poll",
                                         self.adapter.task_type,
                                         task_id,
                                     )
@@ -181,7 +181,7 @@ class TaskPollerService:
                             poll_results.append(True)
                         except Exception as exc:
                             logger.exception(
-                                "[%s] Unexpected error polling task %s: %s",
+                                "[{}] Unexpected error polling task {}: {}",
                                 self.adapter.task_type,
                                 task_id,
                                 self.adapter.sanitize_error_message(str(exc)),
@@ -200,7 +200,7 @@ class TaskPollerService:
                         >= self.adapter.consecutive_failure_alert_threshold
                     ):
                         logger.error(
-                            "[ALERT] %s poller: %d consecutive batches failed.",
+                            "[ALERT] {} poller: {} consecutive batches failed.",
                             self.adapter.task_type,
                             self._consecutive_failures,
                         )

src/services/usage/consumer_streams.py

@@ -493,14 +493,18 @@ class UsageQueueConsumer:
             return
         self._last_metrics_log = now
         try:
-            stream_len = await redis_client.xlen(self._stream_key)
-            pending = await redis_client.xpending(self._stream_key, self._stream_group)
+            # 使用 XINFO GROUPS 获取更准确的 lag（未处理消息数）
+            groups_info = await redis_client.xinfo_groups(self._stream_key)
+            lag = 0
             pending_count = 0
-            if isinstance(pending, dict):
-                pending_count = int(pending.get("pending", 0))
-            elif isinstance(pending, (list, tuple)) and pending:
-                pending_count = int(pending[0])
-            logger.info(f"[usage-queue] backlog={stream_len} pending={pending_count}")
+            for group in groups_info:
+                if isinstance(group, dict) and group.get("name") == self._stream_group:
+                    lag = group.get("lag", 0) or 0
+                    pending_count = group.get("pending", 0) or 0
+                    break
+            # lag=未读消息数, pending=已读但未ACK的消息数
+            if lag > 0 or pending_count > 0:
+                logger.info(f"[usage-queue] lag={lag} pending={pending_count}")
         except Exception as exc:
             logger.debug(f"[usage-queue] metrics log failed: {exc}")
 

src/services/usage/service.py

@@ -1325,7 +1325,7 @@ class UsageService:
             # 避免重复记账：若已结算/作废，直接返回（防止并发重复加计数）
             if getattr(existing_usage, "billing_status", None) in ("settled", "void"):
                 logger.debug(
-                    "record_usage_with_custom_cost: request_id=%s already finalized (billing_status=%s), skip",
+                    "record_usage_with_custom_cost: request_id={} already finalized (billing_status={}), skip",
                     request_id,
                     getattr(existing_usage, "billing_status", None),
                 )
@@ -1599,7 +1599,7 @@ class UsageService:
                         update_params_list.append((record, request_id, params))
                     except Exception as e:
                         skipped_count += 1
-                        logger.warning("批量记录中参数构建失败: %s, request_id=%s", e, request_id)
+                        logger.warning("批量记录中参数构建失败: {}, request_id={}", e, request_id)
 
         insert_params_list: list[tuple[dict[str, Any], str, UsageRecordParams]] = []
         for record in records_to_insert:
@@ -1609,7 +1609,7 @@ class UsageService:
                 insert_params_list.append((record, request_id, params))
             except Exception as e:
                 skipped_count += 1
-                logger.warning("批量记录中参数构建失败: %s, request_id=%s", e, request_id)
+                logger.warning("批量记录中参数构建失败: {}, request_id={}", e, request_id)
 
         # 并行准备所有记录（性能优化）
         all_params = [p for _, _, p in update_params_list] + [p for _, _, p in insert_params_list]
@@ -1659,7 +1659,7 @@ class UsageService:
 
             except Exception as e:
                 skipped_count += 1
-                logger.warning("批量记录中更新失败: %s, request_id=%s", e, request_id)
+                logger.warning("批量记录中更新失败: {}, request_id={}", e, request_id)
                 continue
 
         # 2. 处理需要新建的记录
@@ -1699,7 +1699,7 @@ class UsageService:
 
             except Exception as e:
                 skipped_count += 1
-                logger.warning("批量记录中跳过无效记录: %s, request_id=%s", e, request_id)
+                logger.warning("批量记录中跳过无效记录: {}, request_id={}", e, request_id)
                 continue
 
         # 统计跳过的记录，失败率超过 10% 时提升日志级别
@@ -1707,13 +1707,13 @@ class UsageService:
             skip_ratio = skipped_count / total_count if total_count > 0 else 0
             if skip_ratio > 0.1:
                 logger.error(
-                    "批量记录失败率过高: %d/%d (%.1f%%) 条记录被跳过",
+                    "批量记录失败率过高: {}/{} ({:.1f}%) 条记录被跳过",
                     skipped_count,
                     total_count,
                     skip_ratio * 100,
                 )
             else:
-                logger.warning("批量记录部分失败: %d/%d 条记录被跳过", skipped_count, total_count)
+                logger.warning("批量记录部分失败: {}/{} 条记录被跳过", skipped_count, total_count)
 
         # 批量更新 GlobalModel 使用计数
         for model_name, count in model_counts.items():

src/services/usage/stream.py

@@ -552,7 +552,7 @@ class StreamUsageTracker:
                 logger.error(f"ID:{self.request_id} | {error_msg}")
                 # 设置错误状态，避免被记录为成功
                 self.set_error_status(502, error_msg)
-                # 抛出异常让 FallbackOrchestrator 捕获并触发故障转移
+                # 抛出异常让 TaskService/FailoverEngine 捕获并触发故障转移
                 raise EmptyStreamException(
                     provider_name=self.provider,
                     chunk_count=chunk_count,
@@ -797,7 +797,7 @@ class StreamUsageTracker:
             )
 
             # 记录提供商结果用于动态权重调整
-            # 记录提供商结果的健康监控已由 FallbackOrchestrator 自动处理
+            # 健康监控/自适应调整已由 RequestExecutor/ErrorClassifier 处理
             # 这里不再需要手动记录
         except Exception as e:
             logger.exception(f"Failed to record stream usage: {e}")
@@ -1032,7 +1032,7 @@ class EnhancedStreamUsageTracker(StreamUsageTracker):
                 logger.error(f"ID:{self.request_id} | {error_msg}")
                 # 设置错误状态，避免被记录为成功
                 self.set_error_status(502, error_msg)
-                # 抛出异常让 FallbackOrchestrator 捕获并触发故障转移
+                # 抛出异常让 TaskService/FailoverEngine 捕获并触发故障转移
                 raise EmptyStreamException(
                     provider_name=self.provider,
                     chunk_count=chunk_count,