From 574f0d580e663a2fa8171a3a1c4efbaff268780a Mon Sep 17 00:00:00 2001 From: zyycn Date: Sat, 25 Jul 2026 23:40:15 +0800 Subject: [PATCH] fix(xai): restore Codex compatibility and runtime semantics Preserve transient credential failures as runtime cooldowns, restore account-aware retry and observability facts, and align Grok model/tool adaptation with Codex clients. Keep complete SSE events on the live path while retaining only protocol-required tool and continuation state. Co-authored-by: Codex --- backend/Cargo.lock | 1 + backend/crates/gateway-admin/Cargo.toml | 1 + .../gateway-admin/src/model/accounts.rs | 5 +- .../gateway-admin/src/model/observability.rs | 3 +- .../gateway-admin/src/use_case/accounts.rs | 13 +- .../gateway-admin/tests/use_case/accounts.rs | 96 ++++++- .../crates/gateway-api/src/admin/accounts.rs | 9 +- .../gateway-api/src/admin/observability.rs | 15 +- .../crates/gateway-api/src/openai/models.rs | 85 +++++- .../crates/gateway-api/src/openai/service.rs | 9 +- .../gateway-api/tests/admin/accounts.rs | 10 +- .../crates/gateway-api/tests/openai/models.rs | 51 +++- .../gateway-core/src/engine/coordinator.rs | 48 +--- .../gateway-core/src/engine/execution.rs | 13 +- .../gateway-core/src/engine/provider.rs | 17 +- backend/crates/gateway-core/src/event.rs | 29 --- .../gateway-core/src/provider_ports/mod.rs | 2 +- .../crates/gateway-core/src/routing/mod.rs | 184 +++++++++++++ .../gateway-core/src/routing/snapshot.rs | 66 ++++- .../gateway-core/tests/engine/coordinator.rs | 45 ++++ .../gateway-core/tests/routing/snapshot.rs | 31 ++- .../src/postgres/observability.rs | 27 +- .../src/postgres/provider_accounts.rs | 9 +- .../src/redis/credential_cooldown.rs | 4 +- .../tests/postgres/observability.rs | 65 +++++ .../tests/postgres/provider_accounts.rs | 24 +- .../providers/xai/src/credential/refresh.rs | 84 ++++-- .../providers/xai/src/credential/selector.rs | 83 ++++-- backend/crates/providers/xai/src/lib.rs | 3 + backend/crates/providers/xai/src/provider.rs | 219 +++++++++++----- .../providers/xai/src/transport/network.rs | 15 -- .../providers/xai/src/transport/request.rs | 246 +++++++++++++++--- .../providers/xai/tests/credential/refresh.rs | 15 +- .../xai/tests/credential/selector.rs | 115 ++++++-- .../providers/xai/tests/provider/contract.rs | 126 ++++++++- backend/crates/providers/xai/tests/support.rs | 61 ++++- .../xai/tests/transport/canonical.rs | 102 ++++++++ .../providers/xai/tests/transport/network.rs | 23 +- .../providers/xai/tests/transport/request.rs | 147 +++++++++-- .../components/AccountOverviewCards.vue | 4 +- .../components/AccountUsageWindow.vue | 7 - .../composables/useAccountConnectionTest.ts | 5 +- .../composables/useAccountMutations.ts | 13 +- .../accounts/composables/useAccountsQuery.ts | 2 +- .../views/dashboard/composables/presenter.ts | 10 +- .../usage/components/OpsErrorDetailModal.vue | 2 +- .../views/usage/components/OpsErrorPanel.vue | 4 +- frontend/src/views/usage/constants.ts | 2 +- 48 files changed, 1756 insertions(+), 394 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 25364d8b6..89117ae7f 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1152,6 +1152,7 @@ dependencies = [ "rand_core 0.6.4", "secrecy", "serde", + "serde_json", "subtle", "thiserror", "tokio", diff --git a/backend/crates/gateway-admin/Cargo.toml b/backend/crates/gateway-admin/Cargo.toml index 4355f1623..9a0e9a6af 100644 --- a/backend/crates/gateway-admin/Cargo.toml +++ b/backend/crates/gateway-admin/Cargo.toml @@ -27,6 +27,7 @@ thiserror = "2.0.18" uuid = { version = "1.23.3", features = ["v7"] } [dev-dependencies] +serde_json = { version = "1.0.150", features = ["preserve_order"] } tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt"] } [lints] diff --git a/backend/crates/gateway-admin/src/model/accounts.rs b/backend/crates/gateway-admin/src/model/accounts.rs index 378e02a19..6abea80ee 100644 --- a/backend/crates/gateway-admin/src/model/accounts.rs +++ b/backend/crates/gateway-admin/src/model/accounts.rs @@ -29,7 +29,6 @@ pub enum AccountStatus { QuotaExhausted, Disabled, Banned, - Attention, } /// 账号列表排序字段。 @@ -184,7 +183,7 @@ pub struct AccountSummary { pub total: u64, pub active: u64, pub quota_exhausted: u64, - pub attention: u64, + pub unavailable: u64, } /// 账号启停写入命令。 @@ -222,6 +221,8 @@ pub enum AccountConnectionTestEvent { }, Failed { message: String, + provider_error_code: Option, + provider_error_type: Option, account_status: AccountStatus, }, } diff --git a/backend/crates/gateway-admin/src/model/observability.rs b/backend/crates/gateway-admin/src/model/observability.rs index 450e3e6f9..2c198ba31 100644 --- a/backend/crates/gateway-admin/src/model/observability.rs +++ b/backend/crates/gateway-admin/src/model/observability.rs @@ -718,7 +718,8 @@ pub struct OpsError { pub upstream_model_id: Option, pub upstream_transport: Option, pub failure_kind: String, - pub status_code: Option, + pub client_status_code: Option, + pub upstream_status_code: Option, pub provider_error_code: Option, pub client_response_id: Option, pub upstream_request_id: Option, diff --git a/backend/crates/gateway-admin/src/use_case/accounts.rs b/backend/crates/gateway-admin/src/use_case/accounts.rs index f24573dfc..c8a2be3ed 100644 --- a/backend/crates/gateway-admin/src/use_case/accounts.rs +++ b/backend/crates/gateway-admin/src/use_case/accounts.rs @@ -10,7 +10,6 @@ use gateway_core::{ credential::ProviderAccountId, probe::{AccountProbe, AccountProbeRequest}, }, - error::GatewayErrorKind, routing::{ProviderKind, UpstreamModelId, snapshot::SnapshotControl}, }; @@ -454,12 +453,9 @@ impl AccountsService for DefaultAccountsService { })) .collect(), Err(error) => vec![AccountConnectionTestEvent::Failed { - message: match error.kind() { - GatewayErrorKind::InvalidRequest - | GatewayErrorKind::Unsupported - | GatewayErrorKind::ModelNotFound => error.safe_message().to_owned(), - _ => "Provider connection test failed".to_owned(), - }, + message: error.client_message().to_owned(), + provider_error_code: error.client_error_code().map(ToOwned::to_owned), + provider_error_type: error.client_error_type().map(ToOwned::to_owned), account_status: status, }], } @@ -485,8 +481,7 @@ fn account_status(account: &AccountRecord, now: chrono::DateTime) -> Accoun { AccountStatus::Expired } - crate::model::accounts::AccountAvailability::Ready => AccountStatus::Active, - _ => AccountStatus::Attention, + _ => AccountStatus::Active, } } } diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index 65e386bd4..6fe28595e 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -2,13 +2,14 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use chrono::{TimeDelta, Utc}; -use futures::future::BoxFuture; +use futures::{StreamExt as _, future::BoxFuture}; use gateway_core::{ engine::{ credential::{OpaqueProviderData, ProviderAccountId}, probe::{AccountProbe, AccountProbeRequest, AccountProbeResult}, }, - error::GatewayError, + error::{ClientVisibleUpstreamError, GatewayError, GatewayErrorKind}, + operation::{GenerateRequest, Operation, ProtocolPayload}, routing::ProviderKind, }; @@ -17,9 +18,9 @@ use gateway_admin::{ model::{ MutationContext, Revision, accounts::{ - AccountAvailability, AccountListQuery, AccountPage, AccountRecord, AccountSummary, - AccountUsage, AccountUsageWindowQuery, AccountUsageWindowResult, DeleteAccounts, - SetAccountEnabled, + AccountAvailability, AccountConnectionTestEvent, AccountListQuery, AccountPage, + AccountRecord, AccountSummary, AccountUsage, AccountUsageWindowQuery, + AccountUsageWindowResult, DeleteAccounts, SetAccountEnabled, }, observability::TimeRange, provider_credentials::{ @@ -48,6 +49,7 @@ use gateway_admin::{ }, }, }; +use serde_json::{Map, json}; pub(super) type EventLog = Arc>>; @@ -159,10 +161,22 @@ impl ProviderAdmin for FakeProviderAdmin { fn connection_test_operation( &self, - _: &gateway_core::routing::UpstreamModelId, - _: &str, + model: &gateway_core::routing::UpstreamModelId, + input: &str, ) -> Result { - Err(ProviderAdminError::new(ProviderAdminErrorKind::Unsupported)) + let payload = ProtocolPayload::json_object( + "openai", + Map::from_iter([ + ("model".to_owned(), json!(model.as_str())), + ("input".to_owned(), json!(input)), + ("stream".to_owned(), json!(true)), + ("store".to_owned(), json!(false)), + ]), + ) + .map_err(|_| ProviderAdminError::new(ProviderAdminErrorKind::Invalid))?; + Ok(Operation::Generate(GenerateRequest::from_protocol_payload( + payload, + ))) } fn dashboard_wire_profile( @@ -395,7 +409,7 @@ impl AccountStore for FakeAccountStore { total: 1, active: 1, quota_exhausted: 0, - attention: 0, + unavailable: 0, }, }) } @@ -686,6 +700,37 @@ fn provider_registry_should_reject_duplicate_kind() { )); } +#[tokio::test] +async fn connection_test_should_preserve_safe_provider_error_details() { + let provider = FakeProviderAdmin::new("xai", events()); + let store = FakeAccountStore::new("xai", events()); + let services = + accounts_service_with_probe(provider, store, Arc::new(FailingAccountProbe)).await; + + let events = services + .accounts() + .test_connection( + ProviderAccountId::new("acct_test").expect("account ID"), + gateway_core::routing::UpstreamModelId::new("grok-4.5").expect("model"), + ) + .await + .expect("connection test stream") + .collect::>() + .await; + + assert!(matches!( + events.last(), + Some(AccountConnectionTestEvent::Failed { + message, + provider_error_code: Some(code), + provider_error_type: Some(error_type), + .. + }) if message == "included usage exhausted" + && code == "usage_exhausted" + && error_type == "invalid_request_error" + )); +} + #[tokio::test] async fn accounts_export_should_pass_store_loaded_timestamps_and_material_to_provider() { let provider = FakeProviderAdmin::new("openai", events()); @@ -1005,12 +1050,20 @@ fn rotation_result(command: CredentialRotationCommit) -> CredentialMutationResul async fn accounts_service( provider: Arc, store: Arc, +) -> AdminServices { + accounts_service_with_probe(provider, store, Arc::new(SuccessfulAccountProbe)).await +} + +async fn accounts_service_with_probe( + provider: Arc, + store: Arc, + probe: Arc, ) -> AdminServices { super::AdminHarness::new() .accounts(store) .settings(Arc::new(StaticSettingsStore)) .provider(provider) - .probe(Arc::new(SuccessfulAccountProbe)) + .probe(probe) .build() .await } @@ -1030,6 +1083,29 @@ impl AccountProbe for SuccessfulAccountProbe { } } +struct FailingAccountProbe; + +impl AccountProbe for FailingAccountProbe { + fn probe( + &self, + _: AccountProbeRequest, + ) -> BoxFuture<'_, Result> { + Box::pin(async { + let detail = ClientVisibleUpstreamError::new( + "included usage exhausted", + Some("usage_exhausted".to_owned()), + Some("invalid_request_error".to_owned()), + ) + .expect("safe provider error"); + Err(GatewayError::new( + GatewayErrorKind::RateLimited, + "upstream capacity is temporarily unavailable", + ) + .with_client_visible_upstream_error(detail)) + }) + } +} + fn store_unavailable() -> AdminStoreError { AdminStoreError::new( AdminStoreErrorKind::Unavailable, diff --git a/backend/crates/gateway-api/src/admin/accounts.rs b/backend/crates/gateway-api/src/admin/accounts.rs index 7917455d4..5c4eb0cd2 100644 --- a/backend/crates/gateway-api/src/admin/accounts.rs +++ b/backend/crates/gateway-api/src/admin/accounts.rs @@ -174,7 +174,7 @@ pub struct AccountSummaryView { pub total: u64, pub active: u64, pub quota_exhausted: u64, - pub attention: u64, + pub unavailable: u64, } /// 一条安全账号视图。 @@ -543,10 +543,14 @@ impl From for AccountConnectionTestEvent { }), DomainConnectionTestEvent::Failed { message, + provider_error_code, + provider_error_type, account_status, } => serde_json::json!({ "type": "error", "error": message, + "providerErrorCode": provider_error_code, + "providerErrorType": provider_error_type, "accountStatus": account_status_name(account_status) }), }; @@ -1401,7 +1405,7 @@ fn account_page_data( total: result.summary.total, active: result.summary.active, quota_exhausted: result.summary.quota_exhausted, - attention: result.summary.attention, + unavailable: result.summary.unavailable, }, } } @@ -1798,7 +1802,6 @@ fn account_status_name(status: DomainAccountStatus) -> &'static str { DomainAccountStatus::QuotaExhausted => "quota_exhausted", DomainAccountStatus::Disabled => "disabled", DomainAccountStatus::Banned => "banned", - DomainAccountStatus::Attention => "attention", } } diff --git a/backend/crates/gateway-api/src/admin/observability.rs b/backend/crates/gateway-api/src/admin/observability.rs index f91b82a7a..1bbc05ffc 100644 --- a/backend/crates/gateway-api/src/admin/observability.rs +++ b/backend/crates/gateway-api/src/admin/observability.rs @@ -801,8 +801,8 @@ pub struct DashboardCardsView { pub struct DashboardCredentialsCardView { pub total: String, pub total_value: u64, - pub enabled: String, - pub enabled_value: u64, + pub available: String, + pub available_value: u64, pub unavailable: String, pub unavailable_value: u64, } @@ -1235,7 +1235,6 @@ pub struct OpsErrorView { pub account_id: Option, pub route: String, pub model: Option, - pub status_code: Option, pub client_status_code: Option, pub upstream_status_code: Option, pub transport: Option, @@ -1954,7 +1953,6 @@ fn usage_page_view( } fn ops_error_view(error: domain::OpsError) -> OpsErrorView { - let status = error.status_code.map(i64::from); OpsErrorView { id: error.event_id, request_id: error.request_id, @@ -1964,9 +1962,8 @@ fn ops_error_view(error: domain::OpsError) -> OpsErrorView { account_id: error.provider_account_ref, route: error.operation, model: error.upstream_model_id, - status_code: status, - client_status_code: None, - upstream_status_code: None, + client_status_code: error.client_status_code.map(i64::from), + upstream_status_code: error.upstream_status_code.map(i64::from), transport: error.upstream_transport, attempt_index: error.attempt_index, failure_class: error.failure_kind, @@ -2290,8 +2287,8 @@ fn dashboard_view(result: domain::DashboardResult, kind: TrendKind) -> Dashboard credentials: DashboardCredentialsCardView { total: format_compact_number(provider_accounts.total), total_value: provider_accounts.total, - enabled: format_compact_number(provider_accounts.enabled), - enabled_value: provider_accounts.enabled, + available: format_compact_number(provider_accounts.active), + available_value: provider_accounts.active, unavailable: format_compact_number(provider_accounts.unavailable), unavailable_value: provider_accounts.unavailable, }, diff --git a/backend/crates/gateway-api/src/openai/models.rs b/backend/crates/gateway-api/src/openai/models.rs index 818fca48f..6aea76f91 100644 --- a/backend/crates/gateway-api/src/openai/models.rs +++ b/backend/crates/gateway-api/src/openai/models.rs @@ -6,7 +6,7 @@ use axum::{ http::{HeaderMap, StatusCode}, response::{IntoResponse, Response}, }; -use gateway_core::routing::PublicModelId; +use gateway_core::routing::{PublicModelId, PublicModelProfile}; use serde_json::{Value, json}; use crate::ApiState; @@ -17,6 +17,7 @@ use super::{ }; const MODEL_CREATED_TIMESTAMP: i64 = 1_700_000_000; +const CODEX_BASE_INSTRUCTIONS: &str = "You are Codex, a coding agent. Follow the user's instructions and use the available tools to complete software engineering tasks. Inspect relevant files before editing, preserve unrelated changes, and verify the result."; /// `GET /v1/models`。 pub(crate) async fn models(State(state): State, headers: HeaderMap) -> Response { @@ -30,6 +31,16 @@ pub(crate) async fn models(State(state): State, headers: HeaderMap) -> .into_iter() .map(|model| openai_model_json(&model)) .collect::>(); + let profiles = service.public_model_profiles(&client); + + if !profiles.is_empty() { + let models = profiles + .iter() + .enumerate() + .map(|(index, profile)| codex_model_json(profile, index)) + .collect::>(); + return (StatusCode::OK, Json(json!({ "models": models }))).into_response(); + } ( StatusCode::OK, @@ -134,3 +145,75 @@ fn catalog_model_json(id: &str) -> Value { "source": "gateway", }) } + +fn codex_model_json(profile: &PublicModelProfile, index: usize) -> Value { + let id = profile.model().as_str(); + let presentation = profile.presentation(); + let reasoning_levels = presentation + .supported_reasoning_efforts() + .iter() + .map(|effort| { + json!({ + "effort": effort, + "description": reasoning_effort_description(effort), + }) + }) + .collect::>(); + let reasoning_supported = presentation + .supported_reasoning_efforts() + .iter() + .any(|effort| effort != "none"); + let context_window = presentation.context_window_tokens(); + let input_modalities = if presentation.image_input() { + vec!["text", "image"] + } else { + vec!["text"] + }; + + json!({ + "slug": id, + "display_name": presentation.display_name().unwrap_or(id), + "description": presentation.description(), + "default_reasoning_level": presentation.default_reasoning_effort(), + "supported_reasoning_levels": reasoning_levels, + "shell_type": "shell_command", + "visibility": if presentation.hidden() { "hide" } else { "list" }, + "supported_in_api": true, + "priority": index.saturating_add(1), + "additional_speed_tiers": [], + "service_tiers": [], + "availability_nux": Value::Null, + "upgrade": Value::Null, + "base_instructions": CODEX_BASE_INSTRUCTIONS, + "model_messages": Value::Null, + "include_skills_usage_instructions": false, + "supports_reasoning_summary_parameter": reasoning_supported, + "default_reasoning_summary": "auto", + "support_verbosity": presentation.verbosity(), + "default_verbosity": Value::Null, + "apply_patch_tool_type": presentation.agent_tools().then_some("freeform"), + "web_search_tool_type": "text", + "truncation_policy": { "mode": "tokens", "limit": 10_000 }, + "supports_parallel_tool_calls": presentation.parallel_tool_calls(), + "supports_image_detail_original": presentation.image_detail_original(), + "context_window": context_window, + "max_context_window": context_window, + "effective_context_window_percent": 95, + "experimental_supported_tools": [], + "input_modalities": input_modalities, + "supports_search_tool": presentation.search_tool(), + "use_responses_lite": false, + }) +} + +fn reasoning_effort_description(effort: &str) -> &'static str { + match effort { + "none" => "No reasoning", + "low" => "Fast responses with lighter reasoning", + "medium" => "Balances speed and reasoning depth for everyday tasks", + "high" => "Greater reasoning depth for complex problems", + "xhigh" => "Extra high reasoning depth for complex problems", + "max" => "Maximum reasoning depth for the hardest problems", + _ => "Provider-supported reasoning level", + } +} diff --git a/backend/crates/gateway-api/src/openai/service.rs b/backend/crates/gateway-api/src/openai/service.rs index 65c538711..8093ef07e 100644 --- a/backend/crates/gateway-api/src/openai/service.rs +++ b/backend/crates/gateway-api/src/openai/service.rs @@ -9,7 +9,7 @@ use gateway_core::engine::execution::{ }; use gateway_core::error::{GatewayError, GatewayErrorKind}; use gateway_core::lifecycle::{ConnectionDraining, ConnectionGuard, ConnectionLifecycle}; -use gateway_core::routing::PublicModelId; +use gateway_core::routing::{PublicModelId, PublicModelProfile}; use uuid::Uuid; use super::auth::ClientApiKeyAuthError; @@ -51,6 +51,13 @@ impl OpenAiService { .collect() } + pub(crate) fn public_model_profiles( + &self, + client: &AuthenticatedClient, + ) -> Vec { + self.execution.public_model_profiles(client) + } + pub(crate) fn contains_public_model( &self, client: &AuthenticatedClient, diff --git a/backend/crates/gateway-api/tests/admin/accounts.rs b/backend/crates/gateway-api/tests/admin/accounts.rs index d910850c0..e2d4e1a59 100644 --- a/backend/crates/gateway-api/tests/admin/accounts.rs +++ b/backend/crates/gateway-api/tests/admin/accounts.rs @@ -276,6 +276,8 @@ mod actions { }, DomainConnectionTestEvent::Failed { message: "upstream unavailable".to_owned(), + provider_error_code: Some("usage_exhausted".to_owned()), + provider_error_type: Some("invalid_request_error".to_owned()), account_status: AccountStatus::Active, }, ] @@ -302,7 +304,13 @@ mod actions { }), json!({ "type": "content", "text": "OK" }), json!({ "type": "test_complete", "success": true, "accountStatus": "active" }), - json!({ "type": "error", "error": "upstream unavailable", "accountStatus": "active" }), + json!({ + "type": "error", + "error": "upstream unavailable", + "providerErrorCode": "usage_exhausted", + "providerErrorType": "invalid_request_error", + "accountStatus": "active" + }), ] ); } diff --git a/backend/crates/gateway-api/tests/openai/models.rs b/backend/crates/gateway-api/tests/openai/models.rs index 511eb5e32..9b27c9db6 100644 --- a/backend/crates/gateway-api/tests/openai/models.rs +++ b/backend/crates/gateway-api/tests/openai/models.rs @@ -10,19 +10,28 @@ use gateway_core::engine::execution::{ StartedExecution, }; use gateway_core::error::{GatewayError, GatewayErrorKind}; -use gateway_core::routing::PublicModelId; +use gateway_core::routing::{ModelPresentation, PublicModelId, PublicModelProfile}; use tower::ServiceExt; use super::{api_router, authenticated_client}; struct ModelsExecution { client: AuthenticatedClient, + profiles: bool, } impl ModelsExecution { fn new() -> Arc { Arc::new(Self { client: authenticated_client("sk_models_test"), + profiles: false, + }) + } + + fn with_profiles() -> Arc { + Arc::new(Self { + client: authenticated_client("sk_models_test"), + profiles: true, }) } } @@ -46,6 +55,26 @@ impl ExecutionService for ModelsExecution { .collect() } + fn public_model_profiles(&self, _: &AuthenticatedClient) -> Vec { + if !self.profiles { + return Vec::new(); + } + vec![PublicModelProfile::new( + PublicModelId::new("grok-4.5").expect("model"), + ModelPresentation::new( + Some("Grok 4.5".to_owned()), + Some("xAI Grok 4.5 frontier model.".to_owned()), + ) + .with_reasoning( + Some("medium".to_owned()), + ["low", "medium", "high"].map(str::to_owned).to_vec(), + ) + .with_context_window_tokens(Some(500_000)) + .with_image_input(true) + .with_agent_tools(true, true), + )] + } + fn contains_public_model(&self, _: &AuthenticatedClient, model: &PublicModelId) -> bool { matches!(model.as_str(), "model-a" | "model-b") } @@ -91,6 +120,26 @@ async fn models_should_encode_the_service_visible_catalog() { ); } +#[tokio::test] +async fn models_should_encode_provider_profiles_for_current_codex_clients() { + let response = api_router(ModelsExecution::with_profiles()) + .await + .oneshot(authorized_request("/v1/models?client_version=0.145.0")) + .await + .expect("list Codex models response"); + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("read Codex models body"); + let value: serde_json::Value = serde_json::from_slice(&body).expect("models JSON"); + let model = &value["models"][0]; + + assert_eq!(model["slug"], "grok-4.5"); + assert_eq!(model["default_reasoning_level"], "medium"); + assert_eq!(model["context_window"], 500_000); + assert_eq!(model["apply_patch_tool_type"], "freeform"); + assert_eq!(model["service_tiers"], serde_json::json!([])); +} + #[tokio::test] async fn model_detail_should_keep_the_official_path_id_contract() { let response = api_router(ModelsExecution::new()) diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 41379b654..331836aa9 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -1,6 +1,6 @@ //! 唯一的账号重试、发送与下游 commit barrier owner。 -use std::collections::{BTreeSet, VecDeque}; +use std::collections::BTreeSet; use std::num::NonZeroU32; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime}; @@ -67,12 +67,6 @@ where let continuation_attempt = initial_continuation_attempt(&operation, &plan, continuation.as_ref()); let image_generation_requested = operation.image_generation_requested(); - let commit_policy = if continuation_attempt == ContinuationAttempt::None { - StreamCommitPolicy::FirstCanonicalEvent - } else { - StreamCommitPolicy::UntilOutputOrTerminal - }; - let mut session = ResponseExecutionSession { engine: Arc::clone(&self.engine), request_id, @@ -86,7 +80,6 @@ where required_account, continuation, continuation_attempt, - commit_policy, account_state_owner, cancellation, attempts: 0, @@ -106,7 +99,6 @@ where upstream_response_id: None, last_account_exhaustion: None, provider_attempt_outcomes: Vec::new(), - before_commit: VecDeque::new(), }; if session.cancellation.is_cancelled() { @@ -141,12 +133,6 @@ struct FailureFinalization { provider_response_id: Option, } -#[derive(Clone, Copy)] -enum StreamCommitPolicy { - FirstCanonicalEvent, - UntilOutputOrTerminal, -} - /// API 可逐事件消费的 Core 执行会话。 /// /// API 只能提交下游 delivery 边界；账号重试、断流终结与 @@ -164,7 +150,6 @@ pub struct ResponseExecutionSession { required_account: Option, continuation: Option, continuation_attempt: ContinuationAttempt, - commit_policy: StreamCommitPolicy, account_state_owner: Option, cancellation: CancellationToken, attempts: u32, @@ -184,7 +169,6 @@ pub struct ResponseExecutionSession { upstream_response_id: Option, last_account_exhaustion: Option, provider_attempt_outcomes: Vec, - before_commit: VecDeque, } impl ResponseExecutionSession @@ -203,15 +187,6 @@ where if self.finalized { return Ok(None); } - if self.downstream_committed_at.is_some() - && let Some(event) = self.before_commit.pop_front() - { - return Ok(Some(CoordinatedEvent::single( - event, - CommitRequirement::AlreadyCommitted, - ))); - } - loop { match self.pull().await? { PullOutcome::Event(event) => { @@ -221,25 +196,13 @@ where CommitRequirement::AlreadyCommitted, ))); } - let commit_significant = match self.commit_policy { - StreamCommitPolicy::FirstCanonicalEvent => event.has_canonical_facts(), - StreamCommitPolicy::UntilOutputOrTerminal => event.is_commit_significant(), - }; - self.before_commit.push_back(event); - if !commit_significant { - continue; - } self.delivery_pending = true; - let events = self.before_commit.drain(..).collect(); - return CoordinatedEvent::try_batch( - events, + return Ok(Some(CoordinatedEvent::single( + event, CommitRequirement::CommitBeforeDelivery, - ) - .map(Some); - } - PullOutcome::AttemptDiscarded => { - self.before_commit.clear(); + ))); } + PullOutcome::AttemptDiscarded => {} PullOutcome::End => { if self.downstream_committed_at.is_some() { self.finish_success().await?; @@ -953,7 +916,6 @@ where self.timings.first_text_ms = None; self.timings.first_token_ms = None; self.upstream_complete = false; - self.before_commit.clear(); } async fn finish_success(&mut self) -> Result<(), EngineError> { diff --git a/backend/crates/gateway-core/src/engine/execution.rs b/backend/crates/gateway-core/src/engine/execution.rs index 6ba6b88f6..bff85c85f 100644 --- a/backend/crates/gateway-core/src/engine/execution.rs +++ b/backend/crates/gateway-core/src/engine/execution.rs @@ -29,7 +29,9 @@ use crate::event::{GatewayEvent, ProviderEvent, ProviderResponseHeader}; use crate::operation::{Operation, ProviderSessionState}; use crate::policy::{ClientApiKeyId, ClientPolicy}; use crate::routing::snapshot::RuntimeSnapshotHandle; -use crate::routing::{ProviderKind, PublicModelId, RoutingContext, RuntimeSnapshot}; +use crate::routing::{ + ProviderKind, PublicModelId, PublicModelProfile, RoutingContext, RuntimeSnapshot, +}; const MODEL_REQUEST_DEADLINE: Duration = Duration::from_secs(10 * 60); const CONTINUATION_AFFINITY_TIMEOUT: Duration = Duration::from_millis(100); @@ -139,6 +141,9 @@ pub trait ExecutionService: Send + Sync { plaintext: &str, ) -> Result; fn public_models(&self, client: &AuthenticatedClient) -> Vec; + fn public_model_profiles(&self, _client: &AuthenticatedClient) -> Vec { + Vec::new() + } fn contains_public_model(&self, client: &AuthenticatedClient, model: &PublicModelId) -> bool; fn start( &self, @@ -568,6 +573,12 @@ impl ExecutionService for DefaultExecutionService { .public_models_for_provider(client.policy.provider_kind()) } + fn public_model_profiles(&self, client: &AuthenticatedClient) -> Vec { + client + .snapshot + .public_model_profiles_for_provider(client.policy.provider_kind()) + } + fn contains_public_model(&self, client: &AuthenticatedClient, model: &PublicModelId) -> bool { client .snapshot diff --git a/backend/crates/gateway-core/src/engine/provider.rs b/backend/crates/gateway-core/src/engine/provider.rs index d8ee188ab..42907d57c 100644 --- a/backend/crates/gateway-core/src/engine/provider.rs +++ b/backend/crates/gateway-core/src/engine/provider.rs @@ -20,7 +20,9 @@ use crate::error::{ }; use crate::event::{EventSequenceValidator, ProviderEvent}; use crate::operation::Operation; -use crate::routing::{ModelCapabilities, ProviderCandidate, ProviderKind, UpstreamModelId}; +use crate::routing::{ + ModelCapabilities, ModelPresentation, ProviderCandidate, ProviderKind, UpstreamModelId, +}; /// Box 只出现在 Provider Registry 的统一 event envelope 边界。 pub type EventStream = @@ -433,6 +435,7 @@ pub struct ProviderRequestObservation { pub struct ProviderModelCapabilities { upstream_model: UpstreamModelId, capabilities: ModelCapabilities, + presentation: Option, } /// Provider 实时目录成功发布后的进程内单调代次。 @@ -459,9 +462,16 @@ impl ProviderModelCapabilities { Self { upstream_model, capabilities, + presentation: None, } } + #[must_use] + pub fn with_presentation(mut self, presentation: ModelPresentation) -> Self { + self.presentation = Some(presentation); + self + } + #[must_use] pub const fn upstream_model(&self) -> &UpstreamModelId { &self.upstream_model @@ -471,6 +481,11 @@ impl ProviderModelCapabilities { pub const fn capabilities(&self) -> &ModelCapabilities { &self.capabilities } + + #[must_use] + pub const fn presentation(&self) -> Option<&ModelPresentation> { + self.presentation.as_ref() + } } impl ProviderRequest { diff --git a/backend/crates/gateway-core/src/event.rs b/backend/crates/gateway-core/src/event.rs index 49588ce46..a4f629901 100644 --- a/backend/crates/gateway-core/src/event.rs +++ b/backend/crates/gateway-core/src/event.rs @@ -756,18 +756,6 @@ impl ProviderEvent { pub fn has_client_event(&self) -> bool { !self.canonical.is_empty() || self.wire.is_some() } - - /// 返回是否包含足以冻结下游 commit barrier 的 canonical fact。 - /// - /// 单独的 wire event 不能冻结边界：协议层在看到 `Started` - /// 携带的网关响应身份前无法安全编码它。Core 会保留这些 - /// 事件，直到首个 canonical 可见增量或终态再一并交付。 - #[must_use] - pub fn is_commit_significant(&self) -> bool { - self.canonical - .iter() - .any(GatewayEvent::is_commit_significant) - } } impl From for ProviderEvent { @@ -776,23 +764,6 @@ impl From for ProviderEvent { } } -impl GatewayEvent { - /// 返回该 fact 是否足以冻结下游 commit barrier。 - /// - /// 生命周期、结构与结算 facts 可以在换号前安全丢弃；首个可见增量或 - /// Provider 正常终态才会使客户端输出不可撤回。 - #[must_use] - pub const fn is_commit_significant(&self) -> bool { - matches!( - self, - Self::TextDelta(_) - | Self::ReasoningDelta(_) - | Self::ToolCallDelta(_) - | Self::Completed(_) - ) - } -} - /// Canonical event 顺序错误。 #[derive(Debug, Clone, PartialEq, Eq, Error)] pub enum EventSequenceError { diff --git a/backend/crates/gateway-core/src/provider_ports/mod.rs b/backend/crates/gateway-core/src/provider_ports/mod.rs index d5c78bc07..8bc40f39a 100644 --- a/backend/crates/gateway-core/src/provider_ports/mod.rs +++ b/backend/crates/gateway-core/src/provider_ports/mod.rs @@ -476,7 +476,7 @@ pub trait ProviderCredentialStatePort: Send + Sync { ) -> BoxFuture<'a, Result>; } -/// 临时 cooldown 只保存调度截止时间；原因事实仍由 PostgreSQL 账号状态持有。 +/// 临时 cooldown 只保存可丢失的调度截止时间，不进入账号持久状态。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProviderCooldown { account_id: ProviderAccountId, diff --git a/backend/crates/gateway-core/src/routing/mod.rs b/backend/crates/gateway-core/src/routing/mod.rs index eed3e4bc5..82ce80adc 100644 --- a/backend/crates/gateway-core/src/routing/mod.rs +++ b/backend/crates/gateway-core/src/routing/mod.rs @@ -136,6 +136,177 @@ pub struct ModelCapabilities { upstream_validates_features: bool, } +/// Provider 为客户端模型目录提供的展示与交互能力。 +/// +/// 该值不参与路由，也不把任一 Provider 的 wire 类型带入 Core。 +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ModelPresentation { + display_name: Option, + description: Option, + default_reasoning_effort: Option, + supported_reasoning_efforts: Vec, + context_window_tokens: Option, + image_input: bool, + agent_tools: bool, + parallel_tool_calls: bool, + search_tool: bool, + image_detail_original: bool, + verbosity: bool, + hidden: bool, +} + +impl ModelPresentation { + #[must_use] + pub fn new(display_name: Option, description: Option) -> Self { + Self { + display_name, + description, + ..Self::default() + } + } + + #[must_use] + pub fn with_reasoning( + mut self, + default_effort: Option, + supported_efforts: Vec, + ) -> Self { + self.default_reasoning_effort = default_effort; + self.supported_reasoning_efforts = supported_efforts; + self + } + + #[must_use] + pub const fn with_context_window_tokens(mut self, context_window_tokens: Option) -> Self { + self.context_window_tokens = context_window_tokens; + self + } + + #[must_use] + pub const fn with_image_input(mut self, image_input: bool) -> Self { + self.image_input = image_input; + self + } + + #[must_use] + pub const fn with_agent_tools(mut self, agent_tools: bool, parallel_tool_calls: bool) -> Self { + self.agent_tools = agent_tools; + self.parallel_tool_calls = parallel_tool_calls; + self + } + + #[must_use] + pub const fn with_search_tool(mut self, search_tool: bool) -> Self { + self.search_tool = search_tool; + self + } + + #[must_use] + pub const fn with_image_detail_original(mut self, image_detail_original: bool) -> Self { + self.image_detail_original = image_detail_original; + self + } + + #[must_use] + pub const fn with_verbosity(mut self, verbosity: bool) -> Self { + self.verbosity = verbosity; + self + } + + #[must_use] + pub const fn with_hidden(mut self, hidden: bool) -> Self { + self.hidden = hidden; + self + } + + #[must_use] + pub fn display_name(&self) -> Option<&str> { + self.display_name.as_deref() + } + + #[must_use] + pub fn description(&self) -> Option<&str> { + self.description.as_deref() + } + + #[must_use] + pub fn default_reasoning_effort(&self) -> Option<&str> { + self.default_reasoning_effort.as_deref() + } + + #[must_use] + pub fn supported_reasoning_efforts(&self) -> &[String] { + &self.supported_reasoning_efforts + } + + #[must_use] + pub const fn context_window_tokens(&self) -> Option { + self.context_window_tokens + } + + #[must_use] + pub const fn image_input(&self) -> bool { + self.image_input + } + + #[must_use] + pub const fn agent_tools(&self) -> bool { + self.agent_tools + } + + #[must_use] + pub const fn parallel_tool_calls(&self) -> bool { + self.parallel_tool_calls + } + + #[must_use] + pub const fn search_tool(&self) -> bool { + self.search_tool + } + + #[must_use] + pub const fn image_detail_original(&self) -> bool { + self.image_detail_original + } + + #[must_use] + pub const fn verbosity(&self) -> bool { + self.verbosity + } + + #[must_use] + pub const fn hidden(&self) -> bool { + self.hidden + } +} + +/// 一个公开模型及其 Provider 编译后的客户端画像。 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PublicModelProfile { + model: PublicModelId, + presentation: ModelPresentation, +} + +impl PublicModelProfile { + #[must_use] + pub const fn new(model: PublicModelId, presentation: ModelPresentation) -> Self { + Self { + model, + presentation, + } + } + + #[must_use] + pub const fn model(&self) -> &PublicModelId { + &self.model + } + + #[must_use] + pub const fn presentation(&self) -> &ModelPresentation { + &self.presentation + } +} + impl ModelCapabilities { #[must_use] pub fn new(operations: BTreeSet, max_output_tokens: Option) -> Self { @@ -205,6 +376,7 @@ pub struct ProviderModel { provider: ProviderKind, upstream_model: UpstreamModelId, capabilities: ModelCapabilities, + presentation: Option, } impl ProviderModel { @@ -218,9 +390,16 @@ impl ProviderModel { provider, upstream_model, capabilities, + presentation: None, } } + #[must_use] + pub fn with_presentation(mut self, presentation: ModelPresentation) -> Self { + self.presentation = Some(presentation); + self + } + #[must_use] pub const fn provider(&self) -> &ProviderKind { &self.provider @@ -230,6 +409,11 @@ impl ProviderModel { pub const fn upstream_model(&self) -> &UpstreamModelId { &self.upstream_model } + + #[must_use] + pub const fn presentation(&self) -> Option<&ModelPresentation> { + self.presentation.as_ref() + } } /// 本次请求选择 Provider 时使用的动态过滤事实。 diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index a6f249ba7..5eed44a7e 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -231,11 +231,15 @@ async fn compile_runtime_snapshot( continue; }; provider_models.extend(models.into_iter().map(|model| { - ProviderModel::new( + let compiled = ProviderModel::new( provider.clone(), model.upstream_model().clone(), model.capabilities().clone(), - ) + ); + match model.presentation().cloned() { + Some(presentation) => compiled.with_presentation(presentation), + None => compiled, + } })); } @@ -287,6 +291,8 @@ pub struct RuntimeSnapshot { account_selection_policy: AccountSelectionPolicy, providers: Arc>, provider_models: Arc>>, + provider_model_presentations: + Arc>>, model_mappings: Arc>, provider_catalog_generations: Arc>, client_policies: Arc>, @@ -313,17 +319,24 @@ impl RuntimeSnapshot { let mut model_map = BTreeMap::>::new(); + let mut presentation_map = + BTreeMap::>::new(); for model in provider_models { - if !provider_set.contains(model.provider()) { + let ProviderModel { + provider, + upstream_model, + capabilities, + presentation, + } = model; + if !provider_set.contains(&provider) { return Err(RoutingError::NotFound { entity: "provider", - id: model.provider().to_string(), + id: provider.to_string(), }); } - let models = model_map.entry(model.provider).or_default(); - let upstream_model = model.upstream_model; + let models = model_map.entry(provider.clone()).or_default(); if models - .insert(upstream_model.clone(), model.capabilities) + .insert(upstream_model.clone(), capabilities) .is_some() { return Err(RoutingError::DuplicateEntity { @@ -331,6 +344,12 @@ impl RuntimeSnapshot { id: upstream_model.to_string(), }); } + if let Some(presentation) = presentation { + presentation_map + .entry(provider) + .or_default() + .insert(upstream_model, presentation); + } } let mut client_policy_map = BTreeMap::new(); @@ -350,6 +369,7 @@ impl RuntimeSnapshot { account_selection_policy, providers: Arc::new(provider_set), provider_models: Arc::new(model_map), + provider_model_presentations: Arc::new(presentation_map), model_mappings: Arc::new(BTreeMap::new()), provider_catalog_generations: Arc::new(BTreeMap::new()), client_policies: Arc::new(client_policy_map), @@ -402,6 +422,38 @@ impl RuntimeSnapshot { models.into_iter().collect() } + /// 返回 Provider 已明确声明画像的公开模型；没有画像时不猜测 Provider 语义。 + #[must_use] + pub fn public_model_profiles_for_provider( + &self, + provider: &ProviderKind, + ) -> Vec { + let Some(presentations) = self.provider_model_presentations.get(provider) else { + return Vec::new(); + }; + let mut profiles = BTreeMap::new(); + for (model, presentation) in presentations { + if let Ok(public_model) = PublicModelId::new(model.as_str().to_owned()) { + profiles.insert(public_model, presentation.clone()); + } + } + for alias in self.model_mappings.keys() { + let target = self.mapped_model(alias); + let Some(presentation) = presentations.iter().find_map(|(model, presentation)| { + (model.as_str() == target).then_some(presentation) + }) else { + continue; + }; + if let Ok(public_model) = PublicModelId::new(alias.clone()) { + profiles.insert(public_model, presentation.clone()); + } + } + profiles + .into_iter() + .map(|(model, presentation)| super::PublicModelProfile::new(model, presentation)) + .collect() + } + #[must_use] pub fn public_models(&self) -> Vec { self.providers diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 17643713a..181b0a690 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -1005,6 +1005,51 @@ fn authenticated_native_continuation_reaches_every_attempt_context() { )); } +#[test] +fn streaming_native_continuation_delivers_started_event_before_later_output() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, _, _) = coordinator(vec![Script::Stream { + account_id: "acct_one", + items: complete_stream(None), + }]); + let continuation = NativeContinuationPin::new( + PreviousResponseId::new("previous-response").expect("previous response"), + SafeUpstreamValue::new("upstream-response").expect("upstream response"), + ProviderKind::new("openai").expect("provider"), + ProviderAccountId::new("acct_one").expect("account"), + ); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + Some(ContinuationBinding::Pinned(continuation)), + CancellationToken::new(), + )) + .expect("start execution"); + + let first = block_on(session.next_event()) + .expect("first event") + .expect("started event"); + assert_eq!( + first.commit_requirement(), + CommitRequirement::CommitBeforeDelivery + ); + let events = first.into_provider_events(); + assert!(matches!( + events.as_slice(), + [event] + if matches!(event.canonical_facts(), [GatewayEvent::Started(_)]) + )); + + block_on(session.commit_downstream(Some(200))).expect("commit started event"); + while block_on(session.next_event()) + .expect("remaining event") + .is_some() + {} +} + #[test] fn native_continuation_replays_owner_before_safely_switching_account() { let Operation::Generate(generate) = generate_operation() else { diff --git a/backend/crates/gateway-core/tests/routing/snapshot.rs b/backend/crates/gateway-core/tests/routing/snapshot.rs index 24840266d..4edb1b974 100644 --- a/backend/crates/gateway-core/tests/routing/snapshot.rs +++ b/backend/crates/gateway-core/tests/routing/snapshot.rs @@ -24,8 +24,8 @@ use gateway_core::routing::snapshot::{ SnapshotSubscriptionError, SnapshotSubscriptionPort, runtime_revision_needs_refresh, }; use gateway_core::routing::{ - ConfigRevision, ModelCapabilities, ProviderKind, PublicModelId, RuntimeSnapshot, - UpstreamModelId, + ConfigRevision, ModelCapabilities, ModelPresentation, ProviderKind, PublicModelId, + RuntimeSnapshot, UpstreamModelId, }; use gateway_core::task::WorkerKind; @@ -135,13 +135,19 @@ impl Provider for PublishingCatalogProvider { if self.queries.fetch_add(1, Ordering::SeqCst) == 0 { self.generation.store(1, Ordering::SeqCst); } - Ok(vec![ProviderModelCapabilities::new( - UpstreamModelId::new("upstream-model").expect("model"), - ModelCapabilities::new( - std::collections::BTreeSet::from([OperationKind::Generate]), + Ok(vec![ + ProviderModelCapabilities::new( + UpstreamModelId::new("upstream-model").expect("model"), + ModelCapabilities::new( + std::collections::BTreeSet::from([OperationKind::Generate]), + None, + ), + ) + .with_presentation(ModelPresentation::new( + Some("Upstream Model".to_owned()), None, - ), - )]) + )), + ]) } async fn execute( @@ -215,6 +221,15 @@ fn compiler_retries_when_provider_publishes_catalog_during_compilation() { .map(|generation| generation.get()), Some(1), ); + let profiles = + snapshot.public_model_profiles_for_provider(&ProviderKind::new("alpha").expect("provider")); + assert_eq!( + profiles + .iter() + .map(|profile| profile.model().as_str()) + .collect::>(), + vec!["public-model", "upstream-model"], + ); } #[test] diff --git a/backend/crates/gateway-store/src/postgres/observability.rs b/backend/crates/gateway-store/src/postgres/observability.rs index 8cfbc4438..9a5b35dfe 100644 --- a/backend/crates/gateway-store/src/postgres/observability.rs +++ b/backend/crates/gateway-store/src/postgres/observability.rs @@ -661,7 +661,8 @@ pub struct OpsErrorRecord { pub upstream_model_id: Option, pub upstream_transport: Option, pub failure_kind: String, - pub status_code: Option, + pub client_status_code: Option, + pub upstream_status_code: Option, pub provider_error_code: Option, pub client_response_id: Option, pub upstream_request_id: Option, @@ -1627,7 +1628,8 @@ fn admin_ops_error(error: OpsErrorRecord) -> admin_observability::OpsError { upstream_model_id: error.upstream_model_id, upstream_transport: error.upstream_transport, failure_kind: error.failure_kind, - status_code: error.status_code, + client_status_code: error.client_status_code, + upstream_status_code: error.upstream_status_code, provider_error_code: error.provider_error_code, client_response_id: error.client_response_id, upstream_request_id: error.upstream_request_id, @@ -1978,8 +1980,12 @@ async fn provider_account_metrics( let row = sqlx::query( "select count(*)::bigint as total, count(*) filter (where enabled)::bigint as enabled, - count(*) filter (where not enabled or availability <> 'ready')::bigint - as unavailable, + count(*) filter ( + where not ( + enabled and availability = 'ready' + and (access_token_expires_at is null or access_token_expires_at > $1) + ) + )::bigint as unavailable, count(*) filter ( where enabled and availability = 'ready' and (access_token_expires_at is null or access_token_expires_at > $1) @@ -2922,7 +2928,7 @@ const OPS_ERRORS_CTE: &str = "with errors as ( mr.provider_kind, mr.provider_account_ref, mr.upstream_model_id, mr.upstream_transport, coalesce(mr.error_kind, 'failed') as failure_kind, - coalesce(mr.upstream_status_code, mr.client_status_code) as status_code, + mr.client_status_code, mr.upstream_status_code, mr.provider_error_code, mr.client_response_id, mr.upstream_request_id, mr.latency_ms, coalesce(mr.error_message, mr.error_kind, 'request failed') as message, 1::integer as occurrence_count, @@ -2933,7 +2939,8 @@ const OPS_ERRORS_CTE: &str = "with errors as ( select 'ops_event'::text, oe.id, oe.model_request_id, oe.attempt_index, mr.client_api_key_ref, oe.component, oe.operation, oe.provider_kind, oe.provider_account_ref, - oe.upstream_model_id, null::text, oe.failure_kind, oe.status_code, + oe.upstream_model_id, null::text, oe.failure_kind, + null::integer as client_status_code, oe.status_code as upstream_status_code, oe.provider_error_code, mr.client_response_id, oe.upstream_request_id, oe.latency_ms, oe.message, oe.occurrence_count, oe.created_at, 'ops_event:' || oe.id @@ -2948,7 +2955,8 @@ async fn list_ops_errors(pool: &PgPool, query: OpsErrorQuery) -> StoreResult= ", @@ -3047,7 +3055,7 @@ fn push_ops_filter(statement: &mut QueryBuilder, filter: &OpsErrorFilt statement.push_bind(i32::try_from(index).unwrap_or(i32::MAX)); } if let Some(status) = filter.status_code { - statement.push(" and e.status_code = "); + statement.push(" and e.upstream_status_code = "); statement.push_bind(i32::from(status)); } if let Some(search) = &filter.search { @@ -3148,7 +3156,8 @@ fn ops_error_from_row(row: &sqlx::postgres::PgRow) -> StoreResult &'static str { AdminAccountStatus::QuotaExhausted => "quota_exhausted", AdminAccountStatus::Disabled => "disabled", AdminAccountStatus::Banned => "banned", - AdminAccountStatus::Attention => "attention", } } diff --git a/backend/crates/gateway-store/src/redis/credential_cooldown.rs b/backend/crates/gateway-store/src/redis/credential_cooldown.rs index cd81d4b6e..a6ca9ec04 100644 --- a/backend/crates/gateway-store/src/redis/credential_cooldown.rs +++ b/backend/crates/gateway-store/src/redis/credential_cooldown.rs @@ -1,4 +1,4 @@ -//! PostgreSQL 账号 cooldown 事实的可丢失 Redis 热缓存。 +//! 请求调度用的可丢失账号 cooldown Redis 存储。 use std::{sync::Arc, time::SystemTime}; @@ -243,7 +243,7 @@ impl ProviderCooldownPort for RedisCredentialCooldownRepository { } } -/// PostgreSQL 账号事实到可丢失 cooldown cache 的 write-through adapter。 +/// 账号状态写入到可丢失 cooldown cache 的桥接 adapter。 pub struct CooldownCachingProviderAccountStore { authoritative: Arc, cooldowns: Arc, diff --git a/backend/crates/gateway-store/tests/postgres/observability.rs b/backend/crates/gateway-store/tests/postgres/observability.rs index 3199503d2..5cfab1b76 100644 --- a/backend/crates/gateway-store/tests/postgres/observability.rs +++ b/backend/crates/gateway-store/tests/postgres/observability.rs @@ -62,6 +62,57 @@ fn postgres_admin_observability_adapter_implements_terminal_port() { assert_port::(); } +#[tokio::test] +async fn dashboard_account_metrics_should_partition_available_and_unavailable_accounts() { + let Some(database) = TestDatabase::create("dashboard_account_metrics").await else { + return; + }; + let now = Utc::now(); + sqlx::query( + "insert into provider_accounts ( + id, provider_kind, name, upstream_user_id, authentication_kind, + provider_credentials_json, credential_revision, has_refresh_token, + access_token_expires_at, enabled, availability, + availability_observed_at, created_at, updated_at + ) values + ('acct_available', 'openai', 'available', 'user-available', 'oauth', + '{}'::jsonb, 1, false, $1 + interval '1 day', true, 'ready', $1, $1, $1), + ('acct_expired', 'openai', 'expired', 'user-expired', 'oauth', + '{}'::jsonb, 1, false, $1 - interval '1 day', true, 'ready', $1, $1, $1), + ('acct_banned', 'xai', 'banned', 'user-banned', 'oauth', + '{}'::jsonb, 1, false, $1 + interval '1 day', true, 'banned', $1, $1, $1), + ('acct_disabled', 'xai', 'disabled', 'user-disabled', 'oauth', + '{}'::jsonb, 1, false, $1 + interval '1 day', false, 'ready', $1, $1, $1)", + ) + .bind(now) + .execute(&database.pool) + .await + .expect("seed account metric states"); + let repository = PgObservabilityRepository::new(database.pool.clone()); + let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) + .expect("dashboard range"); + + let metrics = repository + .dashboard_summary(range) + .await + .expect("dashboard summary") + .provider_accounts; + + assert_eq!( + ( + metrics.total, + metrics.enabled, + metrics.active, + metrics.unavailable, + metrics.expired, + metrics.disabled, + metrics.banned, + ), + (4, 3, 1, 3, 1, 1, 1), + ); + database.close().await; +} + #[tokio::test] async fn calculated_usage_billing_facts_keep_only_completed_calculated_costs() { let Some(database) = TestDatabase::create("calculated_usage_billing_facts").await else { @@ -589,6 +640,20 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact .await .expect("ops errors"); assert_eq!(errors.total, 2); + let request_error = errors + .items + .iter() + .find(|error| error.source == "model_request") + .expect("request error"); + assert_eq!(request_error.client_status_code, Some(502)); + assert_eq!(request_error.upstream_status_code, Some(429)); + let attempt_error = errors + .items + .iter() + .find(|error| error.source == "ops_event") + .expect("attempt error"); + assert_eq!(attempt_error.client_status_code, None); + assert_eq!(attempt_error.upstream_status_code, Some(429)); database.close().await; } diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts.rs index a678d57fb..e80223f7f 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts.rs @@ -126,13 +126,13 @@ async fn terminal_admin_list_filters_and_sorts_before_pagination_with_retained_u beta.availability = ProviderAccountAvailability::Banned; let mut charlie = account("acct_charlie", "user-charlie"); charlie.email = Some("charlie@example.invalid".to_owned()); - let mut attention = account("acct_attention", "user-attention"); - attention.email = Some("attention@example.invalid".to_owned()); - attention.availability = ProviderAccountAvailability::Invalid; + let mut invalid = account("acct_invalid", "user-invalid"); + invalid.email = Some("invalid@example.invalid".to_owned()); + invalid.availability = ProviderAccountAvailability::Invalid; let mut quota_exhausted = account("acct_quota_exhausted", "user-quota-exhausted"); quota_exhausted.email = Some("quota-exhausted@example.invalid".to_owned()); quota_exhausted.availability = ProviderAccountAvailability::QuotaExhausted; - for account in [alpha, beta, charlie, attention, quota_exhausted] { + for account in [alpha, beta, charlie, invalid, quota_exhausted] { repository .insert_provider_account(account) .await @@ -214,9 +214,8 @@ async fn terminal_admin_list_filters_and_sorts_before_pagination_with_retained_u assert_eq!(usage_page.config_revision.get(), 1); assert_eq!(usage_page.total, 5); assert_eq!(usage_page.summary.total, 5); - assert_eq!(usage_page.summary.active, 2); + assert_eq!(usage_page.summary.active, 3); assert_eq!(usage_page.summary.quota_exhausted, 1); - assert_eq!(usage_page.summary.attention, 1); assert_eq!( usage_page .items @@ -277,19 +276,6 @@ async fn terminal_admin_list_filters_and_sorts_before_pagination_with_retained_u .await .expect("filter banned accounts"); assert_eq!(banned.items[0].id, "acct_beta"); - let attention = store - .list_accounts(AccountListQuery { - page: 1, - page_size: PageSize::new(10).expect("page size"), - provider_kind: None, - search: None, - status: Some(AccountStatus::Attention), - sort: None, - }) - .await - .expect("filter attention accounts"); - assert_eq!(attention.items[0].id, "acct_attention"); - database.close().await; } diff --git a/backend/crates/providers/xai/src/credential/refresh.rs b/backend/crates/providers/xai/src/credential/refresh.rs index 3cad66f9c..96336caef 100644 --- a/backend/crates/providers/xai/src/credential/refresh.rs +++ b/backend/crates/providers/xai/src/credential/refresh.rs @@ -12,9 +12,10 @@ use gateway_core::engine::credential::{ AccountAvailability, CredentialRevision, ProviderAccountId, }; use gateway_core::provider_ports::{ - ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest, - ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest, ProviderRefreshPolicy, - ProviderRuntimePolicyPort, ProviderStoreError, provider_refresh_retry_at, + ProviderCooldown, ProviderCooldownPort, ProviderLeaseAcquisition, ProviderLeasePort, + ProviderLeaseRequest, ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest, + ProviderRefreshPolicy, ProviderRuntimePolicyPort, ProviderStoreError, + provider_refresh_retry_at, }; use super::catalog::GrokCredentialCatalogService; @@ -36,6 +37,7 @@ const MAX_REFRESH_EXCLUSIONS: usize = 400; const MAX_SECRET_BYTES: usize = 64 * 1_024; const DISCOVERY_NEGATIVE_CACHE_TTL: Duration = Duration::from_secs(5); const REFRESH_RETRY_DELAY: Duration = Duration::from_secs(10 * 60); +const REFRESH_AMBIGUITY_COOLDOWN: Duration = Duration::from_secs(24 * 60 * 60); /// 一个到期且已按 revision 读取明文 RT 的 xAI account。 pub struct DueGrokCredential { @@ -274,6 +276,7 @@ pub struct GrokCredentialRefreshService { refresher: Arc, catalog: Arc, leases: Arc, + cooldowns: Arc, runtime_policy: Arc, } @@ -283,6 +286,7 @@ impl GrokCredentialRefreshService { refresher: Arc, catalog: Arc, leases: Arc, + cooldowns: Arc, runtime_policy: Arc, ) -> Self { Self { @@ -290,6 +294,7 @@ impl GrokCredentialRefreshService { refresher, catalog, leases, + cooldowns, runtime_policy, } } @@ -297,9 +302,6 @@ impl GrokCredentialRefreshService { pub async fn refresh_due( &self, ) -> Result, GrokCredentialRefreshError> { - if !self.prepare_cycle_if_due().await? { - return Ok(Vec::new()); - } self.refresh_due_excluding(&[]).await } @@ -455,8 +457,21 @@ impl GrokCredentialRefreshService { .into_iter() .map(|account_id| GrokCredentialRefreshOutcome::Failed { account_id }) .collect::>(); - outcomes.reserve(batch.credentials.len()); - let refreshed = stream::iter(batch.credentials.into_iter().map(|credential| async move { + let mut credentials = Vec::with_capacity(batch.credentials.len()); + for credential in batch.credentials { + if !self.runtime_cooldown_active(&credential).await { + credentials.push(credential); + } + } + if credentials.is_empty() { + return Ok(outcomes); + } + self.refresher + .prepare_cycle() + .await + .map_err(|_| GrokCredentialRefreshError::Preparation)?; + outcomes.reserve(credentials.len()); + let refreshed = stream::iter(credentials.into_iter().map(|credential| async move { let account_id = credential.account_id.clone(); ( account_id, @@ -539,14 +554,11 @@ impl GrokCredentialRefreshService { ) .await } - Err(GrokRefreshFailure::Ambiguous) => self - .persist_terminal_failure( - credential, - GrokCredentialAvailability::Invalid, - "refresh_ambiguous", - ) - .await - .map(|_| GrokCredentialRefreshOutcome::Ambiguous { account_id }), + Err(GrokRefreshFailure::Ambiguous) => { + self.record_runtime_cooldown(&credential, REFRESH_AMBIGUITY_COOLDOWN) + .await; + Ok(GrokCredentialRefreshOutcome::Ambiguous { account_id }) + } Err(GrokRefreshFailure::Transient) => { if self .persist_backoff(&credential, "refresh-transient") @@ -644,6 +656,36 @@ impl GrokCredentialRefreshService { }) } + async fn runtime_cooldown_active(&self, credential: &DueGrokCredential) -> bool { + let Ok(Some(cooldown)) = self.cooldowns.read(credential.account_id()).await else { + return false; + }; + if cooldown.credential_revision() != credential.credential_revision() { + if cooldown.credential_revision() < credential.credential_revision() { + let _ = self + .cooldowns + .clear(credential.account_id(), credential.credential_revision()) + .await; + } + return false; + } + cooldown.until() > SystemTime::now() + } + + async fn record_runtime_cooldown(&self, credential: &DueGrokCredential, duration: Duration) { + let cooldown_until = SystemTime::now() + .checked_add(duration) + .unwrap_or_else(SystemTime::now); + let _ = self + .cooldowns + .put_if_later(ProviderCooldown::new( + credential.account_id.clone(), + credential.credential_revision, + cooldown_until, + )) + .await; + } + async fn persist_terminal_failure( &self, credential: DueGrokCredential, @@ -721,10 +763,12 @@ impl GrokCredentialRecovery for GrokCredentialRefreshService { GrokCredentialRefreshOutcome::Refreshed { .. } | GrokCredentialRefreshOutcome::Stale { .. }, ) => GrokCredentialRecoveryOutcome::Recovered, - Ok( - GrokCredentialRefreshOutcome::Invalidated { .. } - | GrokCredentialRefreshOutcome::Ambiguous { .. }, - ) => GrokCredentialRecoveryOutcome::Rejected, + Ok(GrokCredentialRefreshOutcome::Invalidated { .. }) => { + GrokCredentialRecoveryOutcome::Rejected + } + Ok(GrokCredentialRefreshOutcome::Ambiguous { .. }) => { + GrokCredentialRecoveryOutcome::Unavailable + } Ok( GrokCredentialRefreshOutcome::Transient { .. } | GrokCredentialRefreshOutcome::Rejected { .. } diff --git a/backend/crates/providers/xai/src/credential/selector.rs b/backend/crates/providers/xai/src/credential/selector.rs index 6616597da..5f8c01b02 100644 --- a/backend/crates/providers/xai/src/credential/selector.rs +++ b/backend/crates/providers/xai/src/credential/selector.rs @@ -7,10 +7,11 @@ use std::time::{Duration, SystemTime}; use chrono::Utc; use gateway_core::engine::credential::{ AccountCandidate, AccountFeedbackStats, AccountSelectionContext, AccountSelector, + ProviderAccount, ProviderAccountId, }; use gateway_core::provider_ports::{ - ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest, - ProviderSchedulingLeaseRequest, + ProviderCooldown, ProviderCooldownPort, ProviderLeaseAcquisition, ProviderLeasePort, + ProviderLeaseRequest, ProviderSchedulingLeaseRequest, }; use gateway_core::routing::ProviderKind; @@ -26,6 +27,7 @@ use crate::{ const DEFAULT_RATE_LIMIT_COOLDOWN: Duration = Duration::from_secs(60); const MAX_RATE_LIMIT_COOLDOWN: Duration = Duration::from_secs(24 * 60 * 60); const STREAM_INTERRUPTION_COOLDOWN: Duration = Duration::from_secs(30); +const UNAUTHORIZED_COOLDOWN: Duration = Duration::from_secs(60); /// 仅经 Core account port、TTL catalog cache 和 Redis lease 选择一个 OAuth session。 pub struct GrokAccountSessionSelector { @@ -34,6 +36,7 @@ pub struct GrokAccountSessionSelector { catalog_cache: Arc, quota: Arc, scheduling: Arc, + cooldowns: Arc, account_feedback: Arc, } @@ -45,6 +48,7 @@ impl GrokAccountSessionSelector { catalog_cache: Arc, quota: Arc, scheduling: Arc, + cooldowns: Arc, account_feedback: Arc, ) -> Self { Self { @@ -53,6 +57,7 @@ impl GrokAccountSessionSelector { catalog_cache, quota, scheduling, + cooldowns, account_feedback, } } @@ -94,8 +99,10 @@ impl GrokAccountSessionSelector { .load_state(&self.provider_kind, &account_ids) .await .map_err(|_| GrokSessionSelectorError::Unavailable)?; + let runtime_cooldowns = self.runtime_cooldowns(&catalog_eligible).await; let mut candidates = catalog_eligible .into_iter() + .filter(|account| !runtime_cooldowns.contains(account.id())) .map(|account| { let health = self .account_feedback @@ -195,6 +202,43 @@ impl GrokAccountSessionSelector { Err(GrokSessionSelectorError::NoEligibleSession) } } + + async fn runtime_cooldowns( + &self, + accounts: &[ProviderAccount], + ) -> std::collections::BTreeSet { + let now = SystemTime::now(); + let mut cooled = std::collections::BTreeSet::new(); + for account in accounts { + let Ok(Some(cooldown)) = self.cooldowns.read(account.id()).await else { + continue; + }; + if cooldown.credential_revision() != account.revision() { + if cooldown.credential_revision() < account.revision() { + let _ = self.cooldowns.clear(account.id(), account.revision()).await; + } + continue; + } + if cooldown.until() > now { + cooled.insert(account.id().clone()); + } + } + cooled + } + + async fn record_runtime_cooldown(&self, session: &SelectedGrokSession, duration: Duration) { + let cooldown_until = SystemTime::now() + .checked_add(duration) + .unwrap_or_else(SystemTime::now); + let _ = self + .cooldowns + .put_if_later(ProviderCooldown::new( + session.account_id().clone(), + session.credential_revision(), + cooldown_until, + )) + .await; + } } impl fmt::Debug for GrokAccountSessionSelector { @@ -205,6 +249,7 @@ impl fmt::Debug for GrokAccountSessionSelector { .field("catalog_cache", &"[TTL_CACHE]") .field("quota", &self.quota) .field("scheduling", &"[SCHEDULING_PORT]") + .field("cooldowns", &"[COOLDOWN_PORT]") .finish() } } @@ -221,12 +266,12 @@ impl GrokSessionSelector for GrokAccountSessionSelector { ) -> GrokCredentialFeedbackFuture<'a> { Box::pin(async move { let observed_at = Utc::now(); - let (availability, reason, cooldown_until) = match failure { - GrokCredentialFailure::Unauthorized => ( - GrokCredentialAvailability::Invalid, - "upstream_unauthorized", - None, - ), + let persistent = match failure { + GrokCredentialFailure::Unauthorized => { + self.record_runtime_cooldown(session, UNAUTHORIZED_COOLDOWN) + .await; + return; + } GrokCredentialFailure::QuotaExhausted => ( GrokCredentialAvailability::QuotaExhausted, "upstream_quota_exhausted", @@ -236,26 +281,16 @@ impl GrokSessionSelector for GrokAccountSessionSelector { let retry_after = retry_after .unwrap_or(DEFAULT_RATE_LIMIT_COOLDOWN) .min(MAX_RATE_LIMIT_COOLDOWN); - let cooldown_until = chrono::Duration::from_std(retry_after) - .ok() - .and_then(|duration| observed_at.checked_add_signed(duration)); - ( - GrokCredentialAvailability::Cooldown, - "upstream_rate_limited", - cooldown_until, - ) + self.record_runtime_cooldown(session, retry_after).await; + return; } GrokCredentialFailure::StreamInterrupted => { - let cooldown_until = chrono::Duration::from_std(STREAM_INTERRUPTION_COOLDOWN) - .ok() - .and_then(|duration| observed_at.checked_add_signed(duration)); - ( - GrokCredentialAvailability::Cooldown, - "upstream_stream_interrupted", - cooldown_until, - ) + self.record_runtime_cooldown(session, STREAM_INTERRUPTION_COOLDOWN) + .await; + return; } }; + let (availability, reason, cooldown_until) = persistent; let _ = self .repository .update_state(&UpdateGrokCredentialState { diff --git a/backend/crates/providers/xai/src/lib.rs b/backend/crates/providers/xai/src/lib.rs index c0556d045..c8cfc7799 100644 --- a/backend/crates/providers/xai/src/lib.rs +++ b/backend/crates/providers/xai/src/lib.rs @@ -103,6 +103,7 @@ pub async fn initialize( ProviderKind::new(XAI_PROVIDER_NAME).map_err(|_| XaiInitializeError::ProviderKind)?; let accounts: Arc = ports.accounts(); let leases = ports.leases(); + let cooldowns = ports.cooldowns(); let account_feedback = ports.account_feedback(); let runtime_policy = ports.runtime_policy(); let repository = GrokCredentialRepository::new(Arc::clone(&accounts)); @@ -135,6 +136,7 @@ pub async fn initialize( catalog_cache, Arc::clone("a), Arc::clone(&leases), + Arc::clone(&cooldowns), Arc::clone(&account_feedback), )); let inference: Arc = Arc::new( @@ -162,6 +164,7 @@ pub async fn initialize( refresher, Arc::clone(&catalog), Arc::clone(&leases), + cooldowns, Arc::clone(&runtime_policy), )); let credential_recovery: Arc = refresh.clone(); diff --git a/backend/crates/providers/xai/src/provider.rs b/backend/crates/providers/xai/src/provider.rs index f474ddd5f..8a0760669 100644 --- a/backend/crates/providers/xai/src/provider.rs +++ b/backend/crates/providers/xai/src/provider.rs @@ -18,13 +18,16 @@ use gateway_core::engine::provider::{ UpstreamTransport, }; use gateway_core::engine::{AttemptContext, ContinuationAttempt, UpstreamSendState}; -use gateway_core::error::{ContinuationFailure, ProviderError, ProviderErrorKind}; +use gateway_core::error::{ + ClientVisibleUpstreamError, ContinuationFailure, ProviderError, ProviderErrorKind, +}; use gateway_core::event::{GatewayEvent, ProviderEvent, ProviderResponseObservation, ResponseMeta}; use gateway_core::operation::{ Feature, GenerateRequest, Operation, OperationKind, ProviderSessionState, }; use gateway_core::routing::{ - ModelCapabilities, ProviderCandidate, ProviderKind, SupportLevel, UpstreamModelId, + ModelCapabilities, ModelPresentation, ProviderCandidate, ProviderKind, SupportLevel, + UpstreamModelId, }; use gateway_core::task::{ ScheduledTask, WorkerContribution, WorkerCycleContext, WorkerDefinitionError, WorkerId, @@ -35,7 +38,6 @@ use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; use url::Url; -use crate::GrokCatalogCapabilityEvidence; use crate::XaiWireProfileState; use crate::credential::{ GrokCredentialCatalogError, GrokCredentialCatalogService, GrokCredentialQuotaService, @@ -54,8 +56,10 @@ use crate::transport::{ GrokRequestEncodeError, GrokResponsesRequest, GrokSessionAffinityKey, GrokSessionSelection, GrokSessionSelector, GrokSessionSelectorError, SelectedGrokSession, }; +use crate::{GrokCatalogCapabilityEvidence, GrokCatalogModel}; const HTTP_SSE_TRANSPORT: &str = "http_sse"; +const DEFAULT_GROK_MODEL: &str = "grok-4.5"; const XAI_SESSION_STATE_MAX_BYTES: usize = 8 * 1024 * 1024; const XAI_SESSION_OUTPUT_LIMIT: usize = 4_096; const MIN_REASONING_CIPHERTEXT_BYTES: usize = 50; @@ -138,38 +142,15 @@ impl Provider for GrokBuildProvider { async fn query_model_capabilities( &self, ) -> Result, ProviderError> { - let models = self.catalog.query_models().await.map_err(|_| { - provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent) - })?; + let Ok(models) = self.catalog.query_models().await else { + return Ok(vec![default_grok_model_capabilities()?]); + }; + if models.is_empty() { + return Ok(vec![default_grok_model_capabilities()?]); + } Ok(models .into_iter() - .map(|model| { - let mut operations = BTreeSet::new(); - if model.capabilities().responses_api() - == GrokCatalogCapabilityEvidence::DeclaredNative - { - operations.insert(OperationKind::Generate); - } - let capabilities = ModelCapabilities::new( - operations, - model - .limits() - .max_output_tokens() - .map(std::num::NonZeroU64::get), - ) - .with_feature( - Feature::Reasoning, - support(model.capabilities().reasoning_effort()), - ) - .with_feature( - Feature::Tools, - tool_support(model.capabilities().streaming_tool_calls()), - ) - .with_feature(Feature::Vision, SupportLevel::Unknown) - .with_feature(Feature::JsonSchema, SupportLevel::Unknown) - .with_feature(Feature::NativeContinuation, SupportLevel::Native); - ProviderModelCapabilities::new(model.request_model().clone(), capabilities) - }) + .map(compile_grok_model_capabilities) .collect()) } @@ -403,6 +384,112 @@ fn tool_support(evidence: GrokCatalogCapabilityEvidence) -> SupportLevel { } } +fn compile_grok_model_capabilities(model: GrokCatalogModel) -> ProviderModelCapabilities { + let mut operations = BTreeSet::new(); + if model.capabilities().responses_api() == GrokCatalogCapabilityEvidence::DeclaredNative { + operations.insert(OperationKind::Generate); + } + let capabilities = ModelCapabilities::new( + operations, + model + .limits() + .max_output_tokens() + .map(std::num::NonZeroU64::get), + ) + .with_feature( + Feature::Reasoning, + support(model.capabilities().reasoning_effort()), + ) + .with_feature( + Feature::Tools, + tool_support(model.capabilities().streaming_tool_calls()), + ) + .with_feature(Feature::Vision, SupportLevel::Unknown) + .with_feature(Feature::JsonSchema, SupportLevel::Unknown) + .with_feature(Feature::NativeContinuation, SupportLevel::Native); + ProviderModelCapabilities::new(model.request_model().clone(), capabilities) + .with_presentation(grok_model_presentation(&model)) +} + +fn default_grok_model_capabilities() -> Result { + let model = UpstreamModelId::new(DEFAULT_GROK_MODEL.to_owned()) + .map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent))?; + let capabilities = ModelCapabilities::new(BTreeSet::from([OperationKind::Generate]), None) + .with_upstream_feature_validation(); + Ok(ProviderModelCapabilities::new(model, capabilities) + .with_presentation(default_grok_model_presentation())) +} + +fn default_grok_model_presentation() -> ModelPresentation { + ModelPresentation::new( + Some("Grok 4.5".to_owned()), + Some("xAI Grok 4.5 frontier model with reasoning and vision.".to_owned()), + ) + .with_reasoning( + Some("medium".to_owned()), + ["low", "medium", "high"].map(str::to_owned).to_vec(), + ) + .with_context_window_tokens(Some(500_000)) + .with_image_input(true) + .with_agent_tools(true, true) +} + +fn grok_model_presentation(model: &GrokCatalogModel) -> ModelPresentation { + let slug = model.request_model().as_str(); + let known_grok_4_5 = matches!( + slug, + DEFAULT_GROK_MODEL | "grok-4.5-latest" | "grok-4.5-build-free" | "grok-build-latest" + ); + let reasoning_evidence = model.capabilities().reasoning_effort(); + let (default_reasoning, reasoning_efforts) = match reasoning_evidence { + GrokCatalogCapabilityEvidence::DeclaredNative => ( + Some("medium".to_owned()), + ["low", "medium", "high"].map(str::to_owned).to_vec(), + ), + GrokCatalogCapabilityEvidence::DeclaredUnsupported => { + (Some("none".to_owned()), vec!["none".to_owned()]) + } + GrokCatalogCapabilityEvidence::Unknown if known_grok_4_5 => ( + Some("medium".to_owned()), + ["low", "medium", "high"].map(str::to_owned).to_vec(), + ), + GrokCatalogCapabilityEvidence::Unknown => (None, Vec::new()), + }; + let context_window_tokens = model + .limits() + .context_window_tokens() + .map(std::num::NonZeroU64::get) + .or(known_grok_4_5.then_some(500_000)); + let tool_evidence = model.capabilities().streaming_tool_calls(); + + ModelPresentation::new( + model + .display_name() + .map(str::to_owned) + .or_else(|| known_grok_4_5.then(|| "Grok 4.5".to_owned())), + model + .metadata() + .description() + .map(str::to_owned) + .or_else(|| { + known_grok_4_5 + .then(|| "xAI Grok 4.5 frontier model with reasoning and vision.".to_owned()) + }), + ) + .with_reasoning(default_reasoning, reasoning_efforts) + .with_context_window_tokens(context_window_tokens) + .with_image_input(known_grok_4_5) + .with_agent_tools( + tool_evidence != GrokCatalogCapabilityEvidence::DeclaredUnsupported, + tool_evidence == GrokCatalogCapabilityEvidence::DeclaredNative + || (known_grok_4_5 && tool_evidence == GrokCatalogCapabilityEvidence::Unknown), + ) + .with_search_tool( + model.capabilities().backend_search() == GrokCatalogCapabilityEvidence::DeclaredNative, + ) + .with_hidden(model.metadata().hidden().unwrap_or(false)) +} + #[derive(Clone, Serialize, Deserialize)] struct XaiSessionState { account_id: String, @@ -692,16 +779,12 @@ fn attach_xai_session_update( .into_iter() .map(XaiReplayItem::ClientInput), ); - transcript.extend( - capture - .output_items - .into_values() - .filter_map(|item| portable_output_item(item, false)) - .map(|item| XaiReplayItem::AccountOutput { - account_id: capture.account_id.clone(), - item, - }), - ); + transcript.extend(capture.output_items.into_values().map(|item| { + XaiReplayItem::AccountOutput { + account_id: capture.account_id.clone(), + item, + } + })); let state = XaiSessionState { account_id: capture.account_id, session_id: capture.session_id, @@ -722,7 +805,11 @@ fn capture_output_item(event: &ProviderEvent, capture: &mut GrokSessionCapture) .or_else(|| wire.data().get("type").and_then(Value::as_str)); if event_type == Some("response.output_item.done") && capture.output_items.len() < XAI_SESSION_OUTPUT_LIMIT - && let Some(item) = wire.data().get("item").cloned() + && let Some(item) = wire + .data() + .get("item") + .cloned() + .and_then(|item| portable_output_item(item, false)) { let index = wire .data() @@ -743,16 +830,17 @@ fn capture_output_item(event: &ProviderEvent, capture: &mut GrokSessionCapture) .and_then(Value::as_array) .filter(|output| !output.is_empty()) { - capture.output_items.clear(); - capture.output_items.extend( - output - .iter() - .take(XAI_SESSION_OUTPUT_LIMIT) - .enumerate() - .filter_map(|(index, item)| { - u32::try_from(index).ok().map(|index| (index, item.clone())) - }), - ); + for (index, item) in output.iter().take(XAI_SESSION_OUTPUT_LIMIT).enumerate() { + let Some(index) = u32::try_from(index).ok() else { + break; + }; + if capture.output_items.contains_key(&index) { + continue; + } + if let Some(item) = portable_output_item(item.clone(), false) { + capture.output_items.insert(index, item); + } + } } } @@ -1351,8 +1439,10 @@ async fn recover_or_record_failure( .await { GrokCredentialRecoveryOutcome::Recovered => error.with_same_account_retry(), - GrokCredentialRecoveryOutcome::Rejected - | GrokCredentialRecoveryOutcome::Unavailable => error, + GrokCredentialRecoveryOutcome::Rejected => error, + GrokCredentialRecoveryOutcome::Unavailable => { + record_failure(selector, session, error).await + } }; } record_failure(selector, session, error).await @@ -1415,10 +1505,23 @@ fn ensure_sent_context(context: &AttemptContext) -> Result<(), ProviderError> { fn map_request_error(error: GrokRequestEncodeError) -> ProviderError { let kind = match error { GrokRequestEncodeError::InvalidProtocolPayload - | GrokRequestEncodeError::InvalidRequestNormalization => ProviderErrorKind::InvalidRequest, + | GrokRequestEncodeError::InvalidRequestNormalization + | GrokRequestEncodeError::InvalidRequestField { .. } => ProviderErrorKind::InvalidRequest, GrokRequestEncodeError::Serialization => ProviderErrorKind::Protocol, }; - provider_error(kind, UpstreamSendState::NotSent) + let provider_error = provider_error(kind, UpstreamSendState::NotSent); + if kind != ProviderErrorKind::InvalidRequest { + return provider_error; + } + let detail = ClientVisibleUpstreamError::new( + error.to_string(), + Some("invalid_request_normalization".to_owned()), + Some("invalid_request_error".to_owned()), + ); + match detail { + Ok(detail) => provider_error.with_client_visible_upstream_error(detail), + Err(_) => provider_error, + } } fn map_selection_error(error: GrokSessionSelectorError) -> ProviderError { diff --git a/backend/crates/providers/xai/src/transport/network.rs b/backend/crates/providers/xai/src/transport/network.rs index 1b5da13b7..b6959d3bf 100644 --- a/backend/crates/providers/xai/src/transport/network.rs +++ b/backend/crates/providers/xai/src/transport/network.rs @@ -356,12 +356,6 @@ impl GrokInferenceTransport for ReqwestGrokInferenceTransport { if !response.status().is_success() { return Err(classify_inference_status(response).await); } - if !is_event_stream(&response) { - return Err(GrokInferenceTransportError::new( - GrokInferenceTransportErrorKind::Protocol, - UpstreamSendState::Sent, - )); - } let http_version = upstream_http_version(response.version()); let status_code = response.status().as_u16(); let request_id = upstream_request_id(&response); @@ -1109,15 +1103,6 @@ fn retry_after(response: &Response) -> Option { .map(Duration::from_secs) } -fn is_event_stream(response: &Response) -> bool { - response - .headers() - .get(CONTENT_TYPE) - .and_then(|value| value.to_str().ok()) - .and_then(|value| value.split(';').next()) - .is_some_and(|value| value.trim().eq_ignore_ascii_case("text/event-stream")) -} - fn is_json_response(response: &Response) -> bool { response .headers() diff --git a/backend/crates/providers/xai/src/transport/request.rs b/backend/crates/providers/xai/src/transport/request.rs index 10e9aa43e..d09960d9d 100644 --- a/backend/crates/providers/xai/src/transport/request.rs +++ b/backend/crates/providers/xai/src/transport/request.rs @@ -1,4 +1,4 @@ -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::fmt; use gateway_core::operation::GenerateRequest; @@ -279,6 +279,18 @@ pub enum GrokRequestEncodeError { /// Responses compatibility fields could not be normalized safely. #[error("Grok Build request normalization failed")] InvalidRequestNormalization, + /// 请求中的具体字段无法安全转换为 Grok Build 接受的形态。 + #[error("Grok Build request field `{field}` could not be normalized safely")] + InvalidRequestField { field: &'static str }, +} + +impl GrokRequestEncodeError { + fn at_field(self, field: &'static str) -> Self { + match self { + Self::InvalidRequestNormalization => Self::InvalidRequestField { field }, + error => error, + } + } } fn sanitize_account_identity(body: &mut Map) { @@ -548,11 +560,12 @@ fn normalize_responses_request( } let text = text .as_object_mut() - .ok_or(GrokRequestEncodeError::InvalidRequestNormalization)?; + .ok_or(GrokRequestEncodeError::InvalidRequestField { field: "text" })?; if text.get("format").is_none_or(Value::is_null) { text.insert( "format".to_owned(), - normalize_response_format(response_format)?, + normalize_response_format(response_format) + .map_err(|error| error.at_field("response_format"))?, ); } } @@ -711,20 +724,22 @@ impl GrokResponseTransform { .stream_calls .get_mut(&primary) .ok_or(GrokRequestEncodeError::InvalidRequestNormalization)?; - if matches!( - state.identity.kind, - ToolKind::ToolSearch | ToolKind::Custom | ToolKind::ApplyPatch - ) { - state.arguments.push_str( - value - .get("delta") - .and_then(Value::as_str) - .unwrap_or_default(), - ); - if state.identity.kind == ToolKind::Custom { - state.last_delta = value.as_object().cloned(); + match state.identity.kind { + ToolKind::Custom => { + state.arguments.push_str( + value + .get("delta") + .and_then(Value::as_str) + .unwrap_or_default(), + ); + state.last_delta = value.as_object().cloned().map(|mut payload| { + payload.remove("delta"); + payload + }); + return Ok(Vec::new()); } - return Ok(Vec::new()); + ToolKind::ToolSearch | ToolKind::ApplyPatch => return Ok(Vec::new()), + ToolKind::Function => {} } } @@ -767,6 +782,10 @@ impl GrokResponseTransform { &input, ), }); + if let Some(state) = self.stream_calls.get_mut(&primary) { + state.arguments.clear(); + state.last_delta = None; + } return Ok(output); } ToolKind::Function => {} @@ -783,10 +802,17 @@ impl GrokResponseTransform { return self.rewrite_apply_patch_done_event(value); } + let completed_call = (event_type == "response.output_item.done") + .then(|| value.get("item").and_then(Value::as_object)) + .flatten() + .and_then(|item| self.primary_stream_call(item)); self.rewrite_response_value(&mut value)?; if let Some(response) = value.get_mut("response").and_then(Value::as_object_mut) { self.restore_visible_tools(response); } + if let Some(primary) = completed_call { + self.take_stream_call(&primary); + } Ok(vec![GrokTransformedWireEvent { event_type: event_type.to_owned(), value, @@ -847,6 +873,18 @@ impl GrokResponseTransform { Some(primary) } + fn primary_stream_call(&self, item: &Map) -> Option { + ["id", "call_id"] + .into_iter() + .filter_map(|field| item.get(field).and_then(Value::as_str)) + .find_map(|key| self.stream_keys.get(key).cloned()) + } + + fn take_stream_call(&mut self, primary: &str) -> Option { + self.stream_keys.retain(|_, owner| owner != primary); + self.stream_calls.remove(primary) + } + fn rewrite_response_value(&self, value: &mut Value) -> Result<(), GrokRequestEncodeError> { match value { Value::Array(values) => { @@ -953,15 +991,11 @@ impl GrokResponseTransform { .and_then(Value::as_object) .cloned() .ok_or(GrokRequestEncodeError::InvalidRequestNormalization)?; - let primary = ["id", "call_id"] - .into_iter() - .find_map(|field| original_item.get(field).and_then(Value::as_str)) - .and_then(|key| self.stream_keys.get(key)) - .cloned(); - let mut added = primary - .as_ref() - .and_then(|key| self.stream_calls.get(key)) - .and_then(|state| state.added_payload.clone()) + let state = self + .primary_stream_call(&original_item) + .and_then(|primary| self.take_stream_call(&primary)); + let mut added = state + .and_then(|state| state.added_payload) .unwrap_or_else(|| { Map::from_iter([( "type".to_owned(), @@ -1148,20 +1182,27 @@ impl ToolNormalizer { mut self, payload: &mut Map, ) -> Result { - let (tools, had_tools) = optional_array(payload.get("tools"))?; + let (tools, had_tools) = + optional_array(payload.get("tools")).map_err(|error| error.at_field("tools"))?; if had_tools { self.response.visible_tools.clone_from(&tools); } - let client_search = inspect_tool_search(&tools)?; - self.normalize_client_search_parallel(payload, client_search)?; + let client_search = inspect_tool_search(&tools).map_err(|error| error.at_field("tools"))?; + self.normalize_client_search_parallel(payload, client_search) + .map_err(|error| error.at_field("parallel_tool_calls"))?; let mut normalized_tools = Vec::with_capacity(tools.len()); for raw_tool in &tools { - normalized_tools.extend(self.normalize_tool(raw_tool, "", client_search, false)?); + normalized_tools.extend( + self.normalize_tool(raw_tool, "", client_search, false) + .map_err(|error| error.at_field("tools"))?, + ); } if let Some(Value::Array(items)) = payload.get("input") { - let normalized = self.normalize_input_items(items)?; + let normalized = self + .normalize_input_items(items) + .map_err(|error| error.at_field("input"))?; normalized_tools.extend(normalized.loaded_tools); self.response.visible_tools.extend(normalized.visible_tools); payload.insert("input".to_owned(), Value::Array(normalized.items)); @@ -1169,11 +1210,14 @@ impl ToolNormalizer { .get("input") .is_some_and(|input| !input.is_null() && !input.is_string()) { - return Err(GrokRequestEncodeError::InvalidRequestNormalization); + return Err(GrokRequestEncodeError::InvalidRequestField { field: "input" }); } if self.client_search_tool.is_some() { - normalized_tools.push(Value::Object(self.build_client_search_function()?)); + normalized_tools.push(Value::Object( + self.build_client_search_function() + .map_err(|error| error.at_field("tools"))?, + )); } normalized_tools = dedupe_normalized_tools(normalized_tools); if normalized_tools.is_empty() { @@ -1184,7 +1228,8 @@ impl ToolNormalizer { } else { payload.insert("tools".to_owned(), Value::Array(normalized_tools.clone())); } - self.normalize_tool_choice(payload, &normalized_tools)?; + self.normalize_tool_choice(payload, &normalized_tools) + .map_err(|error| error.at_field("tool_choice"))?; Ok(self.response) } @@ -1299,6 +1344,11 @@ impl ToolNormalizer { return Ok(Vec::new()); } let mut converted = without_defer_loading(tool); + if let Some(parameters) = converted.get("parameters").cloned() + && let Some(normalized) = normalize_function_parameters_root(¶meters)? + { + converted.insert("parameters".to_owned(), normalized); + } let alias = self.alias(ToolIdentity::new(ToolKind::Function, namespace, name)); converted.insert("name".to_owned(), Value::String(alias)); Ok(vec![Value::Object(converted)]) @@ -1448,6 +1498,136 @@ fn optional_array(value: Option<&Value>) -> Result<(Vec, bool), GrokReque } } +/// Grok Build 要求 function parameters 的根节点必为 object；Codex 会为可选参数 +/// 生成 `object | null`，这里只移除根节点 nullability，嵌套字段保持原样。 +fn normalize_function_parameters_root( + value: &Value, +) -> Result, GrokRequestEncodeError> { + let Some(schema) = value.as_object() else { + return Ok(None); + }; + let mut normalized = schema.clone(); + let mut changed = false; + + if let Some(types) = normalized.get("type").and_then(Value::as_array).cloned() { + let removed_null = types.iter().any(|value| value.as_str() == Some("null")); + if removed_null { + let remaining = types + .into_iter() + .filter(|value| value.as_str() != Some("null")) + .collect::>(); + if remaining.len() != 1 || remaining[0].as_str() != Some("object") { + return Err(invalid_function_parameters_root()); + } + normalized.insert("type".to_owned(), Value::String("object".to_owned())); + changed = true; + } + } + + for keyword in ["anyOf", "oneOf"] { + let Some(branches) = normalized.get(keyword).and_then(Value::as_array).cloned() else { + continue; + }; + let removed_null = branches.iter().any(is_null_only_schema); + if !removed_null { + continue; + } + let remaining = branches + .into_iter() + .filter(|branch| !is_null_only_schema(branch)) + .collect::>(); + if remaining.is_empty() + || remaining.iter().any(|branch| { + branch.as_object().is_none_or(|branch| { + !is_object_root_schema(branch, &normalized, &mut BTreeSet::new()) + }) + }) + { + return Err(invalid_function_parameters_root()); + } + if remaining.len() == 1 && normalized.len() == 1 { + normalized = remaining[0] + .as_object() + .cloned() + .ok_or_else(invalid_function_parameters_root)?; + normalized.insert("type".to_owned(), Value::String("object".to_owned())); + } else { + normalized.insert(keyword.to_owned(), Value::Array(remaining)); + normalized.insert("type".to_owned(), Value::String("object".to_owned())); + } + changed = true; + } + + Ok(changed.then_some(Value::Object(normalized))) +} + +fn is_null_only_schema(value: &Value) -> bool { + let Some(schema) = value.as_object() else { + return false; + }; + match schema.get("type") { + Some(Value::String(kind)) => kind == "null", + Some(Value::Array(types)) if !types.is_empty() => { + types.iter().all(|kind| kind.as_str() == Some("null")) + } + _ => false, + } +} + +fn is_object_root_schema( + schema: &Map, + root: &Map, + visited: &mut BTreeSet, +) -> bool { + match schema.get("type") { + Some(Value::String(kind)) => return kind == "object", + Some(Value::Array(types)) if !types.is_empty() => { + return types.iter().all(|kind| kind.as_str() == Some("object")); + } + Some(_) => return false, + None => {} + } + if schema.contains_key("properties") { + return true; + } + let Some(reference) = schema.get("$ref").and_then(Value::as_str) else { + return false; + }; + if !visited.insert(reference.to_owned()) { + return false; + } + resolve_local_schema_ref(root, reference) + .is_some_and(|resolved| is_object_root_schema(resolved, root, visited)) +} + +fn resolve_local_schema_ref<'a>( + root: &'a Map, + reference: &str, +) -> Option<&'a Map> { + if reference == "#" { + return Some(root); + } + let path = reference.strip_prefix("#/")?; + let mut segments = path.split('/'); + let first = decode_json_pointer_segment(segments.next()?); + let mut current = root.get(&first)?; + for segment in segments { + let segment = decode_json_pointer_segment(segment); + current = current.as_object()?.get(&segment)?; + } + current.as_object() +} + +fn decode_json_pointer_segment(segment: &str) -> String { + segment.replace("~1", "/").replace("~0", "~") +} + +const fn invalid_function_parameters_root() -> GrokRequestEncodeError { + GrokRequestEncodeError::InvalidRequestField { + field: "tools[].parameters", + } +} + fn inspect_tool_search(tools: &[Value]) -> Result { let mut client_search = false; let mut server_search = false; diff --git a/backend/crates/providers/xai/tests/credential/refresh.rs b/backend/crates/providers/xai/tests/credential/refresh.rs index 171816539..8c37ad57f 100644 --- a/backend/crates/providers/xai/tests/credential/refresh.rs +++ b/backend/crates/providers/xai/tests/credential/refresh.rs @@ -20,8 +20,8 @@ use provider_xai::{ }; use crate::support::{ - MemoryGrokCatalogCache, MemoryProviderAccountStore, create_input, credential_object, - runtime_policy, seed_input, + MemoryCooldownPort, MemoryGrokCatalogCache, MemoryProviderAccountStore, create_input, + credential_object, runtime_policy, seed_input, }; const OFFICIAL_FIXTURE: &[u8] = @@ -169,11 +169,13 @@ async fn fixture_many( available: lease_available, calls: AtomicUsize::new(0), }); + let cooldowns = Arc::new(MemoryCooldownPort::default()); let service = GrokCredentialRefreshService::new( repository.clone(), refresher_port, catalog, leases, + cooldowns, runtime_policy(), ); (store, repository, refresher, service) @@ -380,10 +382,11 @@ async fn banned_failure_marks_account_banned() { } #[tokio::test] -async fn ambiguous_refresh_fails_closed_and_is_not_retried() { +async fn ambiguous_refresh_uses_runtime_cooldown_without_persisting_account_state() { let input = due_input("ambiguous"); let id = input.account_id.clone(); - let (store, _, _, service) = fixture(input, [Err(GrokRefreshFailure::Ambiguous)], true).await; + let (store, _, refresher, service) = + fixture(input, [Err(GrokRefreshFailure::Ambiguous)], true).await; let outcomes = service.refresh_due().await.expect("refresh"); assert!(matches!( outcomes.as_slice(), @@ -391,8 +394,9 @@ async fn ambiguous_refresh_fails_closed_and_is_not_retried() { )); assert_eq!( store.account(&id).expect("account").availability(), - AccountAvailability::Invalid + AccountAvailability::Unknown ); + assert_eq!(store.account(&id).expect("account").revision().get(), 1); assert!( service .refresh_due() @@ -400,6 +404,7 @@ async fn ambiguous_refresh_fails_closed_and_is_not_retried() { .expect("second cycle") .is_empty() ); + assert_eq!(refresher.prepare_calls.load(Ordering::SeqCst), 1); } #[tokio::test] diff --git a/backend/crates/providers/xai/tests/credential/selector.rs b/backend/crates/providers/xai/tests/credential/selector.rs index a1033f93b..ecaf12a7b 100644 --- a/backend/crates/providers/xai/tests/credential/selector.rs +++ b/backend/crates/providers/xai/tests/credential/selector.rs @@ -23,7 +23,8 @@ use provider_xai::{ }; use crate::support::{ - MemoryGrokCatalogCache, MemoryProviderAccountStore, account_id, create_input, seed_input, + MemoryCooldownPort, MemoryGrokCatalogCache, MemoryProviderAccountStore, account_id, + create_input, seed_input, }; struct SchedulingCoordinator { @@ -88,6 +89,7 @@ struct SelectorFixture { cache: Arc, selector: GrokAccountSessionSelector, coordinator: Arc, + cooldowns: Arc, feedback: Arc, } @@ -137,6 +139,7 @@ impl SelectorFixture { signals: Mutex::new(signals), denied: Mutex::new(BTreeSet::new()), }); + let cooldowns = Arc::new(MemoryCooldownPort::default()); let catalog_cache: Arc = cache.clone(); let lease_port: Arc = coordinator.clone(); let quota = Arc::new(crate::support::grok_quota_service( @@ -150,6 +153,7 @@ impl SelectorFixture { catalog_cache, quota, lease_port, + cooldowns.clone(), Arc::clone(&feedback), ); Self { @@ -157,6 +161,7 @@ impl SelectorFixture { cache, selector, coordinator, + cooldowns, feedback, } } @@ -257,7 +262,7 @@ async fn required_account_overrides_smart_selection_without_fallback() { } #[tokio::test] -async fn unauthorized_feedback_marks_only_selected_account_invalid() { +async fn unauthorized_feedback_records_runtime_cooldown_without_persisting_account_state() { let fixture = SelectorFixture::new(&["feedback-a", "feedback-b"]).await; let session = fixture .selector @@ -275,7 +280,7 @@ async fn unauthorized_feedback_marks_only_selected_account_invalid() { .account(&selected) .expect("selected") .availability(), - AccountAvailability::Invalid + AccountAvailability::Ready ); let other = [account_id("feedback-a"), account_id("feedback-b")] .into_iter() @@ -285,16 +290,78 @@ async fn unauthorized_feedback_marks_only_selected_account_invalid() { fixture.store.account(&other).expect("other").availability(), AccountAvailability::Ready ); + assert!( + fixture + .cooldowns + .cooldown(&selected) + .is_some_and(|cooldown| cooldown.until() > SystemTime::now()) + ); } #[tokio::test] -async fn rate_limit_feedback_applies_bounded_cooldown() { - let fixture = SelectorFixture::new(&["rate-limit"]).await; +async fn cooldown_for_superseded_credential_does_not_block_rotated_account() { + let fixture = SelectorFixture::new(&["revision-fence"]).await; + let id = account_id("revision-fence"); let session = fixture .selector .select(fixture.request(BTreeSet::new())) .await .expect("session"); + fixture + .selector + .record_failure( + &session, + GrokCredentialFailure::RateLimited { + retry_after: Some(Duration::from_secs(60)), + }, + ) + .await; + + let current = fixture + .store + .load_credential(&id, CredentialRevision::new(1).expect("revision")) + .await + .expect("current credential"); + let prepared = GrokCredentialAdmin + .prepare_rotation(&RotateManagedGrokCredential { + current, + secret: provider_xai::GrokOAuthSecret { + access_token: provider_xai::SecretValue::new("new-access"), + refresh_token: provider_xai::SecretValue::new("new-refresh"), + id_token: None, + scope: provider_xai::OFFICIAL_SCOPES.join(" "), + }, + verified_account: crate::support::profile("subject-revision-fence"), + next_refresh_at: chrono::Utc::now() + chrono::Duration::minutes(30), + }) + .expect("rotate"); + assert!(matches!( + fixture + .store + .compare_and_swap_credential(prepared.credential) + .await + .expect("persist rotation"), + CredentialCasOutcome::Updated(revision) if revision.get() == 2 + )); + + let next = fixture + .selector + .select(fixture.request(BTreeSet::new())) + .await + .expect("replacement credential should not inherit cooldown"); + assert_eq!(next.account_id(), &id); + assert!(fixture.cooldowns.cooldown(&id).is_none()); +} + +#[tokio::test] +async fn rate_limit_feedback_records_runtime_cooldown_without_persisting_account_state() { + let fixture = SelectorFixture::new(&["rate-limit", "available"]).await; + let session = fixture + .selector + .select(fixture.request(BTreeSet::new())) + .await + .expect("session"); + let selected = session.account_id().clone(); fixture .selector .record_failure( @@ -304,20 +371,24 @@ async fn rate_limit_feedback_applies_bounded_cooldown() { }, ) .await; - let account = fixture - .store - .account(session.account_id()) - .expect("account"); - assert_eq!(account.availability(), AccountAvailability::Cooldown); + let account = fixture.store.account(&selected).expect("account"); + assert_eq!(account.availability(), AccountAvailability::Ready); assert!( - account - .cooldown_until() - .is_some_and(|until| until > SystemTime::now()) + fixture + .cooldowns + .cooldown(&selected) + .is_some_and(|cooldown| cooldown.until() > SystemTime::now()) ); + let next = fixture + .selector + .select(fixture.request(BTreeSet::new())) + .await + .expect("runtime cooldown should leave the other account available"); + assert_ne!(next.account_id(), &selected); } #[tokio::test] -async fn interrupted_stream_feedback_applies_account_cooldown() { +async fn interrupted_stream_feedback_records_runtime_cooldown_without_persisting_account_state() { let fixture = SelectorFixture::new(&["stream-interrupted"]).await; let session = fixture .selector @@ -333,12 +404,20 @@ async fn interrupted_stream_feedback_applies_account_cooldown() { .store .account(session.account_id()) .expect("account"); - assert_eq!(account.availability(), AccountAvailability::Cooldown); + assert_eq!(account.availability(), AccountAvailability::Ready); assert!( - account - .cooldown_until() - .is_some_and(|until| until > SystemTime::now()) + fixture + .cooldowns + .cooldown(session.account_id()) + .is_some_and(|cooldown| cooldown.until() > SystemTime::now()) ); + assert!(matches!( + fixture + .selector + .select(fixture.request(BTreeSet::new())) + .await, + Err(GrokSessionSelectorError::NoEligibleSession) + )); } #[tokio::test] diff --git a/backend/crates/providers/xai/tests/provider/contract.rs b/backend/crates/providers/xai/tests/provider/contract.rs index 56d6f2b4b..b235f6a39 100644 --- a/backend/crates/providers/xai/tests/provider/contract.rs +++ b/backend/crates/providers/xai/tests/provider/contract.rs @@ -33,7 +33,8 @@ use provider_xai::{ GrokCredentialFeedbackFuture, GrokCredentialRecovery, GrokCredentialRecoveryOutcome, GrokCredentialRepository, GrokInferenceRequest, GrokInferenceResponse, GrokInferenceTransport, GrokInferenceTransportError, GrokInferenceTransportErrorKind, GrokInferenceTransportFuture, - GrokModelCatalogRequest, GrokModelCatalogTransport, GrokModelCatalogTransportFuture, + GrokModelCatalogRequest, GrokModelCatalogTransport, GrokModelCatalogTransportError, + GrokModelCatalogTransportErrorKind, GrokModelCatalogTransportFuture, GrokModelCatalogTransportResponse, GrokSessionBinding, GrokSessionSelection, GrokSessionSelector, GrokSessionSelectorError, GrokSessionSelectorFuture, SecretValue, SelectedGrokSession, @@ -352,6 +353,18 @@ impl GrokModelCatalogTransport for CatalogWithoutToolMetadataTransport { } } +struct UnavailableCatalogTransport; + +impl GrokModelCatalogTransport for UnavailableCatalogTransport { + fn execute(&self, _: GrokModelCatalogRequest) -> GrokModelCatalogTransportFuture<'_> { + Box::pin(async { + Err(GrokModelCatalogTransportError::new( + GrokModelCatalogTransportErrorKind::Unavailable, + )) + }) + } +} + struct StubRecovery { calls: AtomicUsize, outcome: GrokCredentialRecoveryOutcome, @@ -506,6 +519,19 @@ fn operation_with_reasoning_effort(effort: &str) -> Operation { Operation::Generate(GenerateRequest::from_protocol_payload(payload)) } +fn operation_with_invalid_tools() -> Operation { + let payload = ProtocolPayload::json_object( + "openai", + Map::from_iter([ + ("model".to_owned(), json!("client-model")), + ("input".to_owned(), json!("hello")), + ("tools".to_owned(), json!({"not": "an array"})), + ]), + ) + .expect("OpenAI payload"); + Operation::Generate(GenerateRequest::from_protocol_payload(payload)) +} + fn compaction_operation() -> Operation { let payload = ProtocolPayload::json_object( "openai", @@ -1047,6 +1073,38 @@ async fn first_unauthorized_should_refresh_and_request_one_same_account_retry() assert!(selector.feedback.lock().expect("feedback").is_empty()); } +#[tokio::test] +async fn unavailable_unauthorized_recovery_records_temporary_credential_feedback() { + let selector = StubSelector::success(); + let transport = StubInferenceTransport::error( + GrokInferenceTransportError::new( + GrokInferenceTransportErrorKind::Unauthorized, + UpstreamSendState::Sent, + ) + .with_status(401) + .with_credential_recovery(), + ); + let recovery = StubRecovery::new(GrokCredentialRecoveryOutcome::Unavailable); + let provider = provider_with_recovery(selector.clone(), transport, recovery.clone()).await; + let mut stream = provider + .execute( + provider_request("xai"), + context(CancellationToken::new(), None), + ) + .await + .expect("stream"); + + let error = next_provider_error(&mut stream).await; + + assert_eq!(error.kind(), ProviderErrorKind::Unauthorized); + assert!(!error.retries_same_account()); + assert_eq!(recovery.calls.load(Ordering::SeqCst), 1); + assert_eq!( + selector.feedback.lock().expect("feedback").as_slice(), + &[GrokCredentialFailure::Unauthorized] + ); +} + #[tokio::test] async fn explicit_http_429_marks_provider_error_replay_safe() { let error = mapped_transport_error( @@ -1628,6 +1686,37 @@ async fn provider_rejects_target_owned_by_other_provider() { assert_eq!(selector.calls.load(Ordering::SeqCst), 0); } +#[tokio::test] +async fn provider_should_expose_safe_request_normalization_field_before_selection() { + let selector = StubSelector::success(); + let provider = provider(selector.clone(), StubInferenceTransport::success()).await; + let error = match provider + .execute( + provider_request_with_operation("xai", operation_with_invalid_tools()), + context(CancellationToken::new(), None), + ) + .await + { + Ok(_) => panic!("invalid tools must fail"), + Err(error) => error, + }; + + assert_eq!(error.kind(), ProviderErrorKind::InvalidRequest); + assert_eq!( + error.client_visible_upstream_error().map(|detail| ( + detail.message(), + detail.code(), + detail.error_type() + )), + Some(( + "Grok Build request field `tools` could not be normalized safely", + Some("invalid_request_normalization"), + Some("invalid_request_error"), + )), + ); + assert_eq!(selector.calls.load(Ordering::SeqCst), 0); +} + #[tokio::test] async fn provider_compiles_realtime_catalog_capabilities() { let provider = provider(StubSelector::success(), StubInferenceTransport::success()).await; @@ -1645,6 +1734,41 @@ async fn provider_compiles_realtime_catalog_capabilities() { )) .is_some() ); + let presentation = capabilities[0] + .presentation() + .expect("Grok model presentation"); + assert_eq!(presentation.display_name(), Some("Grok 4.5")); + assert_eq!( + presentation.supported_reasoning_efforts(), + ["low", "medium", "high"] + ); + assert_eq!(presentation.context_window_tokens(), Some(1_000_000)); + assert!(presentation.agent_tools()); +} + +#[tokio::test] +async fn provider_publishes_default_codex_profile_when_catalog_is_unavailable() { + let provider = provider_with_catalog_transport( + StubSelector::success(), + StubInferenceTransport::success(), + StubRecovery::new(GrokCredentialRecoveryOutcome::Unavailable), + Arc::new(UnavailableCatalogTransport), + ) + .await; + let capabilities = provider + .query_model_capabilities() + .await + .expect("fallback capabilities"); + + assert_eq!(capabilities.len(), 1); + assert_eq!(capabilities[0].upstream_model().as_str(), MODEL); + let presentation = capabilities[0] + .presentation() + .expect("fallback Grok model presentation"); + assert_eq!(presentation.display_name(), Some("Grok 4.5")); + assert_eq!(presentation.context_window_tokens(), Some(500_000)); + assert!(presentation.agent_tools()); + assert!(presentation.parallel_tool_calls()); } #[tokio::test] diff --git a/backend/crates/providers/xai/tests/support.rs b/backend/crates/providers/xai/tests/support.rs index 155dc05ce..8a61d07da 100644 --- a/backend/crates/providers/xai/tests/support.rs +++ b/backend/crates/providers/xai/tests/support.rs @@ -15,9 +15,9 @@ use gateway_core::engine::credential::{ }; use gateway_core::error::{StoreError, StoreErrorKind}; use gateway_core::provider_ports::{ - ProviderRefreshPolicy, ProviderRuntimePolicyPort, ProviderSessionAffinityKey, - ProviderSessionAffinityPort, ProviderSessionExclusionPort, ProviderSessionExclusions, - ProviderStoreError, + ProviderCooldown, ProviderCooldownPort, ProviderRefreshPolicy, ProviderRuntimePolicyPort, + ProviderSessionAffinityKey, ProviderSessionAffinityPort, ProviderSessionExclusionPort, + ProviderSessionExclusions, ProviderStoreError, }; use gateway_core::routing::ProviderKind; use provider_xai::{ @@ -550,6 +550,17 @@ pub struct TestSessionAffinity; pub struct TestSessionExclusions; +#[derive(Default)] +pub struct MemoryCooldownPort { + cooldowns: Mutex>, +} + +impl MemoryCooldownPort { + pub fn cooldown(&self, account_id: &ProviderAccountId) -> Option { + lock(&self.cooldowns).get(account_id).cloned() + } +} + impl ProviderSessionAffinityPort for TestSessionAffinity { fn load<'a>( &'a self, @@ -613,6 +624,50 @@ impl ProviderSessionExclusionPort for TestSessionExclusions { } } +impl ProviderCooldownPort for MemoryCooldownPort { + fn put_if_later( + &self, + cooldown: ProviderCooldown, + ) -> futures::future::BoxFuture<'_, Result> { + Box::pin(async move { + let mut cooldowns = lock(&self.cooldowns); + let should_write = cooldowns.get(cooldown.account_id()).is_none_or(|current| { + current.credential_revision() < cooldown.credential_revision() + || (current.credential_revision() == cooldown.credential_revision() + && current.until() < cooldown.until()) + }); + if should_write { + cooldowns.insert(cooldown.account_id().clone(), cooldown); + } + Ok(should_write) + }) + } + + fn read<'a>( + &'a self, + account_id: &'a ProviderAccountId, + ) -> futures::future::BoxFuture<'a, Result, ProviderStoreError>> { + Box::pin(async move { Ok(self.cooldown(account_id)) }) + } + + fn clear<'a>( + &'a self, + account_id: &'a ProviderAccountId, + through_revision: CredentialRevision, + ) -> futures::future::BoxFuture<'a, Result> { + Box::pin(async move { + let mut cooldowns = lock(&self.cooldowns); + let should_remove = cooldowns + .get(account_id) + .is_some_and(|current| current.credential_revision() <= through_revision); + if should_remove { + cooldowns.remove(account_id); + } + Ok(should_remove) + }) + } +} + impl ProviderRuntimePolicyPort for StaticRuntimePolicy { fn load_refresh_policy( &self, diff --git a/backend/crates/providers/xai/tests/transport/canonical.rs b/backend/crates/providers/xai/tests/transport/canonical.rs index 384d30aff..cfeee52f7 100644 --- a/backend/crates/providers/xai/tests/transport/canonical.rs +++ b/backend/crates/providers/xai/tests/transport/canonical.rs @@ -114,6 +114,42 @@ fn decoder_should_normalize_text_usage_and_completion() { )); } +#[test] +fn decoder_should_emit_each_complete_text_event_without_waiting_for_the_stream_end() { + let mut decoder = GrokCanonicalDecoder::new("grok-4.5"); + let created = decoder + .push( + concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_live\",\"model\":\"grok-4.5\"}}\n\n", + ) + .as_bytes(), + ) + .expect("created event"); + let delta = decoder + .push( + concat!( + "event: response.output_text.delta\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"content_index\":0,\"delta\":\"live\"}\n\n", + ) + .as_bytes(), + ) + .expect("text delta"); + + assert!( + created + .iter() + .filter_map(ProviderEvent::wire_event) + .any(|event| event.event_type() == Some("response.created")) + ); + assert!(delta.iter().any(|event| { + event.wire_event().is_some_and(|wire| { + wire.event_type() == Some("response.output_text.delta") + && wire.data().get("delta") == Some(&serde_json::json!("live")) + }) + })); +} + #[test] fn decoder_should_preserve_image_tool_tokens_in_canonical_usage() { let body = concat!( @@ -416,6 +452,72 @@ fn decoder_should_restore_custom_apply_patch_arguments_as_raw_input() { })); } +#[test] +fn decoder_should_buffer_only_custom_arguments_until_the_done_event() { + let request = tool_request(serde_json::json!({ + "model": "client", + "input": "render", + "tools": [{"type": "custom", "name": "render"}] + })); + let mut decoder = GrokCanonicalDecoder::for_request("grok-4.5", &request); + decoder + .push( + concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_render\",\"model\":\"grok-4.5\"}}\n\n", + ) + .as_bytes(), + ) + .expect("custom response created"); + let added = decoder + .push( + concat!( + "event: response.output_item.added\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"item_render\",\"type\":\"function_call\",\"call_id\":\"call_render\",\"name\":\"render\"}}\n\n", + ) + .as_bytes(), + ) + .expect("custom item added"); + let first_delta = decoder + .push( + concat!( + "event: response.function_call_arguments.delta\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"item_render\",\"call_id\":\"call_render\",\"output_index\":0,\"delta\":\"{\\\"input\\\":\\\"hel\"}\n\n", + ) + .as_bytes(), + ) + .expect("first custom argument delta"); + let second_delta = decoder + .push( + concat!( + "event: response.function_call_arguments.delta\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"item_render\",\"call_id\":\"call_render\",\"output_index\":0,\"delta\":\"lo\\\"}\"}\n\n", + ) + .as_bytes(), + ) + .expect("second custom argument delta"); + let done = decoder + .push( + concat!( + "event: response.function_call_arguments.done\n", + "data: {\"type\":\"response.function_call_arguments.done\",\"item_id\":\"item_render\",\"call_id\":\"call_render\",\"output_index\":0,\"arguments\":\"{\\\"input\\\":\\\"hello\\\"}\"}\n\n", + ) + .as_bytes(), + ) + .expect("custom argument done"); + + assert!(wire_events(&added).iter().any(|event| { + event.event_type() == Some("response.output_item.added") + && event.data().pointer("/item/type") == Some(&serde_json::json!("custom_tool_call")) + })); + assert!(first_delta.is_empty()); + assert!(second_delta.is_empty()); + assert!(wire_events(&done).iter().any(|event| { + event.event_type() == Some("response.custom_tool_call_input.done") + && event.data().get("input") == Some(&serde_json::json!("hello")) + })); +} + #[test] fn decoder_should_coalesce_reasoning_item_part_and_summary_index() { let body = concat!( diff --git a/backend/crates/providers/xai/tests/transport/network.rs b/backend/crates/providers/xai/tests/transport/network.rs index 94148c90b..c34abc239 100644 --- a/backend/crates/providers/xai/tests/transport/network.rs +++ b/backend/crates/providers/xai/tests/transport/network.rs @@ -318,27 +318,28 @@ async fn inference_transport_should_classify_reasoning_decode_rejections() { } #[tokio::test] -async fn inference_transport_should_reject_success_with_non_sse_content_type() { +async fn inference_transport_should_not_reject_sse_bytes_only_because_content_type_is_wrong() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/v1/responses")) - .respond_with(ResponseTemplate::new(200).set_body_raw("{}", "application/json")) + .respond_with(ResponseTemplate::new(200).set_body_raw( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\"}}\n\n", + "application/json", + )) .expect(1) .mount(&server) .await; let origin = Url::parse(&server.uri()).expect("wiremock origin"); - let error = inference_transport(&origin) + let chunks = inference_transport(&origin) .execute(inference_request(&origin)) .await - .expect_err("successful non-SSE response must fail closed"); + .expect("body syntax, rather than response metadata, determines the stream") + .into_body() + .collect::>() + .await; - assert_eq!( - (error.kind(), error.send_state()), - ( - GrokInferenceTransportErrorKind::Protocol, - UpstreamSendState::Sent, - ) - ); + assert_eq!(chunks.len(), 1); + assert!(chunks[0].is_ok()); } #[tokio::test] diff --git a/backend/crates/providers/xai/tests/transport/request.rs b/backend/crates/providers/xai/tests/transport/request.rs index 122d37851..6d5e46930 100644 --- a/backend/crates/providers/xai/tests/transport/request.rs +++ b/backend/crates/providers/xai/tests/transport/request.rs @@ -273,6 +273,104 @@ fn explicit_session_should_merge_client_tools_into_the_native_cache_route() { assert_eq!(body.pointer("/tool_choice"), Some(&json!("auto"))); } +#[test] +fn function_parameters_should_remove_only_nullable_object_roots() { + let request = raw_request(json!({ + "model": "client", + "input": "hello", + "tools": [{ + "type": "function", + "name": "automation_update", + "parameters": { + "anyOf": [ + { + "type": "object", + "properties": { + "title": { + "anyOf": [{"type": "string"}, {"type": "null"}] + } + }, + "required": ["title"], + "additionalProperties": false + }, + {"type": "null"} + ] + } + }] + })); + + let encoded = GrokResponsesRequest::encode(&request, "grok-4.5", &client_key()) + .expect("nullable object root"); + let body = Value::Object(encoded.body().clone()); + + assert_eq!( + body.pointer("/tools/0/parameters/type"), + Some(&json!("object")) + ); + assert_eq!(body.pointer("/tools/0/parameters/anyOf"), None); + assert_eq!( + body.pointer("/tools/0/parameters/properties/title/anyOf"), + Some(&json!([{"type": "string"}, {"type": "null"}])) + ); +} + +#[test] +fn function_parameters_should_keep_local_object_refs_after_removing_null() { + let request = raw_request(json!({ + "model": "client", + "input": "hello", + "tools": [{ + "type": "function", + "name": "lookup", + "parameters": { + "$defs": { + "Args": { + "type": "object", + "properties": {"query": {"type": "string"}} + } + }, + "anyOf": [{"$ref": "#/$defs/Args"}, {"type": "null"}] + } + }] + })); + + let encoded = GrokResponsesRequest::encode(&request, "grok-4.5", &client_key()) + .expect("nullable local object ref"); + let body = Value::Object(encoded.body().clone()); + + assert_eq!( + body.pointer("/tools/0/parameters/type"), + Some(&json!("object")) + ); + assert_eq!( + body.pointer("/tools/0/parameters/anyOf"), + Some(&json!([{"$ref": "#/$defs/Args"}])) + ); +} + +#[test] +fn function_parameters_should_report_the_invalid_nullable_root_field() { + let request = raw_request(json!({ + "model": "client", + "input": "hello", + "tools": [{ + "type": "function", + "name": "invalid", + "parameters": { + "anyOf": [{"type": "object"}, {"type": "string"}, {"type": "null"}] + } + }] + })); + + assert_eq!( + GrokResponsesRequest::encode(&request, "grok-4.5", &client_key()) + .expect_err("nullable non-object root"), + GrokRequestEncodeError::InvalidRequestField { + field: "tools[].parameters" + } + ); +} + #[test] fn explicit_session_should_enable_cache_after_codex_additional_tools_normalization() { let request = raw_request(json!({ @@ -721,7 +819,7 @@ fn compaction_history_should_reject_missing_empty_or_non_string_plaintext() { assert_eq!( GrokResponsesRequest::encode(&request, "grok-4.5", &client_key()) .expect_err("invalid compaction plaintext"), - GrokRequestEncodeError::InvalidRequestNormalization + GrokRequestEncodeError::InvalidRequestField { field: "input" } ); } } @@ -765,27 +863,42 @@ fn tool_search_history_should_load_returned_tools_at_the_original_turn() { #[test] fn unsupported_or_ambiguous_tool_contracts_should_fail_before_upstream_io() { - for body in [ - json!({"model": "client", "input": "x", "tools": [{"type": "future_tool"}]}), - json!({"model": "client", "input": "x", "tools": [ - {"type": "tool_search", "execution": "client"}, - {"type": "tool_search", "execution": "server"} - ]}), - json!({"model": "client", "input": "x", "tools": [ - {"type": "shell"}, {"type": "local_shell"} - ]}), - json!({"model": "client", "input": [{ - "type": "apply_patch_call", - "call_id": "patch_1", - "operation": {"type": "update_file", "path": "a.txt"} - }]}), - json!({"model": "client", "input": [{"type": "compaction_trigger"}]}), + for (body, field) in [ + ( + json!({"model": "client", "input": "x", "tools": [{"type": "future_tool"}]}), + "tools", + ), + ( + json!({"model": "client", "input": "x", "tools": [ + {"type": "tool_search", "execution": "client"}, + {"type": "tool_search", "execution": "server"} + ]}), + "tools", + ), + ( + json!({"model": "client", "input": "x", "tools": [ + {"type": "shell"}, {"type": "local_shell"} + ]}), + "tools", + ), + ( + json!({"model": "client", "input": [{ + "type": "apply_patch_call", + "call_id": "patch_1", + "operation": {"type": "update_file", "path": "a.txt"} + }]}), + "input", + ), + ( + json!({"model": "client", "input": [{"type": "compaction_trigger"}]}), + "input", + ), ] { let request = raw_request(body); assert_eq!( GrokResponsesRequest::encode(&request, "grok-4.5", &client_key()) .expect_err("invalid tool contract"), - GrokRequestEncodeError::InvalidRequestNormalization + GrokRequestEncodeError::InvalidRequestField { field } ); } } diff --git a/frontend/src/views/accounts/components/AccountOverviewCards.vue b/frontend/src/views/accounts/components/AccountOverviewCards.vue index d477c4dc0..e76579ada 100644 --- a/frontend/src/views/accounts/components/AccountOverviewCards.vue +++ b/frontend/src/views/accounts/components/AccountOverviewCards.vue @@ -33,8 +33,8 @@ const overviewItems = computed(() => [ icon: Gauge, }, { - label: '需处理', - value: formatCount(props.summary.attention), + label: '不可用账号', + value: formatCount(props.summary.unavailable), caption: '过期 / 禁用 / 封禁', tone: 'danger', icon: AlertTriangle, diff --git a/frontend/src/views/accounts/components/AccountUsageWindow.vue b/frontend/src/views/accounts/components/AccountUsageWindow.vue index ea4129fd5..0bd0a45bf 100644 --- a/frontend/src/views/accounts/components/AccountUsageWindow.vue +++ b/frontend/src/views/accounts/components/AccountUsageWindow.vue @@ -136,7 +136,6 @@ const requestBars = computed(() => { } }) }) -const hasDailyRequests = computed(() => requestBars.value.some(bar => bar.requestCount > 0))
diff --git a/frontend/src/views/accounts/composables/useAccountConnectionTest.ts b/frontend/src/views/accounts/composables/useAccountConnectionTest.ts index ae9e926c6..e2ee7df91 100644 --- a/frontend/src/views/accounts/composables/useAccountConnectionTest.ts +++ b/frontend/src/views/accounts/composables/useAccountConnectionTest.ts @@ -273,7 +273,10 @@ export function useAccountConnectionTest(options: AccountConnectionTestOptions) if (event.type === 'error') { applyAccountStatus(event.accountStatus) connectionTestError.value = event.error || '测试连接失败' - appendConnectionTestLog(connectionTestError.value, 'danger') + const detail = event.providerErrorCode || event.providerErrorType + ? { code: event.providerErrorCode, type: event.providerErrorType } + : undefined + appendConnectionTestLog(connectionTestError.value, 'danger', detail) finishConnectionTest('error') clearConnectionTestRun() } diff --git a/frontend/src/views/accounts/composables/useAccountMutations.ts b/frontend/src/views/accounts/composables/useAccountMutations.ts index 37d533a92..a9304e522 100644 --- a/frontend/src/views/accounts/composables/useAccountMutations.ts +++ b/frontend/src/views/accounts/composables/useAccountMutations.ts @@ -24,7 +24,6 @@ import { useAccountOnboarding } from './useAccountOnboarding' type AccountRow = Awaited>['items'][number] -const attentionAccountStatuses = new Set(['expired', 'disabled', 'banned']) const accountStatusSortRank: Record = { active: 0, quota_exhausted: 1, @@ -250,15 +249,19 @@ export function useAccountMutations(options: { + Number(status === 'quota_exhausted') - Number(current.status === 'quota_exhausted'), ), - attention: Math.max( + unavailable: Math.max( 0, - options.accountSummary.value.attention - + Number(attentionAccountStatuses.has(status)) - - Number(attentionAccountStatuses.has(current.status)), + options.accountSummary.value.unavailable + + Number(isUnavailableAccountStatus(status)) + - Number(isUnavailableAccountStatus(current.status)), ), } } + function isUnavailableAccountStatus(status: string) { + return status === 'expired' || status === 'disabled' || status === 'banned' + } + function sortAccountsByStatus(rows: AccountRow[]) { if (options.sort.value?.key !== 'status') return rows diff --git a/frontend/src/views/accounts/composables/useAccountsQuery.ts b/frontend/src/views/accounts/composables/useAccountsQuery.ts index 41163f4ea..e28b11000 100644 --- a/frontend/src/views/accounts/composables/useAccountsQuery.ts +++ b/frontend/src/views/accounts/composables/useAccountsQuery.ts @@ -17,7 +17,7 @@ export function useAccountsQuery() { total: 0, active: 0, quotaExhausted: 0, - attention: 0, + unavailable: 0, }) const query = usePagedQuery({ diff --git a/frontend/src/views/dashboard/composables/presenter.ts b/frontend/src/views/dashboard/composables/presenter.ts index dc59fb8a7..f0391a813 100644 --- a/frontend/src/views/dashboard/composables/presenter.ts +++ b/frontend/src/views/dashboard/composables/presenter.ts @@ -39,8 +39,8 @@ const emptyCards: DashboardSummary['cards'] = { credentials: { total: '0', totalValue: 0, - enabled: '0', - enabledValue: 0, + available: '0', + availableValue: 0, unavailable: '0', unavailableValue: 0, }, @@ -147,9 +147,9 @@ function metricCards( tone: 'normal', details: [ { - label: '启用', - value: credentials.enabled, - tone: credentials.enabledValue > 0 ? 'success' : 'normal', + label: '可用', + value: credentials.available, + tone: credentials.availableValue > 0 ? 'success' : 'normal', }, { label: '不可用', diff --git a/frontend/src/views/usage/components/OpsErrorDetailModal.vue b/frontend/src/views/usage/components/OpsErrorDetailModal.vue index 9af208085..69724b533 100644 --- a/frontend/src/views/usage/components/OpsErrorDetailModal.vue +++ b/frontend/src/views/usage/components/OpsErrorDetailModal.vue @@ -57,7 +57,7 @@ function latencyDisplay(value: unknown) {