refactor: lazy pool key scheduling

apps/aether-gateway/src/ai_serving/planner/candidate_materialization.rs

@@ -3,7 +3,7 @@ use aether_ai_serving::{
     ai_should_persist_skipped_candidate_for_pool_membership,
     run_ai_available_candidate_persistence, run_ai_candidate_materialization,
     run_ai_skipped_candidate_persistence, AiAvailableCandidatePersistencePort,
-    AiCandidateMaterializationOutcome, AiCandidateMaterializationPort, AiCandidateResolutionMode,
+    AiCandidateMaterializationOutcome, AiCandidateMaterializationPort,
     AiSkippedCandidatePersistencePort,
 };
 use aether_scheduler_core::{ClientSessionAffinity, SchedulerMinimalCandidateSelectionCandidate};
@@ -17,8 +17,6 @@ use uuid::Uuid;
 
 use crate::ai_serving::planner::candidate_affinity_cache::remember_scheduler_affinity_for_candidate;
 use crate::ai_serving::planner::candidate_resolution::{
-    resolve_and_rank_local_execution_candidates,
-    resolve_and_rank_local_execution_candidates_without_transport_pair_gate,
     resolve_and_rank_logical_local_execution_candidates, EligibleLocalExecutionCandidate,
     LocalExecutionCandidateKind, SkippedLocalExecutionCandidate,
 };
@@ -225,37 +223,19 @@ where
         &self,
         candidates: Vec<Self::Candidate>,
     ) -> Result<(Vec<Self::Eligible>, Vec<Self::Skipped>), Self::Error> {
-        let requested_model = self.requested_model.map(str::to_string);
+        let resolved = resolve_and_rank_logical_local_execution_candidates(
-        let resolved = match self.resolution_mode {
+            self.state,
-            AiCandidateResolutionMode::Standard => {
+            candidates,
-                resolve_and_rank_local_execution_candidates(
+            self.client_api_format,
-                    self.state,
+            self.requested_model,
-                    candidates,
+            self.auth_snapshot,
-                    self.client_api_format,
+            self.client_session_affinity,
-                    requested_model.as_deref().unwrap_or_default(),
+            self.required_capabilities,
-                    self.auth_snapshot,
+            self.sticky_session_token,
-                    self.client_session_affinity,
+            self.request_auth_channel,
-                    self.required_capabilities,
+            self.resolution_mode,
-                    self.sticky_session_token,
+        )
-                    self.request_auth_channel,
+        .await;
-                )
-                .await
-            }
-            AiCandidateResolutionMode::WithoutTransportPairGate => {
-                resolve_and_rank_local_execution_candidates_without_transport_pair_gate(
-                    self.state,
-                    candidates,
-                    self.client_api_format,
-                    requested_model.as_deref(),
-                    self.auth_snapshot,
-                    self.client_session_affinity,
-                    self.required_capabilities,
-                    self.sticky_session_token,
-                    self.request_auth_channel,
-                )
-                .await
-            }
-        };
         Ok(resolved)
     }
 
