@@ -5,7 +5,7 @@
 
 #![allow(clippy::result_large_err)]
 
-use crate::persistence::{ObjectType, Store, current_time_ms};
+use crate::persistence::{ObjectType, PersistenceError, Store, WriteCondition, current_time_ms};
 use openshell_core::proto::{
     Provider, ProviderCredentialRefreshStatus, ProviderCredentialRefreshStrategy,
     StoredProviderCredentialRefreshState,
@@ -47,6 +47,41 @@ pub async fn put_refresh_state(
         .map_err(|e| Status::internal(format!("persist provider refresh state failed: {e}")))
 }
 
+/// Persist an updated refresh state only if the row still exists with the
+/// generation read at the start of the rotation.
+///
+/// Uses a version-matched UPDATE, which never inserts — so a refresh deleted
+/// while an STS (or OAuth) request was in flight is not resurrected, and its
+/// stored source-credential material is not recreated (CWE-362). Returns `true`
+/// when persisted, `false` when the refresh was deleted or superseded by a
+/// concurrent write (in which case nothing was written).
+async fn persist_refresh_state_if_current(
+    store: &Store,
+    state: &StoredProviderCredentialRefreshState,
+    expected_version: u64,
+) -> Result<bool, Status> {
+    match store
+        .put_if(
+            StoredProviderCredentialRefreshState::object_type(),
+            state.object_id(),
+            state.object_name(),
+            &state.encode_to_vec(),
+            None,
+            WriteCondition::MatchResourceVersion(expected_version),
+        )
+        .await
+    {
+        Ok(_) => Ok(true),
+        // Version mismatch (superseded) or the row is gone (deleted): the
+        // version-matched UPDATE matched no row, so nothing was written.
+        Err(PersistenceError::Conflict { .. }) => Ok(false),
+        Err(PersistenceError::Database(message)) if message.contains("not found") => Ok(false),
+        Err(e) => Err(Status::internal(format!(
+            "persist provider refresh state failed: {e}"
+        ))),
+    }
+}
+
 pub async fn list_refresh_states_for_provider(
     store: &Store,
     provider_id: &str,
@@ -306,6 +341,13 @@ pub async fn refresh_provider_credential(
     else {
         return Err(Status::not_found("provider refresh state not found"));
     };
+    // Generation of the refresh at the start of the rotation. Terminal persists
+    // match on it so a concurrent delete or rotation is detected rather than
+    // clobbered, and a deleted refresh is never recreated (CWE-362).
+    let expected_version = state
+        .metadata
+        .as_ref()
+        .map_or(0, |meta| meta.resource_version);
 
     info!(
         provider = %state.provider_name,
@@ -326,7 +368,7 @@ pub async fn refresh_provider_credential(
         state.last_error = err.message().to_string();
         state.next_refresh_at_ms =
             now_ms.saturating_add(REFRESH_ERROR_RETRY_SECONDS.saturating_mul(1000));
-        put_refresh_state(store, &state).await?;
+        persist_refresh_state_if_current(store, &state, expected_version).await?;
         warn!(
             provider = %state.provider_name,
             credential_key = %state.credential_key,
@@ -341,14 +383,34 @@ pub async fn refresh_provider_credential(
     match mint_credential(&state).await {
         Ok(minted) => {
             let now_ms = current_time_ms();
+            // A delete-refresh can complete while the mint request is in flight.
+            // Verify the refresh still exists before writing minted credentials
+            // into the provider, so a removed refresh is not silently re-minted
+            // (CWE-362). The version-matched persist below is the authoritative
+            // guard against resurrecting deleted state; this check just avoids
+            // writing credentials we are about to discard.
+            if get_refresh_state(store, provider.object_id(), credential_key)
+                .await?
+                .is_none()
+            {
+                warn!(
+                    provider = %state.provider_name,
+                    credential_key = %state.credential_key,
+                    strategy = %refresh_strategy_name(state.strategy),
+                    "provider credential refresh deleted during rotation; discarding minted credentials"
+                );
+                return Err(Status::aborted(
+                    "provider refresh was deleted during rotation",
+                ));
+            }
             if let Err(err) =
                 apply_minted_credential(store, &provider, credential_key, &minted).await
             {
                 state.status = "error".to_string();
                 state.last_error = err.message().to_string();
                 state.next_refresh_at_ms =
                     now_ms.saturating_add(REFRESH_ERROR_RETRY_SECONDS.saturating_mul(1000));
-                put_refresh_state(store, &state).await?;
+                persist_refresh_state_if_current(store, &state, expected_version).await?;
                 warn!(
                     provider = %state.provider_name,
                     credential_key = %state.credential_key,
@@ -383,7 +445,17 @@ pub async fn refresh_provider_credential(
             state.last_refresh_at_ms = now_ms;
             state.status = "refreshed".to_string();
             state.last_error.clear();
-            put_refresh_state(store, &state).await?;
+            if !persist_refresh_state_if_current(store, &state, expected_version).await? {
+                warn!(
+                    provider = %state.provider_name,
+                    credential_key = %state.credential_key,
+                    strategy = %refresh_strategy_name(state.strategy),
+                    "provider credential refresh deleted or superseded during rotation; not resurrecting refresh state"
+                );
+                return Err(Status::aborted(
+                    "provider refresh was deleted during rotation",
+                ));
+            }
             info!(
                 provider = %state.provider_name,
                 credential_key = %state.credential_key,
@@ -402,7 +474,7 @@ pub async fn refresh_provider_credential(
             state.last_error = err.message().to_string();
             state.next_refresh_at_ms =
                 now_ms.saturating_add(REFRESH_ERROR_RETRY_SECONDS.saturating_mul(1000));
-            put_refresh_state(store, &state).await?;
+            persist_refresh_state_if_current(store, &state, expected_version).await?;
             warn!(
                 provider = %state.provider_name,
                 credential_key = %state.credential_key,
@@ -982,8 +1054,8 @@ async fn run_refresh_worker_tick(store: &Store) -> Result<(), Status> {
 #[cfg(test)]
 mod tests {
     use super::{
-        NewRefreshStateConfig, get_refresh_state, new_refresh_state, put_refresh_state,
-        refresh_provider_credential, refresh_state_name, refresh_strategy_name,
+        NewRefreshStateConfig, delete_refresh_state, get_refresh_state, new_refresh_state,
+        put_refresh_state, refresh_provider_credential, refresh_state_name, refresh_strategy_name,
         run_refresh_worker_tick, seconds_until_ms,
     };
     use crate::persistence::test_store;
@@ -1749,6 +1821,139 @@ mod tests {
         assert!(err.message().contains("AWS_SECRET_ACCESS_KEY"));
     }
 
+    // A wiremock responder that blocks the STS response until the test releases
+    // it, so a delete-refresh can be interleaved deterministically while the
+    // rotation is parked awaiting STS.
+    struct GatedStsResponder {
+        hit: std::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
+        release: std::sync::Mutex<std::sync::mpsc::Receiver<()>>,
+        body: String,
+    }
+
+    impl wiremock::Respond for GatedStsResponder {
+        fn respond(&self, _request: &wiremock::Request) -> ResponseTemplate {
+            let hit = self.hit.lock().unwrap().take();
+            if let Some(hit) = hit {
+                let _ = hit.send(());
+            }
+            let _ = self.release.lock().unwrap().recv();
+            ResponseTemplate::new(200).set_body_string(self.body.clone())
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn rotation_does_not_resurrect_refresh_deleted_mid_flight() {
+        let mock_server = MockServer::start().await;
+        let (hit_tx, hit_rx) = tokio::sync::oneshot::channel::<()>();
+        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+        Mock::given(method("POST"))
+            .and(body_string_contains("Action=AssumeRole"))
+            .respond_with(GatedStsResponder {
+                hit: std::sync::Mutex::new(Some(hit_tx)),
+                release: std::sync::Mutex::new(release_rx),
+                body: r#"<AssumeRoleResponse xmlns="https://sts.amazonaws.com/doc/2011-06-15/">
+  <AssumeRoleResult>
+    <Credentials>
+      <AccessKeyId>ASIAMOCKKEY</AccessKeyId>
+      <SecretAccessKey>MockSecretAccessKey123</SecretAccessKey>
+      <SessionToken>MockSessionTokenXYZ</SessionToken>
+      <Expiration>2099-01-01T00:00:00Z</Expiration>
+    </Credentials>
+  </AssumeRoleResult>
+</AssumeRoleResponse>"#
+                    .to_string(),
+            })
+            .mount(&mock_server)
+            .await;
+
+        let store = test_store().await;
+        crate::grpc::policy::set_global_bool_setting_for_test(
+            &store,
+            openshell_core::settings::PROVIDERS_V2_ENABLED_KEY,
+            true,
+        )
+        .await
+        .unwrap();
+        let prov = provider("aws-race", "aws");
+        store.put_message(&prov).await.unwrap();
+        let provider_id = prov.object_id().to_string();
+
+        let state = new_refresh_state(
+            &prov,
+            "AWS_ACCESS_KEY_ID",
+            NewRefreshStateConfig {
+                additional_output_keys: HashMap::from([
+                    (
+                        "secret_access_key".to_string(),
+                        "AWS_SECRET_ACCESS_KEY".to_string(),
+                    ),
+                    ("session_token".to_string(), "AWS_SESSION_TOKEN".to_string()),
+                ]),
+                strategy: ProviderCredentialRefreshStrategy::AwsStsAssumeRole,
+                material: HashMap::from([
+                    (
+                        "role_arn".to_string(),
+                        "arn:aws:iam::123456789012:role/TestRole".to_string(),
+                    ),
+                    ("aws_access_key_id".to_string(), "AKIATESTKEY".to_string()),
+                    (
+                        "aws_secret_access_key".to_string(),
+                        "TestSecretKey".to_string(),
+                    ),
+                    ("sts_endpoint_url".to_string(), mock_server.uri()),
+                ]),
+                secret_material_keys: vec!["aws_secret_access_key".to_string()],
+                expires_at_ms: 0,
+                token_url: String::new(),
+                scopes: Vec::new(),
+                refresh_before_seconds: 300,
+                max_lifetime_seconds: 3600,
+            },
+        )
+        .unwrap();
+        put_refresh_state(&store, &state).await.unwrap();
+
+        let rotate = refresh_provider_credential(&store, "aws-race", "AWS_ACCESS_KEY_ID");
+        let interfere = async {
+            // Wait until the rotation is inside the STS call (its state read has
+            // already happened), then delete the refresh and release STS.
+            if tokio::time::timeout(std::time::Duration::from_secs(15), hit_rx)
+                .await
+                .is_err()
+            {
+                return;
+            }
+            delete_refresh_state(&store, &provider_id, "AWS_ACCESS_KEY_ID")
+                .await
+                .unwrap();
+            let _ = release_tx.send(());
+        };
+        let (rotate_result, ()) = tokio::join!(rotate, interfere);
+
+        // The rotation must fail rather than complete against a deleted refresh.
+        assert!(
+            rotate_result.is_err(),
+            "rotation should abort when its refresh is deleted mid-flight"
+        );
+        // The deleted refresh state must not be resurrected.
+        assert!(
+            get_refresh_state(&store, &provider_id, "AWS_ACCESS_KEY_ID")
+                .await
+                .unwrap()
+                .is_none(),
+            "deleted refresh state must not be recreated"
+        );
+        // No credentials were minted into the provider.
+        let stored = store
+            .get_message_by_name::<Provider>("aws-race")
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(!stored.credentials.contains_key("AWS_ACCESS_KEY_ID"));
+        assert!(!stored.credentials.contains_key("AWS_SECRET_ACCESS_KEY"));
+        assert!(!stored.credentials.contains_key("AWS_SESSION_TOKEN"));
+    }
+
     fn provider(name: &str, provider_type: &str) -> Provider {
         Provider {
             metadata: Some(ObjectMeta {