codex-proxy-rs/backend/crates/gateway-core/src/engine/execution.rs

//! 数据面执行用例：认证、准入、路由、continuation、circuit 与会话生命周期。

use std::collections::BTreeSet;
use std::fmt;
use std::net::IpAddr;
use std::num::NonZeroU32;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};

use futures::{FutureExt, future::BoxFuture, pin_mut, select_biased};
use futures_timer::Delay;
use uuid::Uuid;

use crate::concurrency::{CapacityWait, ConcurrencyWaitBudget, ConcurrencyWaitQueue};
use crate::engine::admission::{
    ClientAdmissionDecision, ClientAdmissionPort, ClientAdmissionRejection, ClientAdmissionRequest,
};
use crate::engine::budget::{ClientBudgetCharge, ClientBudgetPort};
use crate::engine::continuation::{
    ContinuationBinding, NativeContinuationPin, NativeContinuationPort,
    NativeContinuationStoreErrorKind, PreviousResponseId,
};
use crate::engine::coordinator::ResponseExecutionSession;
use crate::engine::probe::{
    AccountProbe, AccountProbeError, AccountProbeErrorSource, AccountProbeRequest,
    AccountProbeResult, AccountProbeUpstreamResponse,
};
use crate::engine::provider::ProviderRegistry;
use crate::engine::{
    AttemptCoordinator, AttemptRecord, CoordinatedEvent, EngineError, ExecutionStore,
    GatewayEngine, IntermediateFailure, ModelRequestFinalization, ModelRequestId, NewModelRequest,
    ProbeFailure, ProviderAccountId, ProviderAttemptOutcome, RecoveryReport, UpstreamSendState,
};
use crate::error::{GatewayError, GatewayErrorKind, ProviderErrorKind, StoreError};
use crate::event::{GatewayEvent, ProviderEvent, ProviderResponseHeader};
use crate::identity::ProviderKind;
use crate::lifecycle::CancellationToken;
use crate::operation::{Operation, ProviderSessionState};
use crate::policy::{ClientApiKeyId, ClientPolicy};
use crate::routing::{
    ProviderCatalogUnavailable, PublicModelDescriptor, PublicModelId, RoutingContext,
    RuntimeSnapshot, UpstreamModelId,
};
use crate::runtime::RuntimeSnapshotHandle;

const MODEL_REQUEST_DEADLINE: Duration = Duration::from_secs(10 * 60);
const COORDINATION_TIMEOUT: Duration = Duration::from_millis(100);

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ClientTransport {
    HttpJson,
    HttpSse,
    WebSocket,
    InternalProbe,
}

impl ClientTransport {
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::HttpJson => "http_json",
            Self::HttpSse => "http_sse",
            Self::WebSocket => "websocket",
            Self::InternalProbe => "internal",
        }
    }
}

/// API 解码后交给 Core 的稳定请求元数据。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExecutionRequestMetadata {
    pub protocol: String,
    pub endpoint: String,
    pub transport: ClientTransport,
    pub stream: bool,
    pub client_ip: Option<IpAddr>,
    pub user_agent: Option<String>,
    pub previous_response_id: Option<PreviousResponseId>,
}

#[derive(Clone)]
pub struct AuthenticatedClient {
    snapshot: Arc<RuntimeSnapshot>,
    policy: ClientPolicy,
}

impl AuthenticatedClient {
    #[must_use]
    pub const fn snapshot(&self) -> &Arc<RuntimeSnapshot> {
        &self.snapshot
    }

    #[must_use]
    pub const fn policy(&self) -> &ClientPolicy {
        &self.policy
    }
}

