diff --git a/apps/aether-gateway/src/control/route/admin/model_provider_families.rs b/apps/aether-gateway/src/control/route/admin/model_provider_families.rs index 10c378896..b6a89a4b4 100644 --- a/apps/aether-gateway/src/control/route/admin/model_provider_families.rs +++ b/apps/aether-gateway/src/control/route/admin/model_provider_families.rs @@ -22,6 +22,24 @@ pub(super) fn classify_admin_model_provider_family_route( "admin:models", false, )) + } else if method == http::Method::GET && normalized_path == "/api/admin/models/external/config" + { + Some(classified( + "admin_proxy", + "model_external_manage", + "external_config_get", + "admin:models", + false, + )) + } else if method == http::Method::PUT && normalized_path == "/api/admin/models/external/config" + { + Some(classified( + "admin_proxy", + "model_external_manage", + "external_config_set", + "admin:models", + false, + )) } else if method == http::Method::DELETE && normalized_path == "/api/admin/models/external/cache" { diff --git a/apps/aether-gateway/src/control/tests/admin_providers_models.rs b/apps/aether-gateway/src/control/tests/admin_providers_models.rs index 1dba3e340..4f24ef424 100644 --- a/apps/aether-gateway/src/control/tests/admin_providers_models.rs +++ b/apps/aether-gateway/src/control/tests/admin_providers_models.rs @@ -1,5 +1,8 @@ use http::Uri; +use crate::control::GatewayPublicRequestContext; +use crate::handlers::shared::local_proxy_route_requires_buffered_body; + use super::{classify_control_route, headers}; #[test] @@ -488,6 +491,53 @@ fn classifies_admin_external_models_as_admin_proxy_route() { assert!(!decision.is_execution_runtime_candidate()); } +#[test] +fn classifies_admin_external_models_config_routes_as_admin_proxy_routes() { + let headers = headers(&[]); + let uri: Uri = "/api/admin/models/external/config" + .parse() + .expect("uri should parse"); + + for (method, route_kind) in [ + (http::Method::GET, "external_config_get"), + (http::Method::PUT, "external_config_set"), + ] { + let decision = classify_control_route(&method, &uri, &headers) + .unwrap_or_else(|| panic!("{method} route should classify")); + + assert_eq!(decision.route_class.as_deref(), Some("admin_proxy")); + assert_eq!( + decision.route_family.as_deref(), + Some("model_external_manage") + ); + assert_eq!(decision.route_kind.as_deref(), Some(route_kind)); + assert_eq!( + decision.auth_endpoint_signature.as_deref(), + Some("admin:models") + ); + assert!(!decision.is_execution_runtime_candidate()); + } +} + +#[test] +fn admin_external_models_config_update_buffers_request_body() { + let headers = headers(&[]); + let uri: Uri = "/api/admin/models/external/config" + .parse() + .expect("uri should parse"); + let decision = + classify_control_route(&http::Method::PUT, &uri, &headers).expect("route should classify"); + let context = GatewayPublicRequestContext::from_request_parts( + "trace-admin-external-models-config", + &http::Method::PUT, + &uri, + &headers, + Some(decision), + ); + + assert!(local_proxy_route_requires_buffered_body(&context)); +} + #[test] fn classifies_admin_clear_external_models_cache_as_admin_proxy_route() { let headers = headers(&[]); diff --git a/apps/aether-gateway/src/handlers/admin/mod.rs b/apps/aether-gateway/src/handlers/admin/mod.rs index a085d1ace..0626429e8 100644 --- a/apps/aether-gateway/src/handlers/admin/mod.rs +++ b/apps/aether-gateway/src/handlers/admin/mod.rs @@ -18,6 +18,10 @@ mod shared; pub(crate) use self::auth::maybe_build_local_admin_security_response; pub(crate) use self::endpoint::build_admin_endpoint_health_status_payload; pub(crate) use self::features::maybe_build_local_admin_video_tasks_response; +#[cfg(test)] +pub(crate) use self::model::{ + set_admin_external_models_source_url_for_tests, ADMIN_EXTERNAL_MODELS_CONFIG_MUTATION_LOCK_KEY, +}; pub(crate) use self::observability::{ admin_stats_bad_request_response, maybe_build_local_admin_usage_response, parse_bounded_u32, round_to, AdminStatsTimeRange, AdminStatsUsageFilter, @@ -53,4 +57,7 @@ pub(crate) use self::request::{ }; pub(crate) use self::routes::maybe_build_local_admin_response; #[cfg(test)] -pub(crate) use self::system::override_proxy_connectivity_probe_url_for_tests; +pub(crate) use self::system::{ + clear_proxy_node_references_with_cache_failure_for_tests, + override_proxy_connectivity_probe_url_for_tests, +}; diff --git a/apps/aether-gateway/src/handlers/admin/model/catalog_routes.rs b/apps/aether-gateway/src/handlers/admin/model/catalog_routes.rs index 4b0c9a5f9..2a5b37cbc 100644 --- a/apps/aether-gateway/src/handlers/admin/model/catalog_routes.rs +++ b/apps/aether-gateway/src/handlers/admin/model/catalog_routes.rs @@ -1,8 +1,9 @@ use crate::handlers::admin::model::build_admin_model_catalog_payload; use crate::handlers::admin::request::{AdminAppState, AdminRequestContext}; +use crate::handlers::admin::shared::attach_admin_audit_response; use crate::GatewayError; use axum::{ - body::Body, + body::{Body, Bytes}, http, response::{IntoResponse, Response}, Json, @@ -22,6 +23,7 @@ fn build_admin_model_catalog_data_unavailable_response() -> Response { pub(crate) async fn maybe_build_local_admin_model_catalog_response( state: &AdminAppState<'_>, request_context: &AdminRequestContext<'_>, + request_body: Option<&Bytes>, ) -> Result>, GatewayError> { let Some(decision) = request_context.decision() else { return Ok(None); @@ -47,12 +49,15 @@ pub(crate) async fn maybe_build_local_admin_model_catalog_response( && request_context.path() == "/api/admin/models/external" { return Ok(Some( - match state.read_admin_external_models_cache().await? { + match state + .read_admin_external_models_cache(request_context.trace_id()) + .await? + { Some(payload) => Json(payload).into_response(), None => ( http::StatusCode::SERVICE_UNAVAILABLE, Json(json!({ - "detail": "External models catalog requires Rust admin backend" + "detail": "External models catalog unavailable" })), ) .into_response(), @@ -60,6 +65,47 @@ pub(crate) async fn maybe_build_local_admin_model_catalog_response( )); } + if decision.route_family.as_deref() == Some("model_external_manage") + && decision.route_kind.as_deref() == Some("external_config_get") + && request_context.method() == http::Method::GET + && request_context.path() == "/api/admin/models/external/config" + { + return Ok(Some( + Json(state.build_admin_external_models_config_payload().await?).into_response(), + )); + } + + if decision.route_family.as_deref() == Some("model_external_manage") + && decision.route_kind.as_deref() == Some("external_config_set") + && request_context.method() == http::Method::PUT + && request_context.path() == "/api/admin/models/external/config" + { + let Some(request_body) = request_body else { + return Ok(Some( + ( + http::StatusCode::BAD_REQUEST, + Json(json!({ "detail": "请求数据验证失败" })), + ) + .into_response(), + )); + }; + return Ok(Some( + match state + .apply_admin_external_models_config_update(request_body) + .await? + { + Ok(payload) => attach_admin_audit_response( + Json(payload).into_response(), + "admin_external_models_config_updated", + "update_external_models_config", + "external_models_catalog", + "global", + ), + Err((status, payload)) => (status, Json(payload)).into_response(), + }, + )); + } + if decision.route_family.as_deref() == Some("model_external_manage") && decision.route_kind.as_deref() == Some("clear_external_cache") && request_context.method() == http::Method::DELETE diff --git a/apps/aether-gateway/src/handlers/admin/model/external_cache.rs b/apps/aether-gateway/src/handlers/admin/model/external_cache.rs index 790abcdaf..92a16c846 100644 --- a/apps/aether-gateway/src/handlers/admin/model/external_cache.rs +++ b/apps/aether-gateway/src/handlers/admin/model/external_cache.rs @@ -1,13 +1,72 @@ use crate::handlers::admin::request::AdminAppState; use crate::handlers::shared::mark_external_models_official_providers; use crate::GatewayError; -use serde_json::json; +use aether_contracts::{ + ExecutionPlan, ExecutionTimeouts, RequestBody, EXECUTION_REQUEST_FOLLOW_REDIRECTS_HEADER, +}; +use aether_runtime_state::RuntimeLockLease; +use axum::body::Bytes; +use axum::http; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use std::collections::BTreeMap; +use std::time::Duration; use tracing::warn; -const ADMIN_EXTERNAL_MODELS_CACHE_KEY: &str = "aether:external:models_dev"; +const ADMIN_EXTERNAL_MODELS_LEGACY_CACHE_KEY: &str = "aether:external:models_dev"; +const ADMIN_EXTERNAL_MODELS_CACHE_KEY: &str = "aether:external:models_dev:v2"; +const ADMIN_EXTERNAL_MODELS_CACHE_VERSION: u8 = 2; const ADMIN_EXTERNAL_MODELS_CACHE_TTL_SECS: u64 = 15 * 60; const ADMIN_EXTERNAL_MODELS_SOURCE_URL_ENV: &str = "AETHER_GATEWAY_EXTERNAL_MODELS_URL"; const ADMIN_EXTERNAL_MODELS_SOURCE_URL_DEFAULT: &str = "https://models.dev/api.json"; +pub(in crate::handlers::admin) const ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY: &str = + "external_models_proxy_node_id"; +const ADMIN_EXTERNAL_MODELS_CONNECT_TIMEOUT_MS: u64 = 10_000; +const ADMIN_EXTERNAL_MODELS_TOTAL_TIMEOUT_MS: u64 = 300_000; +pub(crate) const ADMIN_EXTERNAL_MODELS_CONFIG_MUTATION_LOCK_KEY: &str = + "admin:external_models_proxy_node_config:mutation"; +const ADMIN_EXTERNAL_MODELS_CONFIG_MUTATION_LOCK_TTL: Duration = Duration::from_secs(10 * 60); + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct AdminExternalModelsCacheEnvelope { + schema_version: u8, + proxy_node_id: Option, + payload: Value, +} + +#[cfg(test)] +pub(crate) struct AdminExternalModelsSourceUrlEnvGuard { + previous: Option, + _lock: std::sync::MutexGuard<'static, ()>, +} + +#[cfg(test)] +impl Drop for AdminExternalModelsSourceUrlEnvGuard { + fn drop(&mut self) { + if let Some(previous) = self.previous.as_deref() { + std::env::set_var(ADMIN_EXTERNAL_MODELS_SOURCE_URL_ENV, previous); + } else { + std::env::remove_var(ADMIN_EXTERNAL_MODELS_SOURCE_URL_ENV); + } + } +} + +#[cfg(test)] +pub(crate) fn set_admin_external_models_source_url_for_tests( + value: &str, +) -> AdminExternalModelsSourceUrlEnvGuard { + static LOCK: std::sync::OnceLock> = std::sync::OnceLock::new(); + let lock = LOCK + .get_or_init(|| std::sync::Mutex::new(())) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let previous = std::env::var(ADMIN_EXTERNAL_MODELS_SOURCE_URL_ENV).ok(); + std::env::set_var(ADMIN_EXTERNAL_MODELS_SOURCE_URL_ENV, value); + AdminExternalModelsSourceUrlEnvGuard { + previous, + _lock: lock, + } +} fn admin_external_models_source_url() -> String { std::env::var(ADMIN_EXTERNAL_MODELS_SOURCE_URL_ENV) @@ -21,12 +80,42 @@ fn normalize_admin_external_models_payload(payload: serde_json::Value) -> serde_ mark_external_models_official_providers(&payload).unwrap_or(payload) } +fn classify_admin_external_models_transport_error(message: &str) -> &'static str { + let message = message.to_ascii_lowercase(); + if message.contains("timed out") || message.contains("timeout") { + "timeout" + } else if message.contains("relay") || message.contains("tunnel") { + "relay" + } else if message.contains("proxy") { + "proxy_config" + } else if message.contains("too large") || message.contains("exceeds") { + "response_too_large" + } else if message.contains("json") { + "invalid_json" + } else if message.contains("decode") || message.contains("content-encoding") { + "response_decode" + } else if message.contains("connect") || message.contains("dns") || message.contains("tcp") { + "connect" + } else if message.contains("header") || message.contains("method") || message.contains("build") + { + "request_build" + } else { + "unknown_transport" + } +} + async fn store_admin_external_models_cache( state: &AdminAppState<'_>, + proxy_node_id: Option<&str>, payload: &serde_json::Value, ) -> Result<(), GatewayError> { + let envelope = AdminExternalModelsCacheEnvelope { + schema_version: ADMIN_EXTERNAL_MODELS_CACHE_VERSION, + proxy_node_id: proxy_node_id.map(ToOwned::to_owned), + payload: payload.clone(), + }; let serialized = - serde_json::to_string(payload).map_err(|err| GatewayError::Internal(err.to_string()))?; + serde_json::to_string(&envelope).map_err(|err| GatewayError::Internal(err.to_string()))?; state .as_ref() .runtime_kv_setex( @@ -38,10 +127,323 @@ async fn store_admin_external_models_cache( Ok(()) } +fn normalize_admin_external_models_proxy_node_id( + value: Option<&Value>, +) -> Result, GatewayError> { + match value { + None | Some(Value::Null) => Ok(None), + Some(Value::String(value)) => { + let value = value.trim(); + if value.is_empty() { + Err(GatewayError::Internal(format!( + "system config '{ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY}' must not be empty" + ))) + } else { + Ok(Some(value.to_string())) + } + } + Some(_) => Err(GatewayError::Internal(format!( + "system config '{ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY}' must be a string or null" + ))), + } +} + +async fn read_admin_external_models_proxy_node_id( + state: &AdminAppState<'_>, +) -> Result, GatewayError> { + let value = state + .read_system_config_json_value_strong(ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY) + .await?; + normalize_admin_external_models_proxy_node_id(value.as_ref()) +} + +pub(crate) async fn build_admin_external_models_config_payload( + state: &AdminAppState<'_>, +) -> Result { + let proxy_node_id = read_admin_external_models_proxy_node_id(state).await?; + Ok(json!({ + "proxy_node_id": proxy_node_id, + })) +} + +fn parse_admin_external_models_config_update( + request_body: &[u8], +) -> Result, (http::StatusCode, Value)> { + let payload = match serde_json::from_slice::(request_body) { + Ok(Value::Object(payload)) => payload, + Ok(_) | Err(_) => { + return Err(( + http::StatusCode::BAD_REQUEST, + json!({ "detail": "请求数据验证失败" }), + )); + } + }; + match payload.get("proxy_node_id") { + Some(Value::Null) => Ok(None), + Some(Value::String(value)) => { + let value = value.trim(); + if value.is_empty() { + return Err(( + http::StatusCode::BAD_REQUEST, + json!({ "detail": "proxy_node_id 不能为空" }), + )); + } + Ok(Some(value.to_string())) + } + Some(_) | None => Err(( + http::StatusCode::BAD_REQUEST, + json!({ "detail": "proxy_node_id 必须是字符串或 null" }), + )), + } +} + +async fn clear_admin_external_models_cache_entries( + state: &AdminAppState<'_>, +) -> Result { + let cleared_current = state + .as_ref() + .runtime_kv_del(ADMIN_EXTERNAL_MODELS_CACHE_KEY) + .await?; + let cleared_legacy = state + .as_ref() + .runtime_kv_del(ADMIN_EXTERNAL_MODELS_LEGACY_CACHE_KEY) + .await?; + Ok(cleared_current || cleared_legacy) +} + +pub(crate) async fn acquire_admin_external_models_config_mutation_lock( + state: &AdminAppState<'_>, +) -> Result { + match state + .app() + .runtime_state() + .lock_try_acquire( + ADMIN_EXTERNAL_MODELS_CONFIG_MUTATION_LOCK_KEY, + state.app().tunnel.local_instance_id(), + ADMIN_EXTERNAL_MODELS_CONFIG_MUTATION_LOCK_TTL, + ) + .await + { + Ok(Some(lock)) => Ok(lock), + Ok(None) => Err(( + http::StatusCode::CONFLICT, + json!({ "detail": "外部模型目录代理配置正在更新，请稍后重试" }), + )), + Err(_) => { + warn!( + runtime_backend = state.app().runtime_state_backend(), + "failed to acquire external models proxy config mutation lock" + ); + Err(( + http::StatusCode::SERVICE_UNAVAILABLE, + json!({ "detail": "外部模型目录代理配置暂时无法更新" }), + )) + } + } +} + +pub(crate) async fn release_admin_external_models_config_mutation_lock( + state: &AdminAppState<'_>, + lock: &RuntimeLockLease, +) { + if state + .app() + .runtime_state() + .lock_release(lock) + .await + .is_err() + { + warn!( + runtime_backend = state.app().runtime_state_backend(), + "failed to release external models proxy config mutation lock" + ); + } +} + +async fn apply_admin_external_models_config_update_locked( + state: &AdminAppState<'_>, + proxy_node_id: Option, +) -> Result, GatewayError> { + if let Some(node_id) = proxy_node_id.as_deref() { + if state.find_proxy_node(node_id).await?.is_none() { + return Ok(Err(( + http::StatusCode::NOT_FOUND, + json!({ "detail": format!("代理节点 '{node_id}' 不存在") }), + ))); + } + if state + .resolve_admin_proxy_node_snapshot(Some(node_id)) + .await + .is_none() + { + return Ok(Err(( + http::StatusCode::CONFLICT, + json!({ "detail": format!("代理节点 '{node_id}' 当前不可用") }), + ))); + } + } + + let config_value = proxy_node_id + .as_ref() + .map_or(Value::Null, |node_id| json!(node_id)); + state + .upsert_system_config_json_value( + ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY, + &config_value, + Some("外部模型目录代理节点 ID"), + ) + .await?; + let cache_cleared = match clear_admin_external_models_cache_entries(state).await { + Ok(cache_cleared) => cache_cleared, + Err(_) => { + // The v2 cache envelope includes the selected node ID, so a stale entry cannot be + // reused after this persisted selector changes. Cache invalidation is best-effort. + warn!( + runtime_backend = state.app().runtime_state_backend(), + "failed to clear external models cache after proxy config update" + ); + false + } + }; + let mut payload = build_admin_external_models_config_payload(state).await?; + payload["cache_cleared"] = json!(cache_cleared); + Ok(Ok(payload)) +} + +pub(crate) async fn apply_admin_external_models_config_update( + state: &AdminAppState<'_>, + request_body: &Bytes, +) -> Result, GatewayError> { + let proxy_node_id = match parse_admin_external_models_config_update(request_body) { + Ok(proxy_node_id) => proxy_node_id, + Err(error) => return Ok(Err(error)), + }; + if !state.app().data.has_system_config_store() { + return Ok(Err(( + http::StatusCode::SERVICE_UNAVAILABLE, + json!({ "detail": "Admin system config data unavailable" }), + ))); + } + if proxy_node_id.is_some() && !state.has_proxy_node_reader() { + return Ok(Err(( + http::StatusCode::SERVICE_UNAVAILABLE, + json!({ "detail": "Admin proxy node data unavailable" }), + ))); + } + + let lock = match acquire_admin_external_models_config_mutation_lock(state).await { + Ok(lock) => lock, + Err(error) => return Ok(Err(error)), + }; + let result = apply_admin_external_models_config_update_locked(state, proxy_node_id).await; + release_admin_external_models_config_mutation_lock(state, &lock).await; + result +} + async fn fetch_admin_external_models_from_source( state: &AdminAppState<'_>, + request_id: &str, + proxy_node_id: Option<&str>, ) -> Result { let url = admin_external_models_source_url(); + if let Some(node_id) = proxy_node_id { + let Some(proxy) = state.resolve_admin_proxy_node_snapshot(Some(node_id)).await else { + warn!( + request_id = %request_id, + proxy_node_id = %node_id, + proxy_mode = "unknown", + transport_error_kind = "node_unavailable", + "external models proxy execution failed" + ); + return Err(GatewayError::Internal( + "external models proxy request failed".to_string(), + )); + }; + let proxy_mode = if proxy.mode.as_deref() == Some("tunnel") { + "tunnel" + } else if proxy.url.is_some() { + "manual" + } else { + "unknown" + }; + let headers = BTreeMap::from([ + ("accept".to_string(), "application/json".to_string()), + ( + "user-agent".to_string(), + "aether-gateway/external-models".to_string(), + ), + ( + EXECUTION_REQUEST_FOLLOW_REDIRECTS_HEADER.to_string(), + "true".to_string(), + ), + ]); + let plan = ExecutionPlan { + request_id: format!("{request_id}:external-models"), + candidate_id: None, + provider_name: Some("external_models_catalog".to_string()), + provider_id: String::new(), + endpoint_id: String::new(), + key_id: String::new(), + method: http::Method::GET.as_str().to_string(), + url, + headers, + content_type: None, + content_encoding: None, + body: RequestBody { + json_body: None, + body_bytes_b64: None, + body_ref: None, + }, + stream: false, + client_api_format: "control:external_models".to_string(), + provider_api_format: "control:external_models".to_string(), + model_name: None, + proxy: Some(proxy), + transport_profile: None, + timeouts: Some(ExecutionTimeouts { + connect_ms: Some(ADMIN_EXTERNAL_MODELS_CONNECT_TIMEOUT_MS), + read_ms: Some(ADMIN_EXTERNAL_MODELS_TOTAL_TIMEOUT_MS), + write_ms: Some(ADMIN_EXTERNAL_MODELS_TOTAL_TIMEOUT_MS), + pool_ms: Some(ADMIN_EXTERNAL_MODELS_CONNECT_TIMEOUT_MS), + total_ms: Some(ADMIN_EXTERNAL_MODELS_TOTAL_TIMEOUT_MS), + ..ExecutionTimeouts::default() + }), + }; + let result = match state + .execute_execution_runtime_sync_plan(Some(request_id), &plan) + .await + { + Ok(result) => result, + Err(err) => { + let error_message = err.clone().into_message(); + let transport_error_kind = + classify_admin_external_models_transport_error(&error_message); + warn!( + request_id = %request_id, + proxy_node_id = %node_id, + proxy_mode, + transport_error_kind, + "external models proxy execution failed" + ); + return Err(GatewayError::Internal( + "external models proxy request failed".to_string(), + )); + } + }; + if !(200..300).contains(&result.status_code) { + return Err(GatewayError::Internal(format!( + "external models source returned HTTP {}", + result.status_code + ))); + } + let payload = result.body.and_then(|body| body.json_body).ok_or_else(|| { + GatewayError::Internal( + "external models source returned a non-JSON response".to_string(), + ) + })?; + return Ok(normalize_admin_external_models_payload(payload)); + } + let response = state .http_client() .get(&url) @@ -60,29 +462,36 @@ async fn fetch_admin_external_models_from_source( pub(crate) async fn read_admin_external_models_cache( state: &AdminAppState<'_>, + request_id: &str, ) -> Result, GatewayError> { + let proxy_node_id = read_admin_external_models_proxy_node_id(state).await?; if let Some(raw) = state .as_ref() .runtime_kv_get(ADMIN_EXTERNAL_MODELS_CACHE_KEY) .await? { - match serde_json::from_str::(&raw) { - Ok(payload) => { - let payload = normalize_admin_external_models_payload(payload); - if let Err(err) = store_admin_external_models_cache(state, &payload).await { - warn!(error = ?err, "failed to refresh external models cache ttl"); - } - return Ok(Some(payload)); + match serde_json::from_str::(&raw) { + Ok(envelope) + if envelope.schema_version == ADMIN_EXTERNAL_MODELS_CACHE_VERSION + && envelope.proxy_node_id == proxy_node_id => + { + return Ok(Some(normalize_admin_external_models_payload( + envelope.payload, + ))); } + Ok(_) => {} Err(err) => { warn!(error = %err, "failed to parse cached external models payload"); } } } - match fetch_admin_external_models_from_source(state).await { + match fetch_admin_external_models_from_source(state, request_id, proxy_node_id.as_deref()).await + { Ok(payload) => { - if let Err(err) = store_admin_external_models_cache(state, &payload).await { + if let Err(err) = + store_admin_external_models_cache(state, proxy_node_id.as_deref(), &payload).await + { warn!(error = ?err, "failed to store fetched external models cache"); } Ok(Some(payload)) @@ -97,10 +506,7 @@ pub(crate) async fn read_admin_external_models_cache( pub(crate) async fn clear_admin_external_models_cache( state: &AdminAppState<'_>, ) -> Result { - let deleted = state - .as_ref() - .runtime_kv_del(ADMIN_EXTERNAL_MODELS_CACHE_KEY) - .await?; + let deleted = clear_admin_external_models_cache_entries(state).await?; Ok(json!({ "cleared": deleted, "message": if deleted { "缓存已清除" } else { "缓存不存在" }, @@ -110,47 +516,15 @@ pub(crate) async fn clear_admin_external_models_cache( #[cfg(test)] mod tests { use super::{ - admin_external_models_source_url, normalize_admin_external_models_payload, - read_admin_external_models_cache, ADMIN_EXTERNAL_MODELS_SOURCE_URL_ENV, + admin_external_models_source_url, classify_admin_external_models_transport_error, + normalize_admin_external_models_payload, normalize_admin_external_models_proxy_node_id, + read_admin_external_models_cache, set_admin_external_models_source_url_for_tests, }; use crate::handlers::admin::request::AdminAppState; use crate::tests::{start_server, AppState}; use axum::routing::get; use axum::{Json, Router}; use serde_json::json; - use std::sync::{Mutex, MutexGuard, OnceLock}; - - fn admin_external_models_env_lock() -> &'static Mutex<()> { - static LOCK: OnceLock> = OnceLock::new(); - LOCK.get_or_init(|| Mutex::new(())) - } - - struct TestEnvVarGuard { - key: &'static str, - previous: Option, - _lock: Option>, - } - - impl Drop for TestEnvVarGuard { - fn drop(&mut self) { - if let Some(previous) = self.previous.as_deref() { - std::env::set_var(self.key, previous); - } else { - std::env::remove_var(self.key); - } - } - } - - fn set_test_env_var(key: &'static str, value: &str) -> TestEnvVarGuard { - let lock = admin_external_models_env_lock().lock().ok(); - let previous = std::env::var(key).ok(); - std::env::set_var(key, value); - TestEnvVarGuard { - key, - previous, - _lock: lock, - } - } #[test] fn normalizes_external_models_payload_with_official_flags() { @@ -172,11 +546,44 @@ mod tests { } #[test] - fn external_models_source_url_uses_env_override_when_present() { - let _guard = set_test_env_var( - ADMIN_EXTERNAL_MODELS_SOURCE_URL_ENV, - "http://127.0.0.1:12345/api", + fn external_models_proxy_config_only_treats_missing_or_null_as_direct() { + assert_eq!( + normalize_admin_external_models_proxy_node_id(None).expect("missing config is direct"), + None ); + assert_eq!( + normalize_admin_external_models_proxy_node_id(Some(&serde_json::Value::Null)) + .expect("null config is direct"), + None + ); + assert!(normalize_admin_external_models_proxy_node_id(Some(&json!(" "))).is_err()); + assert!(normalize_admin_external_models_proxy_node_id(Some(&json!(false))).is_err()); + } + + #[test] + fn classifies_external_models_transport_errors_without_exposing_details() { + for (message, expected) in [ + ("request timeout after 300000ms", "timeout"), + ("hub relay request failed", "relay"), + ("invalid proxy configuration", "proxy_config"), + ("upstream response body exceeds limit", "response_too_large"), + ("upstream response is not valid JSON", "invalid_json"), + ("failed to decode content-encoding gzip", "response_decode"), + ("tcp connect error", "connect"), + ("invalid upstream header value", "request_build"), + ("opaque execution failure", "unknown_transport"), + ] { + assert_eq!( + classify_admin_external_models_transport_error(message), + expected, + "message={message}" + ); + } + } + + #[test] + fn external_models_source_url_uses_env_override_when_present() { + let _guard = set_admin_external_models_source_url_for_tests("http://127.0.0.1:12345/api"); assert_eq!( admin_external_models_source_url(), "http://127.0.0.1:12345/api" @@ -201,16 +608,15 @@ mod tests { }), ); let (upstream_url, upstream_handle) = start_server(upstream).await; - let _guard = set_test_env_var( - ADMIN_EXTERNAL_MODELS_SOURCE_URL_ENV, - &format!("{upstream_url}/api.json"), - ); + let _guard = + set_admin_external_models_source_url_for_tests(&format!("{upstream_url}/api.json")); let state = AppState::new().expect("gateway should build"); - let payload = read_admin_external_models_cache(&AdminAppState::new(&state)) - .await - .expect("external models read should succeed") - .expect("payload should be fetched"); + let payload = + read_admin_external_models_cache(&AdminAppState::new(&state), "external-models-test") + .await + .expect("external models read should succeed") + .expect("payload should be fetched"); assert_eq!(payload["openai"]["official"], json!(true)); assert_eq!(payload["openai"]["models"]["gpt-5"]["name"], json!("GPT-5")); diff --git a/apps/aether-gateway/src/handlers/admin/model/mod.rs b/apps/aether-gateway/src/handlers/admin/model/mod.rs index 265435ad3..20ba0e1a8 100644 --- a/apps/aether-gateway/src/handlers/admin/model/mod.rs +++ b/apps/aether-gateway/src/handlers/admin/model/mod.rs @@ -11,7 +11,14 @@ mod write; pub(super) use self::catalog_routes::maybe_build_local_admin_model_catalog_response; pub(super) use self::external_cache::{ - clear_admin_external_models_cache, read_admin_external_models_cache, + acquire_admin_external_models_config_mutation_lock, apply_admin_external_models_config_update, + build_admin_external_models_config_payload, clear_admin_external_models_cache, + read_admin_external_models_cache, release_admin_external_models_config_mutation_lock, + ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY, +}; +#[cfg(test)] +pub(crate) use self::external_cache::{ + set_admin_external_models_source_url_for_tests, ADMIN_EXTERNAL_MODELS_CONFIG_MUTATION_LOCK_KEY, }; pub(super) use self::global::{ build_admin_global_model_payload, build_admin_global_model_providers_payload, diff --git a/apps/aether-gateway/src/handlers/admin/model/routes.rs b/apps/aether-gateway/src/handlers/admin/model/routes.rs index 403b11b86..84ad36d89 100644 --- a/apps/aether-gateway/src/handlers/admin/model/routes.rs +++ b/apps/aether-gateway/src/handlers/admin/model/routes.rs @@ -7,6 +7,7 @@ pub(crate) async fn maybe_build_local_admin_model_response( if let Some(response) = catalog_routes::maybe_build_local_admin_model_catalog_response( &request.state(), &request.request_context(), + request.request_body(), ) .await? { diff --git a/apps/aether-gateway/src/handlers/admin/request/models.rs b/apps/aether-gateway/src/handlers/admin/request/models.rs index b24daf716..0b63f592d 100644 --- a/apps/aether-gateway/src/handlers/admin/request/models.rs +++ b/apps/aether-gateway/src/handlers/admin/request/models.rs @@ -11,6 +11,7 @@ use aether_admin::provider::{ use aether_data_contracts::repository::global_models::{ AdminProviderModelListQuery, StoredAdminProviderModel, UpsertAdminProviderModelRecord, }; +use axum::http; use serde_json::json; use std::collections::{BTreeMap, BTreeSet}; use uuid::Uuid; @@ -488,8 +489,24 @@ impl<'a> AdminAppState<'a> { pub(crate) async fn read_admin_external_models_cache( &self, + request_id: &str, ) -> Result, GatewayError> { - crate::handlers::admin::model::read_admin_external_models_cache(self).await + crate::handlers::admin::model::read_admin_external_models_cache(self, request_id).await + } + + pub(crate) async fn build_admin_external_models_config_payload( + &self, + ) -> Result { + crate::handlers::admin::model::build_admin_external_models_config_payload(self).await + } + + pub(crate) async fn apply_admin_external_models_config_update( + &self, + request_body: &axum::body::Bytes, + ) -> Result, GatewayError> + { + crate::handlers::admin::model::apply_admin_external_models_config_update(self, request_body) + .await } pub(crate) async fn clear_admin_external_models_cache( diff --git a/apps/aether-gateway/src/handlers/admin/request/system/import.rs b/apps/aether-gateway/src/handlers/admin/request/system/import.rs index 0c34cddd0..d02406e1e 100644 --- a/apps/aether-gateway/src/handlers/admin/request/system/import.rs +++ b/apps/aether-gateway/src/handlers/admin/request/system/import.rs @@ -2,6 +2,7 @@ use super::{AdminAppState, ADMIN_SYSTEM_DATA_EXPORT_VERSION}; use crate::ai_serving::build_provider_key_pool_score_upsert; use crate::api::ai::admin_endpoint_signature_parts; use crate::handlers::admin::admin_provider_pool_config; +use crate::handlers::admin::model::ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY; use crate::handlers::admin::provider::endpoints_admin::payloads::AdminProviderEndpointUpdatePatch; use crate::handlers::admin::provider::shared::payloads::{ AdminProviderCreateRequest, AdminProviderKeyCreateRequest, AdminProviderKeyUpdatePatch, @@ -61,6 +62,15 @@ fn invalid_request(detail: impl Into) -> (http::StatusCode, Value) { ) } +fn normalize_imported_system_config_key(key: &str) -> String { + let normalized = normalize_admin_system_config_key(key); + if normalized.eq_ignore_ascii_case(ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY) { + ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY.to_string() + } else { + normalized + } +} + fn build_admin_system_data_import_part_body( root: &Map, field_name: &str, @@ -1299,6 +1309,82 @@ impl<'a> AdminAppState<'a> { let mut stats = AdminSystemConfigImportStats::default(); + // Proxy nodes are deployment-local resources and are intentionally not imported by the + // Rust admin backend. Apply the external catalog selector before importing any other + // object, and turn a non-empty exported node reference into direct mode. This keeps a + // clean-environment restore portable and prevents a late selector validation failure from + // leaving the rest of the document partially imported. + let (imported_external_models_configs, imported_system_configs): (Vec<_>, Vec<_>) = + imported_system_configs.into_iter().partition(|item| { + normalize_imported_system_config_key(&item.value.key) + == ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY + }); + let mut existing_system_config_keys = self + .list_system_config_entries() + .await? + .into_iter() + .map(|entry| normalize_imported_system_config_key(&entry.key)) + .collect::>(); + for imported_config_item in imported_external_models_configs { + let (_, system_config) = imported_config_item.into_parts(); + let exists = + existing_system_config_keys.contains(ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY); + match (exists, merge_mode) { + (true, AdminImportMergeMode::Skip) => { + stats.system_configs.skipped += 1; + continue; + } + (true, AdminImportMergeMode::Error) => { + return Ok(Err(invalid_request(format!( + "SystemConfig '{ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY}' 已存在" + )))); + } + _ => {} + } + + let imported_proxy_node_id = match system_config.value { + Value::Null => None, + Value::String(value) => { + let value = value.trim(); + if value.is_empty() { + return Ok(Err(invalid_request( + "external_models_proxy_node_id 不能为空", + ))); + } + Some(value.to_string()) + } + _ => { + return Ok(Err(invalid_request( + "external_models_proxy_node_id 必须是字符串或 null", + ))) + } + }; + let request_bytes = Bytes::from( + serde_json::to_vec(&json!({ "proxy_node_id": null })) + .map_err(|err| GatewayError::Internal(err.to_string()))?, + ); + match self + .apply_admin_external_models_config_update(&request_bytes) + .await? + { + Ok(_) => { + if exists { + stats.system_configs.updated += 1; + } else { + stats.system_configs.created += 1; + existing_system_config_keys + .insert(ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY.to_string()); + } + if let Some(node_id) = imported_proxy_node_id { + stats.errors.push(format!( + "外部模型目录代理节点 '{node_id}' 是当前部署的本地引用；代理节点未导入，已切换为直连" + )); + } + } + Err((status, payload)) => return Ok(Err((status, payload))), + } + } + let mut global_models_by_name = self .list_admin_global_models(&AdminGlobalModelListQuery { offset: 0, @@ -2231,15 +2317,14 @@ impl<'a> AdminAppState<'a> { } } - let mut existing_system_config_keys = self - .list_system_config_entries() - .await? - .into_iter() - .map(|entry| normalize_admin_system_config_key(&entry.key)) - .collect::>(); for imported_config_item in imported_system_configs { let (_, system_config) = imported_config_item.into_parts(); - let normalized_key = normalize_admin_system_config_key(&system_config.key); + let ImportedSystemConfig { + key, + value, + description, + } = system_config; + let normalized_key = normalize_imported_system_config_key(&key); let exists = existing_system_config_keys.contains(&normalized_key); match (exists, merge_mode) { (true, AdminImportMergeMode::Skip) => { @@ -2256,13 +2341,14 @@ impl<'a> AdminAppState<'a> { let request_bytes = Bytes::from( serde_json::to_vec(&json!({ - "value": system_config.value, - "description": system_config.description, + "value": value, + "description": description, })) .map_err(|err| GatewayError::Internal(err.to_string()))?, ); - match apply_admin_system_config_update(self, &system_config.key, &request_bytes).await? - { + let update_result = + apply_admin_system_config_update(self, &key, &request_bytes).await?; + match update_result { Ok(_) => { if exists { stats.system_configs.updated += 1; diff --git a/apps/aether-gateway/src/handlers/admin/request/system/mod.rs b/apps/aether-gateway/src/handlers/admin/request/system/mod.rs index 7ab47130b..708a62ad5 100644 --- a/apps/aether-gateway/src/handlers/admin/request/system/mod.rs +++ b/apps/aether-gateway/src/handlers/admin/request/system/mod.rs @@ -29,6 +29,13 @@ impl<'a> AdminAppState<'a> { self.app.read_system_config_json_value(key).await } + pub(crate) async fn read_system_config_json_value_strong( + &self, + key: &str, + ) -> Result, GatewayError> { + self.app.read_system_config_json_value_strong(key).await + } + pub(crate) async fn list_system_config_entries( &self, ) -> Result, GatewayError> { diff --git a/apps/aether-gateway/src/handlers/admin/system/core/mod.rs b/apps/aether-gateway/src/handlers/admin/system/core/mod.rs index 78c6b49ee..f11f8c2a3 100644 --- a/apps/aether-gateway/src/handlers/admin/system/core/mod.rs +++ b/apps/aether-gateway/src/handlers/admin/system/core/mod.rs @@ -84,6 +84,7 @@ pub(crate) async fn maybe_build_local_admin_core_response( crate::handlers::admin::model::maybe_build_local_admin_model_catalog_response( state, request_context, + request_body, ) .await? { diff --git a/apps/aether-gateway/src/handlers/admin/system/mod.rs b/apps/aether-gateway/src/handlers/admin/system/mod.rs index 108115284..e8004221c 100644 --- a/apps/aether-gateway/src/handlers/admin/system/mod.rs +++ b/apps/aether-gateway/src/handlers/admin/system/mod.rs @@ -7,5 +7,8 @@ mod routes; pub(super) mod shared; #[cfg(test)] -pub(crate) use self::proxy_nodes::override_proxy_connectivity_probe_url_for_tests; +pub(crate) use self::proxy_nodes::{ + clear_proxy_node_references_with_cache_failure_for_tests, + override_proxy_connectivity_probe_url_for_tests, +}; pub(super) use self::routes::maybe_build_local_admin_system_response; diff --git a/apps/aether-gateway/src/handlers/admin/system/proxy_nodes.rs b/apps/aether-gateway/src/handlers/admin/system/proxy_nodes.rs index 88ac43c1f..783ccad34 100644 --- a/apps/aether-gateway/src/handlers/admin/system/proxy_nodes.rs +++ b/apps/aether-gateway/src/handlers/admin/system/proxy_nodes.rs @@ -1,6 +1,11 @@ +use std::future::Future; use std::time::{Duration, Instant}; use crate::execution_runtime::transport::format_upstream_request_error; +use crate::handlers::admin::model::{ + acquire_admin_external_models_config_mutation_lock, + release_admin_external_models_config_mutation_lock, +}; use crate::handlers::admin::request::{AdminAppState, AdminRequestContext}; use crate::handlers::admin::shared::query_param_value; use crate::maintenance::{ @@ -31,6 +36,7 @@ use serde::de::DeserializeOwned; use serde::Deserialize; use serde_json::{json, Value}; use sha2::{Digest, Sha256}; +use tracing::warn; use uuid::Uuid; use crate::handlers::public::build_proxy_node_install_session_response; @@ -526,27 +532,43 @@ pub(crate) async fn maybe_build_local_admin_proxy_nodes_response( if decision.route_kind.as_deref() == Some("delete_node") && request_context.method() == http::Method::DELETE { - if !state.has_proxy_node_writer() { + if !state.has_proxy_node_reader() || !state.has_proxy_node_writer() { return Ok(Some(build_admin_proxy_nodes_data_unavailable_response())); } let Some(node_id) = admin_proxy_node_node_id_from_path(request_context.path()) else { return Ok(Some(build_admin_proxy_nodes_not_found_response())); }; - let Some(_deleted_node) = state.delete_proxy_node(&node_id).await? else { - return Ok(Some(build_admin_proxy_nodes_not_found_response())); + let lock = match acquire_admin_external_models_config_mutation_lock(state).await { + Ok(lock) => lock, + Err((status, payload)) => { + return Ok(Some((status, Json(payload)).into_response())); + } }; - let cleanup = clear_deleted_proxy_node_references(state, &node_id).await?; - return Ok(Some( - Json(json!({ + let delete_result: Result, GatewayError> = async { + if state.find_proxy_node(&node_id).await?.is_none() { + return Ok(build_admin_proxy_nodes_not_found_response()); + } + + // Persistent references are cleared before the node itself. If a durable cleanup + // fails, the node remains available and the operation can be retried safely. + let cleanup = clear_proxy_node_references_before_delete(state, &node_id).await?; + let Some(_deleted_node) = state.delete_proxy_node(&node_id).await? else { + return Ok(build_admin_proxy_nodes_not_found_response()); + }; + Ok(Json(json!({ "message": build_delete_proxy_node_message(&cleanup), "node_id": node_id, "cleared_system_proxy": cleanup.cleared_system_proxy, + "cleared_external_models_proxy": cleanup.cleared_external_models_proxy, "cleared_providers": cleanup.cleared_providers, "cleared_endpoints": cleanup.cleared_endpoints, "cleared_keys": cleanup.cleared_keys, })) - .into_response(), - )); + .into_response()) + } + .await; + release_admin_external_models_config_mutation_lock(state, &lock).await; + return delete_result.map(Some); } if decision.route_kind.as_deref() == Some("test_node") @@ -869,6 +891,8 @@ pub(crate) async fn maybe_build_local_admin_proxy_nodes_response( #[derive(Debug, Default)] struct DeletedProxyNodeCleanup { cleared_system_proxy: bool, + cleared_external_models_proxy: bool, + external_models_cache_clear_succeeded: Option, cleared_providers: usize, cleared_endpoints: usize, cleared_keys: usize, @@ -895,15 +919,32 @@ struct NormalizedManualProxyEndpoint { node_port: i32, } -async fn clear_deleted_proxy_node_references( +async fn clear_proxy_node_references_before_delete( state: &AdminAppState<'_>, node_id: &str, ) -> Result { + let external_models_cache_clear = state.clear_admin_external_models_cache(); + clear_proxy_node_references_before_delete_with_cache( + state, + node_id, + external_models_cache_clear, + ) + .await +} + +async fn clear_proxy_node_references_before_delete_with_cache( + state: &AdminAppState<'_>, + node_id: &str, + external_models_cache_clear: F, +) -> Result +where + F: Future>, +{ let mut cleanup = DeletedProxyNodeCleanup::default(); if state.app().data.has_system_config_store() { let is_system_proxy = state - .read_system_config_json_value("system_proxy_node_id") + .read_system_config_json_value_strong("system_proxy_node_id") .await? .and_then(|value| value.as_str().map(str::trim).map(ToOwned::to_owned)) .is_some_and(|value| value == node_id); @@ -917,6 +958,36 @@ async fn clear_deleted_proxy_node_references( .await?; cleanup.cleared_system_proxy = true; } + + let is_external_models_proxy = state + .read_system_config_json_value_strong("external_models_proxy_node_id") + .await? + .and_then(|value| value.as_str().map(str::trim).map(ToOwned::to_owned)) + .is_some_and(|value| value == node_id); + if is_external_models_proxy { + state + .upsert_system_config_json_value( + "external_models_proxy_node_id", + &serde_json::Value::Null, + None, + ) + .await?; + cleanup.external_models_cache_clear_succeeded = + Some(match external_models_cache_clear.await { + Ok(_) => true, + Err(_) => { + // The selector is already persisted as null, and v2 cache entries carry + // their selector. A failed DEL therefore cannot route through this node. + warn!( + runtime_backend = state.app().runtime_state_backend(), + proxy_node_id = %node_id, + "failed to clear external models cache while deleting proxy node" + ); + false + } + }); + cleanup.cleared_external_models_proxy = true; + } } if state.app().has_provider_catalog_data_reader() @@ -978,11 +1049,39 @@ async fn clear_deleted_proxy_node_references( Ok(cleanup) } +#[cfg(test)] +pub(crate) async fn clear_proxy_node_references_with_cache_failure_for_tests( + app: &crate::AppState, + node_id: &str, +) -> Result { + let state = AdminAppState::new(app); + let cleanup = clear_proxy_node_references_before_delete_with_cache(&state, node_id, async { + Err(GatewayError::Internal( + "injected cache delete failure".to_string(), + )) + }) + .await?; + Ok(json!({ + "cleared_system_proxy": cleanup.cleared_system_proxy, + "cleared_external_models_proxy": cleanup.cleared_external_models_proxy, + "external_models_cache_clear_succeeded": cleanup.external_models_cache_clear_succeeded, + "cleared_providers": cleanup.cleared_providers, + "cleared_endpoints": cleanup.cleared_endpoints, + "cleared_keys": cleanup.cleared_keys, + })) +} + fn build_delete_proxy_node_message(cleanup: &DeletedProxyNodeCleanup) -> String { let mut parts = vec!["deleted".to_string()]; if cleanup.cleared_system_proxy { parts.push("system default proxy cleared".to_string()); } + if cleanup.cleared_external_models_proxy { + parts.push("external models proxy cleared".to_string()); + } + if cleanup.external_models_cache_clear_succeeded == Some(false) { + parts.push("external models cache invalidation deferred".to_string()); + } if cleanup.cleared_providers > 0 || cleanup.cleared_endpoints > 0 || cleanup.cleared_keys > 0 { parts.push(format!( "cleared proxy refs from {} provider(s), {} endpoint(s), {} key(s)", diff --git a/apps/aether-gateway/src/handlers/admin/system/shared/configs.rs b/apps/aether-gateway/src/handlers/admin/system/shared/configs.rs index 88da10158..8cd5ad72a 100644 --- a/apps/aether-gateway/src/handlers/admin/system/shared/configs.rs +++ b/apps/aether-gateway/src/handlers/admin/system/shared/configs.rs @@ -1,3 +1,4 @@ +use crate::handlers::admin::model::ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY; use crate::handlers::admin::request::AdminAppState; use crate::handlers::shared::unix_secs_to_rfc3339; use crate::GatewayError; @@ -17,6 +18,30 @@ use axum::body::Bytes; use axum::http; use serde_json::json; +const ADMIN_EXTERNAL_MODELS_CONFIG_ROUTE: &str = "/api/admin/models/external/config"; + +fn is_external_models_proxy_node_config_key(key: &str) -> bool { + key.trim() + .eq_ignore_ascii_case(ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY) +} + +fn external_models_proxy_node_config_owner_error( + key: &str, +) -> Option<(http::StatusCode, serde_json::Value)> { + is_external_models_proxy_node_config_key(key).then(|| { + ( + http::StatusCode::BAD_REQUEST, + json!({ + "detail": format!( + "配置项 '{}' 由模型目录管理，请使用 {}", + ADMIN_EXTERNAL_MODELS_PROXY_NODE_CONFIG_KEY, + ADMIN_EXTERNAL_MODELS_CONFIG_ROUTE, + ) + }), + ) + }) +} + fn normalize_admin_system_config_key(requested_key: &str) -> String { normalize_admin_system_config_key_pure(requested_key) } @@ -51,7 +76,12 @@ fn legacy_admin_system_config_fallback_key(normalized_key: &str) -> Option<&'sta pub(crate) fn build_admin_system_configs_payload( entries: &[aether_data::repository::system::StoredSystemConfigEntry], ) -> serde_json::Value { - build_admin_system_configs_payload_pure(entries) + let visible_entries = entries + .iter() + .filter(|entry| !is_external_models_proxy_node_config_key(&entry.key)) + .cloned() + .collect::>(); + build_admin_system_configs_payload_pure(&visible_entries) } pub(crate) async fn build_admin_system_config_detail_payload( @@ -59,6 +89,9 @@ pub(crate) async fn build_admin_system_config_detail_payload( requested_key: &str, ) -> Result, GatewayError> { let requested_key = requested_key.trim(); + if let Some(error) = external_models_proxy_node_config_owner_error(requested_key) { + return Ok(Err(error)); + } let normalized_key = normalize_admin_system_config_key(requested_key); let mut value = state.read_system_config_json_value(&normalized_key).await?; if value.is_none() { @@ -78,6 +111,9 @@ pub(crate) async fn apply_admin_system_config_update( requested_key: &str, request_body: &Bytes, ) -> Result, GatewayError> { + if let Some(error) = external_models_proxy_node_config_owner_error(requested_key) { + return Ok(Err(error)); + } let update = match parse_admin_system_config_update(requested_key, request_body) { Ok(update) => update, Err(err) => return Ok(Err(err)), @@ -123,6 +159,9 @@ pub(crate) async fn delete_admin_system_config( state: &AdminAppState<'_>, requested_key: &str, ) -> Result, GatewayError> { + if let Some(error) = external_models_proxy_node_config_owner_error(requested_key) { + return Ok(Err(error)); + } let delete_keys = admin_system_config_delete_keys(requested_key); let mut deleted = false; for key in &delete_keys { diff --git a/apps/aether-gateway/src/handlers/shared/request_utils.rs b/apps/aether-gateway/src/handlers/shared/request_utils.rs index a9a7895fb..5c268e02f 100644 --- a/apps/aether-gateway/src/handlers/shared/request_utils.rs +++ b/apps/aether-gateway/src/handlers/shared/request_utils.rs @@ -295,6 +295,7 @@ pub(crate) fn admin_proxy_local_requires_buffered_body( http::Method::POST, Some("import_from_upstream"), ) + | (Some("model_external_manage"), http::Method::PUT, Some("external_config_set")) | ( Some("provider_ops_manage"), http::Method::POST, diff --git a/apps/aether-gateway/src/state/core.rs b/apps/aether-gateway/src/state/core.rs index 105c1d15f..78bd09233 100644 --- a/apps/aether-gateway/src/state/core.rs +++ b/apps/aether-gateway/src/state/core.rs @@ -731,6 +731,16 @@ impl AppState { .await } + pub(crate) async fn read_system_config_json_value_strong( + &self, + key: &str, + ) -> Result, GatewayError> { + self.data + .find_system_config_value_strong(key) + .await + .map_err(|err| GatewayError::Internal(err.to_string())) + } + async fn read_system_config_json_value_with_cache_windows( &self, key: &str, diff --git a/apps/aether-gateway/src/tests/control/admin/models/external.rs b/apps/aether-gateway/src/tests/control/admin/models/external.rs index 26101bdce..710729b19 100644 --- a/apps/aether-gateway/src/tests/control/admin/models/external.rs +++ b/apps/aether-gateway/src/tests/control/admin/models/external.rs @@ -1,35 +1,498 @@ use std::sync::{Arc, Mutex}; +use std::time::Duration; +use aether_data::repository::proxy_nodes::{InMemoryProxyNodeRepository, StoredProxyNode}; use axum::body::Body; +use axum::extract::ws::Message; use axum::routing::any; use axum::{extract::Request, Router}; +use base64::Engine as _; use http::StatusCode; +use serde_json::json; +use tokio::sync::watch; -use super::super::super::{build_router_with_state, start_server, AppState}; +use super::super::super::{build_router_with_state, sample_proxy_node, start_server, AppState}; use crate::constants::{ GATEWAY_HEADER, TRUSTED_ADMIN_SESSION_ID_HEADER, TRUSTED_ADMIN_USER_ID_HEADER, TRUSTED_ADMIN_USER_ROLE_HEADER, }; +use crate::data::GatewayDataState; +use crate::handlers::admin::{ + set_admin_external_models_source_url_for_tests, ADMIN_EXTERNAL_MODELS_CONFIG_MUTATION_LOCK_KEY, +}; +use crate::tunnel::{tunnel_protocol, TunnelProxyConn}; -struct TestEnvVarGuard { - key: &'static str, - previous: Option, +fn trusted_admin(request: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + request + .header(GATEWAY_HEADER, "rust-phase3b") + .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") + .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") + .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123") } -impl Drop for TestEnvVarGuard { - fn drop(&mut self) { - if let Some(previous) = self.previous.as_deref() { - std::env::set_var(self.key, previous); - } else { - std::env::remove_var(self.key); - } +fn online_manual_proxy_node(node_id: &str, proxy_url: impl Into) -> StoredProxyNode { + let mut node = sample_proxy_node(node_id); + node.name = node_id.to_string(); + node.status = "online".to_string(); + node.is_manual = true; + node.tunnel_mode = false; + node.tunnel_connected = false; + node.proxy_url = Some(proxy_url.into()); + node.last_heartbeat_at_unix_secs = None; + node.tunnel_connected_at_unix_secs = None; + node.remote_config = None; + node +} + +#[tokio::test] +async fn gateway_manages_admin_external_models_proxy_config_locally() { + let manual_node = online_manual_proxy_node("manual-node", "http://127.0.0.1:8899"); + let mut offline_node = online_manual_proxy_node("offline-node", "http://127.0.0.1:8900"); + offline_node.status = "offline".to_string(); + let repository = Arc::new(InMemoryProxyNodeRepository::seed(vec![ + manual_node, + offline_node, + ])); + let data_state = GatewayDataState::with_proxy_node_repository_for_tests(repository) + .with_system_config_values_for_tests(Vec::<(String, serde_json::Value)>::new()); + let gateway = build_router_with_state( + AppState::new() + .expect("gateway should build") + .with_data_state_for_tests(data_state), + ); + let (gateway_url, gateway_handle) = start_server(gateway).await; + let client = reqwest::Client::new(); + let config_url = format!("{gateway_url}/api/admin/models/external/config"); + + let response = trusted_admin(client.get(&config_url)) + .send() + .await + .expect("initial config request should succeed"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!(payload["proxy_node_id"], serde_json::Value::Null); + + for invalid_payload in [ + json!({}), + json!({ "proxy_node_id": true }), + json!({ "proxy_node_id": "" }), + json!({ "proxy_node_id": " " }), + ] { + let response = trusted_admin(client.put(&config_url)) + .json(&invalid_payload) + .send() + .await + .expect("invalid config request should complete"); + assert_eq!( + response.status(), + StatusCode::BAD_REQUEST, + "invalid payload should be rejected: {invalid_payload}" + ); } + + let response = trusted_admin(client.put(&config_url)) + .json(&json!({ "proxy_node_id": "missing-node" })) + .send() + .await + .expect("missing-node config request should complete"); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + + let response = trusted_admin(client.put(&config_url)) + .json(&json!({ "proxy_node_id": "offline-node" })) + .send() + .await + .expect("offline-node config request should complete"); + assert_eq!(response.status(), StatusCode::CONFLICT); + + let response = trusted_admin(client.get(&config_url)) + .send() + .await + .expect("config should remain readable"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!(payload["proxy_node_id"], serde_json::Value::Null); + + let response = trusted_admin(client.put(&config_url)) + .json(&json!({ "proxy_node_id": "manual-node" })) + .send() + .await + .expect("manual-node config request should succeed"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!(payload["proxy_node_id"], "manual-node"); + assert!(payload["cache_cleared"].is_boolean()); + + let response = trusted_admin(client.get(&config_url)) + .send() + .await + .expect("saved config request should succeed"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!(payload["proxy_node_id"], "manual-node"); + + let response = trusted_admin(client.put(&config_url)) + .json(&json!({ "proxy_node_id": null })) + .send() + .await + .expect("direct config request should succeed"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!(payload["proxy_node_id"], serde_json::Value::Null); + assert!(payload["cache_cleared"].is_boolean()); + + gateway_handle.abort(); } -fn set_test_env_var(key: &'static str, value: &str) -> TestEnvVarGuard { - let previous = std::env::var(key).ok(); - std::env::set_var(key, value); - TestEnvVarGuard { key, previous } +#[tokio::test] +async fn gateway_serializes_external_models_proxy_selection_with_proxy_node_deletion() { + let node_a = online_manual_proxy_node("node-a", "http://127.0.0.1:8898"); + let node_b = online_manual_proxy_node("node-b", "http://127.0.0.1:8899"); + let repository = Arc::new(InMemoryProxyNodeRepository::seed(vec![node_a, node_b])); + let data_state = GatewayDataState::with_proxy_node_repository_for_tests(repository) + .with_system_config_values_for_tests([( + "external_models_proxy_node_id".to_string(), + json!("node-a"), + )]); + let state = AppState::new() + .expect("gateway should build") + .with_data_state_for_tests(data_state.clone()); + let control_state = state.clone(); + let lock = control_state + .runtime_state() + .lock_try_acquire( + ADMIN_EXTERNAL_MODELS_CONFIG_MUTATION_LOCK_KEY, + "external-models-race-test", + Duration::from_secs(60), + ) + .await + .expect("test mutation lock should be available") + .expect("test mutation lock should be acquired"); + + let gateway = build_router_with_state(state); + let (gateway_url, gateway_handle) = start_server(gateway).await; + let client = reqwest::Client::new(); + let config_url = format!("{gateway_url}/api/admin/models/external/config"); + + let response = trusted_admin(client.put(&config_url)) + .json(&json!({ "proxy_node_id": "node-b" })) + .send() + .await + .expect("contended config request should complete"); + assert_eq!(response.status(), StatusCode::CONFLICT); + + let response = + trusted_admin(client.delete(format!("{gateway_url}/api/admin/proxy-nodes/node-a"))) + .send() + .await + .expect("contended delete request should complete"); + assert_eq!(response.status(), StatusCode::CONFLICT); + assert_eq!( + data_state + .find_system_config_value("external_models_proxy_node_id") + .await + .expect("selector lookup should succeed"), + Some(json!("node-a")) + ); + assert!(data_state + .find_proxy_node("node-a") + .await + .expect("node lookup should succeed") + .is_some()); + + assert!(control_state + .runtime_state() + .lock_release(&lock) + .await + .expect("test mutation lock should release")); + + let response = trusted_admin(client.put(&config_url)) + .json(&json!({ "proxy_node_id": "node-b" })) + .send() + .await + .expect("config request should succeed after lock release"); + assert_eq!(response.status(), StatusCode::OK); + + let response = + trusted_admin(client.delete(format!("{gateway_url}/api/admin/proxy-nodes/node-a"))) + .send() + .await + .expect("delete request should succeed after lock release"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!(payload["cleared_external_models_proxy"], false); + assert_eq!( + data_state + .find_system_config_value("external_models_proxy_node_id") + .await + .expect("selector lookup should succeed"), + Some(json!("node-b")), + "deleting node A must not overwrite a concurrently chosen node B" + ); + + let response = trusted_admin(client.put(&config_url)) + .json(&json!({ "proxy_node_id": "node-a" })) + .send() + .await + .expect("deleted-node config request should complete"); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + assert_eq!( + data_state + .find_system_config_value("external_models_proxy_node_id") + .await + .expect("selector lookup should succeed"), + Some(json!("node-b")), + "a failed save must not leave a dangling deleted node ID" + ); + + gateway_handle.abort(); +} + +#[tokio::test] +async fn gateway_reports_unavailable_external_models_proxy_and_fails_closed() { + let direct_source_hits = Arc::new(Mutex::new(0usize)); + let direct_source_hits_clone = Arc::clone(&direct_source_hits); + let direct_source = Router::new().route( + "/api.json", + any(move |_request: Request| { + let direct_source_hits_inner = Arc::clone(&direct_source_hits_clone); + async move { + *direct_source_hits_inner.lock().expect("mutex should lock") += 1; + ( + StatusCode::OK, + axum::Json(json!({ + "unexpected-direct-provider": { + "name": "Unexpected direct fallback", + "models": {} + } + })), + ) + } + }), + ); + let (direct_source_url, direct_source_handle) = start_server(direct_source).await; + let _guard = + set_admin_external_models_source_url_for_tests(&format!("{direct_source_url}/api.json")); + + let mut offline_node = online_manual_proxy_node("offline-node", "http://127.0.0.1:8900"); + offline_node.status = "offline".to_string(); + let repository = Arc::new(InMemoryProxyNodeRepository::seed(vec![offline_node])); + let data_state = GatewayDataState::with_proxy_node_repository_for_tests(repository) + .with_system_config_values_for_tests([( + "external_models_proxy_node_id".to_string(), + json!("offline-node"), + )]); + let gateway = build_router_with_state( + AppState::new() + .expect("gateway should build") + .with_data_state_for_tests(data_state), + ); + let (gateway_url, gateway_handle) = start_server(gateway).await; + let client = reqwest::Client::new(); + + let response = + trusted_admin(client.get(format!("{gateway_url}/api/admin/models/external/config"))) + .send() + .await + .expect("config request should succeed"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!(payload["proxy_node_id"], "offline-node"); + + let response = trusted_admin(client.get(format!("{gateway_url}/api/admin/models/external"))) + .send() + .await + .expect("catalog request should complete"); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(*direct_source_hits.lock().expect("mutex should lock"), 0); + + gateway_handle.abort(); + direct_source_handle.abort(); +} + +#[tokio::test] +async fn gateway_fetches_external_models_through_connected_tunnel_node() { + let source_url = "https://models.dev.test/api.json"; + let _guard = set_admin_external_models_source_url_for_tests(source_url); + + let mut tunnel_node = sample_proxy_node("tunnel-node"); + tunnel_node.name = "Tunnel Node".to_string(); + tunnel_node.status = "online".to_string(); + tunnel_node.tunnel_mode = true; + tunnel_node.tunnel_connected = true; + tunnel_node.remote_config = None; + let repository = Arc::new(InMemoryProxyNodeRepository::seed(vec![tunnel_node])); + let data_state = GatewayDataState::with_proxy_node_repository_for_tests(repository) + .with_system_config_values_for_tests([( + "external_models_proxy_node_id".to_string(), + json!("tunnel-node"), + )]); + let state = AppState::new() + .expect("gateway should build") + .with_data_state_for_tests(data_state); + let tunnel_state = state.tunnel.app_state(); + let (proxy_tx, mut proxy_rx) = aether_runtime::bounded_queue(8); + let (proxy_close_tx, _) = watch::channel(false); + tunnel_state + .hub + .register_proxy(Arc::new(TunnelProxyConn::new( + 700, + "tunnel-node".to_string(), + "Tunnel Node".to_string(), + proxy_tx, + proxy_close_tx, + 16, + 2, + ))); + + let gateway = build_router_with_state(state); + let (gateway_url, gateway_handle) = start_server(gateway).await; + let request_task = tokio::spawn(async move { + trusted_admin( + reqwest::Client::new().get(format!("{gateway_url}/api/admin/models/external")), + ) + .send() + .await + }); + + let request_headers = match tokio::time::timeout(Duration::from_secs(5), proxy_rx.recv()) + .await + .expect("headers frame should arrive before timeout") + .expect("headers frame should arrive") + { + Message::Binary(data) => data, + other => panic!("unexpected message: {other:?}"), + }; + let request_header = + tunnel_protocol::FrameHeader::parse(&request_headers).expect("request header should parse"); + assert_eq!(request_header.msg_type, tunnel_protocol::REQUEST_HEADERS); + let meta_payload = tunnel_protocol::decode_payload(&request_headers, &request_header) + .expect("request header payload should decode"); + let meta: tunnel_protocol::RequestMeta = + serde_json::from_slice(&meta_payload).expect("request meta should parse"); + assert_eq!(meta.method, "GET"); + assert_eq!(meta.url, source_url); + assert_eq!(meta.follow_redirects, Some(true)); + assert_eq!( + meta.headers.get("accept").map(String::as_str), + Some("application/json") + ); + + let request_body = match tokio::time::timeout(Duration::from_secs(5), proxy_rx.recv()) + .await + .expect("body frame should arrive before timeout") + .expect("body frame should arrive") + { + Message::Binary(data) => data, + other => panic!("unexpected message: {other:?}"), + }; + let request_body_header = + tunnel_protocol::FrameHeader::parse(&request_body).expect("request body should parse"); + assert_eq!(request_body_header.msg_type, tunnel_protocol::REQUEST_BODY); + assert_ne!( + request_body_header.flags & tunnel_protocol::FLAG_END_STREAM, + 0, + "catalog request body frame should close the stream" + ); + + let response_meta = tunnel_protocol::ResponseMeta { + status: 200, + headers: vec![("content-type".to_string(), "application/json".to_string())], + }; + let response_meta_bytes = + serde_json::to_vec(&response_meta).expect("response meta should serialize"); + let mut response_headers_frame = tunnel_protocol::encode_frame( + request_header.stream_id, + tunnel_protocol::RESPONSE_HEADERS, + 0, + &response_meta_bytes, + ); + tunnel_state + .hub + .handle_proxy_frame(700, &mut response_headers_frame) + .await; + + let response_payload = serde_json::to_vec(&json!({ + "tunnel-provider": { + "name": "Tunnel", + "models": { + "tunnel": { + "name": "TUNNEL" + } + } + } + })) + .expect("response payload should serialize"); + let mut response_body_frame = tunnel_protocol::encode_frame( + request_header.stream_id, + tunnel_protocol::RESPONSE_BODY, + 0, + &response_payload, + ); + tunnel_state + .hub + .handle_proxy_frame(700, &mut response_body_frame) + .await; + + let mut response_end_frame = tunnel_protocol::encode_frame( + request_header.stream_id, + tunnel_protocol::STREAM_END, + 0, + &[], + ); + tunnel_state + .hub + .handle_proxy_frame(700, &mut response_end_frame) + .await; + + let response = tokio::time::timeout(Duration::from_secs(5), request_task) + .await + .expect("catalog request should complete before timeout") + .expect("request task should complete") + .expect("catalog request should succeed"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!( + payload["tunnel-provider"]["models"]["tunnel"]["name"], + "TUNNEL" + ); + + gateway_handle.abort(); +} + +#[tokio::test] +async fn gateway_rejects_external_models_proxy_config_when_data_stores_are_unavailable() { + let gateway = build_router_with_state(AppState::new().expect("gateway should build")); + let (gateway_url, gateway_handle) = start_server(gateway).await; + let client = reqwest::Client::new(); + + let response = + trusted_admin(client.put(format!("{gateway_url}/api/admin/models/external/config"))) + .json(&json!({ "proxy_node_id": null })) + .send() + .await + .expect("missing config store request should complete"); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + gateway_handle.abort(); + + let data_state = + GatewayDataState::disabled() + .with_system_config_values_for_tests(Vec::<(String, serde_json::Value)>::new()); + let gateway = build_router_with_state( + AppState::new() + .expect("gateway should build") + .with_data_state_for_tests(data_state), + ); + let (gateway_url, gateway_handle) = start_server(gateway).await; + + let response = + trusted_admin(client.put(format!("{gateway_url}/api/admin/models/external/config"))) + .json(&json!({ "proxy_node_id": "manual-node" })) + .send() + .await + .expect("missing proxy reader request should complete"); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + + gateway_handle.abort(); } #[tokio::test] @@ -47,53 +510,131 @@ async fn gateway_handles_admin_external_models_locally_with_trusted_admin_princi }), ); - let (upstream_url, upstream_handle) = start_server(upstream).await; + let (_upstream_url, upstream_handle) = start_server(upstream).await; + let external_source_hits = Arc::new(Mutex::new(0usize)); + let external_source_hits_clone = Arc::clone(&external_source_hits); let external_source = Router::new().route( "/api.json", - any(|_request: Request| async move { + any(move |_request: Request| { + let external_source_hits_inner = Arc::clone(&external_source_hits_clone); + async move { + *external_source_hits_inner + .lock() + .expect("mutex should lock") += 1; + ( + StatusCode::OK, + axum::Json(json!({ + "direct-provider": { + "name": "Direct", + "models": { + "direct": { + "name": "DIRECT" + } + } + } + })), + ) + } + }), + ); + let (external_source_url, external_source_handle) = start_server(external_source).await; + let proxy_hits = Arc::new(Mutex::new(0usize)); + let proxy_hits_clone = Arc::clone(&proxy_hits); + let proxy_auths = Arc::new(Mutex::new(Vec::>::new())); + let proxy_auths_clone = Arc::clone(&proxy_auths); + let proxy = Router::new().fallback(any(move |request: Request| { + let proxy_hits_inner = Arc::clone(&proxy_hits_clone); + let proxy_auths_inner = Arc::clone(&proxy_auths_clone); + async move { + *proxy_hits_inner.lock().expect("mutex should lock") += 1; + proxy_auths_inner.lock().expect("mutex should lock").push( + request + .headers() + .get("proxy-authorization") + .and_then(|value| value.to_str().ok()) + .map(str::to_string), + ); ( StatusCode::OK, - axum::Json(serde_json::json!({ - "openai": { - "name": "OpenAI", + axum::Json(json!({ + "manual-provider": { + "name": "Manual", "models": { - "gpt-5": { - "name": "GPT-5" + "manual": { + "name": "MANUAL" } } } })), ) - }), - ); - let (external_source_url, external_source_handle) = start_server(external_source).await; - let _guard = set_test_env_var( - "AETHER_GATEWAY_EXTERNAL_MODELS_URL", - &format!("{external_source_url}/api.json"), - ); - let gateway = build_router_with_state(AppState::new().expect("gateway should build")); - let (gateway_url, gateway_handle) = start_server(gateway).await; + } + })); + let (proxy_url, proxy_handle) = start_server(proxy).await; + let _guard = + set_admin_external_models_source_url_for_tests(&format!("{external_source_url}/api.json")); - let response = reqwest::Client::new() - .get(format!("{gateway_url}/api/admin/models/external")) - .header(crate::constants::GATEWAY_HEADER, "rust-phase3b") - .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") - .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") - .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123") + let mut manual_node = online_manual_proxy_node("manual-node", proxy_url); + manual_node.proxy_username = Some("alice".to_string()); + manual_node.proxy_password = Some("supersecret".to_string()); + let repository = Arc::new(InMemoryProxyNodeRepository::seed(vec![manual_node])); + let data_state = GatewayDataState::with_proxy_node_repository_for_tests(repository) + .with_system_config_values_for_tests(Vec::<(String, serde_json::Value)>::new()); + let gateway = build_router_with_state( + AppState::new() + .expect("gateway should build") + .with_data_state_for_tests(data_state), + ); + let (gateway_url, gateway_handle) = start_server(gateway).await; + let client = reqwest::Client::new(); + + let response = trusted_admin(client.get(format!("{gateway_url}/api/admin/models/external"))) .send() .await .expect("request should succeed"); assert_eq!(response.status(), StatusCode::OK); let payload: serde_json::Value = response.json().await.expect("json body should parse"); - assert_eq!(payload["openai"]["official"], serde_json::json!(true)); assert_eq!( - payload["openai"]["models"]["gpt-5"]["name"], - serde_json::json!("GPT-5") + payload["direct-provider"]["models"]["direct"]["name"], + json!("DIRECT") ); + assert_eq!(*external_source_hits.lock().expect("mutex should lock"), 1); assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0); + let response = + trusted_admin(client.put(format!("{gateway_url}/api/admin/models/external/config"))) + .json(&json!({ "proxy_node_id": "manual-node" })) + .send() + .await + .expect("proxy config request should succeed"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!(payload["proxy_node_id"], "manual-node"); + assert_eq!(payload["cache_cleared"], true); + + let response = trusted_admin(client.get(format!("{gateway_url}/api/admin/models/external"))) + .send() + .await + .expect("proxied request should succeed"); + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await.expect("json body should parse"); + assert_eq!( + payload["manual-provider"]["models"]["manual"]["name"], + json!("MANUAL") + ); + assert_eq!(*external_source_hits.lock().expect("mutex should lock"), 1); + assert_eq!(*proxy_hits.lock().expect("mutex should lock"), 1); + let expected_proxy_auth = format!( + "Basic {}", + base64::engine::general_purpose::STANDARD.encode("alice:supersecret") + ); + assert_eq!( + proxy_auths.lock().expect("mutex should lock").as_slice(), + [Some(expected_proxy_auth)] + ); + gateway_handle.abort(); + proxy_handle.abort(); external_source_handle.abort(); upstream_handle.abort(); } @@ -117,15 +658,12 @@ async fn gateway_clears_admin_external_models_cache_locally_with_trusted_admin_p let gateway = build_router_with_state(AppState::new().expect("gateway should build")); let (gateway_url, gateway_handle) = start_server(gateway).await; - let response = reqwest::Client::new() - .delete(format!("{gateway_url}/api/admin/models/external/cache")) - .header(crate::constants::GATEWAY_HEADER, "rust-phase3b") - .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") - .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") - .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123") - .send() - .await - .expect("request should succeed"); + let response = trusted_admin( + reqwest::Client::new().delete(format!("{gateway_url}/api/admin/models/external/cache")), + ) + .send() + .await + .expect("request should succeed"); assert_eq!(response.status(), StatusCode::OK); let payload: serde_json::Value = response.json().await.expect("json body should parse"); diff --git a/apps/aether-gateway/src/tests/control/admin/proxy_nodes.rs b/apps/aether-gateway/src/tests/control/admin/proxy_nodes.rs index 0fd471705..04c069278 100644 --- a/apps/aether-gateway/src/tests/control/admin/proxy_nodes.rs +++ b/apps/aether-gateway/src/tests/control/admin/proxy_nodes.rs @@ -1361,10 +1361,13 @@ async fn gateway_deletes_proxy_nodes_and_clears_proxy_refs_locally() { let data_state = GatewayDataState::with_proxy_node_repository_for_tests(Arc::clone(&proxy_node_repository)) .attach_provider_catalog_repository_for_tests(Arc::clone(&provider_catalog_repository)) - .with_system_config_values_for_tests(vec![( - "system_proxy_node_id".to_string(), - json!("manual-node-1"), - )]); + .with_system_config_values_for_tests(vec![ + ("system_proxy_node_id".to_string(), json!("manual-node-1")), + ( + "external_models_proxy_node_id".to_string(), + json!("manual-node-1"), + ), + ]); let gateway = build_router_with_state( AppState::new() .expect("gateway should build") @@ -1384,6 +1387,7 @@ async fn gateway_deletes_proxy_nodes_and_clears_proxy_refs_locally() { assert_eq!(response.status(), StatusCode::OK); let payload: serde_json::Value = response.json().await.expect("json body should parse"); assert_eq!(payload["cleared_system_proxy"], true); + assert_eq!(payload["cleared_external_models_proxy"], true); assert_eq!(payload["cleared_providers"], 1); assert_eq!(payload["cleared_endpoints"], 1); assert_eq!(payload["cleared_keys"], 1); @@ -1400,6 +1404,13 @@ async fn gateway_deletes_proxy_nodes_and_clears_proxy_refs_locally() { .expect("system config lookup should succeed"), Some(serde_json::Value::Null) ); + assert_eq!( + data_state + .find_system_config_value("external_models_proxy_node_id") + .await + .expect("external models proxy config lookup should succeed"), + Some(serde_json::Value::Null) + ); let provider_ids = vec!["provider-1".to_string()]; let providers = data_state @@ -1421,6 +1432,101 @@ async fn gateway_deletes_proxy_nodes_and_clears_proxy_refs_locally() { gateway_handle.abort(); } +#[tokio::test] +async fn gateway_continues_proxy_reference_cleanup_when_external_models_cache_delete_fails() { + let mut manual_node = sample_proxy_node("manual-node-cache-failure"); + manual_node.status = "online".to_string(); + manual_node.is_manual = true; + manual_node.tunnel_mode = false; + manual_node.tunnel_connected = false; + manual_node.proxy_url = Some("http://127.0.0.1:8899".to_string()); + manual_node.last_heartbeat_at_unix_secs = None; + manual_node.tunnel_connected_at_unix_secs = None; + + let proxy_node_repository = Arc::new(InMemoryProxyNodeRepository::seed(vec![manual_node])); + let mut provider = sample_provider("provider-cache-failure", "OpenAI", 10); + provider.proxy = Some(json!({ + "node_id": "manual-node-cache-failure", + "enabled": true + })); + let mut endpoint = sample_endpoint( + "endpoint-cache-failure", + "provider-cache-failure", + "openai:chat", + "https://example.com/v1", + ); + endpoint.proxy = Some(json!({ + "node_id": "manual-node-cache-failure", + "enabled": true + })); + let mut key = sample_key( + "key-cache-failure", + "provider-cache-failure", + "openai:chat", + "sk-test", + ); + key.proxy = Some(json!({ + "node_id": "manual-node-cache-failure", + "enabled": true + })); + let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed( + vec![provider], + vec![endpoint], + vec![key], + )); + let data_state = + GatewayDataState::with_proxy_node_repository_for_tests(Arc::clone(&proxy_node_repository)) + .attach_provider_catalog_repository_for_tests(Arc::clone(&provider_catalog_repository)) + .with_system_config_values_for_tests([( + "external_models_proxy_node_id".to_string(), + json!("manual-node-cache-failure"), + )]); + let state = AppState::new() + .expect("gateway should build") + .with_data_state_for_tests(data_state.clone()); + + let cleanup = crate::handlers::admin::clear_proxy_node_references_with_cache_failure_for_tests( + &state, + "manual-node-cache-failure", + ) + .await + .expect("cache failure should not abort persistent reference cleanup"); + assert_eq!(cleanup["cleared_external_models_proxy"], true); + assert_eq!(cleanup["external_models_cache_clear_succeeded"], false); + assert_eq!(cleanup["cleared_providers"], 1); + assert_eq!(cleanup["cleared_endpoints"], 1); + assert_eq!(cleanup["cleared_keys"], 1); + assert_eq!( + data_state + .find_system_config_value("external_models_proxy_node_id") + .await + .expect("external models proxy config lookup should succeed"), + Some(serde_json::Value::Null) + ); + assert!(data_state + .find_proxy_node("manual-node-cache-failure") + .await + .expect("node lookup should succeed") + .is_some()); + + let provider_ids = vec!["provider-cache-failure".to_string()]; + let providers = data_state + .list_provider_catalog_providers(false) + .await + .expect("provider list should succeed"); + assert!(providers.iter().all(|provider| provider.proxy.is_none())); + let endpoints = data_state + .list_provider_catalog_endpoints_by_provider_ids(&provider_ids) + .await + .expect("endpoint list should succeed"); + assert!(endpoints.iter().all(|endpoint| endpoint.proxy.is_none())); + let keys = data_state + .list_provider_catalog_keys_by_provider_ids(&provider_ids) + .await + .expect("key list should succeed"); + assert!(keys.iter().all(|key| key.proxy.is_none())); +} + #[tokio::test] async fn gateway_handles_admin_proxy_node_events_locally_with_trusted_admin_principal() { let upstream_hits = Arc::new(Mutex::new(0usize)); diff --git a/apps/aether-gateway/src/tests/control/admin/system.rs b/apps/aether-gateway/src/tests/control/admin/system.rs index a5eb90068..d2763f71a 100644 --- a/apps/aether-gateway/src/tests/control/admin/system.rs +++ b/apps/aether-gateway/src/tests/control/admin/system.rs @@ -1494,6 +1494,10 @@ async fn gateway_handles_admin_system_configs_locally_with_trusted_admin_princip json!("encrypted-turnstile-secret"), ), ("site_name".to_string(), json!("Aether Test")), + ( + "external_models_proxy_node_id".to_string(), + json!("proxy-node-hidden"), + ), ]); let (upstream_url, upstream_handle) = start_server(upstream).await; let gateway = build_router_with_state( @@ -1520,6 +1524,9 @@ async fn gateway_handles_admin_system_configs_locally_with_trusted_admin_princip .iter() .any(|item| item["key"] == "request_record_level")); assert!(!items.iter().any(|item| item["key"] == "request_log_level")); + assert!(!items + .iter() + .any(|item| item["key"] == "external_models_proxy_node_id")); let smtp_password = items .iter() .find(|item| item["key"] == "smtp_password") @@ -1538,6 +1545,82 @@ async fn gateway_handles_admin_system_configs_locally_with_trusted_admin_princip upstream_handle.abort(); } +#[tokio::test] +async fn gateway_rejects_external_models_proxy_through_generic_system_config_routes() { + let data_state = GatewayDataState::disabled().with_system_config_values_for_tests(vec![( + "external_models_proxy_node_id".to_string(), + json!("proxy-node-owned-by-models"), + )]); + let gateway = build_router_with_state( + AppState::new() + .expect("gateway should build") + .with_data_state_for_tests(data_state.clone()), + ); + let (gateway_url, gateway_handle) = start_server(gateway).await; + let client = reqwest::Client::new(); + let config_url = + format!("{gateway_url}/api/admin/system/configs/external_models_proxy_node_id"); + + let get_response = client + .get(&config_url) + .header(crate::constants::GATEWAY_HEADER, "rust-phase3b") + .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") + .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") + .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123") + .send() + .await + .expect("generic config get should complete"); + assert_eq!(get_response.status(), StatusCode::BAD_REQUEST); + let get_payload: serde_json::Value = get_response.json().await.expect("json body should parse"); + assert!(get_payload["detail"] + .as_str() + .is_some_and(|detail| detail.contains("/api/admin/models/external/config"))); + + let put_response = client + .put(&config_url) + .header(crate::constants::GATEWAY_HEADER, "rust-phase3b") + .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") + .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") + .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123") + .json(&json!({ "value": null })) + .send() + .await + .expect("generic config put should complete"); + assert_eq!(put_response.status(), StatusCode::BAD_REQUEST); + let put_payload: serde_json::Value = put_response.json().await.expect("json body should parse"); + assert!(put_payload["detail"] + .as_str() + .is_some_and(|detail| detail.contains("/api/admin/models/external/config"))); + + let delete_response = client + .delete(&config_url) + .header(crate::constants::GATEWAY_HEADER, "rust-phase3b") + .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") + .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") + .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123") + .send() + .await + .expect("generic config delete should complete"); + assert_eq!(delete_response.status(), StatusCode::BAD_REQUEST); + let delete_payload: serde_json::Value = delete_response + .json() + .await + .expect("json body should parse"); + assert!(delete_payload["detail"] + .as_str() + .is_some_and(|detail| detail.contains("/api/admin/models/external/config"))); + + assert_eq!( + data_state + .find_system_config_value_strong("external_models_proxy_node_id") + .await + .expect("external models config should remain readable"), + Some(json!("proxy-node-owned-by-models")) + ); + + gateway_handle.abort(); +} + #[tokio::test] async fn gateway_handles_admin_system_config_detail_locally_with_trusted_admin_principal() { let upstream_hits = Arc::new(Mutex::new(0usize)); diff --git a/apps/aether-gateway/src/tests/control/admin/system_import.rs b/apps/aether-gateway/src/tests/control/admin/system_import.rs index d03cb3cac..174712420 100644 --- a/apps/aether-gateway/src/tests/control/admin/system_import.rs +++ b/apps/aether-gateway/src/tests/control/admin/system_import.rs @@ -350,13 +350,22 @@ async fn gateway_imports_admin_system_config_locally_and_persists_data_impl() { let (gateway_url, gateway_handle) = start_server(gateway).await; let client = reqwest::Client::new(); + let mut import_payload = sample_system_import_payload(); + import_payload["system_configs"] + .as_array_mut() + .expect("system configs should be an array") + .push(json!({ + "key": "external_models_proxy_node_id", + "value": null, + "description": "External models proxy" + })); let response = client .post(format!("{gateway_url}/api/admin/system/config/import")) .header(GATEWAY_HEADER, "rust-phase3b") .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123") - .json(&sample_system_import_payload()) + .json(&import_payload) .send() .await .expect("request should succeed"); @@ -372,7 +381,7 @@ async fn gateway_imports_admin_system_config_locally_and_persists_data_impl() { assert_eq!(payload["stats"]["models"]["created"], json!(1)); assert_eq!(payload["stats"]["ldap"]["created"], json!(1)); assert_eq!(payload["stats"]["oauth"]["created"], json!(1)); - assert_eq!(payload["stats"]["system_configs"]["created"], json!(2)); + assert_eq!(payload["stats"]["system_configs"]["created"], json!(3)); assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0); let global_models = global_model_repository @@ -531,8 +540,16 @@ async fn gateway_imports_admin_system_config_locally_and_persists_data_impl() { .iter() .find(|entry| entry["key"] == "smtp_password") .expect("smtp_password should exist"); + let exported_external_models_proxy = exported_system_configs + .iter() + .find(|entry| entry["key"] == "external_models_proxy_node_id") + .expect("external models proxy should exist"); assert_eq!(exported_site_name["value"], "Imported Aether"); assert_eq!(exported_smtp_password["value"], "smtp-secret"); + assert_eq!( + exported_external_models_proxy["value"], + serde_json::Value::Null + ); gateway_handle.abort(); upstream_handle.abort(); @@ -2459,10 +2476,11 @@ fn gateway_skips_proxy_nodes_during_admin_system_config_import() { } async fn gateway_skips_proxy_nodes_during_admin_system_config_import_impl() { + let data_state = build_empty_admin_system_data_state(); let gateway = build_router_with_state( AppState::new() .expect("gateway should build") - .with_data_state_for_tests(build_empty_admin_system_data_state()), + .with_data_state_for_tests(data_state.clone()), ); let (gateway_url, gateway_handle) = start_server(gateway).await; @@ -2482,6 +2500,11 @@ async fn gateway_skips_proxy_nodes_during_admin_system_config_import_impl() { "name": "Legacy Node", "ip": "127.0.0.1", "port": 8080 + }], + "system_configs": [{ + "key": "external_models_proxy_node_id", + "value": "legacy-node-1", + "description": "External models proxy" }] })) .send() @@ -2491,6 +2514,7 @@ async fn gateway_skips_proxy_nodes_during_admin_system_config_import_impl() { assert_eq!(response.status(), StatusCode::OK); let payload: Value = response.json().await.expect("json body should parse"); assert_eq!(payload["stats"]["proxy_nodes"]["skipped"], json!(1)); + assert_eq!(payload["stats"]["system_configs"]["created"], json!(1)); assert!(payload["stats"]["errors"] .as_array() .expect("errors should be an array") @@ -2498,6 +2522,20 @@ async fn gateway_skips_proxy_nodes_during_admin_system_config_import_impl() { .any(|item| item .as_str() .is_some_and(|value| value.contains("暂不支持导入代理节点")))); + assert!(payload["stats"]["errors"] + .as_array() + .expect("errors should be an array") + .iter() + .any(|item| item + .as_str() + .is_some_and(|value| value.contains("已切换为直连")))); + assert_eq!( + data_state + .find_system_config_value("external_models_proxy_node_id") + .await + .expect("external models proxy config lookup should succeed"), + Some(Value::Null) + ); gateway_handle.abort(); } diff --git a/crates/aether-admin/src/system.rs b/crates/aether-admin/src/system.rs index 219822ed3..b204a8d95 100644 --- a/crates/aether-admin/src/system.rs +++ b/crates/aether-admin/src/system.rs @@ -1749,6 +1749,7 @@ pub fn admin_system_config_default_value(key: &str) -> Option "audit_log_retention_days" => Some(json!(30)), "enable_db_maintenance" => Some(json!(true)), "system_proxy_node_id" => Some(serde_json::Value::Null), + "external_models_proxy_node_id" => Some(serde_json::Value::Null), "smtp_host" => Some(serde_json::Value::Null), "smtp_port" => Some(json!(587)), "smtp_user" => Some(serde_json::Value::Null), diff --git a/frontend/src/api/__tests__/models-dev.spec.ts b/frontend/src/api/__tests__/models-dev.spec.ts index 47e2af766..f3efa6a0d 100644 --- a/frontend/src/api/__tests__/models-dev.spec.ts +++ b/frontend/src/api/__tests__/models-dev.spec.ts @@ -2,23 +2,62 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' const apiMocks = vi.hoisted(() => ({ get: vi.fn(), + put: vi.fn(), delete: vi.fn(), })) vi.mock('@/api/client', () => ({ - default: { get: apiMocks.get, delete: apiMocks.delete }, + default: { get: apiMocks.get, put: apiMocks.put, delete: apiMocks.delete }, })) -import { clearModelsDevCache, getModelsDevList, refreshModelsDevList } from '@/api/models-dev' +import { + clearModelsDevCache, + getExternalModelsAccessConfig, + getModelsDevList, + refreshModelsDevList, + updateExternalModelsAccessConfig, +} from '@/api/models-dev' beforeEach(() => { clearModelsDevCache() localStorage.clear() apiMocks.get.mockReset() + apiMocks.put.mockReset() apiMocks.delete.mockReset() apiMocks.delete.mockResolvedValue({ data: { cleared: true } }) }) +describe('external models access config', () => { + it('reads the configured proxy node', async () => { + apiMocks.get.mockResolvedValue({ data: { proxy_node_id: 'proxy-1' } }) + + await expect(getExternalModelsAccessConfig()).resolves.toEqual({ + proxy_node_id: 'proxy-1', + }) + expect(apiMocks.get).toHaveBeenCalledWith('/api/admin/models/external/config') + }) + + it.each([ + ['a proxy node', 'proxy-1'], + ['direct access', null], + ] as const)('updates %s and clears the browser cache', async (_label, proxyNodeId) => { + localStorage.setItem('models_dev_cache', JSON.stringify({ timestamp: Date.now(), data: {} })) + apiMocks.put.mockResolvedValue({ + data: { proxy_node_id: proxyNodeId, cache_cleared: true }, + }) + + await expect(updateExternalModelsAccessConfig(proxyNodeId)).resolves.toEqual({ + proxy_node_id: proxyNodeId, + cache_cleared: true, + }) + expect(apiMocks.put).toHaveBeenCalledWith( + '/api/admin/models/external/config', + { proxy_node_id: proxyNodeId }, + ) + expect(localStorage.getItem('models_dev_cache')).toBeNull() + }) +}) + describe('getModelsDevList', () => { it('uses current modalities and experimental mode pricing while keeping legacy fallbacks', async () => { apiMocks.get.mockResolvedValue({ diff --git a/frontend/src/api/models-dev.ts b/frontend/src/api/models-dev.ts index 2e45675bb..0bb28219c 100644 --- a/frontend/src/api/models-dev.ts +++ b/frontend/src/api/models-dev.ts @@ -73,6 +73,14 @@ export interface ModelsDevProvider { export type ModelsDevData = Record +export interface ExternalModelsAccessConfig { + proxy_node_id: string | null +} + +export interface ExternalModelsAccessConfigUpdate extends ExternalModelsAccessConfig { + cache_cleared: boolean +} + // 扁平化的模型列表项（用于搜索和选择） export interface ModelsDevModelItem { providerId: string @@ -165,6 +173,29 @@ export async function getModelsDevData(): Promise { return data } +/** + * 获取 models.dev 目录的出站访问配置。 + */ +export async function getExternalModelsAccessConfig(): Promise { + const response = await api.get('/api/admin/models/external/config') + return response.data +} + +/** + * 更新 models.dev 目录的代理节点。null 表示直连。 + * 后端会清理共享目录缓存；前端同时清理内存和 localStorage 缓存。 + */ +export async function updateExternalModelsAccessConfig( + proxyNodeId: string | null, +): Promise { + const response = await api.put( + '/api/admin/models/external/config', + { proxy_node_id: proxyNodeId }, + ) + clearModelsDevCache() + return response.data +} + // 模型列表缓存（避免重复转换） let modelsListCache: ModelsDevModelItem[] | null = null let modelsListCacheTimestamp: number | null = null diff --git a/frontend/src/api/proxy-nodes.ts b/frontend/src/api/proxy-nodes.ts index 4ebf45bc5..4900c5db5 100644 --- a/frontend/src/api/proxy-nodes.ts +++ b/frontend/src/api/proxy-nodes.ts @@ -269,8 +269,18 @@ export const proxyNodesApi = { return response.data }, - async deleteProxyNode(nodeId: string): Promise<{ message: string; node_id: string; cleared_system_proxy: boolean }> { - const response = await apiClient.delete<{ message: string; node_id: string; cleared_system_proxy: boolean }>(`/api/admin/proxy-nodes/${nodeId}`) + async deleteProxyNode(nodeId: string): Promise<{ + message: string + node_id: string + cleared_system_proxy: boolean + cleared_external_models_proxy: boolean + }> { + const response = await apiClient.delete<{ + message: string + node_id: string + cleared_system_proxy: boolean + cleared_external_models_proxy: boolean + }>(`/api/admin/proxy-nodes/${nodeId}`) return response.data }, diff --git a/frontend/src/features/models/components/ExternalModelsAccessControl.vue b/frontend/src/features/models/components/ExternalModelsAccessControl.vue new file mode 100644 index 000000000..461032d75 --- /dev/null +++ b/frontend/src/features/models/components/ExternalModelsAccessControl.vue @@ -0,0 +1,116 @@ + + + diff --git a/frontend/src/features/models/components/__tests__/ExternalModelsAccessControl.spec.ts b/frontend/src/features/models/components/__tests__/ExternalModelsAccessControl.spec.ts new file mode 100644 index 000000000..c2c338a51 --- /dev/null +++ b/frontend/src/features/models/components/__tests__/ExternalModelsAccessControl.spec.ts @@ -0,0 +1,222 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createApp, nextTick } from 'vue' + +const modelsDevMocks = vi.hoisted(() => ({ + getExternalModelsAccessConfig: vi.fn(), + updateExternalModelsAccessConfig: vi.fn(), +})) + +const toastMocks = vi.hoisted(() => ({ + success: vi.fn(), + error: vi.fn(), +})) + +vi.mock('@/api/models-dev', () => modelsDevMocks) + +vi.mock('@/composables/useToast', () => ({ + useToast: () => toastMocks, +})) + +vi.mock('@/components/ui', async () => { + const { defineComponent, h } = await import('vue') + const passthrough = (name: string) => defineComponent({ + name, + inheritAttrs: false, + setup(_, { attrs, slots }) { + return () => h('div', attrs, slots.default?.()) + }, + }) + + return { + Button: defineComponent({ + name: 'ButtonStub', + inheritAttrs: false, + props: { + disabled: { type: Boolean, default: false }, + }, + setup(props, { attrs, slots }) { + return () => h('button', { + ...attrs, + disabled: props.disabled, + type: 'button', + }, slots.default?.()) + }, + }), + Popover: defineComponent({ + name: 'PopoverStub', + inheritAttrs: false, + props: { + open: { type: Boolean, default: false }, + }, + emits: ['update:open'], + setup(props, { attrs, slots }) { + return () => h('div', { + ...attrs, + 'data-popover-open': String(props.open), + }, slots.default?.()) + }, + }), + PopoverTrigger: passthrough('PopoverTriggerStub'), + PopoverContent: passthrough('PopoverContentStub'), + } +}) + +vi.mock('lucide-vue-next', async () => { + const { defineComponent, h } = await import('vue') + return { + Globe: defineComponent({ + name: 'GlobeStub', + setup() { + return () => h('span', { 'data-testid': 'external-models-access-globe' }) + }, + }), + } +}) + +vi.mock('@/features/providers/components/ProxyNodeSelect.vue', async () => { + const { defineComponent, h } = await import('vue') + return { + default: defineComponent({ + name: 'ProxyNodeSelectStub', + inheritAttrs: false, + props: { + modelValue: { type: String, required: true }, + disabled: { type: Boolean, default: false }, + }, + emits: ['update:modelValue'], + setup(props, { attrs, emit }) { + return () => h('div', { + ...attrs, + 'data-model-value': props.modelValue, + 'data-disabled': String(props.disabled), + }, [ + h('button', { + type: 'button', + 'data-testid': 'select-proxy-node', + onClick: () => emit('update:modelValue', 'proxy-1'), + }), + ]) + }, + }), + } +}) + +import ExternalModelsAccessControl from '@/features/models/components/ExternalModelsAccessControl.vue' +import { createI18n, setI18nLocale } from '@/i18n' + +interface MountedControl { + app: ReturnType + root: HTMLDivElement +} + +const mountedControls: MountedControl[] = [] + +function mountControl(): MountedControl { + const root = document.createElement('div') + document.body.appendChild(root) + const app = createApp(ExternalModelsAccessControl) + app.use(createI18n()) + app.mount(root) + const mounted = { app, root } + mountedControls.push(mounted) + return mounted +} + +async function flushAsyncState() { + await Promise.resolve() + await Promise.resolve() + await nextTick() +} + +beforeEach(() => { + setI18nLocale('zh-CN') + modelsDevMocks.getExternalModelsAccessConfig.mockReset() + modelsDevMocks.getExternalModelsAccessConfig.mockResolvedValue({ proxy_node_id: null }) + modelsDevMocks.updateExternalModelsAccessConfig.mockReset() + modelsDevMocks.updateExternalModelsAccessConfig.mockImplementation( + async (proxyNodeId: string | null) => ({ + proxy_node_id: proxyNodeId, + cache_cleared: true, + }), + ) + toastMocks.success.mockReset() + toastMocks.error.mockReset() +}) + +afterEach(() => { + for (const mounted of mountedControls.splice(0)) { + mounted.app.unmount() + mounted.root.remove() + } +}) + +describe('ExternalModelsAccessControl', () => { + it('uses the provider-style globe button as the header control', async () => { + const { root } = mountControl() + await flushAsyncState() + + const trigger = root.querySelector('[data-testid="external-models-access-trigger"]') + expect(trigger?.tagName).toBe('BUTTON') + expect(trigger?.classList.contains('h-8')).toBe(true) + expect(trigger?.classList.contains('w-8')).toBe(true) + expect(root.querySelector('[data-testid="external-models-access-globe"]')).not.toBeNull() + expect(trigger?.getAttribute('aria-label')).toBe('外部模型目录代理节点') + }) + + it('loads the saved node into the shared proxy selector', async () => { + modelsDevMocks.getExternalModelsAccessConfig.mockResolvedValue({ + proxy_node_id: 'proxy-1', + }) + + const { root } = mountControl() + await flushAsyncState() + + const select = root.querySelector('[data-testid="external-models-access-select"]') + expect(select?.getAttribute('data-model-value')).toBe('proxy-1') + expect(select?.getAttribute('data-disabled')).toBe('false') + expect(root.querySelector('[data-testid="external-models-access-trigger"]') + ?.classList.contains('text-blue-500')).toBe(true) + expect(root.querySelector('[data-testid="external-models-access-clear"]')).not.toBeNull() + }) + + it('immediately saves a selected proxy node', async () => { + const { root } = mountControl() + await flushAsyncState() + + ;(root.querySelector('[data-testid="select-proxy-node"]') as HTMLButtonElement).click() + await flushAsyncState() + + expect(modelsDevMocks.updateExternalModelsAccessConfig).toHaveBeenCalledWith('proxy-1') + expect(toastMocks.success).toHaveBeenCalledWith('外部模型目录代理节点已保存') + expect(root.querySelector('[data-testid="external-models-access-select"]') + ?.getAttribute('data-model-value')).toBe('proxy-1') + }) + + it('clears the selected proxy back to direct access', async () => { + modelsDevMocks.getExternalModelsAccessConfig.mockResolvedValue({ + proxy_node_id: 'proxy-1', + }) + const { root } = mountControl() + await flushAsyncState() + + ;(root.querySelector('[data-testid="external-models-access-clear"]') as HTMLButtonElement).click() + await flushAsyncState() + + expect(modelsDevMocks.updateExternalModelsAccessConfig).toHaveBeenCalledWith(null) + expect(root.querySelector('[data-testid="external-models-access-select"]') + ?.getAttribute('data-model-value')).toBe('') + expect(root.querySelector('[data-testid="external-models-access-clear"]')).toBeNull() + }) + + it('keeps the globe trigger and selector disabled when config loading fails', async () => { + modelsDevMocks.getExternalModelsAccessConfig.mockRejectedValue(new Error('load failed')) + const { root } = mountControl() + await flushAsyncState() + + expect(root.querySelector('[data-testid="external-models-access-select"]') + ?.getAttribute('data-disabled')).toBe('true') + expect((root.querySelector('[data-testid="external-models-access-trigger"]') as HTMLButtonElement) + .disabled).toBe(true) + expect(toastMocks.error).toHaveBeenCalledOnce() + }) +}) diff --git a/frontend/src/features/models/components/index.ts b/frontend/src/features/models/components/index.ts index 0ee577d5c..d2c33db16 100644 --- a/frontend/src/features/models/components/index.ts +++ b/frontend/src/features/models/components/index.ts @@ -1,4 +1,5 @@ export { default as GlobalModelFormDialog } from './GlobalModelFormDialog.vue' +export { default as ExternalModelsAccessControl } from './ExternalModelsAccessControl.vue' export { default as ModelDetailDrawer } from './ModelDetailDrawer.vue' export { default as TieredPricingEditor } from './TieredPricingEditor.vue' export { default as ModelMappingsTab } from './ModelMappingsTab.vue' diff --git a/frontend/src/features/providers/components/ProxyNodeSelect.vue b/frontend/src/features/providers/components/ProxyNodeSelect.vue index 22e5c75d4..515f307b4 100644 --- a/frontend/src/features/providers/components/ProxyNodeSelect.vue +++ b/frontend/src/features/providers/components/ProxyNodeSelect.vue @@ -2,10 +2,13 @@