Merge remote-tracking branch 'upstream/main' into feat/356-usage-record-columns

apps/aether-gateway/src/handlers/admin/mod.rs

@@ -6,6 +6,7 @@ pub(super) mod features;
 mod model;
 pub(super) mod observability;
 pub(super) mod provider;
+mod routing;
 mod system;
 mod users;
 
@@ -27,8 +28,7 @@ pub(crate) use self::provider::oauth::provisioning::{
 };
 pub(crate) use self::provider::oauth::quota::dispatch::refresh_provider_pool_quota_locally;
 pub(crate) use self::provider::oauth::quota::shared::{
-    persist_provider_quota_refresh_state, provider_account_self_check_endpoint_for_provider,
-    provider_quota_refresh_endpoint_for_provider, provider_type_supports_account_self_check,
+    persist_provider_quota_refresh_state, provider_quota_refresh_endpoint_for_provider,
     provider_type_supports_quota_refresh,
 };
 pub(crate) use self::provider::oauth::runtime::{

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/codex/mod.rs

@@ -8,7 +8,8 @@ use self::invalid::{
     codex_structured_invalid_reason,
 };
 use self::parse::{
-    parse_codex_backend_me_response, parse_codex_usage_headers, parse_codex_wham_usage_response,
+    build_codex_quota_exhausted_fallback_metadata, parse_codex_usage_headers,
+    parse_codex_wham_usage_response,
 };
 use self::plan::{build_codex_quota_request_spec, execute_codex_quota_plan};
 use super::shared::{
@@ -110,7 +111,7 @@ pub(crate) async fn refresh_codex_provider_quota_locally(
                     "key_id": key.id,
                     "key_name": key.name,
                     "status": "error",
-                    "message": format!("backend-api/me 请求执行失败: {detail}"),
+                    "message": format!("wham/usage 请求执行失败: {detail}"),
                     "status_code": 502,
                 }));
                 continue;