impl fmt::Debug for AuthenticatedClient {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("AuthenticatedClient")
            .field("key_id", &self.policy.key_id())
            .field("revision", &self.snapshot.revision())
            .finish_non_exhaustive()
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum ClientAuthenticationError {
    #[error("client API key is invalid")]
    InvalidKey,
    #[error("runtime snapshot is unavailable")]
    SnapshotUnavailable,
}

pub struct StartExecution {
    pub client: AuthenticatedClient,
    pub public_model: PublicModelId,
    pub operation: Operation,
    pub metadata: ExecutionRequestMetadata,
}

/// 启动一个由协议 adapter 明确绑定到 Provider 自有端点的请求。
pub struct StartProviderExecution {
    pub client: AuthenticatedClient,
    pub provider: ProviderKind,
    pub operation: Operation,
    pub metadata: ExecutionRequestMetadata,
}

enum ExecutionTarget {
    Model(PublicModelId),
    ProviderEndpoint(ProviderKind),
}

impl ExecutionTarget {
    fn into_public_model(self) -> Option<PublicModelId> {
        match self {
            Self::Model(model) => Some(model),
            Self::ProviderEndpoint(_) => None,
        }
    }
}

struct PendingStartExecution {
    client: AuthenticatedClient,
    target: ExecutionTarget,
    operation: Operation,
    metadata: ExecutionRequestMetadata,
}

pub struct StartedExecution {
    pub request_id: ModelRequestId,
    pub created_at: SystemTime,
    pub stream: bool,
    pub session: Box<dyn ExecutionSession>,
}

pub trait ExecutionSession: Send {
    fn trace(&self) -> crate::diagnostics::TraceContext {
        crate::diagnostics::TraceContext::default()
    }
    fn next_event(&mut self) -> BoxFuture<'_, Result<Option<CoordinatedEvent>, EngineError>>;
    fn collect_uncommitted(&mut self) -> BoxFuture<'_, Result<Vec<ProviderEvent>, EngineError>>;
    fn response_headers(&self) -> &[ProviderResponseHeader];
    fn response_status_code(&self) -> Option<u16> {
        None
    }
    fn commit_downstream(
        &mut self,
        client_status_code: Option<u16>,
    ) -> BoxFuture<'_, Result<(), EngineError>>;
    fn record_client_status(
        &mut self,
        client_status_code: u16,
    ) -> BoxFuture<'_, Result<(), EngineError>>;
    /// 执行已终结且结算、准入释放均已返回，协议层才可以放弃清理责任。
    /// 结算失败仍由 Store 保留费用重试，释放失败仍按租约 TTL 收敛。
    fn is_finalized(&self) -> bool;
    fn cancel(&self);
    /// 将会话交给宿主持续驱动清理；取消请求事件的等待不会丢弃已开始的结算。
    fn detach_finalize(self: Box<Self>) -> BoxFuture<'static, ()>;
}

pub trait ExecutionService: Send + Sync {
    fn authenticate(
        &self,
        plaintext: &str,
    ) -> Result<AuthenticatedClient, ClientAuthenticationError>;
    fn public_models(&self, client: &AuthenticatedClient) -> Vec<PublicModelId>;
    fn client_model_catalog<'a>(
        &'a self,
        _client: &'a AuthenticatedClient,
        _protocol: &'a str,
        _client_version: &'a str,
    ) -> BoxFuture<'a, Result<Vec<PublicModelDescriptor>, ProviderCatalogUnavailable>> {
        Box::pin(async { Err(ProviderCatalogUnavailable) })
    }
    fn contains_public_model(&self, client: &AuthenticatedClient, model: &PublicModelId) -> bool;
    fn start(
        &self,
        request: StartExecution,
    ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>>;
    fn start_provider_endpoint(
        &self,
        request: StartProviderExecution,
    ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>>;
}

/// 只验证 Client API Key 并返回稳定 Key ID，不产生请求使用事实。
///
/// 管理侧登录等非数据面场景必须使用该端口，避免把认证本身误记为一次 Key 使用。
pub trait ClientKeyVerifier: Send + Sync {
    fn verify_client_key(
        &self,
        plaintext: &str,
    ) -> Result<ClientApiKeyId, ClientAuthenticationError>;
}

/// 成功认证后的 API Key 使用事实接收器。
///
/// 认证仍是同步快照读取；实现必须自行异步、去重地持久化，不得阻塞客户端请求。
pub trait ClientApiKeyUsageSink: Send + Sync {
    fn record_used(&self, key_id: &ClientApiKeyId);
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderCircuitDecision {
    Allow,
    BlockedUntil(SystemTime),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[error("provider circuit store is unavailable")]
pub struct ProviderCircuitError;

/// Provider circuit 的可重建协调策略；由 Core 拥有并交给 Store adapter 执行。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ProviderCircuitPolicy {
    pub failure_threshold: NonZeroU32,
    pub open_duration: Duration,
}

impl Default for ProviderCircuitPolicy {
    fn default() -> Self {
        Self {
            failure_threshold: NonZeroU32::new(3).unwrap_or(NonZeroU32::MIN),
            open_duration: Duration::from_secs(30),
        }
    }
}

pub trait ProviderCircuitPort: Send + Sync {
    fn decision<'a>(
        &'a self,
        provider_kind: &'a ProviderKind,
    ) -> BoxFuture<'a, Result<ProviderCircuitDecision, ProviderCircuitError>>;
    fn observe_failure<'a>(
        &'a self,
        provider_kind: &'a ProviderKind,
    ) -> BoxFuture<'a, Result<(), ProviderCircuitError>>;
    fn observe_success<'a>(
        &'a self,
        provider_kind: &'a ProviderKind,
    ) -> BoxFuture<'a, Result<(), ProviderCircuitError>>;
}

pub struct DefaultExecutionService {
    snapshots: RuntimeSnapshotHandle,
    coordinator: Arc<AttemptCoordinator<dyn ExecutionStore>>,
    probe_coordinator: Arc<AttemptCoordinator<dyn ExecutionStore>>,
    /// probe 自身走 transient store，探测失败仍写入持久 store 的 ops_events。
    observations: Arc<dyn ExecutionStore>,
    providers: ProviderRegistry,
    admissions: Arc<dyn ClientAdmissionPort>,
    admission_waiting: ConcurrencyWaitQueue<ClientApiKeyId>,
    circuits: Arc<dyn ProviderCircuitPort>,
    continuation: Arc<dyn NativeContinuationPort>,
    client_api_key_usage: Arc<dyn ClientApiKeyUsageSink>,
    budget: Option<Arc<dyn ClientBudgetPort>>,
}

impl DefaultExecutionService {
    #[must_use]
    pub fn new(
        snapshots: RuntimeSnapshotHandle,
        execution: Arc<dyn ExecutionStore>,
        providers: ProviderRegistry,
        admissions: Arc<dyn ClientAdmissionPort>,
        circuits: Arc<dyn ProviderCircuitPort>,
        continuation: Arc<dyn NativeContinuationPort>,
        client_api_key_usage: Arc<dyn ClientApiKeyUsageSink>,
    ) -> Self {
        let observations = Arc::clone(&execution);
        let engine = GatewayEngine::<dyn ExecutionStore>::new(execution, providers.clone());
        let transient: Arc<dyn ExecutionStore> = Arc::new(TransientExecutionStore);
        let probe_engine = GatewayEngine::<dyn ExecutionStore>::new(transient, providers.clone());
        Self {
            snapshots,
            coordinator: Arc::new(AttemptCoordinator::new(engine)),
            probe_coordinator: Arc::new(AttemptCoordinator::new(probe_engine)),
            observations,
            providers,
            admissions,
            admission_waiting: ConcurrencyWaitQueue::default(),
            circuits,
            continuation,
            client_api_key_usage,
            budget: None,
        }
    }

    #[must_use]
    pub fn with_budget(mut self, budget: Arc<dyn ClientBudgetPort>) -> Self {
        self.budget = Some(budget);
        self
    }

