refactor: 重构异步任务系统和计费服务架构

- 重构任务系统：新增 lifecycle (TaskStatus/BillingStatus)、context、application 模块
- 将 video tasks 泛化为 async tasks，支持更通用的异步任务管理
- 新增 Gemini Files 管理模块和管理界面
- 重构 billing 服务：拆分 schema.py 和 service.py
- 新增 candidate 服务模块用于请求候选管理
- 数据库迁移：添加 billing_status、request_id、gemini_file_mappings 表和索引
- 移除废弃的 video_telemetry、task orchestrator 等模块

src/services/candidate/__init__.py

@@ -0,0 +1,28 @@
+"""
+Candidate domain (Phase2)
+
+This package centralizes:
+- candidate resolving (Provider/Endpoint/Key combinations)
+- request_candidates recording & audit
+- failover execution policies
+"""
+
+from src.services.candidate.policy import RetryMode, RetryPolicy, SkipPolicy
+from src.services.candidate.schema import (
+    CANDIDATE_KEY_SCHEMA_VERSION,
+    CandidateKey,
+    CandidateResult,
+)
+from src.services.candidate.service import CandidateService
+
+__all__ = [
+    "CandidateService",
+    # schema
+    "CANDIDATE_KEY_SCHEMA_VERSION",
+    "CandidateKey",
+    "CandidateResult",
+    # policies
+    "RetryMode",
+    "RetryPolicy",
+    "SkipPolicy",
+]

src/services/candidate/failover.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from typing import Any, Protocol
+
+from sqlalchemy.orm import Session
+
+from src.services.cache.aware_scheduler import ProviderCandidate
+
+from .policy import RetryPolicy, SkipPolicy
+from .schema import CandidateKey, CandidateResult
+
+
+class AttemptFunc(Protocol):
+    async def __call__(self, candidate: ProviderCandidate) -> Any: ...
+
+
+class FailoverEngine:
+    """
+    FailoverEngine executes candidate attempts under policies.
+
+    Phase2 scaffolding: implementation will gradually replace legacy orchestrators.
+    """
+
+    def __init__(self, db: Session):
+        self.db = db
+
+    async def execute(
+        self,
+        *,
+        candidates: list[ProviderCandidate],
+        attempt_func: AttemptFunc,
+        retry_policy: RetryPolicy,
+        skip_policy: SkipPolicy,
+        request_id: 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] = []
+        for idx, cand in enumerate(candidates):
+            candidate_keys.append(
+                CandidateKey(
+                    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)),
+                    status="available",
+                )
+            )
+
+        raise NotImplementedError("FailoverEngine.execute is not implemented yet")

src/services/candidate/policy.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+
+
+class RetryMode(str, Enum):
+    """Retry mode for candidate attempts."""
+
+    PRE_EXPAND = "pre_expand"  # pre-create retry slots (sync)
+    ON_DEMAND = "on_demand"  # create retry record only when retry happens
+    DISABLED = "disabled"  # no retry
+
+
+@dataclass(frozen=True)
+class RetryPolicy:
+    """Unified retry policy."""
+
+    mode: RetryMode = RetryMode.DISABLED
+    max_retries: int = 1
+    retry_on_cached_only: bool = True
+
+    @classmethod
+    def for_sync_task(cls) -> "RetryPolicy":
+        return cls(mode=RetryMode.PRE_EXPAND, max_retries=2)
+
+    @classmethod
+    def for_async_task(cls) -> "RetryPolicy":
+        return cls(mode=RetryMode.DISABLED, max_retries=1)
+
+    @classmethod
+    def for_async_submit_with_retry(cls) -> "RetryPolicy":
+        return cls(mode=RetryMode.ON_DEMAND, max_retries=2)
+
+
+@dataclass(frozen=True)
+class SkipPolicy:
+    """Rules for skipping unsupported candidates."""
+
+    allow_format_conversion: bool = True
+    supported_auth_types: set[str] | None = None