@@ -137,9 +138,7 @@ pub(crate) async fn refresh_codex_provider_quota_locally(
                 .as_ref()
                 .and_then(|body| body.json_body.as_ref())
             {
-                if let Some(parsed) = parse_codex_backend_me_response(body_json, now_unix_secs)
-                    .or_else(|| parse_codex_wham_usage_response(body_json, now_unix_secs))
-                {
+                if let Some(parsed) = parse_codex_wham_usage_response(body_json, now_unix_secs) {
                     metadata_update = Some(json!({
                         "codex": merge_codex_quota_metadata(header_metadata.as_ref(), &parsed)
                     }));
@@ -152,21 +151,21 @@ pub(crate) async fn refresh_codex_provider_quota_locally(
                     status = "success".to_string();
                 } else {
                     status = "no_metadata".to_string();
-                    message = Some("backend-api/me 响应中未包含账号信息".to_string());
+                    message = Some("响应中未包含限额信息".to_string());
                 }
             } else {
-                message = Some("无法解析 backend-api/me API 响应".to_string());
+                message = Some("无法解析 wham/usage API 响应".to_string());
             }
         } else {
             let err_msg = extract_execution_error_message(&result);
             message = Some(match err_msg.as_deref() {
                 Some(detail) if !detail.is_empty() => {
                     format!(
-                        "backend-api/me API 返回状态码 {}: {}",
+                        "wham/usage API 返回状态码 {}: {}",
                         result.status_code, detail
                     )
                 }
-                _ => format!("backend-api/me API 返回状态码 {}", result.status_code),
+                _ => format!("wham/usage API 返回状态码 {}", result.status_code),
             });
 
             match result.status_code {
@@ -223,14 +222,26 @@ pub(crate) async fn refresh_codex_provider_quota_locally(
                         oauth_invalid_reason = reason;
                         status = "workspace_deactivated".to_string();
                     } else {
-                        let (at, reason) = codex_build_invalid_state(
-                            &key,
-                            codex_structured_invalid_reason(402, err_msg.as_deref()),
-                            now_unix_secs,
-                        );
-                        oauth_invalid_at_unix_secs = at;
-                        oauth_invalid_reason = reason;
-                        status = "payment_required".to_string();
+                        let plan_type = transport
+                            .key
+                            .decrypted_auth_config
+                            .as_deref()
+                            .and_then(|raw| serde_json::from_str::<serde_json::Value>(raw).ok())
+                            .and_then(|value| {
+                                value
+                                    .get("plan_type")
+                                    .and_then(serde_json::Value::as_str)
+                                    .map(ToOwned::to_owned)
+                            });
+                        metadata_update = Some(json!({
+                            "codex": build_codex_quota_exhausted_fallback_metadata(
+                                plan_type.as_deref(),
+                                now_unix_secs,
+                            )
+                        }));
+                        (oauth_invalid_at_unix_secs, oauth_invalid_reason) =
+                            quota_refresh_success_invalid_state(&key);
+                        status = "quota_exhausted".to_string();
                     }
                 }
                 403 => {

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/codex/parse.rs

@@ -22,13 +22,6 @@ pub(super) fn parse_codex_wham_usage_response(
     admin_provider_quota_pure::parse_codex_wham_usage_response(value, updated_at_unix_secs)
 }
 
-pub(super) fn parse_codex_backend_me_response(
-    value: &serde_json::Value,
-    updated_at_unix_secs: u64,
-) -> Option<serde_json::Value> {
-    admin_provider_quota_pure::parse_codex_backend_me_response(value, updated_at_unix_secs)
-}
-
 pub(super) fn parse_codex_usage_headers(
     headers: &BTreeMap<String, String>,
     updated_at_unix_secs: u64,

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/shared.rs

@@ -84,22 +84,6 @@ pub(crate) fn provider_quota_refresh_endpoint_for_provider(
     )
 }
 
-pub(crate) fn provider_type_supports_account_self_check(provider_type: &str) -> bool {
-    ProviderPoolService::with_builtin_adapters().supports_account_self_check(provider_type)
-}
-
-pub(crate) fn provider_account_self_check_endpoint_for_provider(
-    provider_type: &str,
-    endpoints: &[StoredProviderCatalogEndpoint],
-    include_inactive: bool,
-) -> Option<StoredProviderCatalogEndpoint> {
-    ProviderPoolService::with_builtin_adapters().account_self_check_endpoint_for_provider(
-        provider_type,
-        endpoints,
-        include_inactive,
-    )
-}
-
 pub(crate) fn provider_quota_refresh_missing_endpoint_message(provider_type: &str) -> String {
     ProviderPoolService::with_builtin_adapters()
         .quota_refresh_missing_endpoint_message(provider_type)

apps/aether-gateway/src/handlers/admin/provider/shared/payloads.rs

@@ -226,7 +226,7 @@ pub(crate) struct AdminProviderUpdateRequest {
 
 pub(crate) type AdminProviderUpdatePatch = AdminTypedObjectPatch<AdminProviderUpdateRequest>;
 
-pub(crate) const CODEX_WHAM_USAGE_URL: &str = "https://chatgpt.com/backend-api/me";
+pub(crate) const CODEX_WHAM_USAGE_URL: &str = "https://chatgpt.com/backend-api/wham/usage";
 pub(crate) const KIRO_USAGE_LIMITS_PATH: &str = "/getUsageLimits";
 pub(crate) const KIRO_USAGE_SDK_VERSION: &str = "1.0.0";
 pub(crate) const ANTIGRAVITY_FETCH_AVAILABLE_MODELS_PATH: &str = "/v1internal:fetchAvailableModels";

apps/aether-gateway/src/handlers/admin/provider/write/keys/mod.rs

@@ -1,8 +1,7 @@
-pub(crate) use self::{
-    create::build_admin_create_provider_key_record,
-    payload::{build_admin_provider_keys_page_payload, build_admin_provider_keys_payload},
-    update::build_admin_update_provider_key_record,
-};
+pub(crate) use self::create::build_admin_create_provider_key_record;
+pub(crate) use self::payload::build_admin_provider_keys_page_payload;
+pub(crate) use self::payload::build_admin_provider_keys_payload;
+pub(crate) use self::update::build_admin_update_provider_key_record;
 
 mod create;
 mod payload;

apps/aether-gateway/src/handlers/admin/request/capabilities.rs

@@ -64,6 +64,14 @@ impl<'a> AdminAppState<'a> {
         self.app.has_global_model_data_writer()
     }
 
+    pub(crate) fn has_routing_group_data_reader(&self) -> bool {
+        self.app.has_routing_group_data_reader()
+    }
+
+    pub(crate) fn has_routing_group_data_writer(&self) -> bool {
+        self.app.has_routing_group_data_writer()
+    }
+
     pub(crate) fn has_usage_data_reader(&self) -> bool {
         self.app.has_usage_data_reader()
     }

apps/aether-gateway/src/handlers/admin/request/mod.rs

@@ -9,6 +9,7 @@ mod observability;
 mod provider;
 mod provider_oauth;
 mod route_request;
+mod routing_profiles;
 mod state;
 mod system;
 mod users;

apps/aether-gateway/src/handlers/admin/request/routing_profiles.rs

@@ -0,0 +1,85 @@
+use aether_data_contracts::repository::routing_profiles::{
+    CreateRoutingGroupBindingRecord, CreateRoutingGroupRecord, CreateRoutingGroupVersionRecord,
+    RoutingGroupBindingQuery, RoutingGroupLookupKey, StoredRoutingGroup, StoredRoutingGroupBinding,
+    StoredRoutingGroupVersion, UpdateRoutingGroupBindingRecord, UpdateRoutingGroupRecord,
+};
+
+use super::AdminAppState;
+use crate::GatewayError;
+
+impl<'a> AdminAppState<'a> {
+    pub(crate) async fn list_routing_groups(
+        &self,
+    ) -> Result<Vec<StoredRoutingGroup>, GatewayError> {
+        self.app.list_routing_groups().await
+    }
+
+    pub(crate) async fn find_routing_group(
+        &self,
+        lookup: RoutingGroupLookupKey<'_>,
+    ) -> Result<Option<StoredRoutingGroup>, GatewayError> {
+        self.app.find_routing_group(lookup).await
+    }
+
+    pub(crate) async fn list_routing_group_bindings(
+        &self,
+        query: &RoutingGroupBindingQuery,
+    ) -> Result<Vec<StoredRoutingGroupBinding>, GatewayError> {
+        self.app.list_routing_group_bindings(query).await
+    }
+
+    pub(crate) async fn list_routing_group_versions(
+        &self,
+        group_id: &str,
+    ) -> Result<Vec<StoredRoutingGroupVersion>, GatewayError> {
+        self.app.list_routing_group_versions(group_id).await
+    }
+
+    pub(crate) async fn create_routing_group(
+        &self,
+        record: CreateRoutingGroupRecord,
+    ) -> Result<Option<StoredRoutingGroup>, GatewayError> {
+        self.app.create_routing_group(record).await
+    }
+
+    pub(crate) async fn update_routing_group(
+        &self,
+        id: &str,
+        patch: UpdateRoutingGroupRecord,
+    ) -> Result<Option<StoredRoutingGroup>, GatewayError> {
+        self.app.update_routing_group(id, patch).await
+    }
+
+    pub(crate) async fn delete_routing_group(&self, id: &str) -> Result<bool, GatewayError> {
+        self.app.delete_routing_group(id).await
+    }
+
+    pub(crate) async fn create_routing_group_binding(
+        &self,
+        record: CreateRoutingGroupBindingRecord,
+    ) -> Result<Option<StoredRoutingGroupBinding>, GatewayError> {
+        self.app.create_routing_group_binding(record).await
+    }
+
+    pub(crate) async fn update_routing_group_binding(
+        &self,
+        id: &str,
+        patch: UpdateRoutingGroupBindingRecord,
+    ) -> Result<Option<StoredRoutingGroupBinding>, GatewayError> {
+        self.app.update_routing_group_binding(id, patch).await
+    }
+
+    pub(crate) async fn delete_routing_group_binding(
+        &self,
+        id: &str,
+    ) -> Result<bool, GatewayError> {
+        self.app.delete_routing_group_binding(id).await
+    }
+
+    pub(crate) async fn create_routing_group_version(
+        &self,
+        record: CreateRoutingGroupVersionRecord,
+    ) -> Result<Option<StoredRoutingGroupVersion>, GatewayError> {
+        self.app.create_routing_group_version(record).await
+    }
+}

apps/aether-gateway/src/handlers/admin/routes.rs

@@ -1,6 +1,6 @@
 use super::{
     announcements, auth, billing, endpoint, features, model, observability, provider, request,
-    system, users,
+    routing, system, users,
 };
 
 pub(crate) async fn maybe_build_local_admin_response(
@@ -20,6 +20,10 @@ pub(crate) async fn maybe_build_local_admin_response(
         return Ok(Some(response));
     }
 
+    if let Some(response) = routing::maybe_build_local_admin_routing_response(request).await? {
+        return Ok(Some(response));
+    }
+
     if let Some(response) = auth::maybe_build_local_admin_auth_response(request).await? {
         return Ok(Some(response));
     }

apps/aether-gateway/src/handlers/admin/routing/mod.rs

@@ -0,0 +1,760 @@
+use std::collections::BTreeMap;
+
+use aether_data_contracts::repository::routing_profiles::{
+    CreateRoutingGroupBindingRecord, CreateRoutingGroupRecord, CreateRoutingGroupVersionRecord,
+    RoutingGroupBindingQuery, RoutingGroupBindingSubject, RoutingGroupLookupKey,
+    StoredRoutingGroup, StoredRoutingGroupBinding, StoredRoutingGroupVersion,
+    UpdateRoutingGroupBindingRecord, UpdateRoutingGroupRecord,
+};
+use aether_routing_core::{
+    validate_routing_group_config, MutationPlan, RoutingGroupConfig, RoutingHeaderPatch,
+    RoutingPatchSummary, RoutingRulePhase,
+};
+use axum::{
+    body::{Body, Bytes},
+    http::{self, HeaderMap, HeaderName, HeaderValue},
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde::Deserialize;
+use serde_json::{json, Map, Value};
+use uuid::Uuid;
+
+use crate::clock::current_unix_secs;
+use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
+use crate::handlers::admin::shared::{attach_admin_audit_response, query_param_value};
+use crate::routing::{
+    apply_routing_mutation_plan, build_routing_trace_seed, resolve_gateway_routing_policy,
+    GatewayRoutingPolicyInput,
+};
+use crate::GatewayError;
+
+const ROUTING_GROUPS_ROOT: &str = "/api/admin/routing/groups";
+const ROUTING_BINDINGS_ROOT: &str = "/api/admin/routing/bindings";
+
+#[derive(Debug, Deserialize)]
+struct AdminRoutingGroupCreateRequest {
+    #[serde(default)]
+    id: Option<String>,
+    name: String,
+    #[serde(default)]
+    description: Option<String>,
+    #[serde(default = "default_true")]
+    enabled: bool,
+    #[serde(default)]
+    is_system_default: bool,
+    #[serde(default)]
+    config_json: Option<Value>,
+}
+
+#[derive(Debug, Deserialize)]
+struct AdminRoutingGroupBindingCreateRequest {
+    #[serde(default)]
+    id: Option<String>,
+    group_id: String,
+    subject_type: RoutingGroupBindingSubject,
+    subject_id: String,
+    #[serde(default)]
+    is_default: bool,
+    #[serde(default)]
+    allow_explicit_select: bool,
+}
+
+#[derive(Debug, Deserialize)]
+struct AdminRoutingDryRunRequest {
+    model: String,
+    #[serde(default)]
+    resolved_model: Option<String>,
+    #[serde(default = "default_api_format")]
+    api_format: String,
+    #[serde(default)]
+    user_id: Option<String>,
+    #[serde(default)]
+    api_key_id: Option<String>,
+    #[serde(default)]
+    headers: Option<Value>,
+    #[serde(default)]
+    body: Option<Value>,
+    #[serde(default)]
+    phase: Option<RoutingRulePhase>,
+}
+
+pub(crate) async fn maybe_build_local_admin_routing_response(
+    request: crate::handlers::admin::request::AdminRouteRequest<'_>,
+) -> crate::handlers::admin::request::AdminRouteResult {
+    let state = request.state();
+    let request_context = request.request_context();
+    let request_body = request.request_body();
+
+    if request_context.route_family() != Some("routing_profiles_manage") {
+        return Ok(None);
+    }
+    if !request_context.path().starts_with("/api/admin/routing/") {
+        return Ok(None);
+    }
+    if !state.has_routing_group_data_reader() {
+        return Ok(Some(data_unavailable_response()));
+    }
+
+    let response = if request_context.path().starts_with(ROUTING_GROUPS_ROOT) {
+        maybe_build_routing_groups_response(&state, &request_context, request_body).await?
+    } else if request_context.path().starts_with(ROUTING_BINDINGS_ROOT) {
+        maybe_build_routing_bindings_response(&state, &request_context, request_body).await?
+    } else {
+        None
+    };
+    Ok(response)
+}
+
+async fn maybe_build_routing_groups_response(
+    state: &AdminAppState<'_>,
+    request_context: &AdminRequestContext<'_>,
+    request_body: Option<&Bytes>,
+) -> Result<Option<Response<Body>>, GatewayError> {
+    let path = normalized_admin_path(request_context.path());
+    match (request_context.method(), path.as_str()) {
+        (&http::Method::GET, ROUTING_GROUPS_ROOT) => {
+            let groups = state.list_routing_groups().await?;
+            Ok(Some(
+                Json(json!({
+                    "items": groups.iter().map(routing_group_payload).collect::<Vec<_>>(),
+                    "total": groups.len(),
+                }))
+                .into_response(),
+            ))
+        }
+        (&http::Method::POST, ROUTING_GROUPS_ROOT) => {
+            if !state.has_routing_group_data_writer() {
+                return Ok(Some(data_unavailable_response()));
+            }
+            let payload = parse_json_body::<AdminRoutingGroupCreateRequest>(request_body)?;
+            let config_json = payload.config_json.unwrap_or_else(|| json!({}));
+            validate_config_json(&config_json)?;
+            let now = current_unix_secs() as i64;
+            let record = CreateRoutingGroupRecord {
+                id: payload.id.unwrap_or_else(|| Uuid::new_v4().to_string()),
+                name: payload.name,
+                description: payload.description,
+                enabled: payload.enabled,
+                is_system_default: payload.is_system_default,
+                config_json,
+                version: 1,
+                created_at: now,
+                updated_at: now,
+                published_at: None,
+            };
+            let Some(created) = state.create_routing_group(record).await? else {
+                return Ok(Some(data_unavailable_response()));
+            };
+            Ok(Some(attach_admin_audit_response(
+                Json(routing_group_payload(&created)).into_response(),
+                "admin_routing_group_created",
+                "create_routing_group",
+                "routing_group",
+                &created.id,
+            )))
+        }
+        _ => {
+            let Some((group_id, suffix)) = routing_group_path_parts(path.as_str()) else {
+                return Ok(None);
+            };
+            match (request_context.method(), suffix.as_deref()) {
+                (&http::Method::GET, None) => {
+                    let Some(group) = state
+                        .find_routing_group(RoutingGroupLookupKey::Id(&group_id))
+                        .await?
+                    else {
+                        return Ok(Some(not_found_response(format!(
+                            "routing group {group_id} not found"
+                        ))));
+                    };
+                    Ok(Some(Json(routing_group_payload(&group)).into_response()))
+                }
+                (&http::Method::PATCH, None) => {
+                    if !state.has_routing_group_data_writer() {
+                        return Ok(Some(data_unavailable_response()));
+                    }
+                    let patch = build_routing_group_update_patch(request_body)?;
+                    let Some(updated) = state.update_routing_group(&group_id, patch).await? else {
+                        return Ok(Some(not_found_response(format!(
+                            "routing group {group_id} not found"
+                        ))));
+                    };
+                    Ok(Some(attach_admin_audit_response(
+                        Json(routing_group_payload(&updated)).into_response(),
+                        "admin_routing_group_updated",
+                        "update_routing_group",
+                        "routing_group",
+                        &updated.id,
+                    )))
+                }
+                (&http::Method::DELETE, None) => {
+                    if !state.has_routing_group_data_writer() {
+                        return Ok(Some(data_unavailable_response()));
+                    }
+                    if !state.delete_routing_group(&group_id).await? {
+                        return Ok(Some(not_found_response(format!(
+                            "routing group {group_id} not found"
+                        ))));
+                    }
+                    Ok(Some(attach_admin_audit_response(
+                        http::StatusCode::NO_CONTENT.into_response(),
+                        "admin_routing_group_deleted",
+                        "delete_routing_group",
+                        "routing_group",
+                        &group_id,
+                    )))
+                }
+                (&http::Method::POST, Some("publish")) => {
+                    publish_routing_group(state, &group_id).await
+                }
+                (&http::Method::GET, Some("versions")) => {
+                    let versions = state.list_routing_group_versions(&group_id).await?;
+                    Ok(Some(Json(json!({
+                        "items": versions.iter().map(routing_group_version_payload).collect::<Vec<_>>(),
+                        "total": versions.len(),
+                    })).into_response()))
+                }
+                (&http::Method::POST, Some("dry-run")) => {
+                    dry_run_routing_group(state, &group_id, request_body).await
+                }
+                _ => Ok(None),
+            }
+        }
+    }
+}
+
+async fn maybe_build_routing_bindings_response(
+    state: &AdminAppState<'_>,
+    request_context: &AdminRequestContext<'_>,
+    request_body: Option<&Bytes>,
+) -> Result<Option<Response<Body>>, GatewayError> {
+    let path = normalized_admin_path(request_context.path());
+    match (request_context.method(), path.as_str()) {
+        (&http::Method::GET, ROUTING_BINDINGS_ROOT) => {
+            let query = routing_binding_query_from_request(request_context)?;
+            let bindings = state.list_routing_group_bindings(&query).await?;
+            Ok(Some(
+                Json(json!({
+                    "items": bindings.iter().map(routing_group_binding_payload).collect::<Vec<_>>(),
+                    "total": bindings.len(),
+                }))
+                .into_response(),
+            ))
+        }
+        (&http::Method::POST, ROUTING_BINDINGS_ROOT) => {
+            if !state.has_routing_group_data_writer() {
+                return Ok(Some(data_unavailable_response()));
+            }
+            let payload = parse_json_body::<AdminRoutingGroupBindingCreateRequest>(request_body)?;
+            let now = current_unix_secs() as i64;
+            let record = CreateRoutingGroupBindingRecord {
+                id: payload.id.unwrap_or_else(|| Uuid::new_v4().to_string()),
+                group_id: payload.group_id,
+                subject_type: payload.subject_type,
+                subject_id: payload.subject_id,
+                is_default: payload.is_default,
+                allow_explicit_select: payload.allow_explicit_select,
+                created_at: now,
+                updated_at: now,
+            };
+            let Some(created) = state.create_routing_group_binding(record).await? else {
+                return Ok(Some(data_unavailable_response()));
+            };
+            Ok(Some(attach_admin_audit_response(
+                Json(routing_group_binding_payload(&created)).into_response(),
+                "admin_routing_group_binding_created",
+                "create_routing_group_binding",
+                "routing_group_binding",
+                &created.id,
+            )))
+        }
+        _ => {
+            let Some(binding_id) = routing_binding_id_from_path(path.as_str()) else {
+                return Ok(None);
+            };
+            match *request_context.method() {
+                http::Method::PATCH => {
+                    if !state.has_routing_group_data_writer() {
+                        return Ok(Some(data_unavailable_response()));
+                    }
+                    let patch = build_routing_binding_update_patch(request_body)?;
+                    let Some(updated) = state
+                        .update_routing_group_binding(&binding_id, patch)
+                        .await?
+                    else {
+                        return Ok(Some(not_found_response(format!(
+                            "routing group binding {binding_id} not found"
+                        ))));
+                    };
+                    Ok(Some(attach_admin_audit_response(
+                        Json(routing_group_binding_payload(&updated)).into_response(),
+                        "admin_routing_group_binding_updated",
+                        "update_routing_group_binding",
+                        "routing_group_binding",
+                        &updated.id,
+                    )))
+                }
+                http::Method::DELETE => {
+                    if !state.has_routing_group_data_writer() {
+                        return Ok(Some(data_unavailable_response()));
+                    }
+                    if !state.delete_routing_group_binding(&binding_id).await? {
+                        return Ok(Some(not_found_response(format!(
+                            "routing group binding {binding_id} not found"
+                        ))));
+                    }
+                    Ok(Some(attach_admin_audit_response(
+                        http::StatusCode::NO_CONTENT.into_response(),
+                        "admin_routing_group_binding_deleted",
+                        "delete_routing_group_binding",
+                        "routing_group_binding",
+                        &binding_id,
+                    )))
+                }
+                _ => Ok(None),
+            }
+        }
+    }
+}
+
+async fn publish_routing_group(
+    state: &AdminAppState<'_>,
+    group_id: &str,
+) -> Result<Option<Response<Body>>, GatewayError> {
+    if !state.has_routing_group_data_writer() {
+        return Ok(Some(data_unavailable_response()));
+    }
+    let Some(group) = state
+        .find_routing_group(RoutingGroupLookupKey::Id(group_id))
+        .await?
+    else {
+        return Ok(Some(not_found_response(format!(
+            "routing group {group_id} not found"
+        ))));
+    };
+    validate_config_json(&group.config_json)?;
+    let latest_version = state
+        .list_routing_group_versions(group_id)
+        .await?
+        .into_iter()
+        .map(|version| version.version)
+        .max()
+        .unwrap_or(0);
+    let next_version = group.version.max(latest_version.saturating_add(1));
+    let now = current_unix_secs() as i64;
+    let Some(updated) = state
+        .update_routing_group(
+            group_id,
+            UpdateRoutingGroupRecord {
+                version: Some(next_version),
+                updated_at: now,
+                published_at: Some(Some(now)),
+                ..UpdateRoutingGroupRecord::default()
+            },
+        )
+        .await?
+    else {
+        return Ok(Some(not_found_response(format!(
+            "routing group {group_id} not found"
+        ))));
+    };
+    let _ = state
+        .create_routing_group_version(CreateRoutingGroupVersionRecord {
+            id: Uuid::new_v4().to_string(),
+            group_id: group_id.to_string(),
+            version: next_version,
+            config_json: updated.config_json.clone(),
+            created_at: now,
+            created_by: None,
+        })
+        .await?;
+
+    Ok(Some(attach_admin_audit_response(
+        Json(routing_group_payload(&updated)).into_response(),
+        "admin_routing_group_published",
+        "publish_routing_group",
+        "routing_group",
+        group_id,
+    )))
+}
+
+async fn dry_run_routing_group(
+    state: &AdminAppState<'_>,
+    group_id: &str,
+    request_body: Option<&Bytes>,
+) -> Result<Option<Response<Body>>, GatewayError> {
+    let Some(group) = state
+        .find_routing_group(RoutingGroupLookupKey::Id(group_id))
+        .await?
+    else {
+        return Ok(Some(not_found_response(format!(
+            "routing group {group_id} not found"
+        ))));
+    };
+    let payload = parse_json_body::<AdminRoutingDryRunRequest>(request_body)?;
+    let requested_model = payload.model.trim();
+    if requested_model.is_empty() {
+        return Ok(Some(bad_request_response("model must not be empty")));
+    }
+    let resolved_model = payload
+        .resolved_model
+        .as_deref()
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .unwrap_or(requested_model);
+    let api_format = payload.api_format.trim();
+    let headers_json = payload.headers.unwrap_or_else(|| json!({}));
+    let mut header_map = header_map_from_value(&headers_json)?;
+    let mut body = payload.body.unwrap_or_else(|| json!({}));
+    let policy = resolve_gateway_routing_policy(GatewayRoutingPolicyInput {
+        group_id: Some(group.id.as_str()),
+        group_version: Some(group.version),
+        group_config_json: &group.config_json,
+        selection_source: "admin_dry_run",
+        requested_model,
+        resolved_model,
+        api_format,
+        user_id: payload.user_id.as_deref(),
+        api_key_id: payload.api_key_id.as_deref(),
+        headers: &headers_json,
+        body: &body,
+        phase: payload.phase.unwrap_or(RoutingRulePhase::ClientRequest),
+    })?;
+    let patch_summary = patch_summary(&policy.mutation_plan);
+    apply_routing_mutation_plan(&mut body, &mut header_map, &policy.mutation_plan)?;
+    let mut trace = build_routing_trace_seed(&policy, api_format);
+    trace.client_request_patch_summary = patch_summary.clone();
+
+    Ok(Some(Json(json!({
+        "group": routing_group_payload(&group),
+        "policy": policy,
+        "trace_seed": trace,
+        "patch_summary": patch_summary,
+        "mutated_body": body,
+        "mutated_headers": header_map_payload(&header_map),
+        "candidate_preview": {
+            "status": "policy_only",
+            "ranking_overlay": policy.ranking_overlay,
+            "note": "full candidate preview is produced by runtime materialization once provider/key catalogs are enumerated"
+        }
+    })).into_response()))
+}
+
+fn build_routing_group_update_patch(
+    request_body: Option<&Bytes>,
+) -> Result<UpdateRoutingGroupRecord, GatewayError> {
+    let raw = parse_json_value_body(request_body)?;
+    let Some(object) = raw.as_object() else {
+        return Err(bad_request_error("request body must be a JSON object"));
+    };
+    let mut patch = UpdateRoutingGroupRecord {
+        updated_at: current_unix_secs() as i64,
+        ..UpdateRoutingGroupRecord::default()
+    };
+    if let Some(value) = object.get("name") {
+        patch.name = Some(required_string(value, "name")?);
+    }
+    if let Some(value) = object.get("description") {
+        patch.description = Some(optional_string(value, "description")?);
+    }
+    if let Some(value) = object.get("enabled") {
+        patch.enabled = Some(required_bool(value, "enabled")?);
+    }
+    if let Some(value) = object.get("is_system_default") {
+        patch.is_system_default = Some(required_bool(value, "is_system_default")?);
+    }
+    if let Some(value) = object.get("config_json") {
+        validate_config_json(value)?;
+        patch.config_json = Some(value.clone());
+        patch.version = object
+            .get("version")
+            .and_then(Value::as_i64)
+            .or(Some(current_unix_secs() as i64));
+    } else if let Some(value) = object.get("version") {
+        patch.version = Some(required_i64(value, "version")?.max(1));
+    }
+    if let Some(value) = object.get("published_at") {
+        patch.published_at = Some(optional_i64(value, "published_at")?);
+    }
+    Ok(patch)
+}
+
+fn build_routing_binding_update_patch(
+    request_body: Option<&Bytes>,
+) -> Result<UpdateRoutingGroupBindingRecord, GatewayError> {
+    let raw = parse_json_value_body(request_body)?;
+    let Some(object) = raw.as_object() else {
+        return Err(bad_request_error("request body must be a JSON object"));
+    };
+    let mut patch = UpdateRoutingGroupBindingRecord {
+        updated_at: current_unix_secs() as i64,
+        ..UpdateRoutingGroupBindingRecord::default()
+    };
+    if let Some(value) = object.get("group_id") {
+        patch.group_id = Some(required_string(value, "group_id")?);
+    }
+    if let Some(value) = object.get("subject_type") {
+        patch.subject_type = Some(routing_subject_from_value(value)?);
+    }
+    if let Some(value) = object.get("subject_id") {
+        patch.subject_id = Some(required_string(value, "subject_id")?);
+    }
+    if let Some(value) = object.get("is_default") {
+        patch.is_default = Some(required_bool(value, "is_default")?);
+    }
+    if let Some(value) = object.get("allow_explicit_select") {
+        patch.allow_explicit_select = Some(required_bool(value, "allow_explicit_select")?);
+    }
+    Ok(patch)
+}
+
+fn routing_binding_query_from_request(
+    request_context: &AdminRequestContext<'_>,
+) -> Result<RoutingGroupBindingQuery, GatewayError> {
+    let subject_type = query_param_value(request_context.query_string(), "subject_type")
+        .map(|value| routing_subject_from_str(&value))
+        .transpose()?;
+    Ok(RoutingGroupBindingQuery {
+        group_id: query_param_value(request_context.query_string(), "group_id"),
+        subject_type,
+        subject_id: query_param_value(request_context.query_string(), "subject_id"),
+    })
+}
+
+fn validate_config_json(value: &Value) -> Result<(), GatewayError> {
+    if !value.is_object() {
+        return Err(bad_request_error("config_json must be a JSON object"));
+    }
+    let config = serde_json::from_value::<RoutingGroupConfig>(value.clone())
+        .map_err(|err| bad_request_error(format!("config_json is invalid: {err}")))?;
+    validate_routing_group_config(&config)
+        .map_err(|err| bad_request_error(format!("config_json is invalid: {err}")))
+}
+
+fn parse_json_body<T>(request_body: Option<&Bytes>) -> Result<T, GatewayError>
+where
+    T: for<'de> Deserialize<'de>,
+{
+    let raw = request_body.ok_or_else(|| bad_request_error("request body is required"))?;
+    serde_json::from_slice(raw)
+        .map_err(|err| bad_request_error(format!("request body must be valid JSON: {err}")))
+}
+
+fn parse_json_value_body(request_body: Option<&Bytes>) -> Result<Value, GatewayError> {
+    parse_json_body::<Value>(request_body)
+}
+
+fn header_map_from_value(value: &Value) -> Result<HeaderMap, GatewayError> {
+    let Some(object) = value.as_object() else {
+        return Err(bad_request_error("headers must be a JSON object"));
+    };
+    let mut headers = HeaderMap::new();
+    for (name, value) in object {
+        let Some(value) = value.as_str() else {
+            return Err(bad_request_error(format!(
+                "header {name} must have a string value"
+            )));
+        };
+        let header_name = HeaderName::from_bytes(name.as_bytes())
+            .map_err(|_| bad_request_error(format!("header {name} has invalid name")))?;
+        let header_value = HeaderValue::from_str(value)
+            .map_err(|_| bad_request_error(format!("header {name} has invalid value")))?;
+        headers.insert(header_name, header_value);
+    }
+    Ok(headers)
+}
+
+fn header_map_payload(headers: &HeaderMap) -> BTreeMap<String, String> {
+    headers
+        .iter()
+        .filter_map(|(name, value)| {
+            value
+                .to_str()
+                .ok()
+                .map(|value| (name.as_str().to_string(), value.to_string()))
+        })
+        .collect()
+}
+
+fn patch_summary(plan: &MutationPlan) -> RoutingPatchSummary {
+    RoutingPatchSummary {
+        body_paths: plan
+            .body_patch
+            .iter()
+            .map(|operation| operation.path().to_string())
+            .collect(),
+        header_names: plan
+            .header_patch
+            .iter()
+            .map(|operation| match operation {
+                RoutingHeaderPatch::Set { name, .. } | RoutingHeaderPatch::Remove { name } => {
+                    name.clone()
+                }
+            })
+            .collect(),
+        failed_action: None,
+    }
+}
+
+fn routing_group_payload(group: &StoredRoutingGroup) -> Value {
+    json!({
+        "id": group.id,
+        "name": group.name,
+        "description": group.description,
+        "enabled": group.enabled,
+        "is_system_default": group.is_system_default,
+        "config_json": group.config_json,
+        "version": group.version,
+        "created_at": group.created_at,
+        "updated_at": group.updated_at,
+        "published_at": group.published_at,
+    })
+}
+
+fn routing_group_binding_payload(binding: &StoredRoutingGroupBinding) -> Value {
+    json!({
+        "id": binding.id,
+        "group_id": binding.group_id,
+        "subject_type": binding.subject_type,
+        "subject_id": binding.subject_id,
+        "is_default": binding.is_default,
+        "allow_explicit_select": binding.allow_explicit_select,
+        "created_at": binding.created_at,
+        "updated_at": binding.updated_at,
+    })
+}
+
+fn routing_group_version_payload(version: &StoredRoutingGroupVersion) -> Value {
+    json!({
+        "id": version.id,
+        "group_id": version.group_id,
+        "version": version.version,
+        "config_json": version.config_json,
+        "created_at": version.created_at,
+        "created_by": version.created_by,
+    })
+}
+
+fn normalized_admin_path(path: &str) -> String {
+    let trimmed = path.trim_end_matches('/');
+    if trimmed.is_empty() {
+        "/".to_string()
+    } else {
+        trimmed.to_string()
+    }
+}
+
+fn routing_group_path_parts(path: &str) -> Option<(String, Option<String>)> {
+    let suffix = path.strip_prefix(&(ROUTING_GROUPS_ROOT.to_string() + "/"))?;
+    let mut parts = suffix.split('/');
+    let group_id = parts.next()?.trim();
+    if group_id.is_empty() {
+        return None;
+    }
+    let suffix = parts.next().map(str::to_string);
+    if parts.next().is_some() {
+        return None;
+    }
+    Some((group_id.to_string(), suffix))
+}
+
+fn routing_binding_id_from_path(path: &str) -> Option<String> {
+    let suffix = path.strip_prefix(&(ROUTING_BINDINGS_ROOT.to_string() + "/"))?;
+    if suffix.trim().is_empty() || suffix.contains('/') {
+        return None;
+    }
+    Some(suffix.to_string())
+}
+
+fn routing_subject_from_value(value: &Value) -> Result<RoutingGroupBindingSubject, GatewayError> {
+    let Some(value) = value.as_str() else {
+        return Err(bad_request_error("subject_type must be a string"));
+    };
+    routing_subject_from_str(value)
+}
+
+fn routing_subject_from_str(value: &str) -> Result<RoutingGroupBindingSubject, GatewayError> {
+    match value.trim() {
+        "user" => Ok(RoutingGroupBindingSubject::User),
+        "api_key" => Ok(RoutingGroupBindingSubject::ApiKey),
+        "user_group" => Ok(RoutingGroupBindingSubject::UserGroup),
+        other => Err(bad_request_error(format!(
+            "unsupported subject_type: {other}"
+        ))),
+    }
+}
+
+fn required_string(value: &Value, field: &str) -> Result<String, GatewayError> {
+    value
+        .as_str()
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(str::to_string)
+        .ok_or_else(|| bad_request_error(format!("{field} must be a non-empty string")))
+}
+
+fn optional_string(value: &Value, field: &str) -> Result<Option<String>, GatewayError> {
+    if value.is_null() {
+        return Ok(None);
+    }
+    required_string(value, field).map(Some)
+}
+
+fn required_bool(value: &Value, field: &str) -> Result<bool, GatewayError> {
+    value
+        .as_bool()
+        .ok_or_else(|| bad_request_error(format!("{field} must be a boolean")))
+}
+
+fn required_i64(value: &Value, field: &str) -> Result<i64, GatewayError> {
+    value
+        .as_i64()
+        .ok_or_else(|| bad_request_error(format!("{field} must be an integer")))
+}
+
+fn optional_i64(value: &Value, field: &str) -> Result<Option<i64>, GatewayError> {
+    if value.is_null() {
+        return Ok(None);
+    }
+    required_i64(value, field).map(Some)
+}
+
+fn default_true() -> bool {
+    true
+}
+
+fn default_api_format() -> String {
+    "openai:chat".to_string()
+}
+
+fn bad_request_error(detail: impl Into<String>) -> GatewayError {
+    GatewayError::Client {
+        status: http::StatusCode::BAD_REQUEST,
+        message: detail.into(),
+    }
+}
+
+fn bad_request_response(detail: impl Into<String>) -> Response<Body> {
+    (
+        http::StatusCode::BAD_REQUEST,
+        Json(json!({ "detail": detail.into() })),
+    )
+        .into_response()
+}
+
+fn not_found_response(detail: impl Into<String>) -> Response<Body> {
+    (
+        http::StatusCode::NOT_FOUND,
+        Json(json!({ "detail": detail.into() })),
+    )
+        .into_response()
+}
+
+fn data_unavailable_response() -> Response<Body> {
+    (
+        http::StatusCode::SERVICE_UNAVAILABLE,
+        Json(json!({ "detail": "routing profile data backend is unavailable" })),
+    )
+        .into_response()
+}

apps/aether-gateway/src/handlers/admin/system/core/system_routes.rs

@@ -17,7 +17,9 @@ use crate::handlers::admin::system::shared::settings::{
     build_admin_system_stats_payload, current_aether_version, fetch_latest_admin_system_release,
 };
 use crate::handlers::admin::system::shared::smtp::build_admin_smtp_test_payload;
+use crate::maintenance::{ManualUsageCleanupMode, ManualUsageCleanupOptions};
 use crate::GatewayError;
+use aether_data_contracts::repository::usage::UsageCleanupTargets;
 use axum::{
     body::{Body, Bytes},
     http,
@@ -26,6 +28,7 @@ use axum::{
 };
 use serde_json::json;
 use std::time::Instant;
+use url::form_urlencoded;
 
 pub(super) async fn maybe_build_local_admin_core_system_response(
     state: &AdminAppState<'_>,
@@ -263,7 +266,7 @@ pub(super) async fn maybe_build_local_admin_core_system_response(
         && request_path == "/api/admin/system/cleanup/usage/manual"
     {
         return Ok(Some(
-            build_manual_usage_cleanup_response(state, request_body).await?,
+            build_manual_usage_cleanup_response(state, request_context, request_body).await?,
         ));
     }
 
@@ -625,50 +628,36 @@ async fn build_admin_system_cleanup_payload(
 
 async fn build_manual_usage_cleanup_response(
     state: &AdminAppState<'_>,
+    request_context: &AdminRequestContext<'_>,
     request_body: Option<&Bytes>,
 ) -> Result<Response<Body>, GatewayError> {
-    let older_than_days = match parse_manual_usage_cleanup_request(request_body) {
+    let options = match parse_manual_usage_cleanup_request(request_body) {
         Ok(value) => value,
         Err(response) => return Ok(response),
     };
+    let actor_user_id = request_context
+        .decision()
+        .and_then(|decision| decision.admin_principal.as_ref())
+        .map(|principal| principal.user_id.clone());
 
-    match crate::maintenance::run_manual_usage_cleanup_once(
-        &state.app().data,
-        older_than_days,
-        None,
+    match crate::maintenance::start_manual_usage_cleanup_task(
+        std::sync::Arc::clone(&state.app().data),
+        options,
+        actor_user_id,
     )
     .await
     {
-        Ok(summary) => {
-            let total = summary
-                .body_externalized
-                .saturating_add(summary.legacy_body_refs_migrated)
-                .saturating_add(summary.body_cleaned)
-                .saturating_add(summary.header_cleaned)
-                .saturating_add(summary.keys_cleaned)
-                .saturating_add(summary.records_deleted);
-            let message = match older_than_days {
-                Some(days) => {
-                    format!("请求记录手动清理完成，清理 {days} 天前的记录，影响 {total} 项")
-                }
-                None => format!("请求记录手动清理完成（按当前策略），影响 {total} 项"),
-            };
+        Ok(task) => {
             let payload = json!({
-                "message": message,
-                "requested_older_than_days": older_than_days,
-                "summary": {
-                    "body_externalized": summary.body_externalized,
-                    "legacy_body_refs_migrated": summary.legacy_body_refs_migrated,
-                    "body_cleaned": summary.body_cleaned,
-                    "header_cleaned": summary.header_cleaned,
-                    "keys_cleaned": summary.keys_cleaned,
-                    "records_deleted": summary.records_deleted,
-                },
-                "total_affected": total,
+                "message": task.message,
+                "mode": options.mode.as_str(),
+                "requested_older_than_days": options.requested_older_than_days,
+                "targets": options.targets,
+                "task": task,
             });
             Ok(attach_admin_audit_response(
                 Json(payload).into_response(),
-                "admin_system_usage_cleanup_completed",
+                "admin_system_usage_cleanup_started",
                 "manual_usage_cleanup",
                 "usage_cleanup",
                 "global",
@@ -696,12 +685,20 @@ async fn build_manual_usage_cleanup_preview_response(
         Ok(value) => value,
         Err(response) => return Ok(response),
     };
-    let preview =
-        crate::maintenance::preview_manual_usage_cleanup(&state.app().data, older_than_days)
-            .await
-            .map_err(|err| GatewayError::Internal(err.to_string()))?;
+    let options = match parse_manual_usage_cleanup_query_options(
+        request_context.query_string(),
+        older_than_days,
+    ) {
+        Ok(value) => value,
+        Err(response) => return Ok(response),
+    };
+    let preview = crate::maintenance::preview_manual_usage_cleanup(&state.app().data, options)
+        .await
+        .map_err(|err| GatewayError::Internal(err.to_string()))?;
     Ok(Json(json!({
+        "mode": preview.mode.as_str(),
         "requested_older_than_days": preview.requested_older_than_days,
+        "targets": preview.targets,
         "effective_cutoffs": {
             "detail": preview.detail_cutoff,
             "compressed": preview.compressed_cutoff,
@@ -720,12 +717,12 @@ async fn build_manual_usage_cleanup_preview_response(
 
 fn parse_manual_usage_cleanup_request(
     request_body: Option<&Bytes>,
-) -> Result<Option<u32>, Response<Body>> {
+) -> Result<ManualUsageCleanupOptions, Response<Body>> {
     let Some(body) = request_body else {
-        return Ok(None);
+        return Ok(ManualUsageCleanupOptions::policy());
     };
     if body.is_empty() {
-        return Ok(None);
+        return Ok(ManualUsageCleanupOptions::policy());
     }
     let parsed: serde_json::Value = match serde_json::from_slice(body) {
         Ok(value) => value,
@@ -738,45 +735,208 @@ fn parse_manual_usage_cleanup_request(
         }
     };
     let Some(object) = parsed.as_object() else {
-        return Ok(None);
+        return Err(bad_manual_cleanup_request("请求体必须为 JSON 对象"));
     };
-    match object.get("older_than_days") {
-        None | Some(serde_json::Value::Null) => Ok(None),
-        Some(value) => value
-            .as_u64()
-            .and_then(|value| u32::try_from(value).ok())
-            .filter(|days| *days >= 1)
-            .map(Some)
-            .ok_or_else(|| {
-                (
-                    http::StatusCode::BAD_REQUEST,
-                    Json(json!({
-                        "detail": "older_than_days 必须为正整数",
-                    })),
-                )
-                    .into_response()
-            }),
+    parse_manual_usage_cleanup_options(
+        object.get("mode").and_then(serde_json::Value::as_str),
+        object.get("older_than_days"),
+        object.get("targets"),
+    )
+}
+
+fn parse_manual_usage_cleanup_query_options(
+    query_string: Option<&str>,
+    older_than_days: Option<u32>,
+) -> Result<ManualUsageCleanupOptions, Response<Body>> {
+    let mode = query_param(query_string, "mode");
+    let targets = query_param(query_string, "targets").map(serde_json::Value::String);
+    let older_value =
+        older_than_days.map(|days| serde_json::Value::Number(serde_json::Number::from(days)));
+    parse_manual_usage_cleanup_options(mode.as_deref(), older_value.as_ref(), targets.as_ref())
+}
+
+fn parse_manual_usage_cleanup_options(
+    raw_mode: Option<&str>,
+    older_than_days: Option<&serde_json::Value>,
+    targets_value: Option<&serde_json::Value>,
+) -> Result<ManualUsageCleanupOptions, Response<Body>> {
+    let (requested_older_than_days, requested_before_now) =
+        parse_manual_cleanup_older_than_days(older_than_days)?;
+    let mode =
+        parse_manual_cleanup_mode(raw_mode, requested_older_than_days, requested_before_now)?;
+
+    if mode == ManualUsageCleanupMode::OlderThanDays && requested_older_than_days.is_none() {
+        return Err(bad_manual_cleanup_request(
+            "older_than_days 模式必须提供正整数天数",
+        ));
+    }
+    if mode == ManualUsageCleanupMode::BeforeNow && requested_older_than_days.is_some() {
+        return Err(bad_manual_cleanup_request(
+            "before_now 模式不能同时提供 older_than_days",
+        ));
+    }
+    if raw_mode.is_some() && requested_before_now && mode != ManualUsageCleanupMode::BeforeNow {
+        return Err(bad_manual_cleanup_request(
+            "older_than_days 为 0 时必须使用 before_now 模式",
+        ));
+    }
+    if raw_mode.is_some()
+        && mode == ManualUsageCleanupMode::Policy
+        && requested_older_than_days.is_some()
+    {
+        return Err(bad_manual_cleanup_request(
+            "policy 模式不能同时提供 older_than_days",
+        ));
+    }
+
+    let targets = parse_manual_cleanup_targets(targets_value, mode)?;
+    if !targets.any_selected() {
+        return Err(bad_manual_cleanup_request("至少选择一个清理范围"));
+    }
+    if mode == ManualUsageCleanupMode::BeforeNow
+        && (targets.headers || targets.records || targets.expired_keys)
+    {
+        return Err(bad_manual_cleanup_request(
+            "清理当前时刻之前只允许选择详细请求体和压缩请求体",
+        ));
+    }
+
+    Ok(ManualUsageCleanupOptions {
+        mode,
+        requested_older_than_days,
+        targets,
+    })
+}
+
+fn parse_manual_cleanup_mode(
+    raw_mode: Option<&str>,
+    requested_older_than_days: Option<u32>,
+    requested_before_now: bool,
+) -> Result<ManualUsageCleanupMode, Response<Body>> {
+    let Some(raw) = raw_mode.map(str::trim).filter(|value| !value.is_empty()) else {
+        return Ok(if requested_before_now {
+            ManualUsageCleanupMode::BeforeNow
+        } else if requested_older_than_days.is_some() {
+            ManualUsageCleanupMode::OlderThanDays
+        } else {
+            ManualUsageCleanupMode::Policy
+        });
+    };
+    match raw {
+        "policy" => Ok(ManualUsageCleanupMode::Policy),
+        "older_than_days" => Ok(ManualUsageCleanupMode::OlderThanDays),
+        "before_now" => Ok(ManualUsageCleanupMode::BeforeNow),
+        _ => Err(bad_manual_cleanup_request(
+            "mode 必须为 policy、older_than_days 或 before_now",
+        )),
     }
 }
 
-fn parse_older_than_days_query(query_string: Option<&str>) -> Result<Option<u32>, Response<Body>> {
-    let Some(query) = query_string.filter(|value| !value.is_empty()) else {
-        return Ok(None);
-    };
-    let value = query
-        .split('&')
-        .filter_map(|pair| pair.split_once('='))
-        .find_map(|(key, value)| {
-            if key == "older_than_days" && !value.is_empty() {
-                Some(value)
-            } else {
-                None
+fn parse_manual_cleanup_older_than_days(
+    value: Option<&serde_json::Value>,
+) -> Result<(Option<u32>, bool), Response<Body>> {
+    match value {
+        None | Some(serde_json::Value::Null) => Ok((None, false)),
+        Some(value) => {
+            let Some(raw) = value.as_u64() else {
+                return Err(bad_manual_cleanup_request("older_than_days 必须为非负整数"));
+            };
+            if raw == 0 {
+                return Ok((None, true));
+            }
+            let days = u32::try_from(raw)
+                .ok()
+                .filter(|days| *days >= 1)
+                .ok_or_else(|| bad_manual_cleanup_request("older_than_days 必须为正整数"))?;
+            Ok((Some(days), false))
+        }
+    }
+}
+
+fn parse_manual_cleanup_targets(
+    value: Option<&serde_json::Value>,
+    mode: ManualUsageCleanupMode,
+) -> Result<UsageCleanupTargets, Response<Body>> {
+    let Some(value) = value else {
+        return Ok(match mode {
+            ManualUsageCleanupMode::BeforeNow => UsageCleanupTargets::body_targets(),
+            ManualUsageCleanupMode::Policy | ManualUsageCleanupMode::OlderThanDays => {
+                UsageCleanupTargets::all_policy_targets()
             }
         });
-    let Some(raw) = value else {
+    };
+    if value.is_null() {
+        return Ok(match mode {
+            ManualUsageCleanupMode::BeforeNow => UsageCleanupTargets::body_targets(),
+            ManualUsageCleanupMode::Policy | ManualUsageCleanupMode::OlderThanDays => {
+                UsageCleanupTargets::all_policy_targets()
+            }
+        });
+    }
+
+    let raw_targets = match value {
+        serde_json::Value::Array(items) => items
+            .iter()
+            .map(|item| {
+                item.as_str()
+                    .map(str::to_string)
+                    .ok_or_else(|| bad_manual_cleanup_request("targets 必须为字符串数组"))
+            })
+            .collect::<Result<Vec<_>, _>>()?,
+        serde_json::Value::String(raw) => raw
+            .split(',')
+            .map(str::trim)
+            .filter(|item| !item.is_empty())
+            .map(str::to_string)
+            .collect(),
+        _ => return Err(bad_manual_cleanup_request("targets 必须为字符串数组")),
+    };
+
+    let mut targets = UsageCleanupTargets {
+        detail_body: false,
+        compressed_body: false,
+        headers: false,
+        records: false,
+        expired_keys: false,
+    };
+    for raw in raw_targets {
+        match raw.as_str() {
+            "detail_body" | "detail" | "raw_body" => targets.detail_body = true,
+            "compressed_body" | "compressed" => targets.compressed_body = true,
+            "headers" | "header" => targets.headers = true,
+            "records" | "log" | "logs" => targets.records = true,
+            "expired_keys" => targets.expired_keys = true,
+            "all" => targets = UsageCleanupTargets::all_policy_targets(),
+            _ => {
+                return Err(bad_manual_cleanup_request(
+                    "targets 只能包含 detail_body、compressed_body、headers、records",
+                ))
+            }
+        }
+    }
+    Ok(targets)
+}
+
+fn bad_manual_cleanup_request(detail: impl Into<String>) -> Response<Body> {
+    (
+        http::StatusCode::BAD_REQUEST,
+        Json(json!({ "detail": detail.into() })),
+    )
+        .into_response()
+}
+
+fn query_param(query_string: Option<&str>, name: &str) -> Option<String> {
+    let query = query_string.filter(|value| !value.is_empty())?;
+    form_urlencoded::parse(query.as_bytes())
+        .find_map(|(key, value)| (key == name && !value.is_empty()).then(|| value.into_owned()))
+}
+
+fn parse_older_than_days_query(query_string: Option<&str>) -> Result<Option<u32>, Response<Body>> {
+    let Some(value) = query_param(query_string, "older_than_days") else {
         return Ok(None);
     };
-    raw.parse::<u32>()
+    value
+        .parse::<u32>()
         .ok()
         .filter(|days| *days >= 1)
         .map(Some)
@@ -790,3 +950,55 @@ fn parse_older_than_days_query(query_string: Option<&str>) -> Result<Option<u32>
                 .into_response()
         })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn manual_cleanup_request_defaults_to_policy_targets() {
+        let options = parse_manual_usage_cleanup_request(None).expect("default request is valid");
+
+        assert_eq!(options.mode, ManualUsageCleanupMode::Policy);
+        assert_eq!(options.requested_older_than_days, None);
+        assert_eq!(options.targets, UsageCleanupTargets::all_policy_targets());
+    }
+
+    #[test]
+    fn manual_cleanup_request_treats_zero_days_as_before_now_body_only() {
+        let body = Bytes::from_static(br#"{"older_than_days":0}"#);
+
+        let options =
+            parse_manual_usage_cleanup_request(Some(&body)).expect("before-now request is valid");
+
+        assert_eq!(options.mode, ManualUsageCleanupMode::BeforeNow);
+        assert_eq!(options.requested_older_than_days, None);
+        assert_eq!(options.targets, UsageCleanupTargets::body_targets());
+    }
+
+    #[test]
+    fn manual_cleanup_request_rejects_before_now_headers() {
+        let body = Bytes::from_static(br#"{"mode":"before_now","targets":["headers"]}"#);
+
+        assert!(parse_manual_usage_cleanup_request(Some(&body)).is_err());
+    }
+
+    #[test]
+    fn manual_cleanup_preview_query_decodes_comma_separated_targets() {
+        let options = parse_manual_usage_cleanup_query_options(
+            Some("mode=before_now&targets=detail_body%2Ccompressed_body"),
+            None,
+        )
+        .expect("encoded targets query is valid");
+
+        assert_eq!(options.mode, ManualUsageCleanupMode::BeforeNow);
+        assert_eq!(options.targets, UsageCleanupTargets::body_targets());
+    }
+
+    #[test]
+    fn manual_cleanup_request_rejects_non_object_body() {
+        let body = Bytes::from_static(br#"[]"#);
+
+        assert!(parse_manual_usage_cleanup_request(Some(&body)).is_err());
+    }
+}

apps/aether-gateway/src/handlers/proxy/finalize.rs

@@ -7,7 +7,7 @@ use crate::constants::{
 };
 use crate::control::GatewayControlDecision;
 use crate::control::GatewayPublicRequestContext;
-use crate::middleware::{should_downgrade_access_log, RequestLogEmitted};
+use crate::middleware::{sanitize_access_log_path, should_downgrade_access_log, RequestLogEmitted};
 use crate::AppState;
 use aether_runtime::{maybe_hold_axum_response_permit, AdmissionPermit};
 use axum::body::{Body, Bytes};
@@ -95,11 +95,12 @@ pub(super) fn finalize_gateway_response(
         .map(|auth_context| auth_context.api_key_id.as_str())
         .unwrap_or("-");
     let status_code = response.status().as_u16();
+    let sanitized_path_and_query = sanitize_access_log_path(path_and_query);
     emit_admin_audit(
         &mut response,
         trace_id,
         method,
-        path_and_query,
+        sanitized_path_and_query.as_str(),
         control_decision,
     );
     if response.status().is_server_error() {
@@ -112,7 +113,7 @@ pub(super) fn finalize_gateway_response(
             request_id,
             remote_addr = %remote_addr,
             method = %method,
-            path = %path_and_query,
+            path = %sanitized_path_and_query,
             user_id,
             api_key_id,
             route_class,
@@ -122,7 +123,7 @@ pub(super) fn finalize_gateway_response(
             elapsed_ms,
             "gateway request failed"
         );
-    } else if should_downgrade_access_log(method, path_and_query) {
+    } else if should_downgrade_access_log(method, sanitized_path_and_query.as_str()) {
         trace!(
             event_name = "http_request_completed",
             log_type = "access",
@@ -132,7 +133,7 @@ pub(super) fn finalize_gateway_response(
             request_id,
             remote_addr = %remote_addr,
             method = %method,
-            path = %path_and_query,
+            path = %sanitized_path_and_query,
             user_id,
             api_key_id,
             route_class,
@@ -152,7 +153,7 @@ pub(super) fn finalize_gateway_response(
             request_id,
             remote_addr = %remote_addr,
             method = %method,
-            path = %path_and_query,
+            path = %sanitized_path_and_query,
             user_id,
             api_key_id,
             route_class,
@@ -254,3 +255,106 @@ pub(super) fn finalize_gateway_response_with_context(
         request_permit,
     )
 }
+
+#[cfg(test)]
+mod tests {
+    use super::finalize_gateway_response;
+    use crate::control::GatewayControlDecision;
+    use crate::AppState;
+    use axum::body::Body;
+    use axum::http::{Method, Response, StatusCode};
+    use std::sync::{Arc, Mutex};
+    use std::time::Instant;
+    use tracing_subscriber::filter::LevelFilter;
+    use tracing_subscriber::prelude::*;
+
+    #[derive(Clone, Default)]
+    struct SharedBuffer(Arc<Mutex<Vec<u8>>>);
+
+    struct SharedBufferWriter(Arc<Mutex<Vec<u8>>>);
+
+    impl SharedBuffer {
+        fn lines(&self) -> Vec<serde_json::Value> {
+            String::from_utf8(self.0.lock().expect("buffer should lock").clone())
+                .expect("buffer should contain valid utf-8")
+                .lines()
+                .filter(|line| !line.trim().is_empty())
+                .map(|line| serde_json::from_str(line).expect("json log line should parse"))
+                .collect()
+        }
+    }
+
+    impl std::io::Write for SharedBufferWriter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0
+                .lock()
+                .expect("buffer should lock")
+                .extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> tracing_subscriber::fmt::writer::MakeWriter<'a> for SharedBuffer {
+        type Writer = SharedBufferWriter;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            SharedBufferWriter(Arc::clone(&self.0))
+        }
+    }
+
+    #[test]
+    fn finalize_gateway_response_logs_sanitized_path_and_query() {
+        let state = AppState::new().expect("gateway state should build");
+        let writer = SharedBuffer::default();
+        let subscriber = tracing_subscriber::registry().with(
+            tracing_subscriber::fmt::layer()
+                .json()
+                .flatten_event(true)
+                .with_current_span(false)
+                .with_span_list(false)
+                .with_writer(writer.clone())
+                .with_filter(LevelFilter::INFO),
+        );
+        let dispatch = tracing::Dispatch::new(subscriber);
+        let _guard = tracing::dispatcher::set_default(&dispatch);
+        let remote_addr = "127.0.0.1:8080"
+            .parse()
+            .expect("remote address should parse");
+        let control_decision = GatewayControlDecision::synthetic(
+            "/v1beta/models/gemini-3-flash-preview:generateContent",
+            Some("ai_public".to_string()),
+            Some("gemini".to_string()),
+            Some("generate_content".to_string()),
+            Some("gemini:generate_content".to_string()),
+        );
+
+        let response = Response::builder()
+            .status(StatusCode::OK)
+            .body(Body::empty())
+            .expect("response should build");
+
+        let _response = finalize_gateway_response(
+            &state,
+            response,
+            "trace-finalize",
+            &remote_addr,
+            &Method::GET,
+            "/v1beta/models/gemini-3-flash-preview:generateContent?key=secret&alt=sse",
+            Some(&control_decision),
+            "execution_runtime_sync",
+            &Instant::now(),
+            None,
+        );
+
+        let logs = writer.lines();
+        assert_eq!(logs.len(), 1);
+        assert_eq!(
+            logs[0]["path"],
+            "/v1beta/models/gemini-3-flash-preview:generateContent?alt=sse"
+        );
+    }
+}

apps/aether-gateway/src/handlers/shared/request_utils.rs

@@ -300,6 +300,11 @@ pub(crate) fn admin_proxy_local_requires_buffered_body(
                     http::Method::POST,
                     Some("query_models" | "test_model" | "test_model_failover"),
                 )
+                | (Some("routing_profiles_manage"), http::Method::POST, Some("create_group"))
+                | (Some("routing_profiles_manage"), http::Method::PATCH, Some("update_group"))
+                | (Some("routing_profiles_manage"), http::Method::POST, Some("dry_run_group"))
+                | (Some("routing_profiles_manage"), http::Method::POST, Some("create_binding"))
+                | (Some("routing_profiles_manage"), http::Method::PATCH, Some("update_binding"))
                 | (Some("billing_manage"), http::Method::POST, Some("apply_preset"))
                 | (Some("billing_manage"), http::Method::POST, Some("create_rule"))
                 | (Some("billing_manage"), http::Method::PUT, Some("update_rule"))