    fn authenticate_without_usage(
        &self,
        plaintext: &str,
    ) -> Result<AuthenticatedClient, ClientAuthenticationError> {
        let snapshot = self
            .snapshots
            .acquire()
            .map_err(|_| ClientAuthenticationError::SnapshotUnavailable)?;
        let policy = snapshot
            .client_policies()
            .filter(|policy| {
                constant_time_equal(plaintext, policy.plaintext_key().expose_for_auth())
            })
            .find(|policy| policy.authorize().is_ok())
            .cloned()
            .ok_or(ClientAuthenticationError::InvalidKey)?;
        Ok(AuthenticatedClient { snapshot, policy })
    }

    async fn start_inner(&self, request: StartExecution) -> Result<StartedExecution, GatewayError> {
        let StartExecution {
            client,
            public_model,
            operation,
            metadata,
        } = request;
        self.start_inner_with_target(PendingStartExecution {
            client,
            target: ExecutionTarget::Model(public_model),
            operation,
            metadata,
        })
        .await
    }

    async fn start_provider_endpoint_inner(
        &self,
        request: StartProviderExecution,
    ) -> Result<StartedExecution, GatewayError> {
        let StartProviderExecution {
            client,
            provider,
            operation,
            metadata,
        } = request;
        self.start_inner_with_target(PendingStartExecution {
            client,
            target: ExecutionTarget::ProviderEndpoint(provider),
            operation,
            metadata,
        })
        .await
    }

    async fn start_inner_with_target(
        &self,
        mut request: PendingStartExecution,
    ) -> Result<StartedExecution, GatewayError> {
        // 长连接每次执行都重新鉴权并冻结当前策略，确保限额和授权变更对新请求生效。
        request.client = self
            .authenticate(request.client.policy.plaintext_key().expose_for_auth())
            .map_err(|error| match error {
                ClientAuthenticationError::InvalidKey => {
                    GatewayError::new(GatewayErrorKind::Unauthorized, "client API key is invalid")
                }
                ClientAuthenticationError::SnapshotUnavailable => GatewayError::new(
                    GatewayErrorKind::Internal,
                    "runtime snapshot is unavailable",
                ),
            })?;
        let started_at = SystemTime::now();
        let deadline_at = started_at
            .checked_add(MODEL_REQUEST_DEADLINE)
            .ok_or_else(|| {
                GatewayError::new(GatewayErrorKind::Internal, "system clock is invalid")
            })?;
        let request_id = new_request_id()?;
        let routing_context = self
            .route_context(request.client.policy.account_scope().provider_kinds())
            .await?;
        let account_scope = Arc::clone(request.client.policy.account_scope());
        let plan = match &request.target {
            ExecutionTarget::ProviderEndpoint(provider) => {
                request.client.snapshot.plan_provider_endpoint(
                    provider,
                    &request.operation,
                    account_scope,
                    &routing_context,
                )
            }
            ExecutionTarget::Model(public_model) => request.client.snapshot.plan(
                public_model,
                &request.operation,
                account_scope,
                &routing_context,
            ),
        }
        .map_err(map_routing_error)?;
        let continuation = match request.metadata.previous_response_id.as_ref() {
            Some(previous) => {
                let resolve = self
                    .continuation
                    .resolve(request.client.policy.key_id(), previous)
                    .fuse();
                let timeout = Delay::new(COORDINATION_TIMEOUT).fuse();
                pin_mut!(resolve, timeout);
                let pin = select_biased! {
                    result = resolve => match result {
                        Ok(pin) => pin,
                        Err(error)
                            if error.kind() == NativeContinuationStoreErrorKind::Unavailable =>
                        {
                            tracing::debug!(
                                request_id = request_id.as_str(),
                                %error,
                                "Continuation affinity 查询失败，退化为外部续接"
                            );
                            None
                        }
                        Err(error)
                            if error.kind()
                                == NativeContinuationStoreErrorKind::OwnershipMismatch =>
                        {
                            return Err(GatewayError::new(
                                GatewayErrorKind::PolicyDenied,
                                "continuation does not belong to this client API key",
                            ));
                        }
                        Err(error) => {
                            tracing::warn!(
                                request_id = request_id.as_str(),
                                %error,
                                "Continuation affinity 记录无效，已拒绝续接"
                            );
                            return Err(GatewayError::new(
                                GatewayErrorKind::Internal,
                                "continuation state is invalid",
                            ));
                        }
                    },
                    _ = timeout => None,
                };
                match pin {
                    Some(pin) if !pin.matches_client(request.client.policy.key_id()) => {
                        return Err(GatewayError::new(
                            GatewayErrorKind::PolicyDenied,
                            "continuation does not belong to this client API key",
                        ));
                    }
                    Some(pin) if !request.client.policy.account_scope().allows(pin.account()) => {
                        return Err(GatewayError::new(
                            GatewayErrorKind::PolicyDenied,
                            "continuation account is outside the client account scope",
                        ));
                    }
                    Some(pin)
                        if !plan
                            .candidates()
                            .iter()
                            .any(|candidate| candidate.provider() == pin.provider()) =>
                    {
                        return Err(GatewayError::new(
                            GatewayErrorKind::NoAvailableProvider,
                            "continuation provider is not available",
                        ));
                    }
                    Some(pin) => {
                        attach_continuation_session_state(&mut request.operation, &pin);
                        ContinuationBinding::Pinned(pin)
                    }
                    None => ContinuationBinding::External(previous.clone()),
                }
            }
            None => {
                return self
                    .start_without_continuation(
                        request,
                        request_id,
                        started_at,
                        deadline_at,
                        plan,
                        None,
                    )
                    .await;
            }
        };
        self.start_without_continuation(
            request,
            request_id,
            started_at,
            deadline_at,
            plan,
            Some(continuation),
        )
        .await
    }