src/services/candidate/recorder.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from src.models.database import RequestCandidate
+
+from .schema import CandidateKey
+
+
+class CandidateRecorder:
+    """Read helpers for RequestCandidate audit data."""
+
+    def __init__(self, db: Session):
+        self.db = db
+
+    def get_candidate_keys(self, request_id: str) -> list[CandidateKey]:
+        rows: list[RequestCandidate] = (
+            self.db.query(RequestCandidate)
+            .filter(RequestCandidate.request_id == request_id)
+            .order_by(RequestCandidate.candidate_index.asc(), RequestCandidate.retry_index.asc())
+            .all()
+        )
+
+        result: list[CandidateKey] = []
+        for row in rows:
+            provider_name = None
+            if getattr(row, "provider", None) is not None:
+                provider_name = getattr(row.provider, "name", None)
+
+            key_name = None
+            auth_type = None
+            priority = None
+            if getattr(row, "key", None) is not None:
+                key_name = getattr(row.key, "name", None)
+                auth_type = getattr(row.key, "auth_type", None)
+                priority = getattr(row.key, "priority", None)
+
+            result.append(
+                CandidateKey(
+                    candidate_index=int(row.candidate_index or 0),
+                    retry_index=int(row.retry_index or 0),
+                    provider_id=str(row.provider_id) if row.provider_id else None,
+                    provider_name=str(provider_name) if provider_name else None,
+                    endpoint_id=str(row.endpoint_id) if row.endpoint_id else None,
+                    key_id=str(row.key_id) if row.key_id else None,
+                    key_name=str(key_name) if key_name else None,
+                    auth_type=str(auth_type) if auth_type else None,
+                    priority=int(priority) if priority is not None else None,
+                    is_cached=bool(getattr(row, "is_cached", False)),
+                    status=str(getattr(row, "status", "") or "pending"),
+                    skip_reason=getattr(row, "skip_reason", None),
+                    error_type=getattr(row, "error_type", None),
+                    error_message=getattr(row, "error_message", None),
+                    status_code=getattr(row, "status_code", None),
+                    latency_ms=getattr(row, "latency_ms", None),
+                )
+            )
+
+        return result

src/services/candidate/resolver.py

@@ -0,0 +1,10 @@
+"""
+CandidateResolver facade import.
+
+Phase2 keeps the implementation in `services/orchestration/` for compatibility,
+and gradually migrates it into this package.
+"""
+
+from src.services.orchestration.candidate_resolver import CandidateResolver
+
+__all__ = ["CandidateResolver"]

src/services/candidate/schema.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from src.services.cache.aware_scheduler import ProviderCandidate
+
+CANDIDATE_KEY_SCHEMA_VERSION = "1.0"
+
+
+@dataclass(frozen=True)
+class CandidateKey:
+    """Stable candidate key snapshot for audit."""
+
+    schema_version: str = CANDIDATE_KEY_SCHEMA_VERSION
+
+    candidate_index: int = 0
+    retry_index: int = 0
+
+    provider_id: str | None = None
+    provider_name: str | None = None
+    endpoint_id: str | None = None
+    key_id: str | None = None
+    key_name: str | None = None
+    auth_type: str | None = None
+    priority: int | None = None
+    is_cached: bool = False
+
+    status: str = "pending"  # pending/success/failed/skipped/available/...
+    skip_reason: str | None = None
+    error_type: str | None = None
+    error_message: str | None = None
+    status_code: int | None = None
+    latency_ms: int | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        data: dict[str, Any] = {
+            "schema_version": self.schema_version,
+            "candidate_index": self.candidate_index,
+            "retry_index": self.retry_index,
+            "provider_id": self.provider_id,
+            "provider_name": self.provider_name,
+            "endpoint_id": self.endpoint_id,
+            "key_id": self.key_id,
+            "key_name": self.key_name,
+            "auth_type": self.auth_type,
+            "priority": self.priority,
+            "is_cached": self.is_cached,
+            "status": self.status,
+            "skip_reason": self.skip_reason,
+            "error_type": self.error_type,
+            "error_message": self.error_message,
+            "status_code": self.status_code,
+            "latency_ms": self.latency_ms,
+        }
+        # drop Nones for compact audit payload
+        return {k: v for k, v in data.items() if v is not None}
+
+
+@dataclass(slots=True)
+class CandidateResult:
+    """Failover execution result."""
+
+    success: bool
+    selected: ProviderCandidate | None
+    selected_index: int | None
+    candidate_keys: list[CandidateKey]
+
+    external_task_id: str | None = None
+    error: Exception | None = None
+    last_status_code: int | None = None

src/services/candidate/service.py

