diff --git a/apps/aether-gateway/src/maintenance/runtime/oauth_token_refresh.rs b/apps/aether-gateway/src/maintenance/runtime/oauth_token_refresh.rs
index 93629953f..15d1a9fa4 100644
--- a/apps/aether-gateway/src/maintenance/runtime/oauth_token_refresh.rs
+++ b/apps/aether-gateway/src/maintenance/runtime/oauth_token_refresh.rs
@@ -51,9 +51,15 @@ pub(crate) async fn perform_oauth_token_refresh_once(
     let endpoints = state
         .list_provider_catalog_endpoints_by_provider_ids(&provider_ids)
         .await?;
+    // Read the catalog rows without opening/decrypting credentials in bulk.
+    // A single legacy/plaintext row must not abort refresh for every healthy
+    // key, and this maintenance scan must not trigger the normal lazy v2
+    // credential rewrite path. Each candidate is opened in isolation below.
     let keys = state
+        .data
         .list_provider_catalog_keys_by_provider_ids(&provider_ids)
-        .await?;
+        .await
+        .map_err(|err| GatewayError::Internal(err.to_string()))?;
     let endpoints_by_provider = group_endpoints_by_provider(endpoints);
     let keys_by_provider = group_keys_by_provider(keys);
     let mut summary = OAuthTokenRefreshRunSummary::default();
@@ -85,12 +91,31 @@ pub(crate) async fn perform_oauth_token_refresh_once(
                 continue;
             };
 
-            let Some(transport) = state
+            let transport = match state
                 .read_provider_transport_snapshot(&provider.id, &endpoint.id, &key.id)
-                .await?
-            else {
-                summary.skipped = summary.skipped.saturating_add(1);
-                continue;
+                .await
+            {
+                Ok(Some(transport)) => transport,
+                Ok(None) => {
+                    summary.skipped = summary.skipped.saturating_add(1);
+                    continue;
+                }
+                Err(err) if is_nonfatal_legacy_credential_error(&err) => {
+                    // Keep malformed historical credentials untouched. They
+                    // are intentionally skipped while other keys continue.
+                    summary.skipped = summary.skipped.saturating_add(1);
+                    warn!(
+                        event_name = "oauth_token_refresh_skipped_invalid_credential",
+                        log_type = "ops",
+                        worker = "oauth_token_refresh",
+                        provider_id = %provider.id,
+                        key_id = %key.id,
+                        reason = "invalid_stored_credential",
+                        "gateway skipped oauth refresh for an invalid stored credential"
+                    );
+                    continue;
+                }
+                Err(err) => return Err(err),
             };
             let is_agent_identity =
                 crate::provider_transport::is_codex_agent_identity_transport(&transport);
@@ -236,8 +261,10 @@ async fn provider_key_credentials_changed(
     before: &StoredProviderCatalogKey,
 ) -> Result<bool, GatewayError> {
     let Some(after) = state
+        .data
         .list_provider_catalog_keys_by_ids(std::slice::from_ref(&before.id))
-        .await?
+        .await
+        .map_err(|err| GatewayError::Internal(err.to_string()))?
         .into_iter()
         .next()
     else {
@@ -274,6 +301,29 @@ fn now_unix_secs() -> u64 {
         .unwrap_or_default()
 }
 
+/// Credential decoding errors are expected for rows written by older
+/// versions of the service. They are non-fatal for a best-effort maintenance
+/// scan, but normal request/admin paths still fail closed on the same error.
+fn is_nonfatal_legacy_credential_error(error: &GatewayError) -> bool {
+    let GatewayError::Internal(message) = error else {
+        return false;
+    };
+    let message = message.to_ascii_lowercase();
+    // Missing encryption configuration is an operational failure and must
+    // remain fail-closed.  Only errors that identify a stored field or a
+    // malformed legacy ciphertext are safe to isolate to one key.
+    if message.contains("encryption key is not configured") {
+        return false;
+    }
+    message.contains("provider_api_keys.api_key")
+        || message.contains("provider_api_keys.auth_config")
+        || message.contains("legacy provider catalog credential")
+        || message.contains("provider catalog credential is not an authenticated ciphertext")
+        || message.contains("provider catalog credential contains reserved framing")
+        || message.contains("provider catalog credential authentication failed")
+        || message.contains("provider catalog credential envelope")
+}
+
 #[cfg(test)]
 mod tests {
     use aether_data_contracts::repository::provider_catalog::{
@@ -281,8 +331,10 @@ mod tests {
     };
 
     use super::{
-        agent_identity_needs_task_recovery, auth_config_has_refresh_token, oauth_refresh_candidate,
+        agent_identity_needs_task_recovery, auth_config_has_refresh_token,
+        is_nonfatal_legacy_credential_error, oauth_refresh_candidate,
     };
+    use crate::GatewayError;
 
     #[test]
     fn legacy_antigravity_refresh_token_is_refreshable() {
@@ -335,4 +387,27 @@ mod tests {
             Some("[REFRESH_FAILED] temporary"),
         ));
     }
+
+    #[test]
+    fn only_stored_provider_credential_errors_are_non_fatal() {
+        assert!(is_nonfatal_legacy_credential_error(
+            &GatewayError::Internal(
+                "provider catalog credential is not an authenticated ciphertext".to_string(),
+            )
+        ));
+        assert!(is_nonfatal_legacy_credential_error(
+            &GatewayError::Internal(
+                "provider_api_keys.auth_config has an invalid provider catalog credential envelope"
+                    .to_string(),
+            )
+        ));
+        assert!(!is_nonfatal_legacy_credential_error(
+            &GatewayError::Internal("postgres error: connection refused".to_string(),)
+        ));
+        assert!(!is_nonfatal_legacy_credential_error(
+            &GatewayError::Internal(
+                "provider catalog credential encryption key is not configured".to_string(),
+            )
+        ));
+    }
 }
diff --git a/apps/aether-gateway/src/model_fetch/runtime.rs b/apps/aether-gateway/src/model_fetch/runtime.rs
index 9af23e144..ce6cde53a 100644
--- a/apps/aether-gateway/src/model_fetch/runtime.rs
+++ b/apps/aether-gateway/src/model_fetch/runtime.rs
@@ -153,12 +153,10 @@ where
             .push(endpoint);
     }
     let mut keys_by_provider = HashMap::<String, Vec<StoredProviderCatalogKey>>::new();
-    for key in <S as ModelFetchAssociationStore>::list_provider_catalog_keys_by_provider_ids(
-        state,
-        &provider_ids,
-    )
-    .await
-    .map_err(GatewayError::Internal)?
+    for key in state
+        .list_provider_catalog_keys_for_model_fetch(&provider_ids)
+        .await
+        .map_err(GatewayError::Internal)?
     {
         keys_by_provider
             .entry(key.provider_id.clone())
@@ -180,6 +178,7 @@ where
                 continue;
             }
             let selected_endpoints = selected_models_fetch_endpoints(&endpoints, &key);
+            let key = sanitize_model_fetch_key(key);
             targets.push(SelectedFetchTarget {
                 provider: provider.clone(),
                 key,
@@ -190,6 +189,56 @@ where
     Ok(targets)
 }
 
+/// Keep only the key metadata needed after target collection. Raw catalog
+/// rows contain encrypted credentials and transport secrets; model discovery
+/// reopens a single snapshot by id when it actually needs to make a request.
+fn sanitize_model_fetch_key(mut key: StoredProviderCatalogKey) -> StoredProviderCatalogKey {
+    // `SelectedFetchTarget` lives across endpoint selection and the complete
+    // fetch/persist operation.  Keep only fields consumed by that operation;
+    // in particular, do not retain historical diagnostics, scheduling state,
+    // usage counters, or transport configuration copied from a raw database
+    // row.  The actual credential/proxy snapshot is reopened by id for one
+    // endpoint at a time.
+    key.capabilities = None;
+    key.auth_type_by_format = None;
+    key.allow_auth_channel_mismatch_formats = None;
+    key.encrypted_api_key = None;
+    key.encrypted_auth_config = None;
+    key.note = None;
+    key.internal_priority = 0;
+    key.rate_multipliers = None;
+    key.global_priority_by_format = None;
+    key.expires_at_unix_secs = None;
+    key.cache_ttl_minutes = 0;
+    key.max_probe_interval_minutes = 0;
+    key.proxy = None;
+    key.fingerprint = None;
+    key.rpm_limit = None;
+    key.concurrent_limit = None;
+    key.learned_rpm_limit = None;
+    key.concurrent_429_count = None;
+    key.rpm_429_count = None;
+    key.last_429_at_unix_secs = None;
+    key.last_429_type = None;
+    key.adjustment_history = None;
+    key.utilization_samples = None;
+    key.last_probe_increase_at_unix_secs = None;
+    key.last_rpm_peak = None;
+    key.request_count = None;
+    key.total_tokens = 0;
+    key.total_cost_usd = 0.0;
+    key.success_count = None;
+    key.error_count = None;
+    key.total_response_time_ms = None;
+    key.last_used_at_unix_secs = None;
+    key.last_models_fetch_at_unix_secs = None;
+    key.last_models_fetch_error = None;
+    key.oauth_invalid_at_unix_secs = None;
+    key.oauth_invalid_reason = None;
+    key.status_snapshot = None;
+    key
+}
+
 async fn execute_fetch_targets<S>(
     state: &S,
     targets: Vec<SelectedFetchTarget>,
@@ -293,13 +342,14 @@ async fn fetch_and_persist_key_models(
     }
 
     let mut transports = Vec::new();
+    let mut skipped_invalid_credential = false;
     for endpoint in &target.endpoints {
         match state
             .read_provider_transport_snapshot(&target.provider.id, &endpoint.id, &target.key.id)
-            .await?
+            .await
         {
-            Some(transport) => transports.push(transport),
-            None => {
+            Ok(Some(transport)) => transports.push(transport),
+            Ok(None) => {
                 warn!(
                     provider_id = %target.provider.id,
                     endpoint_id = %endpoint.id,
@@ -307,9 +357,29 @@ async fn fetch_and_persist_key_models(
                     "gateway model fetch transport snapshot unavailable"
                 );
             }
+            Err(error) if is_nonfatal_legacy_credential_error(&error) => {
+                skipped_invalid_credential = true;
+                warn!(
+                    event_name = "model_fetch_skipped_invalid_credential",
+                    log_type = "ops",
+                    provider_id = %target.provider.id,
+                    endpoint_id = %endpoint.id,
+                    key_id = %target.key.id,
+                    reason = "invalid_stored_credential",
+                    "gateway skipped model fetch for an invalid stored credential"
+                );
+            }
+            Err(error) => return Err(error),
         }
     }
 
+    // A malformed legacy credential is isolated to its key. Do not turn it
+    // into a cycle-wide failure (or rewrite the row merely to record a fetch
+    // error), and let other eligible keys continue through the worker.
+    if transports.is_empty() && skipped_invalid_credential {
+        return Ok(KeyFetchDisposition::Skipped);
+    }
+
     if transports.is_empty() {
         persist_key_fetch_failure(
             state,
@@ -482,6 +552,30 @@ pub(crate) fn safe_model_fetch_error(error: &str) -> String {
     "Upstream models fetch failed".to_string()
 }
 
+/// Credential decoding failures from old catalog rows are isolated by the
+/// background model-fetch worker. Normal request/admin paths remain fail
+/// closed; this predicate only controls whether one maintenance item may be
+/// skipped without aborting the whole cycle.
+fn is_nonfatal_legacy_credential_error(error: &GatewayError) -> bool {
+    let GatewayError::Internal(message) = error else {
+        return false;
+    };
+    let message = message.to_ascii_lowercase();
+    // Missing encryption configuration is an operational failure and must
+    // remain fail-closed.  Only errors that identify a stored field or a
+    // malformed legacy ciphertext are safe to isolate to one key.
+    if message.contains("encryption key is not configured") {
+        return false;
+    }
+    message.contains("provider_api_keys.api_key")
+        || message.contains("provider_api_keys.auth_config")
+        || message.contains("legacy provider catalog credential")
+        || message.contains("provider catalog credential is not an authenticated ciphertext")
+        || message.contains("provider catalog credential contains reserved framing")
+        || message.contains("provider catalog credential authentication failed")
+        || message.contains("provider catalog credential envelope")
+}
+
 fn model_fetch_error_http_status(error: &str) -> Option<u16> {
     [
         "http ",
@@ -553,7 +647,8 @@ fn now_unix_secs() -> u64 {
 #[cfg(test)]
 mod tests {
     use super::{
-        perform_model_fetch_once_with_state, safe_model_fetch_error, state::ModelFetchRuntimeState,
+        perform_model_fetch_once_with_state, safe_model_fetch_error, sanitize_model_fetch_key,
+        state::ModelFetchRuntimeState,
     };
     use aether_contracts::{ExecutionPlan, ExecutionResult, ProxySnapshot};
     use aether_data_contracts::repository::global_models::{
@@ -585,6 +680,7 @@ mod tests {
         endpoints: Arc<Vec<StoredProviderCatalogEndpoint>>,
         keys: Arc<Mutex<Vec<StoredProviderCatalogKey>>>,
         transports: Arc<HashMap<(String, String, String), GatewayProviderTransportSnapshot>>,
+        transport_errors: Arc<HashMap<(String, String, String), String>>,
         execution_results: Arc<Mutex<VecDeque<ExecutionResult>>>,
         executed_plans: Arc<Mutex<Vec<ExecutionPlan>>>,
         cached_models: Arc<Mutex<HashMap<(String, String), Vec<Value>>>>,
@@ -604,6 +700,7 @@ mod tests {
                 endpoints: Arc::new(endpoints),
                 keys: Arc::new(Mutex::new(keys)),
                 transports: Arc::new(transports),
+                transport_errors: Arc::new(HashMap::new()),
                 execution_results: Arc::new(Mutex::new(VecDeque::from(execution_results))),
                 executed_plans: Arc::new(Mutex::new(Vec::new())),
                 cached_models: Arc::new(Mutex::new(HashMap::new())),
@@ -611,6 +708,14 @@ mod tests {
             }
         }
 
+        fn with_transport_errors(
+            mut self,
+            transport_errors: HashMap<(String, String, String), String>,
+        ) -> Self {
+            self.transport_errors = Arc::new(transport_errors);
+            self
+        }
+
         fn key(&self, key_id: &str) -> StoredProviderCatalogKey {
             self.keys
                 .lock()
@@ -758,6 +863,13 @@ mod tests {
             endpoint_id: &str,
             key_id: &str,
         ) -> Result<Option<GatewayProviderTransportSnapshot>, GatewayError> {
+            if let Some(error) = self.transport_errors.get(&(
+                provider_id.to_string(),
+                endpoint_id.to_string(),
+                key_id.to_string(),
+            )) {
+                return Err(GatewayError::Internal(error.clone()));
+            }
             Ok(self
                 .transports
                 .get(&(
@@ -1044,6 +1156,15 @@ mod tests {
         }
     }
 
+    #[test]
+    fn invalid_credential_classifier_does_not_swallow_missing_key_configuration() {
+        assert!(!super::is_nonfatal_legacy_credential_error(
+            &GatewayError::Internal(
+                "provider catalog credential encryption key is not configured".to_string(),
+            )
+        ));
+    }
+
     #[tokio::test]
     async fn gateway_runtime_state_supports_shared_models_fetch_plan_builder() {
         let state = TestState::default();
@@ -1383,6 +1504,114 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn model_fetch_isolates_malformed_legacy_key_from_healthy_key() {
+        let provider = sample_provider("provider-openai", "openai");
+        let endpoint = sample_endpoint(
+            "endpoint-openai-responses",
+            "provider-openai",
+            "openai:responses",
+        );
+        let mut malformed = sample_key(
+            "key-openai-malformed",
+            "provider-openai",
+            "api_key",
+            &["openai:responses"],
+        );
+        malformed.encrypted_api_key = Some("legacy-plaintext-or-corrupt".to_string());
+        malformed.allowed_models = Some(json!(["legacy-model"]));
+        malformed.last_models_fetch_error = Some("previous error".to_string());
+        let malformed_ciphertext = malformed.encrypted_api_key.clone();
+        let healthy = sample_key(
+            "key-openai-healthy",
+            "provider-openai",
+            "api_key",
+            &["openai:responses"],
+        );
+        let healthy_transport = sample_transport(
+            "openai",
+            "provider-openai",
+            "endpoint-openai-responses",
+            "key-openai-healthy",
+            "openai:responses",
+            "api_key",
+            None,
+        );
+        let state = TestState::new(
+            vec![provider],
+            vec![endpoint],
+            vec![malformed, healthy],
+            HashMap::from([(
+                (
+                    "provider-openai".to_string(),
+                    "endpoint-openai-responses".to_string(),
+                    "key-openai-healthy".to_string(),
+                ),
+                healthy_transport,
+            )]),
+            vec![execution_result(json!({
+                "data": [{"id": "gpt-healthy"}]
+            }))],
+        )
+        .with_transport_errors(HashMap::from([(
+            (
+                "provider-openai".to_string(),
+                "endpoint-openai-responses".to_string(),
+                "key-openai-malformed".to_string(),
+            ),
+            "provider_api_keys.api_key is not an authenticated ciphertext".to_string(),
+        )]));
+
+        let summary = perform_model_fetch_once_with_state(&state)
+            .await
+            .expect("one malformed key must not abort the cycle");
+
+        assert_eq!(summary.attempted, 2);
+        assert_eq!(summary.skipped, 1);
+        assert_eq!(summary.succeeded, 1);
+        let malformed_after = state.key("key-openai-malformed");
+        assert_eq!(malformed_after.encrypted_api_key, malformed_ciphertext);
+        assert_eq!(
+            malformed_after.allowed_models,
+            Some(json!(["legacy-model"]))
+        );
+        assert_eq!(
+            malformed_after.last_models_fetch_error.as_deref(),
+            Some("previous error")
+        );
+        assert_eq!(
+            state.key("key-openai-healthy").allowed_models,
+            Some(json!(["gpt-healthy"]))
+        );
+    }
+
+    #[test]
+    fn sanitized_model_fetch_key_drops_raw_transport_and_diagnostic_state() {
+        let mut key = sample_key("key-sanitize", "provider", "api_key", &["openai:responses"]);
+        key.capabilities = Some(json!({"secret": "capability"}));
+        key.note = Some("operator note".to_string());
+        key.proxy = Some(json!({"url": "http://user:pass@example.test"}));
+        key.last_models_fetch_error = Some("upstream detail".to_string());
+        key.oauth_invalid_reason = Some("token detail".to_string());
+        key.allowed_models = Some(json!(["keep-model-filter"]));
+        key.upstream_metadata = Some(json!({"provider": {"quota": 1}}));
+
+        let sanitized = sanitize_model_fetch_key(key);
+
+        assert_eq!(sanitized.encrypted_api_key, None);
+        assert_eq!(sanitized.encrypted_auth_config, None);
+        assert_eq!(sanitized.proxy, None);
+        assert_eq!(sanitized.fingerprint, None);
+        assert_eq!(sanitized.note, None);
+        assert_eq!(sanitized.last_models_fetch_error, None);
+        assert_eq!(sanitized.oauth_invalid_reason, None);
+        assert_eq!(sanitized.allowed_models, Some(json!(["keep-model-filter"])));
+        assert_eq!(
+            sanitized.upstream_metadata,
+            Some(json!({"provider": {"quota": 1}}))
+        );
+    }
+
     #[tokio::test]
     async fn model_fetch_failure_does_not_persist_upstream_error_body_credentials() {
         const UPSTREAM_SECRET: &str = "upstream-secret-token-value";
diff --git a/apps/aether-gateway/src/model_fetch/runtime/state.rs b/apps/aether-gateway/src/model_fetch/runtime/state.rs
index e90d720fa..49ce24f37 100644
--- a/apps/aether-gateway/src/model_fetch/runtime/state.rs
+++ b/apps/aether-gateway/src/model_fetch/runtime/state.rs
@@ -31,6 +31,20 @@ pub(crate) trait ModelFetchRuntimeState:
         provider_ids: &[String],
     ) -> Result<Vec<StoredProviderCatalogEndpoint>, GatewayError>;
 
+    /// Return raw catalog key rows for the background fetcher. Production
+    /// implementations should avoid the normal bulk credential-opening
+    /// wrapper here: one malformed legacy row must not prevent healthy keys
+    /// from being considered, and a maintenance scan must not lazily rewrite
+    /// historical ciphertext. Test implementations can use the association
+    /// store's existing method via this default.
+    async fn list_provider_catalog_keys_for_model_fetch(
+        &self,
+        provider_ids: &[String],
+    ) -> Result<Vec<StoredProviderCatalogKey>, String> {
+        self.list_provider_catalog_keys_by_provider_ids(provider_ids)
+            .await
+    }
+
     async fn read_provider_transport_snapshot(
         &self,
         provider_id: &str,
diff --git a/apps/aether-gateway/src/state/integrations.rs b/apps/aether-gateway/src/state/integrations.rs
index 49951b70a..3ebade0de 100644
--- a/apps/aether-gateway/src/state/integrations.rs
+++ b/apps/aether-gateway/src/state/integrations.rs
@@ -404,6 +404,16 @@ impl ModelFetchRuntimeState for AppState {
         AppState::list_provider_catalog_endpoints_by_provider_ids(self, provider_ids).await
     }
 
+    async fn list_provider_catalog_keys_for_model_fetch(
+        &self,
+        provider_ids: &[String],
+    ) -> Result<Vec<StoredProviderCatalogKey>, String> {
+        self.data
+            .list_provider_catalog_keys_by_provider_ids(provider_ids)
+            .await
+            .map_err(|err| err.to_string())
+    }
+
     async fn read_provider_transport_snapshot(
         &self,
         provider_id: &str,