refactor(rules): 规则引擎容错优化，无效规则条目跳过而非中止整个规则集

- header/body rules 的 _are_locally_supported 简化为仅检查是否为数组 - apply 逻辑中遇到格式错误/不支持的规则条目改为 continue 跳过，而非 return false - 允许非字符串 header value，自动序列化为 JSON 字符串 - 宽松处理无效 regex flag，不再拒绝整条规则 fix(gateway): Claude CLI 路由仅检查 bearer 头，不排斥同时携带 x-api-key 的请求 feat(observability): 监控链路候选展示解密 auth_config 的账号标签和 OAuth 计划类型
2026-09-02 01:10:23 +08:00 · 2026-04-25 19:46:52 +08:00
parent 00744c0ce5
commit 912a92cd1a
7 changed files with 438 additions and 184 deletions
--- a/apps/aether-gateway/src/control/route/mod.rs
+++ b/apps/aether-gateway/src/control/route/mod.rs
@@ -185,10 +185,7 @@ pub(super) fn is_claude_cli_request(headers: &http::HeaderMap) -> bool {
    let auth_header = header_value_str(headers, http::header::AUTHORIZATION.as_str())
        .unwrap_or_default()
        .to_ascii_lowercase();
-    let has_bearer = auth_header.starts_with("bearer ");
-    let has_api_key =
-        header_value_str(headers, "x-api-key").is_some_and(|value| !value.trim().is_empty());
-    has_bearer && !has_api_key
+    auth_header.starts_with("bearer ")
 }

 pub(super) fn is_gemini_cli_request(headers: &http::HeaderMap) -> bool {
--- a/apps/aether-gateway/src/control/tests/ai.rs
+++ b/apps/aether-gateway/src/control/tests/ai.rs
@@ -52,6 +52,41 @@ fn classifies_claude_messages_cli_when_bearer_without_api_key() {
    assert!(decision.is_execution_runtime_candidate());
 }

+#[test]
+fn classifies_claude_messages_cli_when_bearer_is_present_even_with_api_key() {
+    let headers = headers(&[
+        ("authorization", "Bearer token-123"),
+        ("x-api-key", "sk-client"),
+    ]);
+    let uri: Uri = "/v1/messages".parse().expect("uri should parse");
+    let decision =
+        classify_control_route(&http::Method::POST, &uri, &headers).expect("route should classify");
+
+    assert_eq!(decision.route_family.as_deref(), Some("claude"));
+    assert_eq!(decision.route_kind.as_deref(), Some("cli"));
+    assert_eq!(
+        decision.auth_endpoint_signature.as_deref(),
+        Some("claude:cli")
+    );
+    assert!(decision.is_execution_runtime_candidate());
+}
+
+#[test]
+fn classifies_claude_messages_chat_when_api_key_without_bearer() {
+    let headers = headers(&[("x-api-key", "sk-client")]);
+    let uri: Uri = "/v1/messages".parse().expect("uri should parse");
+    let decision =
+        classify_control_route(&http::Method::POST, &uri, &headers).expect("route should classify");
+
+    assert_eq!(decision.route_family.as_deref(), Some("claude"));
+    assert_eq!(decision.route_kind.as_deref(), Some("chat"));
+    assert_eq!(
+        decision.auth_endpoint_signature.as_deref(),
+        Some("claude:chat")
+    );
+    assert!(decision.is_execution_runtime_candidate());
+}
+
 #[test]
 fn classifies_gemini_cli_generate_content_when_x_app_contains_cli() {
    let headers = headers(&[("x-app", "Gemini-CLI")]);
--- a/apps/aether-gateway/src/handlers/admin/observability/monitoring/tests/trace.rs
+++ b/apps/aether-gateway/src/handlers/admin/observability/monitoring/tests/trace.rs
@@ -2,7 +2,9 @@ use super::super::test_support::{
    request_context, sample_candidate, sample_endpoint, sample_key, sample_provider, sample_usage,
 };
 use super::local_monitoring_response;
+use crate::data::GatewayDataState;
 use crate::AppState;
+use aether_crypto::{encrypt_python_fernet_plaintext, DEVELOPMENT_ENCRYPTION_KEY};
 use aether_data_contracts::repository::candidates::RequestCandidateStatus;
 use axum::body::to_bytes;
 use serde_json::json;
@@ -76,6 +78,78 @@ async fn admin_monitoring_trace_request_returns_local_payload() {
    assert_eq!(payload["candidates"][0]["status_code"], json!(502));
 }

+#[tokio::test]
+async fn admin_monitoring_trace_request_returns_oauth_account_label_from_auth_config() {
+    let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![
+        sample_candidate(
+            "cand-used",
+            "request-1",
+            0,
+            RequestCandidateStatus::Failed,
+            Some(101),
+            Some(33),
+            Some(502),
+        ),
+    ]));
+    let auth_config = json!({
+        "provider_type": "codex",
+        "email": "codex_alice@example.com",
+        "plan_type": "plus",
+        "refresh_token": "rt-test"
+    })
+    .to_string();
+    let oauth_key = sample_key()
+        .with_transport_fields(
+            None,
+            encrypt_python_fernet_plaintext(DEVELOPMENT_ENCRYPTION_KEY, "__placeholder__")
+                .expect("placeholder should encrypt"),
+            Some(
+                encrypt_python_fernet_plaintext(DEVELOPMENT_ENCRYPTION_KEY, &auth_config)
+                    .expect("auth config should encrypt"),
+            ),
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+        )
+        .expect("key transport fields should build");
+    let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
+        vec![sample_provider()],
+        vec![sample_endpoint()],
+        vec![oauth_key],
+    ));
+    let data_state = GatewayDataState::with_decision_trace_readers_for_tests(
+        request_candidates,
+        provider_catalog,
+    )
+    .with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY);
+    let state = AppState::new()
+        .expect("state should build")
+        .with_data_state_for_tests(data_state);
+    let context = request_context(http::Method::GET, "/api/admin/monitoring/trace/request-1");
+
+    let response = local_monitoring_response(&state, &context)
+        .await
+        .expect("handler should not error")
+        .expect("route should be handled locally");
+
+    assert_eq!(response.status(), http::StatusCode::OK);
+    let body = to_bytes(response.into_body(), usize::MAX)
+        .await
+        .expect("body should read");
+    let payload: serde_json::Value = serde_json::from_slice(&body).expect("json body should parse");
+    assert_eq!(
+        payload["candidates"][0]["key_account_label"],
+        json!("codex_alice@example.com")
+    );
+    assert_eq!(
+        payload["candidates"][0]["key_oauth_plan_type"],
+        json!("plus")
+    );
+}
+
 #[tokio::test]
 async fn admin_monitoring_trace_final_status_prefers_failed_over_stale_pending() {
    let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![
--- a/apps/aether-gateway/src/handlers/admin/observability/monitoring/trace.rs
+++ b/apps/aether-gateway/src/handlers/admin/observability/monitoring/trace.rs
@@ -6,19 +6,26 @@ use aether_admin::observability::monitoring::{
    admin_monitoring_bad_request_response, admin_monitoring_trace_not_found_response,
    admin_monitoring_trace_provider_id_from_path, admin_monitoring_trace_request_id_from_path,
    build_admin_monitoring_trace_provider_stats_payload_response,
-    build_admin_monitoring_trace_request_payload_response, parse_admin_monitoring_attempted_only,
+    build_admin_monitoring_trace_request_payload_response_with_key_accounts,
+    parse_admin_monitoring_attempted_only, AdminMonitoringKeyAccountDisplay,
+};
+use aether_data_contracts::repository::{
+    candidates::{DecisionTrace, RequestCandidateStatus},
+    provider_catalog::StoredProviderCatalogKey,
 };
-use aether_data_contracts::repository::candidates::{DecisionTrace, RequestCandidateStatus};
 use axum::{
    body::Body,
    response::{IntoResponse, Response},
 };
+use serde_json::{Map, Value};
+use std::collections::BTreeMap;
 use tracing::debug;

 pub(super) async fn build_admin_monitoring_trace_request_response(
    state: &AdminAppState<'_>,
    request_context: &AdminRequestContext<'_>,
 ) -> Result<Response<Body>, GatewayError> {
+    let admin_state = state;
    let state = state.as_ref();
    let Some(request_id) =
        admin_monitoring_trace_request_id_from_path(&request_context.request_path)
@@ -56,11 +63,105 @@ pub(super) async fn build_admin_monitoring_trace_request_response(
        .read_request_usage_audit(&request_id)
        .await
        .map_err(|err| GatewayError::Internal(err.to_string()))?;
+    let key_accounts = build_admin_monitoring_key_account_display_map(admin_state, &trace).await?;

-    Ok(build_admin_monitoring_trace_request_payload_response(
-        &trace,
-        usage.as_ref(),
-    ))
+    Ok(
+        build_admin_monitoring_trace_request_payload_response_with_key_accounts(
+            &trace,
+            usage.as_ref(),
+            &key_accounts,
+        ),
+    )
+}
+
+async fn build_admin_monitoring_key_account_display_map(
+    state: &AdminAppState<'_>,
+    trace: &DecisionTrace,
+) -> Result<BTreeMap<String, AdminMonitoringKeyAccountDisplay>, GatewayError> {
+    let key_ids = trace
+        .candidates
+        .iter()
+        .filter_map(|item| item.candidate.key_id.as_deref())
+        .filter(|value| !value.trim().is_empty())
+        .map(ToOwned::to_owned)
+        .collect::<std::collections::BTreeSet<_>>()
+        .into_iter()
+        .collect::<Vec<_>>();
+    if key_ids.is_empty() {
+        return Ok(BTreeMap::new());
+    }
+
+    let keys = state.read_provider_catalog_keys_by_ids(&key_ids).await?;
+    Ok(keys
+        .into_iter()
+        .filter_map(|key| {
+            let display = resolve_admin_monitoring_key_account_display(state, &key)?;
+            Some((key.id, display))
+        })
+        .collect())
+}
+
+fn resolve_admin_monitoring_key_account_display(
+    state: &AdminAppState<'_>,
+    key: &StoredProviderCatalogKey,
+) -> Option<AdminMonitoringKeyAccountDisplay> {
+    let auth_config = parse_admin_monitoring_key_auth_config(state, key);
+    let label = auth_config
+        .as_ref()
+        .and_then(|config| {
+            first_non_empty_json_string([
+                config.get("email"),
+                config.get("account_name"),
+                config.get("accountName"),
+                config.get("client_email"),
+                config.get("account_id"),
+                config.get("accountId"),
+            ])
+        })
+        .or_else(|| {
+            key.upstream_metadata.as_ref().and_then(|metadata| {
+                first_non_empty_json_string([
+                    metadata.get("email"),
+                    metadata.get("account_name"),
+                    metadata.get("accountName"),
+                    metadata.get("account_id"),
+                    metadata.get("accountId"),
+                ])
+            })
+        });
+    let oauth_plan_type = auth_config.as_ref().and_then(|config| {
+        first_non_empty_json_string([config.get("plan_type"), config.get("planType")])
+    });
+
+    if label.is_none() && oauth_plan_type.is_none() {
+        return None;
+    }
+
+    Some(AdminMonitoringKeyAccountDisplay {
+        label,
+        oauth_plan_type,
+    })
+}
+
+fn parse_admin_monitoring_key_auth_config(
+    state: &AdminAppState<'_>,
+    key: &StoredProviderCatalogKey,
+) -> Option<Map<String, Value>> {
+    let ciphertext = key.encrypted_auth_config.as_deref()?;
+    let plaintext = state.decrypt_catalog_secret_with_fallbacks(ciphertext)?;
+    serde_json::from_str::<Value>(&plaintext)
+        .ok()?
+        .as_object()
+        .cloned()
+}
+
+fn first_non_empty_json_string<'a>(
+    values: impl IntoIterator<Item = Option<&'a Value>>,
+) -> Option<String> {
+    values.into_iter().find_map(|value| {
+        let text = value?.as_str()?.trim();
+        (!text.is_empty()).then(|| text.to_string())
+    })
 }

 pub(super) async fn build_admin_monitoring_trace_provider_stats_response(