@@ -0,0 +1,515 @@
+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.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
+
+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:
+    """
+    CandidateService (Facade).
+
+    Phase2 note: this is introduced as a new domain entrypoint. Legacy orchestrators
+    still exist and will be migrated gradually to use this service.
+    """
+
+    def __init__(self, db: Session, redis_client: Any | None = None) -> None:
+        self.db = db
+        self.redis = redis_client
+        self._cache_scheduler = None
+        self._resolver: CandidateResolver | None = None
+        self._error_classifier: ErrorClassifier | None = None
+        self._recorder = CandidateRecorder(db)
+
+    async def _ensure_initialized(self) -> None:
+        if self._cache_scheduler is not None:
+            return
+
+        priority_mode = SystemConfigService.get_config(
+            self.db,
+            "provider_priority_mode",
+            "provider",
+        )
+        scheduling_mode = SystemConfigService.get_config(
+            self.db,
+            "scheduling_mode",
+            "cache_affinity",
+        )
+        self._cache_scheduler = await get_cache_aware_scheduler(
+            self.redis,
+            priority_mode=priority_mode,
+            scheduling_mode=scheduling_mode,
+        )
+        self._resolver = CandidateResolver(db=self.db, cache_scheduler=self._cache_scheduler)
+        self._error_classifier = ErrorClassifier(db=self.db, cache_scheduler=self._cache_scheduler)
+
+    async def resolve(
+        self,
+        *,
+        api_format: str,
+        model_name: str,
+        affinity_key: str,
+        user_api_key: ApiKey | None = None,
+        request_id: str | None = None,
+        is_stream: bool = False,
+        capability_requirements: dict[str, bool] | None = None,
+        preferred_key_ids: list[str] | None = None,
+    ) -> tuple[list[ProviderCandidate], str]:
+        await self._ensure_initialized()
+        assert self._resolver is not None
+        return 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=is_stream,
+            capability_requirements=capability_requirements,
+            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,
+        *,
+        candidates: list[ProviderCandidate],
+        request_id: str,
+        user_api_key: ApiKey,
+        required_capabilities: dict[str, bool] | None = None,
+        expand_retries: bool = True,
+    ) -> dict[tuple[int, int], str]:
+        # CandidateResolver.create_candidate_records is currently the canonical implementation
+        assert self._resolver is not None, "Call resolve() once before create_candidate_records()"
+        return self._resolver.create_candidate_records(
+            all_candidates=candidates,
+            request_id=request_id,
+            user_id=str(user_api_key.user_id),
+            user_api_key=user_api_key,
+            required_capabilities=required_capabilities,
+            expand_retries=expand_retries,
+        )
+
+    def get_candidate_keys(self, request_id: str) -> list["CandidateKey"]:
+        return self._recorder.get_candidate_keys(request_id)

src/services/candidate/submit.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Protocol, runtime_checkable
+
+import httpx
+
+from src.services.billing.rule_service import BillingRuleLookupResult
+from src.services.cache.aware_scheduler import ProviderCandidate
+
+
+@runtime_checkable
+class SubmitFunc(Protocol):
+    async def __call__(self, candidate: ProviderCandidate) -> httpx.Response: ...
+
+
+@runtime_checkable
+class ExtractExternalTaskIdFunc(Protocol):
+    def __call__(self, payload: dict[str, Any]) -> str | None: ...
+
+
+class UpstreamClientRequestError(RuntimeError):
+    """可判定为客户端请求问题（不应 failover）的上游错误。"""
+
+    def __init__(
+        self,
+        *,
+        response: httpx.Response,
+        candidate_keys: list[dict[str, Any]],
+    ) -> None:
+        self.response = response
+        self.candidate_keys = candidate_keys
+        super().__init__(f"Upstream client error: HTTP {response.status_code}")
+
+
+class AllCandidatesFailedError(RuntimeError):
+    def __init__(
+        self,
+        *,
+        reason: str,
+        candidate_keys: list[dict[str, Any]],
+        last_status_code: int | None = None,
+    ) -> None:
+        self.reason = reason
+        self.candidate_keys = candidate_keys
+        self.last_status_code = last_status_code
+        super().__init__(f"All candidates failed: {reason}")
+
+
+class CandidateUnsupportedError(RuntimeError):
+    """候选不被当前任务支持（如 auth_type/格式转换需求不支持）。"""
+
+
+class CandidateSubmissionError(RuntimeError):
+    """候选提交异常（网络/解密/解析等）。"""
+
+
+@dataclass(slots=True)
+class SubmitOutcome:
+    candidate: ProviderCandidate
+    candidate_keys: list[dict[str, Any]]
+    external_task_id: str
+    rule_lookup: BillingRuleLookupResult | None
+    upstream_payload: dict[str, Any] | None = None
+    upstream_headers: dict[str, str] | None = None
+    upstream_status_code: int | None = None