    async fn start_without_continuation(
        &self,
        request: PendingStartExecution,
        request_id: ModelRequestId,
        started_at: SystemTime,
        deadline_at: SystemTime,
        plan: crate::routing::RoutingPlan,
        continuation: Option<ContinuationBinding>,
    ) -> Result<StartedExecution, GatewayError> {
        let PendingStartExecution {
            client,
            target,
            operation,
            metadata,
        } = request;
        let admission_started_at = Instant::now();
        let concurrency_wait_budget = ConcurrencyWaitBudget::default();
        let admission = self
            .acquire_client_admission(&client, &request_id, deadline_at, &concurrency_wait_budget)
            .await?;
        let admission_decision_ms = duration_ms(admission_started_at.elapsed());
        if let Some(budget) = &self.budget
            && let Err(error) = budget.admit(client.policy.key_id().clone()).await
        {
            admission.release().await;
            return Err(error);
        }
        let observation = plan
            .candidates()
            .first()
            .map_or_else(Default::default, |candidate| {
                self.providers.request_observation(
                    candidate.provider(),
                    &operation,
                    client.policy.key_id(),
                )
            });
        let new_request = NewModelRequest {
            id: request_id.clone(),
            client_api_key_id: Some(client.policy.key_id().clone()),
            client_api_key_ref: client.policy.key_id().clone(),
            config_revision: plan.config_revision(),
            routing: client.policy.account_scope().routing_snapshot(),
            protocol: metadata.protocol,
            operation: operation.kind(),
            endpoint: metadata.endpoint,
            client_transport: metadata.transport.as_str().to_owned(),
            requested_model: target.into_public_model().or(observation.requested_model),
            client_ip: metadata.client_ip,
            user_agent: metadata.user_agent,
            reasoning_effort: observation.reasoning_effort,
            reasoning_preset: observation.reasoning_preset,
            request_kind: observation.request_kind,
            subagent_kind: observation.subagent_kind,
            compact: observation.compact,
            continuation: observation.continuation,
            image_generation_requested: operation.image_generation_requested(),
            admission_decision_ms: Some(admission_decision_ms),
            started_at,
            deadline_at,
        };
        let core = match self
            .coordinator
            .start(
                new_request,
                operation,
                plan,
                None,
                continuation,
                CancellationToken::new(),
            )
            .await
        {
            Ok(core) => core.with_concurrency_wait_budget(concurrency_wait_budget),
            Err(error) => {
                if let Some(budget) = &self.budget {
                    settle_budget(
                        budget.as_ref(),
                        ClientBudgetCharge {
                            key_id: client.policy.key_id().clone(),
                            request_id: request_id.clone(),
                            amount_usd: crate::metering::Decimal::ZERO,
                            completed_at: SystemTime::now(),
                        },
                    )
                    .await;
                }
                admission.release().await;
                return Err(gateway_error_from_engine(&error));
            }
        };
        Ok(StartedExecution {
            request_id,
            created_at: started_at,
            stream: metadata.stream,
            session: Box::new(DefaultExecutionSession::new(
                core,
                admission,
                Arc::clone(&self.circuits),
                Arc::clone(&self.continuation),
                self.budget.clone(),
            )),
        })
    }

    async fn acquire_client_admission(
        &self,
        client: &AuthenticatedClient,
        request_id: &ModelRequestId,
        deadline_at: SystemTime,
        budget: &ConcurrencyWaitBudget,
    ) -> Result<AdmissionLease, GatewayError> {
        let policy = client.snapshot.client_queue_policy();
        let limits = client.policy.limits();
        let key = client.policy.key_id();
        let mut waiting = CapacityWait::new(&self.admission_waiting, policy, deadline_at, budget);
        let mut admission = AdmissionLease {
            port: Arc::clone(&self.admissions),
            client_api_key_id: key.clone(),
            model_request_id: request_id.clone(),
            armed: false,
        };
        loop {
            let remaining = deadline_at
                .duration_since(SystemTime::now())
                .unwrap_or_default();
            if remaining.is_zero() {
                return Err(GatewayError::new(
                    GatewayErrorKind::Timeout,
                    "request deadline elapsed",
                ));
            }
            // 在发送原子准入之前接管取消清理，覆盖 Redis 已取得租约但返回尚未被观察的窗口。
            admission.armed = true;
            let acquire = self
                .admissions
                .admit(ClientAdmissionRequest {
                    model_request_id: request_id.clone(),
                    client_api_key_id: key.clone(),
                    lease_ttl: remaining,
                    allow_concurrency_acquire: limits.max_concurrency == 0 || waiting.can_try(key),
                    limits,
                })
                .fuse();
            let timeout = Delay::new(remaining).fuse();
            pin_mut!(acquire, timeout);
            let decision = select_biased! {
                result = acquire => result.map_err(|_| GatewayError::new(GatewayErrorKind::NoAvailableProvider, "request admission is temporarily unavailable"))?,
                _ = timeout => return Err(GatewayError::new(GatewayErrorKind::Timeout, "request deadline elapsed")),
            };
            match decision {
                ClientAdmissionDecision::Granted => {
                    if !waiting.elapsed().is_zero() {
                        tracing::info!(
                            request_id = request_id.as_str(),
                            queue_layer = "client_key",
                            queue_wait_ms = duration_ms(waiting.elapsed()),
                            "排队请求已取得 Key 并发槽位"
                        );
                    }
                    return Ok(admission);
                }
                ClientAdmissionDecision::Rejected(reason) => {
                    admission.armed = false;
                    if reason == ClientAdmissionRejection::RateLimited || policy.max_waiting == 0 {
                        return Err(GatewayError::new(
                            GatewayErrorKind::RateLimited,
                            "request exceeds client API key limits",
                        ));
                    }
                    if waiting.elapsed().is_zero()
                        && let Some(budget) = &self.budget
                    {
                        budget.admit(key.clone()).await?;
                    }
                    waiting.wait(std::slice::from_ref(key)).await.map_err(|error| {
                        tracing::info!(request_id = request_id.as_str(), queue_layer = "client_key", queue_wait_ms = duration_ms(waiting.elapsed()), reason = %error, "Key 排队请求被拒绝");
                        error.gateway_error()
                    })?;
                }
            }
        }
    }