@@ -278,11 +258,14 @@ where
         &self,
         candidates: Vec<Self::Eligible>,
     ) -> Result<Vec<Self::Attempt>, Self::Error> {
-        Ok(persist_available_local_execution_candidates_with_context(
+        Ok(materialize_logical_local_execution_candidate_attempts(
             self.state,
             self.trace_id,
             self.persistence_policy.available,
             candidates,
+            self.sticky_session_token,
+            self.requested_model,
+            self.request_auth_channel,
             &self.build_available_extra_data,
         )
         .await)
@@ -491,6 +474,7 @@ where
     F: Fn(&EligibleLocalExecutionCandidate) -> Option<Value> + Send + Sync,
     G: Fn(SkippedLocalExecutionCandidate) -> SkippedLocalExecutionCandidate + Send + Sync,
 {
+    let _ = build_available_extra_data;
     let (candidates, resolved_skipped) = resolve_and_rank_logical_local_execution_candidates(
         state,
         candidates,
@@ -901,6 +885,64 @@ where
     .await
 }
 
+#[allow(clippy::too_many_arguments)]
+async fn materialize_logical_local_execution_candidate_attempts<F>(
+    state: PlannerAppState<'_>,
+    trace_id: &str,
+    context: LocalAvailableCandidatePersistenceContext<'_>,
+    candidates: Vec<EligibleLocalExecutionCandidate>,
+    sticky_session_token: Option<&str>,
+    requested_model: Option<&str>,
+    request_auth_channel: Option<&str>,
+    build_extra_data: &F,
+) -> Vec<LocalExecutionCandidateAttempt>
+where
+    F: Fn(&EligibleLocalExecutionCandidate) -> Option<Value> + Send + Sync,
+{
+    let mut attempts = Vec::new();
+
+    for (candidate_index, candidate) in candidates.into_iter().enumerate() {
+        let candidate_index = u32::try_from(candidate_index).unwrap_or(u32::MAX);
+        match candidate.kind {
+            LocalExecutionCandidateKind::SingleKey => {
+                attempts.extend(
+                    persist_available_local_execution_candidate_at_index(
+                        state,
+                        trace_id,
+                        context,
+                        candidate,
+                        candidate_index,
+                        build_extra_data,
+                    )
+                    .await,
+                );
+            }
+            LocalExecutionCandidateKind::PoolGroup => {
+                let mut cursor = PoolKeyCursor::new(
+                    state,
+                    candidate,
+                    sticky_session_token,
+                    requested_model,
+                    request_auth_channel,
+                );
+                let attempt_count_before_pool = attempts.len();
+                while let Some(candidate) = cursor.next_key().await {
+                    attempts.extend(build_unpersisted_local_execution_candidate_attempts(
+                        candidate,
+                        candidate_index,
+                    ));
+                }
+                let _ = cursor.take_skipped_candidates();
+                if attempts.len() == attempt_count_before_pool {
+                    cursor.log_exhausted();
+                }
+            }
+        }
+    }
+
+    attempts
+}
+
 async fn persist_available_local_execution_candidate_at_index<F>(
     state: PlannerAppState<'_>,
     trace_id: &str,
@@ -1167,7 +1209,10 @@ mod tests {
     use std::collections::VecDeque;
     use std::sync::Arc;
 
+    use aether_data::repository::auth::InMemoryAuthApiKeySnapshotRepository;
+    use aether_data::repository::candidate_selection::InMemoryMinimalCandidateSelectionReadRepository;
     use aether_data::repository::candidates::InMemoryRequestCandidateRepository;
+    use aether_data::repository::provider_catalog::InMemoryProviderCatalogReadRepository;
     use aether_provider_transport::snapshot::{
         GatewayProviderTransportEndpoint, GatewayProviderTransportKey,
         GatewayProviderTransportProvider,
@@ -1278,6 +1323,7 @@ mod tests {
             orchestration: LocalExecutionCandidateMetadata {
                 candidate_group_id: pool_key_index.map(|_| "pool-group".to_string()),
                 pool_key_index,
+                pool_key_lease: None,
             },
             ranking: None,
         }
@@ -1320,6 +1366,61 @@ mod tests {
         assert_eq!(stored[0].candidate_index, 2);
     }
 
+    #[tokio::test]
+    async fn logical_materialization_does_not_persist_pool_group_representative() {
+        let request_candidate_repository = Arc::new(InMemoryRequestCandidateRepository::default());
+        let app = AppState::new()
+            .expect("state should build")
+            .with_data_state_for_tests(
+                GatewayDataState::with_auth_candidate_selection_provider_catalog_and_request_candidate_repository_for_tests(
+                    Arc::new(InMemoryAuthApiKeySnapshotRepository::default()),
+                    Arc::new(InMemoryMinimalCandidateSelectionReadRepository::default()),
+                    Arc::new(InMemoryProviderCatalogReadRepository::seed(
+                        Vec::new(),
+                        Vec::new(),
+                        Vec::new(),
+                    )),
+                    Arc::clone(&request_candidate_repository),
+                    "test-encryption-key",
+                ),
+            );
+        let mut pool_group = sample_eligible("pool-group", None);
+        pool_group.kind = LocalExecutionCandidateKind::PoolGroup;
+        pool_group.transport = sample_transport(
+            "pool-group",
+            Some(json!({ "pool_advanced": { "scheduling_presets": [] } })),
+        );
+
+        let attempts = materialize_logical_local_execution_candidate_attempts(
+            PlannerAppState::new(&app),
+            "trace-logical-pool",
+            LocalAvailableCandidatePersistenceContext {
+                user_id: "user-1",
+                api_key_id: "api-key-1",
+                required_capabilities: None,
+                error_context: "persist should not fail",
+            },
+            vec![pool_group, sample_eligible("normal-key", None)],
+            None,
+            Some("gpt-5"),
+            None,
+            &|_| None,
+        )
+        .await;
+
+        assert_eq!(attempts.len(), 1);
+        assert_eq!(attempts[0].candidate_index, 1);
+        assert_eq!(attempts[0].eligible.candidate.key_id, "normal-key");
+
+        let stored = app
+            .read_request_candidates_by_request_id("trace-logical-pool")
+            .await
+            .expect("request candidates should read");
+        assert_eq!(stored.len(), 1);
+        assert_eq!(stored[0].key_id.as_deref(), Some("normal-key"));
+        assert_eq!(stored[0].candidate_index, 1);
+    }
+
     #[test]
     fn pool_key_attempts_use_distinct_effective_retry_indices() {
         let first = build_unpersisted_local_execution_candidate_attempts(

apps/aether-gateway/src/ai_serving/planner/passthrough/provider/family/payload.rs

@@ -105,6 +105,7 @@ pub(crate) async fn maybe_build_local_same_format_provider_decision_payload_for_
                 client_api_format: spec_metadata.api_format,
                 mapped_model: Some(&resolved.mapped_model),
                 candidate_group_id: eligible.orchestration.candidate_group_id.as_deref(),
+                pool_key_lease: eligible.orchestration.pool_key_lease.as_ref(),
                 ranking: eligible.ranking.as_ref(),
                 upstream_url: Some(&resolved.upstream_url),
                 header_rules: resolved.transport.endpoint.header_rules.as_ref(),

apps/aether-gateway/src/ai_serving/planner/report_context.rs

@@ -6,6 +6,7 @@ use aether_ai_serving::{
     provider_stream_event_api_format_for_provider_type as ai_provider_stream_event_api_format_for_provider_type,
     AiExecutionReportContextParts, AiRequestOrigin,
 };
+use aether_runtime_state::RuntimeLockLease;
 use aether_scheduler_core::{ClientSessionAffinity, SchedulerRankingOutcome};
 use serde_json::{Map, Value};
 
@@ -17,7 +18,7 @@ use crate::ai_serving::{
 use crate::client_session_affinity::{
     client_session_affinity_report_context_value, CLIENT_SESSION_AFFINITY_REPORT_CONTEXT_FIELD,
 };
-use crate::orchestration::ExecutionAttemptIdentity;
+use crate::orchestration::{insert_pool_key_lease_report_context_fields, ExecutionAttemptIdentity};
 
 pub(crate) struct LocalExecutionReportContextParts<'a> {
     pub(crate) auth_context: &'a ExecutionRuntimeAuthContext,
@@ -37,6 +38,7 @@ pub(crate) struct LocalExecutionReportContextParts<'a> {
     pub(crate) client_api_format: &'a str,
     pub(crate) mapped_model: Option<&'a str>,
     pub(crate) candidate_group_id: Option<&'a str>,
+    pub(crate) pool_key_lease: Option<&'a RuntimeLockLease>,
     pub(crate) ranking: Option<&'a SchedulerRankingOutcome>,
     pub(crate) upstream_url: Option<&'a str>,
     pub(crate) header_rules: Option<&'a Value>,
@@ -86,6 +88,7 @@ pub(crate) fn build_local_execution_report_context(
     {
         merge_incoming_tls_fingerprint(&mut extra_fields, incoming_tls);
     }
+    insert_pool_key_lease_report_context_fields(&mut extra_fields, parts.pool_key_lease);
     insert_request_path_fields(
         &mut extra_fields,
         parts.request_path,
@@ -258,6 +261,7 @@ mod tests {
                 client_api_format: "openai:chat",
                 mapped_model: None,
                 candidate_group_id: None,
+                pool_key_lease: None,
                 ranking: None,
                 upstream_url: None,
                 header_rules: None,
@@ -337,6 +341,7 @@ mod tests {
                 client_api_format: "gemini:generate_content",
                 mapped_model: None,
                 candidate_group_id: None,
+                pool_key_lease: None,
                 ranking: None,
                 upstream_url: None,
                 header_rules: None,
@@ -405,6 +410,7 @@ mod tests {
                 client_api_format: "openai:chat",
                 mapped_model: None,
                 candidate_group_id: None,
+                pool_key_lease: None,
                 ranking: None,
                 upstream_url: None,
                 header_rules: None,

apps/aether-gateway/src/ai_serving/planner/specialized/files/decision.rs

@@ -87,6 +87,7 @@ pub(super) async fn maybe_build_local_gemini_files_decision_payload_for_candidat
         client_api_format: GEMINI_FILES_CLIENT_API_FORMAT,
         mapped_model: None,
         candidate_group_id: eligible.orchestration.candidate_group_id.as_deref(),
+        pool_key_lease: eligible.orchestration.pool_key_lease.as_ref(),
         ranking: eligible.ranking.as_ref(),
         upstream_url: None,
         header_rules: transport.endpoint.header_rules.as_ref(),

apps/aether-gateway/src/ai_serving/planner/specialized/image/decision.rs

@@ -107,6 +107,7 @@ pub(super) async fn maybe_build_local_openai_image_decision_payload_for_candidat
             client_api_format: spec_metadata.api_format,
             mapped_model: Some(&resolved.mapped_model),
             candidate_group_id: eligible.orchestration.candidate_group_id.as_deref(),
+            pool_key_lease: eligible.orchestration.pool_key_lease.as_ref(),
             ranking: eligible.ranking.as_ref(),
             upstream_url: Some(&resolved.upstream_url),
             header_rules: transport.endpoint.header_rules.as_ref(),

apps/aether-gateway/src/ai_serving/planner/specialized/video/decision.rs

@@ -68,6 +68,7 @@ pub(super) async fn maybe_build_local_video_create_decision_payload_for_candidat
         client_api_format: spec_metadata.api_format,
         mapped_model: Some(&resolved.mapped_model),
         candidate_group_id: eligible.orchestration.candidate_group_id.as_deref(),
+        pool_key_lease: eligible.orchestration.pool_key_lease.as_ref(),
         ranking: eligible.ranking.as_ref(),
         upstream_url: None,
         header_rules: transport.endpoint.header_rules.as_ref(),

apps/aether-gateway/src/ai_serving/planner/standard/family/payload.rs

@@ -113,6 +113,7 @@ pub(super) async fn maybe_build_local_standard_decision_payload_for_candidate(
                 client_api_format: spec_metadata.api_format,
                 mapped_model: Some(&resolved.mapped_model),
                 candidate_group_id: eligible.orchestration.candidate_group_id.as_deref(),
+                pool_key_lease: eligible.orchestration.pool_key_lease.as_ref(),
                 ranking: eligible.ranking.as_ref(),
                 upstream_url: Some(&resolved.upstream_url),
                 header_rules: resolved.transport.endpoint.header_rules.as_ref(),

apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/payload.rs

@@ -102,6 +102,7 @@ pub(crate) async fn maybe_build_local_openai_chat_decision_payload_for_candidate
                 client_api_format: "openai:chat",
                 mapped_model: Some(&resolved.mapped_model),
                 candidate_group_id: eligible.orchestration.candidate_group_id.as_deref(),
+                pool_key_lease: eligible.orchestration.pool_key_lease.as_ref(),
                 ranking: eligible.ranking.as_ref(),
                 upstream_url: Some(&resolved.upstream_url),
                 header_rules: resolved.transport.endpoint.header_rules.as_ref(),

apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/decision/payload.rs

@@ -98,6 +98,7 @@ pub(crate) async fn maybe_build_local_openai_responses_decision_payload_for_cand
                 client_api_format: spec_metadata.api_format,
                 mapped_model: Some(&resolved.mapped_model),
                 candidate_group_id: eligible.orchestration.candidate_group_id.as_deref(),
+                pool_key_lease: eligible.orchestration.pool_key_lease.as_ref(),
                 ranking: eligible.ranking.as_ref(),
                 upstream_url: Some(&resolved.upstream_url),
                 header_rules: resolved.transport.endpoint.header_rules.as_ref(),

apps/aether-gateway/src/executor/candidate_loop.rs

@@ -16,6 +16,7 @@ use crate::clock::current_unix_ms;
 use crate::control::GatewayControlDecision;
 use crate::execution_runtime::{execute_execution_runtime_stream, execute_execution_runtime_sync};
 use crate::executor::{build_local_execution_exhaustion, LocalExecutionRequestOutcome};
+use crate::handlers::shared::provider_pool::release_admin_provider_pool_key_lease;
 use crate::log_ids::short_request_id;
 use crate::orchestration::local_execution_candidate_metadata_from_report_context;
 use crate::request_candidate_runtime::{
@@ -403,7 +404,19 @@ where
 {
     for plan_and_report in remaining {
         let report_context = plan_and_report.report_context();
-        if should_skip_unused_persistence(report_context.as_ref()) {
+        let metadata =
+            local_execution_candidate_metadata_from_report_context(report_context.as_ref());
+        if let Some(lease) = metadata.pool_key_lease.as_ref() {
+            if let Err(err) =
+                release_admin_provider_pool_key_lease(state.runtime_state.as_ref(), lease).await
+            {
+                warn!(
+                    error = ?err,
+                    "gateway candidate loop: failed to release unused pool key lease"
+                );
+            }
+        }
+        if should_skip_unused_persistence_from_metadata(&metadata) {
             continue;
         }
         record_local_request_candidate_status(
@@ -426,6 +439,12 @@ where
 
 fn should_skip_unused_persistence(report_context: Option<&serde_json::Value>) -> bool {
     let metadata = local_execution_candidate_metadata_from_report_context(report_context);
+    should_skip_unused_persistence_from_metadata(&metadata)
+}
+
+fn should_skip_unused_persistence_from_metadata(
+    metadata: &crate::orchestration::LocalExecutionCandidateMetadata,
+) -> bool {
     metadata.candidate_group_id.is_some() && metadata.pool_key_index.is_some()
 }
 

apps/aether-gateway/src/handlers/admin/provider/pool/runtime/keys.rs

@@ -14,6 +14,10 @@ pub(super) fn pool_cooldown_key(provider_id: &str, key_id: &str) -> String {
     format!("ap:{provider_id}:cooldown:{key_id}")
 }
 
+pub(super) fn pool_lease_key(provider_id: &str, key_id: &str) -> String {
+    format!("ap:{provider_id}:lease:{key_id}")
+}
+
 pub(super) fn pool_cooldown_index_key(provider_id: &str) -> String {
     format!("ap:{provider_id}:cooldown_idx")
 }

apps/aether-gateway/src/handlers/admin/provider/pool/runtime/leases.rs

@@ -0,0 +1,27 @@
+use super::keys::pool_lease_key;
+use aether_runtime_state::{DataLayerError, RuntimeLockLease, RuntimeState};
+use std::time::Duration;
+
+pub(crate) const ADMIN_PROVIDER_POOL_KEY_LEASE_TTL_MS: u64 = 15 * 60 * 1000;
+
+pub(crate) async fn try_claim_admin_provider_pool_key(
+    runtime: &RuntimeState,
+    provider_id: &str,
+    key_id: &str,
+    owner: &str,
+) -> Result<Option<RuntimeLockLease>, DataLayerError> {
+    runtime
+        .lock_try_acquire(
+            &pool_lease_key(provider_id, key_id),
+            owner,
+            Duration::from_millis(ADMIN_PROVIDER_POOL_KEY_LEASE_TTL_MS),
+        )
+        .await
+}
+
+pub(crate) async fn release_admin_provider_pool_key_lease(
+    runtime: &RuntimeState,
+    lease: &RuntimeLockLease,
+) -> Result<bool, DataLayerError> {
+    runtime.lock_release(lease).await
+}

apps/aether-gateway/src/handlers/admin/provider/pool/runtime/mod.rs

@@ -1,9 +1,14 @@
 mod keys;
+mod leases;
 mod mutations;
 mod reads;
 mod status;
 mod writes;
 
+pub(crate) use self::leases::{
+    release_admin_provider_pool_key_lease, try_claim_admin_provider_pool_key,
+    ADMIN_PROVIDER_POOL_KEY_LEASE_TTL_MS,
+};
 pub(crate) use self::mutations::{
     clear_admin_provider_pool_cooldown, reset_admin_provider_pool_cost,
 };

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

@@ -2,7 +2,8 @@ pub(crate) use super::super::admin::provider::pool::config::admin_provider_pool_
 pub(crate) use super::super::admin::provider::pool::runtime::{
     admin_provider_pool_key_circuit_breaker_reason, read_admin_provider_pool_runtime_state,
     record_admin_provider_pool_error, record_admin_provider_pool_stream_timeout,
-    record_admin_provider_pool_success,
+    record_admin_provider_pool_success, release_admin_provider_pool_key_lease,
+    try_claim_admin_provider_pool_key, ADMIN_PROVIDER_POOL_KEY_LEASE_TTL_MS,
 };
 pub(crate) use super::super::admin::provider::shared::support::{
     AdminProviderPoolConfig, AdminProviderPoolRuntimeState, AdminProviderPoolSchedulingPreset,

apps/aether-gateway/src/orchestration/attempt.rs

@@ -1,3 +1,4 @@
+use aether_runtime_state::RuntimeLockLease;
 use aether_scheduler_core::parse_request_candidate_report_context;
 use serde_json::Value;
 
@@ -29,8 +30,14 @@ impl ExecutionAttemptIdentity {
 pub(crate) struct LocalExecutionCandidateMetadata {
     pub(crate) candidate_group_id: Option<String>,
     pub(crate) pool_key_index: Option<u32>,
+    pub(crate) pool_key_lease: Option<RuntimeLockLease>,
 }
 
+pub(crate) const POOL_KEY_LEASE_KEY_REPORT_FIELD: &str = "pool_key_lease_key";
+pub(crate) const POOL_KEY_LEASE_OWNER_REPORT_FIELD: &str = "pool_key_lease_owner";
+pub(crate) const POOL_KEY_LEASE_TOKEN_REPORT_FIELD: &str = "pool_key_lease_token";
+pub(crate) const POOL_KEY_LEASE_TTL_MS_REPORT_FIELD: &str = "pool_key_lease_ttl_ms";
+
 pub(crate) fn attempt_identity_from_report_context(
     report_context: Option<&Value>,
 ) -> Option<ExecutionAttemptIdentity> {
@@ -57,9 +64,65 @@ pub(crate) fn local_execution_candidate_metadata_from_report_context(
             .and_then(|value| value.get("pool_key_index"))
             .and_then(Value::as_u64)
             .and_then(|value| u32::try_from(value).ok()),
+        pool_key_lease: pool_key_lease_from_report_context(report_context),
     }
 }
 
+pub(crate) fn insert_pool_key_lease_report_context_fields(
+    extra_fields: &mut serde_json::Map<String, Value>,
+    lease: Option<&RuntimeLockLease>,
+) {
+    let Some(lease) = lease else {
+        return;
+    };
+    extra_fields.insert(
+        POOL_KEY_LEASE_KEY_REPORT_FIELD.to_string(),
+        Value::String(lease.key.clone()),
+    );
+    extra_fields.insert(
+        POOL_KEY_LEASE_OWNER_REPORT_FIELD.to_string(),
+        Value::String(lease.owner.clone()),
+    );
+    extra_fields.insert(
+        POOL_KEY_LEASE_TOKEN_REPORT_FIELD.to_string(),
+        Value::String(lease.token.clone()),
+    );
+    extra_fields.insert(
+        POOL_KEY_LEASE_TTL_MS_REPORT_FIELD.to_string(),
+        Value::Number(lease.ttl_ms.into()),
+    );
+}
+
+fn pool_key_lease_from_report_context(report_context: Option<&Value>) -> Option<RuntimeLockLease> {
+    let report_context = report_context?;
+    let key = report_context
+        .get(POOL_KEY_LEASE_KEY_REPORT_FIELD)
+        .and_then(Value::as_str)
+        .map(str::trim)
+        .filter(|value| !value.is_empty())?;
+    let owner = report_context
+        .get(POOL_KEY_LEASE_OWNER_REPORT_FIELD)
+        .and_then(Value::as_str)
+        .map(str::trim)
+        .filter(|value| !value.is_empty())?;
+    let token = report_context
+        .get(POOL_KEY_LEASE_TOKEN_REPORT_FIELD)
+        .and_then(Value::as_str)
+        .map(str::trim)
+        .filter(|value| !value.is_empty())?;
+    let ttl_ms = report_context
+        .get(POOL_KEY_LEASE_TTL_MS_REPORT_FIELD)
+        .and_then(Value::as_u64)
+        .filter(|value| *value > 0)?;
+
+    Some(RuntimeLockLease {
+        key: key.to_string(),
+        owner: owner.to_string(),
+        token: token.to_string(),
+        ttl_ms,
+    })
+}
+
 pub(crate) fn build_local_attempt_identities(
     candidate_index: u32,
     transport: &GatewayProviderTransportSnapshot,
@@ -127,6 +190,7 @@ mod tests {
         GatewayProviderTransportEndpoint, GatewayProviderTransportKey,
         GatewayProviderTransportProvider, GatewayProviderTransportSnapshot,
     };
+    use aether_runtime_state::RuntimeLockLease;
 
     fn sample_transport(
         provider_max_retries: Option<i32>,
@@ -412,6 +476,10 @@ mod tests {
         let metadata = local_execution_candidate_metadata_from_report_context(Some(&json!({
             "candidate_group_id": "group-1",
             "pool_key_index": 3,
+            "pool_key_lease_key": "ap:provider-1:lease:key-1",
+            "pool_key_lease_owner": "gateway-1",
+            "pool_key_lease_token": "gateway-1:token-1",
+            "pool_key_lease_ttl_ms": 900000,
         })));
 
         assert_eq!(
@@ -419,6 +487,12 @@ mod tests {
             LocalExecutionCandidateMetadata {
                 candidate_group_id: Some("group-1".to_string()),
                 pool_key_index: Some(3),
+                pool_key_lease: Some(RuntimeLockLease {
+                    key: "ap:provider-1:lease:key-1".to_string(),
+                    owner: "gateway-1".to_string(),
+                    token: "gateway-1:token-1".to_string(),
+                    ttl_ms: 900000,
+                }),
             }
         );
     }

apps/aether-gateway/src/orchestration/effects.rs

@@ -27,8 +27,9 @@ use crate::handlers::shared::provider_pool::admin_provider_pool_config_from_conf
 use crate::handlers::shared::provider_pool::{
     admin_provider_pool_key_circuit_breaker_reason, record_admin_provider_pool_error,
     record_admin_provider_pool_stream_timeout, record_admin_provider_pool_success,
-    AdminProviderPoolConfig,
+    release_admin_provider_pool_key_lease, AdminProviderPoolConfig,
 };
+use crate::orchestration::local_execution_candidate_metadata_from_report_context;
 use crate::scheduler::affinity::SCHEDULER_AFFINITY_TTL;
 use crate::AppState;
 
@@ -129,19 +130,40 @@ pub(crate) async fn apply_local_execution_effect(
         }
         LocalExecutionEffect::PoolSuccessSync { payload } => {
             record_sync_pool_success_effect(state, context, payload).await;
+            release_pool_key_lease_effect(state, context).await;
         }
         LocalExecutionEffect::PoolSuccessStream { payload } => {
             record_stream_pool_success_effect(state, context, payload).await;
+            release_pool_key_lease_effect(state, context).await;
         }
         LocalExecutionEffect::PoolError(effect) => {
             record_pool_error_effect(state, context, effect).await;
+            release_pool_key_lease_effect(state, context).await;
         }
         LocalExecutionEffect::PoolStreamTimeout => {
             record_pool_stream_timeout_effect(state, context).await;
+            release_pool_key_lease_effect(state, context).await;
         }
     }
 }
 
+async fn release_pool_key_lease_effect(state: &AppState, context: LocalExecutionEffectContext<'_>) {
+    let metadata = local_execution_candidate_metadata_from_report_context(context.report_context);
+    let Some(lease) = metadata.pool_key_lease else {
+        return;
+    };
+    if let Err(err) =
+        release_admin_provider_pool_key_lease(state.runtime_state.as_ref(), &lease).await
+    {
+        warn!(
+            error = ?err,
+            provider_id = %context.plan.provider_id,
+            key_id = %context.plan.key_id,
+            "gateway orchestration effects: failed to release pool key lease"
+        );
+    }
+}
+
 fn report_context_string_field<'a>(
     report_context: Option<&'a Value>,
     field: &str,

apps/aether-gateway/src/orchestration/mod.rs

@@ -17,7 +17,8 @@ pub(crate) use self::adaptive::{
     LocalAdaptiveRateLimitProjection, LocalAdaptiveSuccessProjection,
 };
 pub(crate) use self::attempt::{
-    attempt_identity_from_report_context, build_local_attempt_identities, local_attempt_slot_count,
+    attempt_identity_from_report_context, build_local_attempt_identities,
+    insert_pool_key_lease_report_context_fields, local_attempt_slot_count,
     local_execution_candidate_metadata_from_report_context, ExecutionAttemptIdentity,
     LocalExecutionCandidateMetadata,
 };

crates/aether-ai-serving/src/pool_scheduler.rs

@@ -206,6 +206,10 @@ fn schedule_pool_group<Candidate>(
         .unwrap_or_default();
     let active_presets =
         normalize_enabled_pool_presets(&pool_config.scheduling_presets, provider_type.as_str());
+    let lru_distribution_enabled = pool_config.lru_enabled
+        && !active_presets
+            .iter()
+            .any(|preset| pool_preset_mutex_group(&preset.preset).is_some());
 
     let mut available = Vec::new();
     let mut skipped = Vec::new();
@@ -285,7 +289,7 @@ fn schedule_pool_group<Candidate>(
         let sort_vectors = build_pool_sort_vectors(
             &available,
             &active_presets,
-            pool_config.lru_enabled,
+            lru_distribution_enabled,
             group_sort_seed(
                 provider_type.as_str(),
                 available.first().map(|item| &item.item.facts),
@@ -300,7 +304,7 @@ fn schedule_pool_group<Candidate>(
                 .cmp(&sort_vectors.get(&right.item.facts.key_id))
                 .then(left.original_index.cmp(&right.original_index))
         });
-    } else if pool_config.lru_enabled {
+    } else if lru_distribution_enabled {
         let lru_ranks = lru_rank_indices(&available, false);
         available.sort_by(|left, right| {
             lru_ranks
@@ -359,6 +363,16 @@ fn build_pool_sort_vectors<Candidate>(
     let lru_ranks = lru_rank_indices(items, false);
     let cache_affinity_ranks = lru_rank_indices(items, true);
 
+    if lru_enabled {
+        for item in items {
+            let key_id = item.item.facts.key_id.clone();
+            vectors
+                .entry(key_id.clone())
+                .or_default()
+                .push(*lru_ranks.get(&key_id).unwrap_or(&0));
+        }
+    }
+
     for preset in presets {
         let ranks = match preset.preset.as_str() {
             "cache_affinity" => cache_affinity_ranks.clone(),
@@ -385,16 +399,6 @@ fn build_pool_sort_vectors<Candidate>(
         }
     }
 
-    if lru_enabled {
-        for item in items {
-            let key_id = item.item.facts.key_id.clone();
-            vectors
-                .entry(key_id.clone())
-                .or_default()
-                .push(*lru_ranks.get(&key_id).unwrap_or(&0));
-        }
-    }
-
     vectors
 }
 
@@ -408,13 +412,13 @@ fn lru_rank_indices<Candidate>(
 
 fn priority_first_ranks<Candidate>(
     items: &[PoolGroupCandidateOrdering<Candidate>],
-    lru_ranks: &BTreeMap<String, usize>,
+    _lru_ranks: &BTreeMap<String, usize>,
 ) -> BTreeMap<String, usize> {
     let scores = collect_metric_scores(items, |item| {
         Some(f64::from(item.item.facts.key_internal_priority))
     });
     if !score_map_has_variation(&scores) {
-        return lru_ranks.clone();
+        return neutral_rank_indices(items);
     }
     rank_indices_from_score_map(items, &scores, false)
 }
@@ -422,27 +426,35 @@ fn priority_first_ranks<Candidate>(
 fn single_account_ranks<Candidate>(
     items: &[PoolGroupCandidateOrdering<Candidate>],
 ) -> BTreeMap<String, usize> {
-    let n = items.len().saturating_sub(1).max(1) as f64;
-    let priority_scores = collect_metric_scores(items, |item| {
-        Some(f64::from(item.item.facts.key_internal_priority))
-    });
-    let priority_ranks = rank_indices_from_score_map(items, &priority_scores, false);
     let lru_desc_ranks = lru_rank_indices(items, true);
-    let combined_scores = items
+    let mut decorated = items
         .iter()
         .map(|item| {
             let key_id = item.item.facts.key_id.clone();
-            let priority_rank = *priority_ranks.get(&key_id).unwrap_or(&0) as f64 / n;
+            (
-            let lru_rank = *lru_desc_ranks.get(&key_id).unwrap_or(&0) as f64 / n;
+                item.item.facts.key_internal_priority,
-            (key_id, Some((priority_rank * 0.75) + (lru_rank * 0.25)))
+                *lru_desc_ranks.get(&key_id).unwrap_or(&0),
+                item.original_index,
+                key_id,
+            )
         })
-        .collect::<BTreeMap<_, _>>();
+        .collect::<Vec<_>>();
-    rank_indices_from_score_map(items, &combined_scores, false)
+    decorated.sort_by(|left, right| {
+        left.0
+            .cmp(&right.0)
+            .then(left.1.cmp(&right.1))
+            .then(left.2.cmp(&right.2))
+    });
+    decorated
+        .into_iter()
+        .enumerate()
+        .map(|(rank, (_, _, _, key_id))| (key_id, rank))
+        .collect()
 }
 
 fn plan_ranks<Candidate>(
     items: &[PoolGroupCandidateOrdering<Candidate>],
-    lru_ranks: &BTreeMap<String, usize>,
+    _lru_ranks: &BTreeMap<String, usize>,
     mode: Option<&str>,
 ) -> BTreeMap<String, usize> {
     let scores = items
@@ -458,14 +470,14 @@ fn plan_ranks<Candidate>(
         })
         .collect::<BTreeMap<_, _>>();
     if !score_map_has_variation(&scores) {
-        return lru_ranks.clone();
+        return neutral_rank_indices(items);
     }
     rank_indices_from_score_map(items, &scores, false)
 }
 
 fn health_first_ranks<Candidate>(
     items: &[PoolGroupCandidateOrdering<Candidate>],
-    lru_ranks: &BTreeMap<String, usize>,
+    _lru_ranks: &BTreeMap<String, usize>,
 ) -> BTreeMap<String, usize> {
     let scores = collect_metric_scores(items, |item| {
         item.item
@@ -474,39 +486,39 @@ fn health_first_ranks<Candidate>(
             .map(|score| 1.0 - score.clamp(0.0, 1.0))
     });
     if !score_map_has_signal(&scores) {
-        return lru_ranks.clone();
+        return neutral_rank_indices(items);
     }
     rank_indices_from_score_map(items, &scores, false)
 }
 
 fn latency_first_ranks<Candidate>(
     items: &[PoolGroupCandidateOrdering<Candidate>],
-    lru_ranks: &BTreeMap<String, usize>,
+    _lru_ranks: &BTreeMap<String, usize>,
 ) -> BTreeMap<String, usize> {
     let scores = collect_metric_scores(items, |item| item.item.key_context.latency_avg_ms);
     if !score_map_has_signal(&scores) {
-        return lru_ranks.clone();
+        return neutral_rank_indices(items);
     }
     rank_indices_from_score_map(items, &scores, false)
 }
 
 fn cost_first_ranks<Candidate>(
     items: &[PoolGroupCandidateOrdering<Candidate>],
-    lru_ranks: &BTreeMap<String, usize>,
+    _lru_ranks: &BTreeMap<String, usize>,
     cost_limit_per_key_tokens: Option<u64>,
 ) -> BTreeMap<String, usize> {
     let scores = collect_metric_scores(items, |item| {
         cost_penalty(item, cost_limit_per_key_tokens).or(item.item.key_context.quota_usage_ratio)
     });
     if !score_map_has_signal(&scores) {
-        return lru_ranks.clone();
+        return neutral_rank_indices(items);
     }
     rank_indices_from_score_map(items, &scores, false)
 }
 
 fn quota_balanced_ranks<Candidate>(
     items: &[PoolGroupCandidateOrdering<Candidate>],
-    lru_ranks: &BTreeMap<String, usize>,
+    _lru_ranks: &BTreeMap<String, usize>,
     cost_limit_per_key_tokens: Option<u64>,
 ) -> BTreeMap<String, usize> {
     let scores = collect_metric_scores(items, |item| {
@@ -516,18 +528,18 @@ fn quota_balanced_ranks<Candidate>(
             .or_else(|| cost_penalty(item, cost_limit_per_key_tokens))
     });
     if !score_map_has_signal(&scores) {
-        return lru_ranks.clone();
+        return neutral_rank_indices(items);
     }
     rank_indices_from_score_map(items, &scores, false)
 }
 
 fn recent_refresh_ranks<Candidate>(
     items: &[PoolGroupCandidateOrdering<Candidate>],
-    lru_ranks: &BTreeMap<String, usize>,
+    _lru_ranks: &BTreeMap<String, usize>,
 ) -> BTreeMap<String, usize> {
     let scores = collect_metric_scores(items, |item| item.item.key_context.quota_reset_seconds);
     if !score_map_has_signal(&scores) {
-        return lru_ranks.clone();
+        return neutral_rank_indices(items);
     }
     rank_indices_from_score_map(items, &scores, false)
 }
@@ -646,6 +658,15 @@ fn rank_indices_from_score_map<Candidate>(
         .collect()
 }
 
+fn neutral_rank_indices<Candidate>(
+    items: &[PoolGroupCandidateOrdering<Candidate>],
+) -> BTreeMap<String, usize> {
+    items
+        .iter()
+        .map(|item| (item.item.facts.key_id.clone(), 0))
+        .collect()
+}
+
 fn cost_penalty<Candidate>(
     item: &PoolGroupCandidateOrdering<Candidate>,
     cost_limit_per_key_tokens: Option<u64>,
@@ -740,47 +761,42 @@ fn normalize_enabled_pool_presets(
         entries.push((entries.len(), "recent_refresh".to_string(), true, None));
     }
 
-    let mut group_anchor_index = BTreeMap::<String, usize>::new();
+    let mut distribution_mode = None::<(usize, String, Option<String>)>;
-    for (index, preset, _, _) in &entries {
+    let mut strategy_presets = Vec::<(usize, String, Option<String>)>::new();
-        let Some(mutex_group) = pool_preset_mutex_group(preset) else {
-            continue;
-        };
-        group_anchor_index
-            .entry(mutex_group.to_string())
-            .or_insert(*index);
-    }
-
-    let mut ordered_enabled = Vec::<(usize, usize, String, Option<String>)>::new();
-    let mut group_enabled = BTreeMap::<String, (usize, usize, String, Option<String>)>::new();
 
     for (index, preset, enabled, mode) in entries {
-        if !enabled
+        if !enabled || !pool_preset_supported_for_provider(&preset, &provider_type) {
-            || preset == "lru"
-            || !pool_preset_supported_for_provider(&preset, &provider_type)
-        {
             continue;
         }
 
         let Some(mutex_group) = pool_preset_mutex_group(&preset) else {
-            ordered_enabled.push((index, index, preset, mode));
+            strategy_presets.push((index, preset, mode));
             continue;
         };
-        let anchor = group_anchor_index
+
-            .get(mutex_group)
+        if mutex_group == "distribution_mode"
-            .copied()
+            && distribution_mode
-            .unwrap_or(index);
+                .as_ref()
-        let existing = group_enabled.get(mutex_group);
+                .is_none_or(|current| index < current.0)
-        if existing.is_none_or(|current| index < current.1) {
+        {
-            group_enabled.insert(mutex_group.to_string(), (anchor, index, preset, mode));
+            distribution_mode = Some((index, preset, mode));
         }
     }
 
-    ordered_enabled.extend(group_enabled.into_values());
+    let mut normalized = Vec::new();
-    ordered_enabled.sort_by(|left, right| left.0.cmp(&right.0).then(left.1.cmp(&right.1)));
+
-    ordered_enabled
+    if let Some((_, preset, mode)) = distribution_mode.filter(|(_, preset, _)| preset != "lru") {
-        .into_iter()
+        normalized.push(NormalizedPoolPreset { preset, mode });
-        .map(|(_, _, preset, mode)| NormalizedPoolPreset { preset, mode })
+    }
-        .collect()
+
+    strategy_presets.sort_by_key(|left| left.0);
+    normalized.extend(
+        strategy_presets
+            .into_iter()
+            .map(|(_, preset, mode)| NormalizedPoolPreset { preset, mode }),
+    );
+
+    normalized
 }
 
 fn pool_preset_supported_for_provider(preset: &str, provider_type: &str) -> bool {
@@ -959,7 +975,193 @@ mod tests {
     }
 
     #[test]
-    fn normalizes_distribution_mutex_group_to_first_enabled_member() {
+    fn pool_scheduler_applies_distribution_mode_before_strategy_presets() {
+        let key_cache_hit =
+            sample_candidate("provider-pool", "endpoint-1", "key-cache-hit", 50, true)
+                .with_presets(vec![
+                    AiPoolSchedulingPreset {
+                        preset: "cache_affinity".to_string(),
+                        enabled: true,
+                        mode: None,
+                    },
+                    AiPoolSchedulingPreset {
+                        preset: "priority_first".to_string(),
+                        enabled: true,
+                        mode: None,
+                    },
+                ]);
+        let key_high_priority =
+            sample_candidate("provider-pool", "endpoint-1", "key-high-priority", 10, true)
+                .with_presets(vec![
+                    AiPoolSchedulingPreset {
+                        preset: "cache_affinity".to_string(),
+                        enabled: true,
+                        mode: None,
+                    },
+                    AiPoolSchedulingPreset {
+                        preset: "priority_first".to_string(),
+                        enabled: true,
+                        mode: None,
+                    },
+                ]);
+
+        let runtime_by_provider = BTreeMap::from([(
+            "provider-pool".to_string(),
+            AiPoolRuntimeState {
+                lru_score_by_key: BTreeMap::from([
+                    ("key-cache-hit".to_string(), 200.0),
+                    ("key-high-priority".to_string(), 10.0),
+                ]),
+                ..AiPoolRuntimeState::default()
+            },
+        )]);
+
+        let outcome = run_ai_pool_scheduler(
+            vec![key_cache_hit, key_high_priority],
+            &runtime_by_provider,
+            "seed",
+        );
+
+        assert!(outcome.skipped_candidates.is_empty());
+        assert_eq!(
+            outcome
+                .candidates
+                .iter()
+                .map(|item| item.candidate.as_str())
+                .collect::<Vec<_>>(),
+            vec!["key-cache-hit", "key-high-priority"]
+        );
+    }
+
+    #[test]
+    fn load_balance_distribution_is_not_overridden_by_priority_strategy() {
+        let key_random_first =
+            sample_candidate("provider-pool", "endpoint-1", "key-random-first", 50, true)
+                .with_presets(vec![
+                    AiPoolSchedulingPreset {
+                        preset: "load_balance".to_string(),
+                        enabled: true,
+                        mode: None,
+                    },
+                    AiPoolSchedulingPreset {
+                        preset: "priority_first".to_string(),
+                        enabled: true,
+                        mode: None,
+                    },
+                ]);
+        let key_high_priority =
+            sample_candidate("provider-pool", "endpoint-1", "key-high-priority", 10, true)
+                .with_presets(vec![
+                    AiPoolSchedulingPreset {
+                        preset: "load_balance".to_string(),
+                        enabled: true,
+                        mode: None,
+                    },
+                    AiPoolSchedulingPreset {
+                        preset: "priority_first".to_string(),
+                        enabled: true,
+                        mode: None,
+                    },
+                ]);
+        let nonce = (0..1000)
+            .map(|index| format!("seed-{index}"))
+            .find(|nonce| {
+                let group_seed = format!("codex:provider-pool:endpoint-1:model-1:gpt-5:{nonce}");
+                stable_hash_score(format!("{group_seed}:key-random-first").as_str())
+                    < stable_hash_score(format!("{group_seed}:key-high-priority").as_str())
+            })
+            .expect("test seed should exist");
+
+        let outcome = run_ai_pool_scheduler(
+            vec![key_random_first, key_high_priority],
+            &BTreeMap::new(),
+            nonce.as_str(),
+        );
+
+        assert!(outcome.skipped_candidates.is_empty());
+        assert_eq!(
+            outcome
+                .candidates
+                .iter()
+                .map(|item| item.candidate.as_str())
+                .collect::<Vec<_>>(),
+            vec!["key-random-first", "key-high-priority"]
+        );
+    }
+
+    #[test]
+    fn single_account_distribution_orders_by_priority_then_reverse_lru() {
+        let key_priority_old =
+            sample_candidate("provider-pool", "endpoint-1", "key-priority-old", 10, true)
+                .with_presets(vec![AiPoolSchedulingPreset {
+                    preset: "single_account".to_string(),
+                    enabled: true,
+                    mode: None,
+                }]);
+        let key_priority_recent = sample_candidate(
+            "provider-pool",
+            "endpoint-1",
+            "key-priority-recent",
+            10,
+            true,
+        )
+        .with_presets(vec![AiPoolSchedulingPreset {
+            preset: "single_account".to_string(),
+            enabled: true,
+            mode: None,
+        }]);
+        let key_lower_priority_recent = sample_candidate(
+            "provider-pool",
+            "endpoint-1",
+            "key-lower-priority-recent",
+            50,
+            true,
+        )
+        .with_presets(vec![AiPoolSchedulingPreset {
+            preset: "single_account".to_string(),
+            enabled: true,
+            mode: None,
+        }]);
+
+        let runtime_by_provider = BTreeMap::from([(
+            "provider-pool".to_string(),
+            AiPoolRuntimeState {
+                lru_score_by_key: BTreeMap::from([
+                    ("key-priority-old".to_string(), 10.0),
+                    ("key-priority-recent".to_string(), 200.0),
+                    ("key-lower-priority-recent".to_string(), 500.0),
+                ]),
+                ..AiPoolRuntimeState::default()
+            },
+        )]);
+
+        let outcome = run_ai_pool_scheduler(
+            vec![
+                key_priority_old,
+                key_lower_priority_recent,
+                key_priority_recent,
+            ],
+            &runtime_by_provider,
+            "seed",
+        );
+
+        assert!(outcome.skipped_candidates.is_empty());
+        assert_eq!(
+            outcome
+                .candidates
+                .iter()
+                .map(|item| item.candidate.as_str())
+                .collect::<Vec<_>>(),
+            vec![
+                "key-priority-recent",
+                "key-priority-old",
+                "key-lower-priority-recent"
+            ]
+        );
+    }
+
+    #[test]
+    fn normalizes_distribution_mode_before_strategy_presets() {
         let presets = normalize_enabled_ai_pool_presets(
             &[
                 AiPoolSchedulingPreset {
@@ -989,6 +1191,32 @@ mod tests {
         assert_eq!(presets, ["single_account", "priority_first"]);
     }
 
+    #[test]
+    fn normalizes_lru_as_mutually_exclusive_distribution_mode() {
+        let presets = normalize_enabled_ai_pool_presets(
+            &[
+                AiPoolSchedulingPreset {
+                    preset: "lru".to_string(),
+                    enabled: true,
+                    mode: None,
+                },
+                AiPoolSchedulingPreset {
+                    preset: "cache_affinity".to_string(),
+                    enabled: true,
+                    mode: None,
+                },
+                AiPoolSchedulingPreset {
+                    preset: "priority_first".to_string(),
+                    enabled: true,
+                    mode: None,
+                },
+            ],
+            "openai",
+        );
+
+        assert_eq!(presets, ["priority_first"]);
+    }
+
     fn sample_candidate(
         provider_id: &str,
         endpoint_id: &str,

crates/aether-data-contracts/src/repository/candidate_selection/mod.rs

@@ -2,6 +2,7 @@ mod types;
 
 pub use types::{
     MinimalCandidateSelectionReadRepository, MinimalCandidateSelectionRepository,
-    StoredMinimalCandidateSelectionRow, StoredPoolKeyCandidateRowsQuery,
+    StoredMinimalCandidateSelectionRow, StoredPoolKeyCandidateOrder,
-    StoredProviderModelMapping, StoredRequestedModelCandidateRowsQuery,
+    StoredPoolKeyCandidateRowsQuery, StoredProviderModelMapping,
+    StoredRequestedModelCandidateRowsQuery,
 };

crates/aether-data-contracts/src/repository/candidate_selection/types.rs

@@ -42,6 +42,18 @@ pub struct StoredMinimalCandidateSelectionRow {
     pub model_is_available: bool,
 }
 
+#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum StoredPoolKeyCandidateOrder {
+    #[default]
+    InternalPriority,
+    Lru,
+    CacheAffinity,
+    SingleAccount,
+    LoadBalance {
+        seed: String,
+    },
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct StoredPoolKeyCandidateRowsQuery {
     pub api_format: String,
@@ -49,6 +61,8 @@ pub struct StoredPoolKeyCandidateRowsQuery {
     pub endpoint_id: String,
     pub model_id: String,
     pub selected_provider_model_name: String,
+    #[serde(default)]
+    pub order: StoredPoolKeyCandidateOrder,
     pub offset: u32,
     pub limit: u32,
 }

crates/aether-data/src/repository/candidate_selection/memory.rs

@@ -4,7 +4,8 @@ use async_trait::async_trait;
 
 use super::{
     MinimalCandidateSelectionReadRepository, StoredMinimalCandidateSelectionRow,
-    StoredPoolKeyCandidateRowsQuery, StoredRequestedModelCandidateRowsQuery,
+    StoredPoolKeyCandidateOrder, StoredPoolKeyCandidateRowsQuery,
+    StoredRequestedModelCandidateRowsQuery,
 };
 use crate::DataLayerError;
 
@@ -130,11 +131,7 @@ impl MinimalCandidateSelectionReadRepository for InMemoryMinimalCandidateSelecti
                     && row.model_id == query.model_id
             })
             .collect::<Vec<_>>();
-        rows.sort_by(|left, right| {
+        sort_pool_key_rows(&mut rows, &query.order);
-            left.key_internal_priority
-                .cmp(&right.key_internal_priority)
-                .then(left.key_id.cmp(&right.key_id))
-        });
         Ok(rows
             .into_iter()
             .skip(query.offset as usize)
@@ -143,6 +140,38 @@ impl MinimalCandidateSelectionReadRepository for InMemoryMinimalCandidateSelecti
     }
 }
 
+fn sort_pool_key_rows(
+    rows: &mut [StoredMinimalCandidateSelectionRow],
+    order: &StoredPoolKeyCandidateOrder,
+) {
+    rows.sort_by(|left, right| match order {
+        StoredPoolKeyCandidateOrder::LoadBalance { seed } => {
+            stable_pool_key_hash(seed.as_str(), left.key_id.as_str())
+                .cmp(&stable_pool_key_hash(seed.as_str(), right.key_id.as_str()))
+                .then(left.key_id.cmp(&right.key_id))
+        }
+        _ => left
+            .key_internal_priority
+            .cmp(&right.key_internal_priority)
+            .then(left.key_id.cmp(&right.key_id)),
+    });
+}
+
+fn stable_pool_key_hash(seed: &str, key_id: &str) -> u64 {
+    let mut hash = 0xcbf29ce484222325u64;
+    for byte in seed
+        .as_bytes()
+        .iter()
+        .copied()
+        .chain(std::iter::once(b':'))
+        .chain(key_id.as_bytes().iter().copied())
+    {
+        hash ^= u64::from(byte);
+        hash = hash.wrapping_mul(0x100000001b3);
+    }
+    hash
+}
+
 fn normalize_api_format(value: &str) -> String {
     aether_ai_formats::normalize_api_format_alias(value)
 }
@@ -215,7 +244,8 @@ mod tests {
     use super::InMemoryMinimalCandidateSelectionReadRepository;
     use crate::repository::candidate_selection::{
         MinimalCandidateSelectionReadRepository, StoredMinimalCandidateSelectionRow,
-        StoredPoolKeyCandidateRowsQuery, StoredRequestedModelCandidateRowsQuery,
+        StoredPoolKeyCandidateOrder, StoredPoolKeyCandidateRowsQuery,
+        StoredRequestedModelCandidateRowsQuery,
     };
 
     fn sample_row(
@@ -402,6 +432,7 @@ mod tests {
                 endpoint_id: "endpoint-pool".to_string(),
                 model_id: "model-pool".to_string(),
                 selected_provider_model_name: "gpt-5".to_string(),
+                order: StoredPoolKeyCandidateOrder::InternalPriority,
                 offset: 2,
                 limit: 2,
             })

crates/aether-data/src/repository/candidate_selection/mod.rs

@@ -6,8 +6,9 @@ mod sqlite;
 #[allow(unused_imports)]
 pub(crate) use aether_data_contracts::repository::candidate_selection::{
     MinimalCandidateSelectionReadRepository, MinimalCandidateSelectionRepository,
-    StoredMinimalCandidateSelectionRow, StoredPoolKeyCandidateRowsQuery,
+    StoredMinimalCandidateSelectionRow, StoredPoolKeyCandidateOrder,
-    StoredProviderModelMapping, StoredRequestedModelCandidateRowsQuery,
+    StoredPoolKeyCandidateRowsQuery, StoredProviderModelMapping,
+    StoredRequestedModelCandidateRowsQuery,
 };
 pub use memory::InMemoryMinimalCandidateSelectionReadRepository;
 pub use mysql::MysqlMinimalCandidateSelectionReadRepository;

crates/aether-data/src/repository/candidate_selection/mysql.rs

@@ -5,7 +5,7 @@ use sqlx::{mysql::MySqlRow, MySql, QueryBuilder, Row};
 
 use super::{
     MinimalCandidateSelectionReadRepository, StoredMinimalCandidateSelectionRow,
-    StoredPoolKeyCandidateRowsQuery, StoredProviderModelMapping,
+    StoredPoolKeyCandidateOrder, StoredPoolKeyCandidateRowsQuery, StoredProviderModelMapping,
     StoredRequestedModelCandidateRowsQuery,
 };
 use crate::driver::mysql::MysqlPool;
@@ -35,6 +35,7 @@ SELECT
   pak.capabilities AS key_capabilities,
   pak.internal_priority AS key_internal_priority,
   pak.global_priority_by_format AS key_global_priority_by_format,
+  pak.last_used_at AS key_last_used_at_unix_secs,
   m.id AS model_id,
   m.global_model_id AS global_model_id,
   gm.name AS global_model_name,
@@ -67,6 +68,7 @@ struct CandidateSelectionRow {
     row: StoredMinimalCandidateSelectionRow,
     provider_pool_enabled: bool,
     key_auth_config: Option<String>,
+    key_last_used_at_unix_secs: Option<u64>,
 }
 
 impl MysqlMinimalCandidateSelectionReadRepository {
@@ -180,18 +182,18 @@ impl MinimalCandidateSelectionReadRepository for MysqlMinimalCandidateSelectionR
             .load_rows_for_api_format(&query.api_format)
             .await?
             .into_iter()
-            .map(|item| item.row)
             .filter(|row| {
-                row.provider_id == query.provider_id
+                row.row.provider_id == query.provider_id
-                    && row.endpoint_id == query.endpoint_id
+                    && row.row.endpoint_id == query.endpoint_id
-                    && row.model_id == query.model_id
+                    && row.row.model_id == query.model_id
             })
             .collect::<Vec<_>>();
-        let mut rows = sort_pool_key_rows(rows);
+        let mut rows = sort_pool_key_rows(rows, &query.order);
         Ok(rows
             .drain(..)
             .skip(query.offset as usize)
             .take(query.limit as usize)
+            .map(|item| item.row)
             .collect())
     }
 }
@@ -246,16 +248,66 @@ fn sort_rows(
 }
 
 fn sort_pool_key_rows(
-    mut rows: Vec<StoredMinimalCandidateSelectionRow>,
+    mut rows: Vec<CandidateSelectionRow>,
-) -> Vec<StoredMinimalCandidateSelectionRow> {
+    order: &StoredPoolKeyCandidateOrder,
-    rows.sort_by(|left, right| {
+) -> Vec<CandidateSelectionRow> {
-        left.key_internal_priority
+    rows.sort_by(|left, right| match order {
-            .cmp(&right.key_internal_priority)
+        StoredPoolKeyCandidateOrder::InternalPriority => compare_pool_key_internal(left, right),
-            .then(left.key_id.cmp(&right.key_id))
+        StoredPoolKeyCandidateOrder::Lru => left
+            .key_last_used_at_unix_secs
+            .cmp(&right.key_last_used_at_unix_secs)
+            .then_with(|| compare_pool_key_internal(left, right)),
+        StoredPoolKeyCandidateOrder::CacheAffinity => right
+            .key_last_used_at_unix_secs
+            .cmp(&left.key_last_used_at_unix_secs)
+            .then_with(|| compare_pool_key_internal(left, right)),
+        StoredPoolKeyCandidateOrder::SingleAccount => left
+            .row
+            .key_internal_priority
+            .cmp(&right.row.key_internal_priority)
+            .then_with(|| {
+                right
+                    .key_last_used_at_unix_secs
+                    .cmp(&left.key_last_used_at_unix_secs)
+            })
+            .then(left.row.key_id.cmp(&right.row.key_id)),
+        StoredPoolKeyCandidateOrder::LoadBalance { seed } => {
+            stable_pool_key_hash(seed.as_str(), left.row.key_id.as_str())
+                .cmp(&stable_pool_key_hash(
+                    seed.as_str(),
+                    right.row.key_id.as_str(),
+                ))
+                .then(left.row.key_id.cmp(&right.row.key_id))
+        }
     });
     rows
 }
 
+fn compare_pool_key_internal(
+    left: &CandidateSelectionRow,
+    right: &CandidateSelectionRow,
+) -> std::cmp::Ordering {
+    left.row
+        .key_internal_priority
+        .cmp(&right.row.key_internal_priority)
+        .then(left.row.key_id.cmp(&right.row.key_id))
+}
+
+fn stable_pool_key_hash(seed: &str, key_id: &str) -> u64 {
+    let mut hash = 0xcbf29ce484222325u64;
+    for byte in seed
+        .as_bytes()
+        .iter()
+        .copied()
+        .chain(std::iter::once(b':'))
+        .chain(key_id.as_bytes().iter().copied())
+    {
+        hash ^= u64::from(byte);
+        hash = hash.wrapping_mul(0x100000001b3);
+    }
+    hash
+}
+
 fn row_matches_requested_model(
     row: &StoredMinimalCandidateSelectionRow,
     requested_model_name: &str,
@@ -394,6 +446,10 @@ fn map_candidate_selection_row(row: &MySqlRow) -> Result<CandidateSelectionRow,
         },
         provider_pool_enabled,
         key_auth_config: row.try_get("key_auth_config").map_sql_err()?,
+        key_last_used_at_unix_secs: row
+            .try_get::<Option<i64>, _>("key_last_used_at_unix_secs")
+            .map_sql_err()?
+            .and_then(|value| u64::try_from(value).ok()),
     })
 }
 

crates/aether-data/src/repository/candidate_selection/postgres.rs

@@ -5,7 +5,7 @@ use std::collections::BTreeSet;
 
 use super::{
     MinimalCandidateSelectionReadRepository, StoredMinimalCandidateSelectionRow,
-    StoredPoolKeyCandidateRowsQuery, StoredProviderModelMapping,
+    StoredPoolKeyCandidateOrder, StoredPoolKeyCandidateRowsQuery, StoredProviderModelMapping,
     StoredRequestedModelCandidateRowsQuery,
 };
 use crate::{error::SqlxResultExt, DataLayerError};
@@ -505,6 +505,36 @@ LIMIT $7
 OFFSET $8
 "#;
 
+fn pool_key_candidate_order_by_sql(order: &StoredPoolKeyCandidateOrder) -> &'static str {
+    match order {
+        StoredPoolKeyCandidateOrder::InternalPriority => {
+            "ORDER BY\n  pak.internal_priority ASC,\n  pak.id ASC"
+        }
+        StoredPoolKeyCandidateOrder::Lru => {
+            "ORDER BY\n  pak.last_used_at ASC NULLS FIRST,\n  pak.internal_priority ASC,\n  pak.id ASC"
+        }
+        StoredPoolKeyCandidateOrder::CacheAffinity => {
+            "ORDER BY\n  pak.last_used_at DESC NULLS LAST,\n  pak.internal_priority ASC,\n  pak.id ASC"
+        }
+        StoredPoolKeyCandidateOrder::SingleAccount => {
+            "ORDER BY\n  pak.internal_priority ASC,\n  pak.last_used_at DESC NULLS LAST,\n  pak.id ASC"
+        }
+        StoredPoolKeyCandidateOrder::LoadBalance { .. } => {
+            "ORDER BY\n  md5($9 || ':' || pak.id) ASC,\n  pak.id ASC"
+        }
+    }
+}
+
+fn pool_key_candidate_selection_sql(order: &StoredPoolKeyCandidateOrder) -> String {
+    let default_order =
+        "ORDER BY\n  pak.internal_priority ASC,\n  pak.id ASC\nLIMIT $7\nOFFSET $8\n";
+    let replacement = format!(
+        "{}\nLIMIT $7\nOFFSET $8\n",
+        pool_key_candidate_order_by_sql(order)
+    );
+    LIST_POOL_KEYS_FOR_GROUP_SQL.replace(default_order, &replacement)
+}
+
 #[derive(Debug, Clone)]
 pub struct SqlxMinimalCandidateSelectionReadRepository {
     pool: PgPool,
@@ -650,19 +680,23 @@ impl SqlxMinimalCandidateSelectionReadRepository {
         let sql_match_aliases = sql_match_aliases(&storage_aliases);
         let limit = i64::from(query.limit.max(1));
         let offset = i64::from(query.offset);
+        let sql = pool_key_candidate_selection_sql(&query.order);
         for api_format in storage_aliases {
+            let mut query_builder = sqlx::query(sql.as_str())
+                .bind(api_format)
+                .bind(query.provider_id.as_str())
+                .bind(query.endpoint_id.as_str())
+                .bind(query.model_id.as_str())
+                .bind(sql_match_aliases.clone())
+                .bind(canonical_api_format.clone())
+                .bind(limit)
+                .bind(offset);
+            if let StoredPoolKeyCandidateOrder::LoadBalance { seed } = &query.order {
+                query_builder = query_builder.bind(seed.as_str());
+            }
             rows.extend(
                 Self::collect_query_rows(
-                    sqlx::query(LIST_POOL_KEYS_FOR_GROUP_SQL)
+                    query_builder.fetch(&self.pool),
-                        .bind(api_format)
-                        .bind(query.provider_id.as_str())
-                        .bind(query.endpoint_id.as_str())
-                        .bind(query.model_id.as_str())
-                        .bind(sql_match_aliases.clone())
-                        .bind(canonical_api_format.clone())
-                        .bind(limit)
-                        .bind(offset)
-                        .fetch(&self.pool),
                     map_candidate_selection_row,
                 )
                 .await?,
@@ -1039,13 +1073,16 @@ mod tests {
     use serde_json::json;
 
     use super::{
-        parse_provider_model_mappings, parse_string_list, requested_model_selection_page_sql,
+        parse_provider_model_mappings, parse_string_list, pool_key_candidate_selection_sql,
-        requested_model_selection_sql, SqlxMinimalCandidateSelectionReadRepository,
+        requested_model_selection_page_sql, requested_model_selection_sql,
+        SqlxMinimalCandidateSelectionReadRepository,
         LIST_FOR_EXACT_API_FORMAT_AND_GLOBAL_MODEL_SQL, LIST_FOR_EXACT_API_FORMAT_SQL,
         LIST_POOL_KEYS_FOR_GROUP_SQL,
     };
     use crate::driver::postgres::{PostgresPoolConfig, PostgresPoolFactory};
-    use crate::repository::candidate_selection::StoredProviderModelMapping;
+    use crate::repository::candidate_selection::{
+        StoredPoolKeyCandidateOrder, StoredProviderModelMapping,
+    };
 
     #[tokio::test]
     async fn repository_constructs_from_lazy_pool() {
@@ -1100,6 +1137,23 @@ mod tests {
         assert!(sql.ends_with("LIMIT $5\nOFFSET $6"));
     }
 
+    #[test]
+    fn pool_key_selection_sql_applies_query_order() {
+        let load_balance_sql =
+            pool_key_candidate_selection_sql(&StoredPoolKeyCandidateOrder::LoadBalance {
+                seed: "seed".to_string(),
+            });
+        assert!(load_balance_sql.contains("md5($9 || ':' || pak.id) ASC"));
+        assert!(load_balance_sql.ends_with("LIMIT $7\nOFFSET $8\n"));
+
+        let lru_sql = pool_key_candidate_selection_sql(&StoredPoolKeyCandidateOrder::Lru);
+        assert!(lru_sql.contains("pak.last_used_at ASC NULLS FIRST"));
+
+        let cache_affinity_sql =
+            pool_key_candidate_selection_sql(&StoredPoolKeyCandidateOrder::CacheAffinity);
+        assert!(cache_affinity_sql.contains("pak.last_used_at DESC NULLS LAST"));
+    }
+
     #[test]
     fn parse_string_list_accepts_stringified_array() {
         let parsed = parse_string_list(

crates/aether-data/src/repository/candidate_selection/sqlite.rs

@@ -5,7 +5,7 @@ use sqlx::{sqlite::SqliteRow, QueryBuilder, Row, Sqlite};
 
 use super::{
     MinimalCandidateSelectionReadRepository, StoredMinimalCandidateSelectionRow,
-    StoredPoolKeyCandidateRowsQuery, StoredProviderModelMapping,
+    StoredPoolKeyCandidateOrder, StoredPoolKeyCandidateRowsQuery, StoredProviderModelMapping,
     StoredRequestedModelCandidateRowsQuery,
 };
 use crate::driver::sqlite::SqlitePool;
@@ -35,6 +35,7 @@ SELECT
   pak.capabilities AS key_capabilities,
   pak.internal_priority AS key_internal_priority,
   pak.global_priority_by_format AS key_global_priority_by_format,
+  pak.last_used_at AS key_last_used_at_unix_secs,
   m.id AS model_id,
   m.global_model_id AS global_model_id,
   gm.name AS global_model_name,
@@ -67,6 +68,7 @@ struct CandidateSelectionRow {
     row: StoredMinimalCandidateSelectionRow,
     provider_pool_enabled: bool,
     key_auth_config: Option<String>,
+    key_last_used_at_unix_secs: Option<u64>,
 }
 
 impl SqliteMinimalCandidateSelectionReadRepository {
@@ -180,18 +182,18 @@ impl MinimalCandidateSelectionReadRepository for SqliteMinimalCandidateSelection
             .load_rows_for_api_format(&query.api_format)
             .await?
             .into_iter()
-            .map(|item| item.row)
             .filter(|row| {
-                row.provider_id == query.provider_id
+                row.row.provider_id == query.provider_id
-                    && row.endpoint_id == query.endpoint_id
+                    && row.row.endpoint_id == query.endpoint_id
-                    && row.model_id == query.model_id
+                    && row.row.model_id == query.model_id
             })
             .collect::<Vec<_>>();
-        let mut rows = sort_pool_key_rows(rows);
+        let mut rows = sort_pool_key_rows(rows, &query.order);
         Ok(rows
             .drain(..)
             .skip(query.offset as usize)
             .take(query.limit as usize)
+            .map(|item| item.row)
             .collect())
     }
 }
@@ -246,16 +248,66 @@ fn sort_rows(
 }
 
 fn sort_pool_key_rows(
-    mut rows: Vec<StoredMinimalCandidateSelectionRow>,
+    mut rows: Vec<CandidateSelectionRow>,
-) -> Vec<StoredMinimalCandidateSelectionRow> {
+    order: &StoredPoolKeyCandidateOrder,
-    rows.sort_by(|left, right| {
+) -> Vec<CandidateSelectionRow> {
-        left.key_internal_priority
+    rows.sort_by(|left, right| match order {
-            .cmp(&right.key_internal_priority)
+        StoredPoolKeyCandidateOrder::InternalPriority => compare_pool_key_internal(left, right),
-            .then(left.key_id.cmp(&right.key_id))
+        StoredPoolKeyCandidateOrder::Lru => left
+            .key_last_used_at_unix_secs
+            .cmp(&right.key_last_used_at_unix_secs)
+            .then_with(|| compare_pool_key_internal(left, right)),
+        StoredPoolKeyCandidateOrder::CacheAffinity => right
+            .key_last_used_at_unix_secs
+            .cmp(&left.key_last_used_at_unix_secs)
+            .then_with(|| compare_pool_key_internal(left, right)),
+        StoredPoolKeyCandidateOrder::SingleAccount => left
+            .row
+            .key_internal_priority
+            .cmp(&right.row.key_internal_priority)
+            .then_with(|| {
+                right
+                    .key_last_used_at_unix_secs
+                    .cmp(&left.key_last_used_at_unix_secs)
+            })
+            .then(left.row.key_id.cmp(&right.row.key_id)),
+        StoredPoolKeyCandidateOrder::LoadBalance { seed } => {
+            stable_pool_key_hash(seed.as_str(), left.row.key_id.as_str())
+                .cmp(&stable_pool_key_hash(
+                    seed.as_str(),
+                    right.row.key_id.as_str(),
+                ))
+                .then(left.row.key_id.cmp(&right.row.key_id))
+        }
     });
     rows
 }
 
+fn compare_pool_key_internal(
+    left: &CandidateSelectionRow,
+    right: &CandidateSelectionRow,
+) -> std::cmp::Ordering {
+    left.row
+        .key_internal_priority
+        .cmp(&right.row.key_internal_priority)
+        .then(left.row.key_id.cmp(&right.row.key_id))
+}
+
+fn stable_pool_key_hash(seed: &str, key_id: &str) -> u64 {
+    let mut hash = 0xcbf29ce484222325u64;
+    for byte in seed
+        .as_bytes()
+        .iter()
+        .copied()
+        .chain(std::iter::once(b':'))
+        .chain(key_id.as_bytes().iter().copied())
+    {
+        hash ^= u64::from(byte);
+        hash = hash.wrapping_mul(0x100000001b3);
+    }
+    hash
+}
+
 fn row_matches_requested_model(
     row: &StoredMinimalCandidateSelectionRow,
     requested_model_name: &str,
@@ -394,6 +446,10 @@ fn map_candidate_selection_row(row: &SqliteRow) -> Result<CandidateSelectionRow,
         },
         provider_pool_enabled,
         key_auth_config: row.try_get("key_auth_config").map_sql_err()?,
+        key_last_used_at_unix_secs: row
+            .try_get::<Option<i64>, _>("key_last_used_at_unix_secs")
+            .map_sql_err()?
+            .and_then(|value| u64::try_from(value).ok()),
     })
 }
 
@@ -620,8 +676,8 @@ mod tests {
     use super::SqliteMinimalCandidateSelectionReadRepository;
     use crate::lifecycle::migrate::run_sqlite_migrations;
     use crate::repository::candidate_selection::{
-        MinimalCandidateSelectionReadRepository, StoredPoolKeyCandidateRowsQuery,
+        MinimalCandidateSelectionReadRepository, StoredPoolKeyCandidateOrder,
-        StoredRequestedModelCandidateRowsQuery,
+        StoredPoolKeyCandidateRowsQuery, StoredRequestedModelCandidateRowsQuery,
     };
 
     #[tokio::test]
@@ -673,6 +729,7 @@ mod tests {
                 endpoint_id: "endpoint-1".to_string(),
                 model_id: "model-1".to_string(),
                 selected_provider_model_name: "provider-model".to_string(),
+                order: StoredPoolKeyCandidateOrder::InternalPriority,
                 offset: 1,
                 limit: 1,
             })