    async fn route_context(
        &self,
        provider_kinds: &BTreeSet<ProviderKind>,
    ) -> Result<RoutingContext, GatewayError> {
        let decisions = futures::future::join_all(provider_kinds.iter().map(|provider_kind| {
            let circuits = Arc::clone(&self.circuits);
            async move {
                let decision = circuits.decision(provider_kind).fuse();
                let timeout = Delay::new(COORDINATION_TIMEOUT).fuse();
                pin_mut!(decision, timeout);
                let decision = select_biased! {
                    result = decision => Some(result),
                    _ = timeout => None,
                };
                (provider_kind, decision)
            }
        }))
        .await;
        let mut blocked_providers = BTreeSet::new();
        for (provider_kind, decision) in decisions {
            match decision {
                Some(Ok(ProviderCircuitDecision::BlockedUntil(_))) => {
                    blocked_providers.insert(provider_kind.clone());
                }
                Some(Err(error)) => tracing::warn!(
                    provider = provider_kind.as_str(),
                    %error,
                    "Provider circuit 读取失败，按可重建协调状态 fail-open"
                ),
                None => tracing::warn!(
                    provider = provider_kind.as_str(),
                    "Provider circuit 读取超时，按可重建协调状态 fail-open"
                ),
                Some(Ok(ProviderCircuitDecision::Allow)) => {}
            }
        }
        Ok(RoutingContext {
            required_provider: None,
            blocked_providers,
        })
    }

    async fn probe_inner(
        &self,
        request: AccountProbeRequest,
    ) -> Result<AccountProbeResult, AccountProbeError> {
        let AccountProbeRequest {
            account_id,
            provider_kind,
            upstream_model,
            operation,
        } = request;
        let observed = ProbeObservation {
            provider_kind: provider_kind.clone(),
            account_id: account_id.clone(),
            upstream_model: upstream_model.clone(),
        };
        let snapshot = self.snapshots.acquire().map_err(|_| {
            GatewayError::new(
                GatewayErrorKind::Internal,
                "runtime snapshot is unavailable",
            )
        })?;
        let public_model =
            PublicModelId::new(upstream_model.as_str().to_owned()).map_err(|_| {
                GatewayError::new(GatewayErrorKind::Unsupported, "requested model is invalid")
            })?;
        let routing_context = RoutingContext {
            required_provider: Some(provider_kind),
            ..RoutingContext::default()
        };
        let plan = snapshot
            .plan(
                &public_model,
                &operation,
                snapshot.all_account_scope(),
                &routing_context,
            )
            .map_err(map_routing_error)?;
        let started_at = SystemTime::now();
        let deadline_at = started_at
            .checked_add(MODEL_REQUEST_DEADLINE)
            .ok_or_else(|| {
                GatewayError::new(GatewayErrorKind::Internal, "system clock is invalid")
            })?;
        let request_id = new_request_id()?;
        let actor = ClientApiKeyId::new("admin_connection_test")
            .map_err(|_| GatewayError::new(GatewayErrorKind::Internal, "invalid admin actor"))?;
        let new_request = NewModelRequest {
            id: request_id,
            client_api_key_id: None,
            client_api_key_ref: actor,
            config_revision: plan.config_revision(),
            routing: crate::routing::AccountRoutingSnapshot::all(),
            protocol: "admin_connection_test".to_owned(),
            operation: operation.kind(),
            endpoint: "/api/admin/accounts/connection-test".to_owned(),
            client_transport: ClientTransport::InternalProbe.as_str().to_owned(),
            requested_model: Some(public_model),
            client_ip: None,
            user_agent: None,
            reasoning_effort: None,
            reasoning_preset: None,
            request_kind: Some("account_connection_test".to_owned()),
            subagent_kind: None,
            compact: false,
            continuation: Default::default(),
            image_generation_requested: false,
            admission_decision_ms: None,
            started_at,
            deadline_at,
        };
        let mut session = match self
            .probe_coordinator
            .start_diagnostic(
                new_request,
                operation,
                plan,
                account_id,
                None,
                CancellationToken::new(),
            )
            .await
        {
            Ok(session) => session,
            Err(error) => {
                return Err(self
                    .observe_probe_failure(&observed, started_at, &error)
                    .await);
            }
        };
        let events = session.collect_uncommitted().await;
        publish_provider_attempt_outcomes(
            self.circuits.as_ref(),
            session.provider_attempt_outcomes(),
        )
        .await;
        let events = match events {
            Ok(events) => events,
            Err(error) => {
                return Err(self
                    .observe_probe_failure(&observed, started_at, &error)
                    .await);
            }
        };
        if let Err(error) = session.commit_downstream(Some(200)).await {
            return Err(self
                .observe_probe_failure(&observed, started_at, &error)
                .await);
        }
        Ok(AccountProbeResult {
            text: events
                .into_iter()
                .flat_map(|event| event.into_parts().0)
                .filter_map(|fact| match fact {
                    GatewayEvent::TextDelta(delta) => Some(delta.text),
                    _ => None,
                })
                .collect(),
        })
    }

    /// 探测失败先记录脱敏分类事实，再把请求局部的原始上游响应交给认证管理端。
    async fn observe_probe_failure(
        &self,
        observed: &ProbeObservation,
        started_at: SystemTime,
        error: &EngineError,
    ) -> AccountProbeError {
        let (source, send_state, upstream_response) = match error {
            EngineError::Provider(provider_error) => {
                let upstream_response = provider_error
                    .client_visible_upstream_response()
                    .map(AccountProbeUpstreamResponse::from_client_response);
                let has_upstream_facts = provider_error.send_state() != UpstreamSendState::NotSent
                    || provider_error.upstream_status().is_some()
                    || provider_error.upstream_code().is_some()
                    || provider_error.client_visible_upstream_error().is_some()
                    || upstream_response.is_some();
                let source = if has_upstream_facts {
                    AccountProbeErrorSource::Upstream
                } else {
                    AccountProbeErrorSource::Provider
                };
                (source, Some(provider_error.send_state()), upstream_response)
            }
            _ => (AccountProbeErrorSource::Gateway, None, None),
        };
        if let EngineError::Provider(provider_error) = error {
            let latency = started_at.elapsed().unwrap_or_default();
            let latency_ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX);
            tracing::warn!(
                target: "gateway_probe",
                provider_kind = observed.provider_kind.as_str(),
                account_id = observed.account_id.as_str(),
                upstream_model = observed.upstream_model.as_str(),
                failure_kind = provider_error.kind().as_str(),
                send_state = ?provider_error.send_state(),
                upstream_status = ?provider_error.upstream_status(),
                provider_error_code = ?provider_error.upstream_code().map(|code| code.as_str()),
                latency_ms,
                "账号连接测试失败"
            );
            if let Err(store_error) = self
                .observations
                .record_probe_failure(ProbeFailure {
                    provider_kind: observed.provider_kind.clone(),
                    account_id: observed.account_id.clone(),
                    upstream_model_id: observed.upstream_model.clone(),
                    error: provider_error.clone(),
                    latency,
                })
                .await
            {
                tracing::warn!(
                    operation = "record_probe_failure",
                    provider_kind = observed.provider_kind.as_str(),
                    account_id = observed.account_id.as_str(),
                    error_kind = ?store_error.kind(),
                    "账号连接测试观测写入失败，测试结果不受影响"
                );
            }
        }
        AccountProbeError::new(
            gateway_error_from_engine(error),
            source,
            send_state,
            upstream_response,
        )
    }
}

struct ProbeObservation {
    provider_kind: ProviderKind,
    account_id: ProviderAccountId,
    upstream_model: UpstreamModelId,
}

struct TransientExecutionStore;

#[async_trait::async_trait]
impl ExecutionStore for TransientExecutionStore {
    async fn create_model_request(&self, _: NewModelRequest) -> Result<(), StoreError> {
        Ok(())
    }

    async fn record_attempt(&self, _: AttemptRecord) -> Result<(), StoreError> {
        Ok(())
    }

    async fn mark_send_state(
        &self,
        _: &ModelRequestId,
        _: UpstreamSendState,
    ) -> Result<(), StoreError> {
        Ok(())
    }

    async fn mark_downstream_committed(
        &self,
        _: &ModelRequestId,
        _: SystemTime,
        _: Option<u16>,
    ) -> Result<(), StoreError> {
        Ok(())
    }

    async fn record_client_status(&self, _: &ModelRequestId, _: u16) -> Result<(), StoreError> {
        Ok(())
    }

    async fn record_intermediate_failure(&self, _: IntermediateFailure) -> Result<(), StoreError> {
        Ok(())
    }

    async fn finalize_model_request(&self, _: ModelRequestFinalization) -> Result<(), StoreError> {
        Ok(())
    }

    async fn recover_expired(&self, _: SystemTime) -> Result<RecoveryReport, StoreError> {
        Ok(RecoveryReport::default())
    }
}

impl ExecutionService for DefaultExecutionService {
    fn authenticate(
        &self,
        plaintext: &str,
    ) -> Result<AuthenticatedClient, ClientAuthenticationError> {
        let client = self.authenticate_without_usage(plaintext)?;
        self.client_api_key_usage
            .record_used(client.policy().key_id());
        Ok(client)
    }

    fn public_models(&self, client: &AuthenticatedClient) -> Vec<PublicModelId> {
        client
            .snapshot
            .public_models_for_scope(client.policy.account_scope())
    }

    fn client_model_catalog<'a>(
        &'a self,
        client: &'a AuthenticatedClient,
        protocol: &'a str,
        client_version: &'a str,
    ) -> BoxFuture<'a, Result<Vec<PublicModelDescriptor>, ProviderCatalogUnavailable>> {
        Box::pin(async move {
            let scope = client.policy.account_scope();
            let mut result = Vec::new();
            let mut seen = BTreeSet::new();
            for kind in scope.provider_kinds() {
                let provider = self.providers.get(kind).ok_or(ProviderCatalogUnavailable)?;
                let Some(models) = provider
                    .query_client_model_catalog(scope, protocol, client_version)
                    .await?
                else {
                    for profile in client.snapshot.public_model_profiles_for_provider(kind) {
                        if !scope.allows_provider_model(
                            kind,
                            &client.snapshot.mapped_model(profile.model().as_str()),
                        ) {
                            continue;
                        }
                        if seen.insert(profile.model().clone()) {
                            result.push(PublicModelDescriptor::Adapted(profile));
                        }
                    }
                    continue;
                };
                // 保留上游顺序；映射只选择完整的目标对象，不能跨账号/模型混拼字段。
                let by_id = models
                    .iter()
                    .map(|entry| (entry.model.as_str(), entry))
                    .collect::<std::collections::BTreeMap<_, _>>();
                for entry in &models {
                    let target = client.snapshot.mapped_model(entry.model.as_str());
                    if !scope.allows_provider_model(kind, &target) {
                        continue;
                    }
                    let Some(source) = by_id.get(target.as_str()) else {
                        continue;
                    };
                    let model = PublicModelId::new(entry.model.as_str().to_owned())
                        .map_err(|_| ProviderCatalogUnavailable)?;
                    // 原生目录可能比路由快照更新，不能向客户端公布当前已知不可路由的模型。
                    if !client
                        .snapshot
                        .contains_public_model_for_provider(&model, kind)
                    {
                        continue;
                    }
                    if seen.insert(model.clone()) {
                        result.push(source.content.for_public_model(model));
                    }
                }
                for model in client.snapshot.public_models_for_provider(kind) {
                    let target = client.snapshot.mapped_model(model.as_str());
                    if !scope.allows_provider_model(kind, &target) {
                        continue;
                    }
                    if target == model.as_str() || seen.contains(&model) {
                        continue;
                    }
                    if !client
                        .snapshot
                        .contains_public_model_for_provider(&model, kind)
                    {
                        continue;
                    }
                    if let Some(source) = by_id.get(target.as_str()) {
                        seen.insert(model.clone());
                        result.push(source.content.for_public_model(model));
                    }
                }
            }
            Ok(result)
        })
    }

    fn contains_public_model(&self, client: &AuthenticatedClient, model: &PublicModelId) -> bool {
        client
            .snapshot
            .contains_public_model_for_scope(model, client.policy.account_scope())
    }

    fn start(
        &self,
        request: StartExecution,
    ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
        Box::pin(async move { self.start_inner(request).await })
    }

    fn start_provider_endpoint(
        &self,
        request: StartProviderExecution,
    ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
        Box::pin(async move { self.start_provider_endpoint_inner(request).await })
    }
}

impl ClientKeyVerifier for DefaultExecutionService {
    fn verify_client_key(
        &self,
        plaintext: &str,
    ) -> Result<ClientApiKeyId, ClientAuthenticationError> {
        self.authenticate_without_usage(plaintext)
            .map(|client| client.policy().key_id().clone())
    }
}

impl AccountProbe for DefaultExecutionService {
    fn probe(
        &self,
        request: AccountProbeRequest,
    ) -> BoxFuture<'_, Result<AccountProbeResult, AccountProbeError>> {
        Box::pin(async move { self.probe_inner(request).await })
    }
}

struct AdmissionLease {
    armed: bool,
    port: Arc<dyn ClientAdmissionPort>,
    client_api_key_id: ClientApiKeyId,
    model_request_id: ModelRequestId,
}

async fn settle_budget(port: &dyn ClientBudgetPort, charge: ClientBudgetCharge) {
    if let Err(error) = port.settle(charge).await {
        tracing::error!(%error, "Client budget settlement failed; storage will retry on the next request");
    }
}

impl AdmissionLease {
    async fn release(mut self) {
        if let Err(error) = self
            .port
            .release(&self.client_api_key_id, &self.model_request_id)
            .await
        {
            tracing::warn!(%error, "Client admission 释放失败，依赖租约 TTL 收敛");
        }
        self.armed = false;
    }
}

impl Drop for AdmissionLease {
    fn drop(&mut self) {
        if self.armed {
            self.port
                .abandon(&self.client_api_key_id, &self.model_request_id);
        }
    }
}

struct DefaultExecutionSession {
    core: ResponseExecutionSession<dyn ExecutionStore>,
    admission: Option<AdmissionLease>,
    cleanup: Option<BoxFuture<'static, ()>>,
    circuits: Arc<dyn ProviderCircuitPort>,
    continuation: Arc<dyn NativeContinuationPort>,
    observed_provider_outcomes: usize,
    continuation_recorded: bool,
    budget: Option<Arc<dyn ClientBudgetPort>>,
}

impl DefaultExecutionSession {
    fn new(
        core: ResponseExecutionSession<dyn ExecutionStore>,
        admission: AdmissionLease,
        circuits: Arc<dyn ProviderCircuitPort>,
        continuation: Arc<dyn NativeContinuationPort>,
        budget: Option<Arc<dyn ClientBudgetPort>>,
    ) -> Self {
        Self {
            core,
            admission: Some(admission),
            cleanup: None,
            circuits,
            continuation,
            observed_provider_outcomes: 0,
            continuation_recorded: false,
            budget,
        }
    }

    async fn settle_if_finalized(&mut self) {
        if self.core.is_finalized()
            && let Some(admission) = self.admission.take()
        {
            let budget = self.budget.take();
            let charge = self.core.budget_charge();
            // 在首次 await 前把完整清理责任留在会话内。事件等待被取消后，后续 poll
            // 或 detach 继续同一个 future，既不丢失费用，也不重启已完成的结算。
            self.cleanup = Some(Box::pin(async move {
                if let Some(budget) = budget {
                    settle_budget(budget.as_ref(), charge).await;
                }
                admission.release().await;
            }));
        }
        if let Some(cleanup) = self.cleanup.as_mut() {
            cleanup.await;
            self.cleanup = None;
        }
    }

    async fn observe_provider_outcomes(&mut self) {
        let outcomes = self.core.provider_attempt_outcomes();
        let new_outcomes = outcomes
            .get(self.observed_provider_outcomes..)
            .unwrap_or_default()
            .to_vec();
        self.observed_provider_outcomes = outcomes.len();
        publish_provider_attempt_outcomes(self.circuits.as_ref(), &new_outcomes).await;
    }

    async fn record_continuation(&mut self, state: Option<&ProviderSessionState>) {
        if self.continuation_recorded {
            return;
        }
        let Some(state) = state else {
            return;
        };
        let Some(pin) = self.core.native_continuation_pin(state) else {
            return;
        };
        self.continuation_recorded = true;
        record_native_continuation(self.continuation.as_ref(), pin).await;
    }

    async fn finalize_detached(&mut self) {
        if let Err(error) = self.core.cancel_and_finalize().await {
            tracing::warn!(%error, "Detached execution 终态收敛失败");
        }
        self.observe_provider_outcomes().await;
        self.settle_if_finalized().await;
    }
}

impl Drop for DefaultExecutionSession {
    fn drop(&mut self) {
        self.core.cancel();
    }
}

impl ExecutionSession for DefaultExecutionSession {
    fn trace(&self) -> crate::diagnostics::TraceContext {
        self.core.trace()
    }
    fn next_event(&mut self) -> BoxFuture<'_, Result<Option<CoordinatedEvent>, EngineError>> {
        Box::pin(async move {
            let result = self.core.next_event().await;
            if let Ok(Some(event)) = result.as_ref() {
                self.record_continuation(event.session_update()).await;
            }
            self.observe_provider_outcomes().await;
            self.settle_if_finalized().await;
            result
        })
    }

    fn collect_uncommitted(&mut self) -> BoxFuture<'_, Result<Vec<ProviderEvent>, EngineError>> {
        Box::pin(async move {
            let result = self.core.collect_uncommitted().await;
            if let Ok(events) = result.as_ref() {
                let state = events.iter().find_map(ProviderEvent::session_update);
                self.record_continuation(state).await;
            }
            self.observe_provider_outcomes().await;
            self.settle_if_finalized().await;
            result
        })
    }

    fn response_headers(&self) -> &[ProviderResponseHeader] {
        self.core.response_headers()
    }

    fn response_status_code(&self) -> Option<u16> {
        self.core.response_status_code()
    }

    fn commit_downstream(
        &mut self,
        client_status_code: Option<u16>,
    ) -> BoxFuture<'_, Result<(), EngineError>> {
        Box::pin(async move {
            let result = self.core.commit_downstream(client_status_code).await;
            self.observe_provider_outcomes().await;
            self.settle_if_finalized().await;
            result
        })
    }

    fn record_client_status(
        &mut self,
        client_status_code: u16,
    ) -> BoxFuture<'_, Result<(), EngineError>> {
        Box::pin(async move {
            let result = self.core.record_client_status(client_status_code).await;
            self.observe_provider_outcomes().await;
            self.settle_if_finalized().await;
            result
        })
    }

    fn is_finalized(&self) -> bool {
        self.core.is_finalized() && self.admission.is_none() && self.cleanup.is_none()
    }

    fn cancel(&self) {
        self.core.cancel();
    }

    fn detach_finalize(mut self: Box<Self>) -> BoxFuture<'static, ()> {
        Box::pin(async move { self.finalize_detached().await })
    }
}

#[must_use]
pub const fn provider_failure_affects_circuit(error_kind: ProviderErrorKind) -> bool {
    matches!(
        error_kind,
        ProviderErrorKind::Timeout
            | ProviderErrorKind::Transport
            | ProviderErrorKind::Protocol
            | ProviderErrorKind::Unavailable
    )
}

async fn publish_provider_attempt_outcomes(
    circuits: &dyn ProviderCircuitPort,
    outcomes: &[ProviderAttemptOutcome],
) {
    for outcome in outcomes {
        let result = match outcome.error_kind() {
            None => circuits.observe_success(outcome.provider_kind()).await,
            Some(kind) if provider_failure_affects_circuit(kind) => {
                circuits.observe_failure(outcome.provider_kind()).await
            }
            Some(_) => continue,
        };
        if let Err(error) = result {
            tracing::warn!(
                provider = outcome.provider_kind().as_str(),
                %error,
                "Provider circuit feedback 写入失败，数据面不受影响"
            );
        }
    }
}

async fn record_native_continuation(
    continuation: &dyn NativeContinuationPort,
    pin: NativeContinuationPin,
) {
    let provider = pin.provider().as_str().to_owned();
    let account = pin.account().as_str().to_owned();
    let record = continuation.record(pin).fuse();
    let timeout = Delay::new(COORDINATION_TIMEOUT).fuse();
    pin_mut!(record, timeout);
    select_biased! {
        result = record => {
            if let Err(error) = result {
                tracing::warn!(
                    provider = %provider,
                    account = %account,
                    %error,
                    "Continuation affinity 写入失败，后续请求将退化为外部续接"
                );
            }
        },
        _ = timeout => {
            tracing::warn!(
                provider = %provider,
                account = %account,
                "Continuation affinity 后台写入超时，已丢弃本次亲和记录"
            );
        },
    }
}

fn constant_time_equal(left: &str, right: &str) -> bool {
    if left.len() != right.len() {
        return false;
    }
    left.as_bytes()
        .iter()
        .zip(right.as_bytes())
        .fold(0_u8, |difference, (left, right)| {
            difference | (left ^ right)
        })
        == 0
}

fn duration_ms(duration: Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}

fn new_request_id() -> Result<ModelRequestId, GatewayError> {
    ModelRequestId::new(format!("req_{}", Uuid::now_v7().simple()))
        .map_err(|_| GatewayError::new(GatewayErrorKind::Internal, "failed to allocate request ID"))
}

fn map_routing_error(error: crate::validation::RoutingError) -> GatewayError {
    match error {
        crate::validation::RoutingError::ModelNotFound {
            model,
            mapped_model,
        } => GatewayError::new(
            GatewayErrorKind::ModelNotFound,
            if model == mapped_model {
                "the requested model was not found in the provider catalogs available to this API key; check the model name"
            } else {
                "the requested model maps to an upstream model that was not found in the provider catalogs available to this API key; check the configured model mapping"
            },
        ),
        crate::validation::RoutingError::NoCapableProvider { .. }
        | crate::validation::RoutingError::NoCapableProviderEndpoint { .. }
        | crate::validation::RoutingError::EmptyAccountScope => GatewayError::new(
            GatewayErrorKind::NoAvailableProvider,
            "no provider can execute this request",
        ),
        _ => GatewayError::new(
            GatewayErrorKind::Internal,
            "runtime routing configuration is invalid",
        ),
    }
}

fn attach_continuation_session_state(operation: &mut Operation, pin: &NativeContinuationPin) {
    let Some(state) = pin.session_state() else {
        return;
    };
    if state.provider() != pin.provider().as_str()
        || operation.provider_session_state(state.provider()).is_some()
    {
        return;
    }
    operation.set_provider_session_state(state.clone());
}

pub fn gateway_error_from_engine(error: &EngineError) -> GatewayError {
    match error {
        EngineError::Cancelled => {
            GatewayError::new(GatewayErrorKind::Cancelled, "request was cancelled")
        }
        EngineError::Deadline => {
            GatewayError::new(GatewayErrorKind::Timeout, "request deadline elapsed")
        }
        EngineError::Provider(provider) => GatewayError::from_provider(provider),
        EngineError::EmptyRoutingPlan | EngineError::ProviderNotRegistered { .. } => {
            GatewayError::new(
                GatewayErrorKind::NoAvailableProvider,
                "no provider is available",
            )
        }
        _ => GatewayError::new(GatewayErrorKind::Internal, "request execution failed"),
    }
}