OpenShell/crates/openshell-server/src/provider_refresh.rs

// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! Provider credential refresh state.

#![allow(clippy::result_large_err)]

use crate::credentials::RefreshMaterialScope;
use crate::persistence::{
    ObjectListQuery, ObjectType, PersistenceError, Store, WriteCondition, current_time_ms,
};
use openshell_core::ObjectWorkspace;
use openshell_core::proto::{
    CredentialHandle, Provider, ProviderCredentialRefreshRecoveryAction,
    ProviderCredentialRefreshStatus, ProviderCredentialRefreshStrategy,
};
use openshell_core::{ObjectId, ObjectName};
use prost::Message;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::time::Duration;
use tonic::{Code, Status};
use tracing::{info, warn};

use crate::storage_proto::{
    StoredProviderCredentialRefreshStateV2 as StoredProviderCredentialRefreshState,
    StoredRefreshMaterialDeletion,
};

const DEFAULT_REFRESH_BEFORE_SECONDS: i64 = 300;
const EXPIRATION_PRESENT_ANNOTATION: &str = "openshell.nvidia.com/refresh-expiration-present";
const DEFAULT_MAX_LIFETIME_SECONDS: i64 = 3600;
const REFRESH_ERROR_RETRY_SECONDS: i64 = 60;
const REFRESH_CONFIGURATION_RETRY_SECONDS: i64 = 60 * 60;
const MAX_OAUTH_ERROR_RESPONSE_BYTES: usize = 8 * 1024;

pub fn refresh_material_scope(
    state: &StoredProviderCredentialRefreshState,
) -> RefreshMaterialScope<'_> {
    RefreshMaterialScope {
        provider_name: &state.provider_name,
        workspace: state.object_workspace(),
        provider_id: &state.provider_id,
        credential_key: &state.credential_key,
    }
}

impl ObjectType for StoredProviderCredentialRefreshState {
    fn object_type() -> &'static str {
        "provider_credential_refresh_state"
    }
}

pub fn refresh_state_name(provider_id: &str, credential_key: &str) -> String {
    let mut key = String::with_capacity(credential_key.len() * 2);
    for byte in credential_key.as_bytes() {
        use std::fmt::Write as _;
        write!(&mut key, "{byte:02x}").expect("writing to String cannot fail");
    }
    format!("provider-refresh-{provider_id}-{key}")
}

/// Return the durable authorization epoch for one configured refresh grant.
///
/// Records created before the explicit epoch field was introduced use their
/// gateway-generated object ID as a stable migration epoch. An explicit
/// reconfiguration writes a new random epoch while preserving object metadata,
/// so reauthorization still revokes handles derived from the legacy value.
pub fn effective_authorization_epoch(
    state: &StoredProviderCredentialRefreshState,
) -> Result<&str, Status> {
    if !state.authorization_epoch.is_empty() {
        return Ok(&state.authorization_epoch);
    }
    state
        .metadata
        .as_ref()
        .map(|metadata| metadata.id.as_str())
        .filter(|id| !id.is_empty())
        .ok_or_else(|| {
            Status::failed_precondition("provider refresh state has no authorization epoch")
        })
}

#[cfg(test)]
pub async fn put_refresh_state(
    store: &Store,
    state: &StoredProviderCredentialRefreshState,
) -> Result<(), Status> {
    store
        .put_scoped_message(state, &state.provider_id)
        .await
        .map_err(|e| Status::internal(format!("persist provider refresh state failed: {e}")))
}

/// Atomically claim a new provider-and-credential refresh identity.
///
/// The refresh name is unique within a workspace. A concurrent creator must
/// lose instead of overwriting the winner so its caller can delete any secret
/// material staged before this write.
pub async fn create_refresh_state(
    store: &Store,
    state: &StoredProviderCredentialRefreshState,
) -> Result<(), Status> {
    match store
        .create_scoped(
            StoredProviderCredentialRefreshState::object_type(),
            state.object_id(),
            state.object_name(),
            state.object_workspace(),
            &state.provider_id,
            &state.encode_to_vec(),
            None,
        )
        .await
    {
        Ok(_) => Ok(()),
        Err(PersistenceError::UniqueViolation { .. }) => Err(Status::aborted(
            "provider refresh was concurrently configured",
        )),
        Err(err) => Err(Status::internal(format!(
            "create provider refresh state failed: {err}"
        ))),
    }
}

/// 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 the new
/// resource version when persisted, or `None` 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<Option<u64>, Status> {
    match store
        .put_if(
            StoredProviderCredentialRefreshState::object_type(),
            state.object_id(),
            state.object_name(),
            state.object_workspace(),
            &state.encode_to_vec(),
            None,
            WriteCondition::MatchResourceVersion(expected_version),
        )
        .await
    {
        Ok(result) => Ok(Some(result.resource_version)),
        // The version-matched UPDATE matched no row: the refresh was deleted
        // (current version `None`) or superseded by a concurrent write (a
        // different current version). Either way nothing was written.
        Err(PersistenceError::Conflict { .. }) => Ok(None),
        Err(e) => Err(Status::internal(format!(
            "persist provider refresh state failed: {e}"
        ))),
    }
}

pub async fn replace_refresh_state_if_current(
    store: &Store,
    state: &StoredProviderCredentialRefreshState,
    expected_version: u64,
) -> Result<bool, Status> {
    Ok(
        persist_refresh_state_if_current(store, state, expected_version)
            .await?
            .is_some(),
    )
}

pub async fn list_refresh_states_for_provider(
    store: &Store,
    provider_id: &str,
) -> Result<Vec<StoredProviderCredentialRefreshState>, Status> {
    store
        .collect_messages(ObjectListQuery::Scope(provider_id))
        .await
        .map_err(|e| Status::internal(format!("list provider refresh states failed: {e}")))
}

pub async fn list_all_refresh_states(
    store: &Store,
) -> Result<Vec<StoredProviderCredentialRefreshState>, Status> {
    store
        .collect_messages(ObjectListQuery::AllWorkspaces)
        .await
        .map_err(|e| Status::internal(format!("list provider refresh states failed: {e}")))
}

pub async fn get_refresh_state(
    store: &Store,
    workspace: &str,
    provider_id: &str,
    credential_key: &str,
) -> Result<Option<StoredProviderCredentialRefreshState>, Status> {
    let name = refresh_state_name(provider_id, credential_key);
    store
        .get_message_by_name::<StoredProviderCredentialRefreshState>(workspace, &name)
        .await
        .map_err(|e| Status::internal(format!("fetch provider refresh state failed: {e}")))
}

pub async fn delete_refresh_state_with_credentials(
    store: &Store,
    credentials: &crate::credentials::CredentialRuntime,
    workspace: &str,
    provider_id: &str,
    credential_key: &str,
) -> Result<bool, Status> {
    let Some(state) = get_refresh_state(store, workspace, provider_id, credential_key).await?
    else {
        return Ok(false);
    };
    delete_observed_refresh_state_with_credentials(store, credentials, state).await?;
    Ok(true)
}

/// Delete the observed refresh identity without resolving its name again.
pub async fn delete_observed_refresh_state_with_credentials(
    store: &Store,
    credentials: &crate::credentials::CredentialRuntime,
    mut state: StoredProviderCredentialRefreshState,
) -> Result<(), Status> {
    let mut version = state
        .metadata
        .as_ref()
        .map_or(0, |metadata| metadata.resource_version);
    if state
        .metadata
        .as_ref()
        .is_some_and(|metadata| metadata.deletion_time.is_none())
    {
        if let Some(metadata) = state.metadata.as_mut() {
            metadata.deletion_time =
                openshell_core::time::timestamp_from_millis(current_time_ms()).ok();
        }
        state.authorization_epoch = uuid::Uuid::new_v4().to_string();
        state.status = "deleting".to_string();
        state.next_refresh_at_ms = i64::MAX;
        let Some(current_version) =
            persist_refresh_state_if_current(store, &state, version).await?
        else {
            if store
                .get_message::<StoredProviderCredentialRefreshState>(state.object_id())
                .await
                .map_err(|err| {
                    Status::internal(format!("fetch provider refresh state failed: {err}"))
                })?
                .is_none()
            {
                return Ok(());
            }
            return Err(Status::aborted(
                "provider refresh was concurrently modified during deletion",
            ));
        };
        version = current_version;
        if let Some(metadata) = state.metadata.as_mut() {
            metadata.resource_version = version;
        }
    }

    delete_pending_secret_handles(credentials, &state).await?;
    credentials
        .delete_refresh_material_handles(
            refresh_material_scope(&state),
            &state.secret_material_handles,
        )
        .await?;
    store
        .delete_if(
            StoredProviderCredentialRefreshState::object_type(),
            state.object_id(),
            version,
        )
        .await
        .map_err(|err| match err {
            PersistenceError::Conflict { .. } => {
                Status::aborted("provider refresh was concurrently modified during deletion")
            }
            other => Status::internal(format!("delete provider refresh state failed: {other}")),
        })?;
    Ok(())
}

pub async fn delete_refresh_states_for_provider_with_credentials(
    store: &Store,
    credentials: &crate::credentials::CredentialRuntime,
    provider_id: &str,
) -> Result<u64, Status> {
    let states = list_refresh_states_for_provider(store, provider_id).await?;
    let mut deleted = 0;
    for state in &states {
        if delete_refresh_state_with_credentials(
            store,
            credentials,
            state.object_workspace(),
            provider_id,
            &state.credential_key,
        )
        .await?
        {
            deleted += 1;
        }
    }
    Ok(deleted)
}

pub fn refresh_status_from_state(
    state: &StoredProviderCredentialRefreshState,
) -> ProviderCredentialRefreshStatus {
    ProviderCredentialRefreshStatus {
        provider: state.provider_name.clone(),
        provider_id: state.provider_id.clone(),
        credential_key: state.credential_key.clone(),
        strategy: state.strategy,
        status: state.status.clone(),
        expiration_time: if refresh_has_expiration(state) {
            openshell_core::time::timestamp_from_millis(state.expires_at_ms).ok()
        } else {
            None
        },
        next_refresh_time: if state.next_refresh_at_ms == i64::MAX {
            None
        } else {
            openshell_core::time::optional_timestamp_from_legacy_millis(state.next_refresh_at_ms)
                .ok()
                .flatten()
        },
        last_refresh_time: openshell_core::time::optional_timestamp_from_legacy_millis(
            state.last_refresh_at_ms,
        )
        .ok()
        .flatten(),
        last_error: state.last_error.clone(),
        recovery_action: state.recovery_action,
        failure_code: state.failure_code.clone(),
        provider_error_subtype: state.provider_error_subtype.clone(),
        last_error_time: openshell_core::time::optional_timestamp_from_legacy_millis(
            state.last_error_at_ms,
        )
        .ok()
        .flatten(),
    }
}

/// Whether a refresh expiration was explicitly provided.
///
/// Legacy records infer presence from a nonzero millisecond value. New records
/// use a private metadata annotation only for the ambiguous Unix epoch value so
/// the frozen storage protobuf schema does not need to change.
pub fn refresh_has_expiration(state: &StoredProviderCredentialRefreshState) -> bool {
    state.expires_at_ms != 0
        || state.metadata.as_ref().is_some_and(|metadata| {
            metadata
                .annotations
                .get(EXPIRATION_PRESENT_ANNOTATION)
                .is_some_and(|value| value == "true")
        })
}

pub fn set_refresh_expiration_presence(
    state: &mut StoredProviderCredentialRefreshState,
    present: bool,
) {
    let Some(metadata) = state.metadata.as_mut() else {
        return;
    };
    if present && state.expires_at_ms == 0 {
        metadata.annotations.insert(
            EXPIRATION_PRESENT_ANNOTATION.to_string(),
            "true".to_string(),
        );
    } else {
        metadata.annotations.remove(EXPIRATION_PRESENT_ANNOTATION);
    }
}

pub struct NewRefreshStateConfig {
    pub strategy: ProviderCredentialRefreshStrategy,
    pub material: HashMap<String, String>,
    pub secret_material_keys: Vec<String>,
    pub expires_at_ms: i64,
    pub token_url: String,
    pub scopes: Vec<String>,
    pub refresh_before: Option<prost_types::Duration>,
    pub max_lifetime: Option<prost_types::Duration>,
    /// Resolved semantic output id -> concrete env key for credentials this
    /// refresh co-mints beyond its primary. Pinned from the profile's
    /// `additional_outputs` at configure time.
    pub additional_output_keys: HashMap<String, String>,
}

#[allow(clippy::unnecessary_wraps)]
pub fn new_refresh_state(
    provider: &Provider,
    workspace: &str,
    credential_key: &str,
    config: NewRefreshStateConfig,
) -> Result<StoredProviderCredentialRefreshState, Status> {
    if let Some(value) = config.refresh_before.as_ref() {
        openshell_core::time::duration_to_std(value)
            .map_err(|error| Status::invalid_argument(error.to_string()))?;
    }
    if let Some(value) = config.max_lifetime.as_ref() {
        let duration = openshell_core::time::duration_to_std(value)
            .map_err(|error| Status::invalid_argument(error.to_string()))?;
        if duration.is_zero() {
            return Err(Status::invalid_argument(
                "max_lifetime must be greater than zero when present",
            ));
        }
    }
    let provider_id = provider.object_id().to_string();
    let provider_name = provider.object_name().to_string();
    let now_ms = current_time_ms();
    let next_refresh_at_ms =
        next_refresh_at_ms(config.expires_at_ms, config.refresh_before.as_ref());
    Ok(StoredProviderCredentialRefreshState {
        metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta {
            id: uuid::Uuid::new_v4().to_string(),
            name: refresh_state_name(&provider_id, credential_key),
            created_time: openshell_core::time::timestamp_from_millis(now_ms).ok(),
            labels: HashMap::new(),
            resource_version: 0,
            annotations: HashMap::new(),
            workspace: workspace.to_string(),
            deletion_time: None,
        }),
        provider_id,
        provider_name,
        credential_key: credential_key.to_string(),
        strategy: config.strategy as i32,
        material: config.material,
        secret_material_keys: config.secret_material_keys,
        expires_at_ms: config.expires_at_ms,
        next_refresh_at_ms,
        last_refresh_at_ms: 0,
        status: "configured".to_string(),
        last_error: String::new(),
        token_url: config.token_url,
        scopes: config.scopes,
        refresh_before: config.refresh_before,
        max_lifetime: config.max_lifetime,
        additional_output_keys: config.additional_output_keys,
        authorization_epoch: uuid::Uuid::new_v4().to_string(),
        secret_material_handles: HashMap::new(),
        pending_secret_deletions: Vec::new(),
        recovery_action: ProviderCredentialRefreshRecoveryAction::Unspecified as i32,
        failure_code: String::new(),
        provider_error_subtype: String::new(),
        last_error_at_ms: 0,
    })
}

#[derive(Debug)]
struct MintedCredential {
    access_token: String,
    expires_at_ms: i64,
    refresh_token: Option<String>,
    additional_credentials: HashMap<String, String>,
}

#[derive(Debug, Deserialize)]
struct TokenResponse {
    access_token: String,
    expires_in: Option<i64>,
    refresh_token: Option<String>,
}

#[derive(Debug, Deserialize)]
struct OAuthErrorResponse {
    error: String,
    #[serde(default, deserialize_with = "deserialize_optional_oauth_string")]
    error_subtype: Option<String>,
}

fn deserialize_optional_oauth_string<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let value = serde_json::Value::deserialize(deserializer)?;
    Ok(value.as_str().map(str::to_owned))
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum OAuthGrantKind {
    UserRefreshToken,
    NonInteractive,
}

#[derive(Debug)]
struct RefreshFailure {
    status: Status,
    recovery_action: ProviderCredentialRefreshRecoveryAction,
    failure_code: &'static str,
    provider_error_subtype: Option<&'static str>,
    retry_schedule: RefreshRetrySchedule,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RefreshRetrySchedule {
    Short,
    Configuration,
    Parked,
}

impl RefreshFailure {
    fn retryable(status: Status, failure_code: &'static str) -> Self {
        Self {
            status,
            recovery_action: ProviderCredentialRefreshRecoveryAction::Retry,
            failure_code,
            provider_error_subtype: None,
            retry_schedule: RefreshRetrySchedule::Short,
        }
    }

    fn investigate(status: Status, failure_code: &'static str) -> Self {
        Self {
            status,
            recovery_action: ProviderCredentialRefreshRecoveryAction::Investigate,
            failure_code,
            provider_error_subtype: None,
            retry_schedule: RefreshRetrySchedule::Short,
        }
    }

    fn reauthorize(
        status: Status,
        failure_code: &'static str,
        provider_error_subtype: Option<&'static str>,
    ) -> Self {
        Self {
            status,
            recovery_action: ProviderCredentialRefreshRecoveryAction::Reauthorize,
            failure_code,
            provider_error_subtype,
            retry_schedule: RefreshRetrySchedule::Parked,
        }
    }

    fn fix_configuration(status: Status, failure_code: &'static str) -> Self {
        Self {
            status,
            recovery_action: ProviderCredentialRefreshRecoveryAction::FixConfiguration,
            failure_code,
            provider_error_subtype: None,
            retry_schedule: RefreshRetrySchedule::Configuration,
        }
    }

    fn fix_configuration_with_subtype(
        status: Status,
        failure_code: &'static str,
        provider_error_subtype: &'static str,
    ) -> Self {
        Self {
            status,
            recovery_action: ProviderCredentialRefreshRecoveryAction::FixConfiguration,
            failure_code,
            provider_error_subtype: Some(provider_error_subtype),
            retry_schedule: RefreshRetrySchedule::Configuration,
        }
    }

    fn into_status(self) -> Status {
        self.status
    }

    fn from_status(status: &Status) -> Self {
        Self::from(Status::new(status.code(), status.message().to_string()))
    }
}

impl From<Status> for RefreshFailure {
    fn from(status: Status) -> Self {
        match status.code() {
            Code::InvalidArgument
            | Code::FailedPrecondition
            | Code::PermissionDenied
            | Code::Unauthenticated => {
                Self::fix_configuration(status, "refresh_configuration_invalid")
            }
            Code::Unavailable
            | Code::DeadlineExceeded
            | Code::ResourceExhausted
            | Code::Aborted
            | Code::Internal => Self::retryable(status, "refresh_failed"),
            _ => Self::investigate(status, "refresh_failed"),
        }
    }
}

#[derive(Debug, Serialize)]
struct GoogleServiceAccountClaims<'a> {
    iss: &'a str,
    scope: String,
    aud: &'a str,
    iat: i64,
    exp: i64,
    #[serde(skip_serializing_if = "Option::is_none")]
    sub: Option<&'a str>,
}

pub fn next_refresh_at_ms(
    expires_at_ms: i64,
    refresh_before: Option<&prost_types::Duration>,
) -> i64 {
    let refresh_before = refresh_before
        .and_then(|value| openshell_core::time::duration_to_std(value).ok())
        .unwrap_or_else(|| {
            Duration::from_secs(u64::try_from(DEFAULT_REFRESH_BEFORE_SECONDS).unwrap_or(u64::MAX))
        });
    let refresh_before_ms = refresh_before.as_millis().saturating_add(u128::from(
        !refresh_before.subsec_nanos().is_multiple_of(1_000_000),
    ));
    let refresh_before_ms = i64::try_from(refresh_before_ms).unwrap_or(i64::MAX);
    if expires_at_ms > 0 {
        return expires_at_ms.saturating_sub(refresh_before_ms);
    }
    0
}

// OAuth/JWT expiry fields use whole seconds. Round an exact positive profile
// duration up only at that protocol boundary so a fractional lifetime never
// collapses into the absent/default sentinel.
fn max_lifetime_seconds(state: &StoredProviderCredentialRefreshState) -> i64 {
    let Some(value) = state.max_lifetime.as_ref() else {
        return DEFAULT_MAX_LIFETIME_SECONDS;
    };
    let Ok(duration) = openshell_core::time::duration_to_std(value) else {
        return DEFAULT_MAX_LIFETIME_SECONDS;
    };
    let seconds = duration
        .as_secs()
        .saturating_add(u64::from(duration.subsec_nanos() != 0));
    i64::try_from(seconds).unwrap_or(i64::MAX).max(1)
}

fn seconds_until_ms(now_ms: i64, target_ms: i64) -> i64 {
    if target_ms <= 0 {
        return 0;
    }
    target_ms.saturating_sub(now_ms).max(0) / 1000
}

pub fn refresh_strategy_name(strategy: i32) -> &'static str {
    match ProviderCredentialRefreshStrategy::try_from(strategy)
        .unwrap_or(ProviderCredentialRefreshStrategy::Unspecified)
    {
        ProviderCredentialRefreshStrategy::Static => "static",
        ProviderCredentialRefreshStrategy::External => "external",
        ProviderCredentialRefreshStrategy::Oauth2RefreshToken => "oauth2_refresh_token",
        ProviderCredentialRefreshStrategy::Oauth2ClientCredentials => "oauth2_client_credentials",
        ProviderCredentialRefreshStrategy::GoogleServiceAccountJwt => "google_service_account_jwt",
        ProviderCredentialRefreshStrategy::AwsStsAssumeRole => "aws_sts_assume_role",
        ProviderCredentialRefreshStrategy::Unspecified => "unspecified",
    }
}

pub use openshell_providers::is_gateway_mintable_strategy;

/// Secret source-material fields that are security-sensitive by strategy even
/// when a direct API caller omits `secret_material_keys`.
pub fn strategy_secret_material_keys(
    strategy: ProviderCredentialRefreshStrategy,
) -> &'static [&'static str] {
    match strategy {
        ProviderCredentialRefreshStrategy::Oauth2RefreshToken => {
            &["refresh_token", "client_secret"]
        }
        ProviderCredentialRefreshStrategy::Oauth2ClientCredentials => &["client_secret"],
        ProviderCredentialRefreshStrategy::GoogleServiceAccountJwt => &["private_key"],
        ProviderCredentialRefreshStrategy::AwsStsAssumeRole => {
            &["aws_secret_access_key", "aws_session_token"]
        }
        ProviderCredentialRefreshStrategy::Static
        | ProviderCredentialRefreshStrategy::External
        | ProviderCredentialRefreshStrategy::Unspecified => &[],
    }
}

async fn resolve_refresh_material(
    credentials: Option<&crate::credentials::CredentialRuntime>,
    state: &StoredProviderCredentialRefreshState,
) -> Result<StoredProviderCredentialRefreshState, Status> {
    if state.secret_material_handles.is_empty() {
        return Ok(state.clone());
    }
    let credentials = credentials.ok_or_else(|| {
        Status::failed_precondition(
            "provider refresh material requires the configured credential runtime",
        )
    })?;
    let resolved = credentials
        .resolve_refresh_material(
            refresh_material_scope(state),
            &state.secret_material_handles,
        )
        .await?;
    let mut transient = state.clone();
    transient.material.extend(resolved);
    Ok(transient)
}

pub fn enqueue_pending_secret_deletion(
    state: &mut StoredProviderCredentialRefreshState,
    material_key: &str,
    handle: CredentialHandle,
) {
    state
        .pending_secret_deletions
        .push(StoredRefreshMaterialDeletion {
            material_key: material_key.to_string(),
            handle: Some(handle),
        });
}

async fn delete_pending_secret_handles(
    credentials: &crate::credentials::CredentialRuntime,
    state: &StoredProviderCredentialRefreshState,
) -> Result<(), Status> {
    credentials
        .delete_refresh_material_deletions(
            refresh_material_scope(state),
            &state.pending_secret_deletions,
        )
        .await
}

async fn cleanup_pending_secret_deletions(
    store: &Store,
    credentials: Option<&crate::credentials::CredentialRuntime>,
    state: &mut StoredProviderCredentialRefreshState,
    expected_version: u64,
) -> Result<u64, Status> {
    if state.pending_secret_deletions.is_empty() {
        return Ok(expected_version);
    }
    let credentials = credentials.ok_or_else(|| {
        Status::failed_precondition(
            "provider refresh cleanup requires the configured credential runtime",
        )
    })?;
    delete_pending_secret_handles(credentials, state).await?;
    // Keep the caller's in-memory state unchanged unless the CAS succeeds. A
    // failed cleanup must not let the live refresh path persist a locally
    // cleared tombstone list over the durable retry references.
    let mut cleaned = state.clone();
    cleaned.pending_secret_deletions.clear();
    let new_version = persist_refresh_state_if_current(store, &cleaned, expected_version)
        .await?
        .ok_or_else(|| {
            Status::aborted("provider refresh was deleted or superseded during secret cleanup")
        })?;
    if let Some(metadata) = cleaned.metadata.as_mut() {
        metadata.resource_version = new_version;
    }
    *state = cleaned;
    Ok(new_version)
}

async fn persist_retryable_refresh_error_state(
    store: &Store,
    state: &mut StoredProviderCredentialRefreshState,
    expected_version: u64,
    error: &Status,
) -> Result<u64, Status> {
    let failure = RefreshFailure::retryable(
        Status::new(error.code(), error.message().to_string()),
        "refresh_failed",
    );
    persist_refresh_failure_state(store, state, expected_version, &failure).await
}

async fn persist_refresh_failure_state(
    store: &Store,
    state: &mut StoredProviderCredentialRefreshState,
    expected_version: u64,
    failure: &RefreshFailure,
) -> Result<u64, Status> {
    let now_ms = current_time_ms();
    state.status = match failure.recovery_action {
        ProviderCredentialRefreshRecoveryAction::Retry
        | ProviderCredentialRefreshRecoveryAction::Unspecified => "error",
        ProviderCredentialRefreshRecoveryAction::Reauthorize => "reauthorization_required",
        ProviderCredentialRefreshRecoveryAction::FixConfiguration => "configuration_required",
        ProviderCredentialRefreshRecoveryAction::Investigate => "investigation_required",
    }
    .to_string();
    state.last_error = failure.status.message().to_string();
    state.recovery_action = failure.recovery_action as i32;
    state.failure_code = failure.failure_code.to_string();
    state.provider_error_subtype = failure
        .provider_error_subtype
        .unwrap_or_default()
        .to_string();
    state.last_error_at_ms = now_ms;
    state.next_refresh_at_ms = match failure.retry_schedule {
        RefreshRetrySchedule::Short => {
            now_ms.saturating_add(REFRESH_ERROR_RETRY_SECONDS.saturating_mul(1000))
        }
        RefreshRetrySchedule::Configuration => {
            now_ms.saturating_add(REFRESH_CONFIGURATION_RETRY_SECONDS.saturating_mul(1000))
        }
        RefreshRetrySchedule::Parked => i64::MAX,
    };
    let new_version = persist_refresh_state_if_current(store, state, expected_version)
        .await?
        .ok_or_else(|| {
            Status::aborted(
                "provider refresh was deleted or superseded while recording a refresh error",
            )
        })?;
    if let Some(metadata) = state.metadata.as_mut() {
        metadata.resource_version = new_version;
    }
    Ok(new_version)
}

fn validate_secret_material_references(
    state: &StoredProviderCredentialRefreshState,
) -> Result<(), Status> {
    let mut missing: Vec<_> = state
        .secret_material_keys
        .iter()
        .filter(|key| {
            !state.material.contains_key(*key) && !state.secret_material_handles.contains_key(*key)
        })
        .cloned()
        .collect();
    if missing.is_empty() {
        return Ok(());
    }
    missing.sort();
    missing.dedup();
    Err(Status::failed_precondition(format!(
        "provider refresh secret material is missing both inline values and credential handles for {}; a mixed-version gateway upgrade may have discarded the handles, so restore the refresh state or reconfigure the grant",
        missing.join(", ")
    )))
}

pub async fn refresh_provider_credential(
    store: &Store,
    workspace: &str,
    credentials: &crate::credentials::CredentialRuntime,
    compute: Option<&crate::compute::ComputeRuntime>,
    provider_name: &str,
    credential_key: &str,
) -> Result<StoredProviderCredentialRefreshState, Status> {
    let provider = store
        .get_message_by_name::<Provider>(workspace, provider_name)
        .await
        .map_err(|e| Status::internal(format!("fetch provider failed: {e}")))?
        .ok_or_else(|| Status::not_found("provider not found"))?;
    let Some(state) =
        get_refresh_state(store, workspace, provider.object_id(), credential_key).await?
    else {
        return Err(Status::not_found("provider refresh state not found"));
    };
    if state
        .metadata
        .as_ref()
        .is_some_and(|metadata| metadata.deletion_time.is_some())
    {
        return Err(Status::failed_precondition(
            "provider refresh is being deleted",
        ));
    }
    let mut state = state;
    validate_secret_material_references(&state)?;
    // 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);
    let expected_version = match cleanup_pending_secret_deletions(
        store,
        Some(credentials),
        &mut state,
        expected_version,
    )
    .await
    {
        Ok(new_version) => new_version,
        Err(err) => {
            warn!(
                provider = %state.provider_name,
                credential_key = %state.credential_key,
                error = %err,
                "provider refresh material cleanup failed; continuing with live refresh"
            );
            expected_version
        }
    };

    info!(
        provider = %state.provider_name,
        credential_key = %state.credential_key,
        strategy = %refresh_strategy_name(state.strategy),
        status = %state.status,
        expires_at_ms = state.expires_at_ms,
        next_refresh_at_ms = state.next_refresh_at_ms,
        "provider credential refresh started"
    );

    let mint_result = match resolve_refresh_material(Some(credentials), &state).await {
        Ok(transient_state) => mint_credential(&transient_state).await,
        Err(err) => Err(err.into()),
    };
    match mint_result {
        Ok(minted) => {
            let now_ms = current_time_ms();
            let mut staged_refresh_token_handles = HashMap::new();
            if let Some(ref refresh_token) = minted.refresh_token {
                if !state
                    .secret_material_keys
                    .iter()
                    .any(|key| key == "refresh_token")
                {
                    state.secret_material_keys.push("refresh_token".to_string());
                }
                let material =
                    HashMap::from([("refresh_token".to_string(), refresh_token.clone())]);
                let staging_id = format!(
                    "{}-refresh-material-{}",
                    state.object_id(),
                    uuid::Uuid::new_v4()
                );
                staged_refresh_token_handles = match credentials
                    .store_refresh_material_with_object_id(
                        refresh_material_scope(&state),
                        &staging_id,
                        &material,
                        &HashMap::new(),
                    )
                    .await
                {
                    Ok(handles) => handles,
                    Err(store_err) => {
                        let failure = RefreshFailure::reauthorize(
                            Status::failed_precondition(format!(
                                "the OAuth provider rotated the refresh token, but the replacement could not be stored; the grant must be re-authorized: {}",
                                store_err.message()
                            )),
                            "oauth_rotated_refresh_token_store_failed",
                            None,
                        );
                        persist_refresh_failure_state(
                            store,
                            &mut state,
                            expected_version,
                            &failure,
                        )
                        .await?;
                        return Err(failure.into_status());
                    }
                };
                let Some(handle) = staged_refresh_token_handles.get("refresh_token").cloned()
                else {
                    let failure = RefreshFailure::reauthorize(
                        Status::failed_precondition(
                            "the OAuth provider rotated the refresh token, but the credential driver returned no replacement handle; the grant must be re-authorized",
                        ),
                        "oauth_rotated_refresh_token_handle_missing",
                        None,
                    );
                    cleanup_staged_refresh_material_handles(
                        credentials,
                        &state,
                        &staged_refresh_token_handles,
                    )
                    .await;
                    persist_refresh_failure_state(store, &mut state, expected_version, &failure)
                        .await?;
                    return Err(failure.into_status());
                };
                if let Some(previous) = state
                    .secret_material_handles
                    .insert("refresh_token".to_string(), handle)
                {
                    enqueue_pending_secret_deletion(&mut state, "refresh_token", previous);
                }
                state.material.remove("refresh_token");
            }
            state.expires_at_ms = minted.expires_at_ms;
            set_refresh_expiration_presence(&mut state, true);
            state.next_refresh_at_ms =
                next_refresh_at_ms(minted.expires_at_ms, state.refresh_before.as_ref());
            state.last_refresh_at_ms = now_ms;
            state.status = "refreshed".to_string();
            state.last_error.clear();
            state.recovery_action = ProviderCredentialRefreshRecoveryAction::Unspecified as i32;
            state.failure_code.clear();
            state.provider_error_subtype.clear();
            state.last_error_at_ms = 0;

            // Claim the refresh generation with a version-matched write BEFORE
            // touching the provider. It succeeds only if the refresh still holds
            // the generation we started from; if it was deleted, recreated, or a
            // concurrent rotation won, it returns `None` and we leave both the
            // refresh state and the provider unchanged — no credentials minted
            // from a stale generation are written, and a deleted refresh is not
            // resurrected (CWE-362). This makes generation ownership the gate on
            // the provider credential write.
            let new_version = match persist_refresh_state_if_current(
                store,
                &state,
                expected_version,
            )
            .await
            {
                Ok(Some(new_version)) => new_version,
                Ok(None) => {
                    if !staged_refresh_token_handles.is_empty() {
                        cleanup_staged_refresh_material_handles(
                            credentials,
                            &state,
                            &staged_refresh_token_handles,
                        )
                        .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; discarding minted credentials"
                    );
                    return Err(Status::aborted(
                        "provider refresh was deleted or superseded during rotation",
                    ));
                }
                Err(err) => {
                    // The replacement refresh token is already in credential
                    // storage. Retry the same CAS with error/backoff state so a
                    // transient database failure does not discard the only
                    // upstream-valid grant. If that also fails, leave the
                    // staged object intact for operator recovery rather than
                    // deleting an irreplaceable rotated token.
                    persist_retryable_refresh_error_state(
                        store,
                        &mut state,
                        expected_version,
                        &err,
                    )
                    .await?;
                    return Err(err);
                }
            };

            // Generation is ours; write the minted credentials into the provider.
            if let Err(err) = apply_minted_credential(
                store,
                workspace,
                Some(credentials),
                compute,
                &provider,
                credential_key,
                &minted,
            )
            .await
            {
                // Reflect the failure on the state we just wrote; skip silently
                // if it was deleted concurrently (it is not recreated).
                let failure = RefreshFailure::from_status(&err);
                persist_refresh_failure_state(store, &mut state, new_version, &failure).await?;
                warn!(
                    provider = %state.provider_name,
                    credential_key = %state.credential_key,
                    strategy = %refresh_strategy_name(state.strategy),
                    status = %state.status,
                    next_refresh_at_ms = state.next_refresh_at_ms,
                    seconds_until_refresh = seconds_until_ms(now_ms, state.next_refresh_at_ms),
                    error = %err,
                    "provider credential refresh errored"
                );
                return Err(err);
            }
            info!(
                provider = %state.provider_name,
                credential_key = %state.credential_key,
                strategy = %refresh_strategy_name(state.strategy),
                status = %state.status,
                expires_at_ms = state.expires_at_ms,
                next_refresh_at_ms = state.next_refresh_at_ms,
                seconds_until_refresh = seconds_until_ms(now_ms, state.next_refresh_at_ms),
                "provider credential refresh completed"
            );
            if !state.pending_secret_deletions.is_empty()
                && let Err(err) = cleanup_pending_secret_deletions(
                    store,
                    Some(credentials),
                    &mut state,
                    new_version,
                )
                .await
            {
                warn!(
                    provider = %state.provider_name,
                    credential_key = %state.credential_key,
                    error = %err,
                    "failed to clean up replaced refresh material; retrying on the next sweep"
                );
            }
            Ok(state)
        }
        Err(failure) => {
            let now_ms = current_time_ms();
            persist_refresh_failure_state(store, &mut state, expected_version, &failure).await?;
            warn!(
                provider = %state.provider_name,
                credential_key = %state.credential_key,
                strategy = %refresh_strategy_name(state.strategy),
                status = %state.status,
                next_refresh_at_ms = state.next_refresh_at_ms,
                seconds_until_refresh = seconds_until_ms(now_ms, state.next_refresh_at_ms),
                recovery_action = ?failure.recovery_action,
                failure_code = failure.failure_code,
                error = %failure.status,
                "provider credential refresh errored"
            );
            Err(failure.into_status())
        }
    }
}

async fn cleanup_staged_refresh_material_handles(
    credentials: &crate::credentials::CredentialRuntime,
    state: &StoredProviderCredentialRefreshState,
    handles: &HashMap<String, CredentialHandle>,
) {
    if let Err(err) = credentials
        .delete_refresh_material_handles(refresh_material_scope(state), handles)
        .await
    {
        warn!(
            provider = %state.provider_name,
            credential_key = %state.credential_key,
            error = %err,
            "failed to clean up staged provider refresh material"
        );
    }
}

async fn apply_minted_credential(
    store: &Store,
    workspace: &str,
    credentials: Option<&crate::credentials::CredentialRuntime>,
    compute: Option<&crate::compute::ComputeRuntime>,
    provider: &Provider,
    credential_key: &str,
    minted: &MintedCredential,
) -> Result<(), Status> {
    let mut updated = provider.clone();
    let staging_id = format!("{}-refresh-{}", provider.object_id(), uuid::Uuid::new_v4());
    let staged_handles = if let Some(credentials) = credentials
        && credentials.stores_provider_credentials()
    {
        if let Some(compute) = compute {
            compute.ensure_workspace(workspace).await?;
        }
        let mut creds_to_store =
            HashMap::from([(credential_key.to_string(), minted.access_token.clone())]);
        for (key, value) in &minted.additional_credentials {
            creds_to_store.insert(key.clone(), value.clone());
        }
        // Stage under new handles with a unique staging ID to ensure we don't overwrite
        // the still-committed values before validation/CAS succeeds
        let staged = credentials
            .store_provider_credentials_with_object_id(
                provider.object_name(),
                provider.object_workspace(),
                provider.object_id(),
                &staging_id,
                &creds_to_store,
                &HashMap::new(), // Empty map forces creation of new handles
            )
            .await?;
        if !staged.contains_key(credential_key) {
            cleanup_staged_refresh_handles(credentials, provider, &staged).await;
            return Err(Status::internal(
                "credential driver did not return refreshed credential handle",
            ));
        }
        for (key, handle) in &staged {
            updated.credentials.remove(key);
            updated
                .credential_handles
                .insert(key.clone(), handle.clone());
        }
        Some(staged)
    } else {
        updated
            .credentials
            .insert(credential_key.to_string(), minted.access_token.clone());
        for (key, value) in &minted.additional_credentials {
            updated.credentials.insert(key.clone(), value.clone());
        }
        None
    };
    let credential_expiration_time =
        openshell_core::time::optional_timestamp_from_legacy_millis(minted.expires_at_ms)
            .map_err(|error| Status::internal(error.to_string()))?;
    if let Some(expiration_time) = credential_expiration_time.as_ref() {
        updated
            .credential_expiration_times
            .insert(credential_key.to_string(), *expiration_time);
        for key in minted.additional_credentials.keys() {
            updated
                .credential_expiration_times
                .insert(key.clone(), *expiration_time);
        }
    } else {
        updated.credential_expiration_times.remove(credential_key);
        for key in minted.additional_credentials.keys() {
            updated.credential_expiration_times.remove(key);
        }
    }
    // Acquire the shared sandbox mutation boundary only around validation and
    // persistence, after any remote minting or credential staging. This
    // prevents route status from committing against the old provider revision
    // after the rotation writes, without holding the guard across network I/O.
    let _sandbox_sync_guard = if let Some(compute) = compute {
        Some(compute.sandbox_sync_guard().await)
    } else {
        None
    };
    if let Err(err) = crate::grpc::provider::validate_provider_update_against_attached_sandboxes(
        store, workspace, &updated,
    )
    .await
    {
        if let Some(credentials) = credentials
            && let Some(handles) = &staged_handles
        {
            cleanup_staged_refresh_handles(credentials, provider, handles).await;
        }
        return Err(err);
    }

    // Capture only handles actually replaced in the CAS snapshot. This avoids
    // deleting unchanged sibling handles and remains correct if another refresh
    // updated the provider after this refresh began.
    let mut old_handles_to_delete = HashMap::new();
    let cas_result = store
        .update_message_cas::<Provider, _>(provider.object_id(), 0, |current| {
            if let Some(handles) = staged_handles.clone() {
                for (key, handle) in &handles {
                    current.credentials.remove(key);
                    if let Some(old_handle) = current
                        .credential_handles
                        .insert(key.clone(), handle.clone())
                        && old_handle != *handle
                    {
                        old_handles_to_delete.insert(key.clone(), old_handle);
                    }
                }
            } else {
                current
                    .credentials
                    .insert(credential_key.to_string(), minted.access_token.clone());
                for (key, value) in &minted.additional_credentials {
                    current.credentials.insert(key.clone(), value.clone());
                }
            }
            if let Some(expiration_time) = credential_expiration_time.as_ref() {
                current
                    .credential_expiration_times
                    .insert(credential_key.to_string(), *expiration_time);
                for key in minted.additional_credentials.keys() {
                    current
                        .credential_expiration_times
                        .insert(key.clone(), *expiration_time);
                }
            } else {
                current.credential_expiration_times.remove(credential_key);
                for key in minted.additional_credentials.keys() {
                    current.credential_expiration_times.remove(key);
                }
            }
        })
        .await
        .map(|_| ())
        .map_err(|e| {
            Status::internal(format!("persist refreshed provider credential failed: {e}"))
        });
    if cas_result.is_err()
        && let Some(credentials) = credentials
        && let Some(ref handles) = staged_handles
    {
        cleanup_staged_refresh_handles(credentials, provider, handles).await;
    }

    // If CAS succeeded and we have old handles to delete, clean them up
    if cas_result.is_ok()
        && !old_handles_to_delete.is_empty()
        && let Some(credentials) = credentials
        && let Err(cleanup_err) = credentials
            .delete_provider_credential_handles(
                provider.object_name(),
                provider.object_workspace(),
                provider.object_id(),
                &old_handles_to_delete,
            )
            .await
    {
        warn!(
            provider_name = %provider.object_name(),
            error = %cleanup_err,
            "failed to clean up old provider credential handles after successful refresh"
        );
        // Don't fail the operation - the refresh succeeded, this is just cleanup
    }

    cas_result
}

async fn cleanup_staged_refresh_handles(
    credentials: &crate::credentials::CredentialRuntime,
    provider: &Provider,
    handles: &HashMap<String, CredentialHandle>,
) {
    if let Err(cleanup_err) = credentials
        .delete_provider_credential_handles(
            provider.object_name(),
            provider.object_workspace(),
            provider.object_id(),
            handles,
        )
        .await
    {
        warn!(
            provider_name = %provider.object_name(),
            error = %cleanup_err,
            "failed to clean up staged provider credentials after refresh failure"
        );
    }
}

async fn mint_credential(
    state: &StoredProviderCredentialRefreshState,
) -> Result<MintedCredential, RefreshFailure> {
    let strategy = ProviderCredentialRefreshStrategy::try_from(state.strategy)
        .unwrap_or(ProviderCredentialRefreshStrategy::Unspecified);
    match strategy {
        ProviderCredentialRefreshStrategy::Oauth2RefreshToken => {
            mint_oauth2_refresh_token(state).await
        }
        ProviderCredentialRefreshStrategy::Oauth2ClientCredentials => {
            mint_oauth2_client_credentials(state).await
        }
        ProviderCredentialRefreshStrategy::GoogleServiceAccountJwt => {
            mint_google_service_account_jwt(state).await
        }
        ProviderCredentialRefreshStrategy::AwsStsAssumeRole => {
            mint_aws_sts_assume_role(state).await
        }
        ProviderCredentialRefreshStrategy::External
        | ProviderCredentialRefreshStrategy::Static
        | ProviderCredentialRefreshStrategy::Unspecified => Err(Status::failed_precondition(
            format!("refresh strategy '{strategy:?}' cannot be minted by the gateway"),
        )
        .into()),
    }
}

async fn mint_oauth2_refresh_token(
    state: &StoredProviderCredentialRefreshState,
) -> Result<MintedCredential, RefreshFailure> {
    let token_url = oauth2_token_url(state)?;
    let client_id = required_material(&state.material, "client_id")?;
    let refresh_token = required_material(&state.material, "refresh_token")?;
    let mut form = vec![
        ("grant_type".to_string(), "refresh_token".to_string()),
        ("client_id".to_string(), client_id),
        ("refresh_token".to_string(), refresh_token),
    ];
    if let Some(client_secret) = material_value(&state.material, &["client_secret"]) {
        form.push(("client_secret".to_string(), client_secret));
    }
    let scope = refresh_scopes(state).join(" ");
    if !scope.is_empty() {
        form.push(("scope".to_string(), scope));
    }

    request_token(
        &token_url,
        &form,
        max_lifetime_seconds(state),
        OAuthGrantKind::UserRefreshToken,
    )
    .await
}

async fn mint_oauth2_client_credentials(
    state: &StoredProviderCredentialRefreshState,
) -> Result<MintedCredential, RefreshFailure> {
    let token_url = oauth2_token_url(state)?;
    let client_id = required_material(&state.material, "client_id")?;
    let client_secret = required_material(&state.material, "client_secret")?;
    let mut form = vec![
        ("grant_type".to_string(), "client_credentials".to_string()),
        ("client_id".to_string(), client_id),
        ("client_secret".to_string(), client_secret),
    ];
    let scope = refresh_scopes(state).join(" ");
    if !scope.is_empty() {
        form.push(("scope".to_string(), scope));
    }

    request_token(
        &token_url,
        &form,
        max_lifetime_seconds(state),
        OAuthGrantKind::NonInteractive,
    )
    .await
}

async fn mint_google_service_account_jwt(
    state: &StoredProviderCredentialRefreshState,
) -> Result<MintedCredential, RefreshFailure> {
    crate::install_jsonwebtoken_crypto_provider();
    let token_url = google_token_url(state);
    let client_email = required_material(&state.material, "client_email")?;
    let private_key = required_material(&state.material, "private_key")?;
    let scopes = refresh_scopes(state);
    if scopes.is_empty() {
        return Err(Status::invalid_argument(
            "google_service_account_jwt requires at least one scope",
        )
        .into());
    }
    let now_ms = current_time_ms();
    let now_secs = now_ms / 1000;
    let lifetime_secs = max_lifetime_seconds(state).min(DEFAULT_MAX_LIFETIME_SECONDS);
    let subject = material_value(&state.material, &["subject", "sub"]);
    let claims = GoogleServiceAccountClaims {
        iss: &client_email,
        scope: scopes.join(" "),
        aud: &token_url,
        iat: now_secs,
        exp: now_secs.saturating_add(lifetime_secs),
        sub: subject.as_deref(),
    };
    let assertion = jsonwebtoken::encode(
        &jsonwebtoken::Header::new(jsonwebtoken::Algorithm::RS256),
        &claims,
        &jsonwebtoken::EncodingKey::from_rsa_pem(private_key.as_bytes()).map_err(|_| {
            Status::invalid_argument("google_service_account_jwt private_key must be RSA PEM")
        })?,
    )
    .map_err(|_| Status::internal("sign google service account jwt failed"))?;
    let form = vec![
        (
            "grant_type".to_string(),
            "urn:ietf:params:oauth:grant-type:jwt-bearer".to_string(),
        ),
        ("assertion".to_string(), assertion),
    ];
    request_token(
        &token_url,
        &form,
        lifetime_secs,
        OAuthGrantKind::NonInteractive,
    )
    .await
}

async fn mint_aws_sts_assume_role(
    state: &StoredProviderCredentialRefreshState,
) -> Result<MintedCredential, RefreshFailure> {
    let role_arn = required_material(&state.material, "role_arn")?;
    let session_name = material_value(&state.material, &["session_name"])
        .unwrap_or_else(|| "openshell-sandbox".to_string());
    let external_id = material_value(&state.material, &["external_id"]);
    let region =
        material_value(&state.material, &["aws_region"]).unwrap_or_else(|| "us-east-1".to_string());

    let region_provider = aws_sdk_sts::config::Region::new(region);
    let mut config_loader =
        aws_config::defaults(aws_config::BehaviorVersion::latest()).region(region_provider);

    // Explicit source credentials are all-or-nothing. A lone key must not
    // silently fall through to the gateway's ambient identity (CWE-20): the
    // caller asked for a specific principal, so an incomplete pair is an error.
    // An optional session token supports temporary source credentials (SSO or a
    // prior AssumeRole); it requires the access/secret pair.
    let session_token = material_value(&state.material, &["aws_session_token"]);
    match (
        material_value(&state.material, &["aws_access_key_id"]),
        material_value(&state.material, &["aws_secret_access_key"]),
    ) {
        (Some(access_key), Some(secret_key)) => {
            let creds = aws_sdk_sts::config::Credentials::new(
                access_key,
                secret_key,
                session_token,
                None,
                "openshell-provider-refresh",
            );
            config_loader = config_loader.credentials_provider(creds);
        }
        (None, None) if session_token.is_some() => {
            return Err(Status::invalid_argument(
                "aws_session_token requires aws_access_key_id and aws_secret_access_key",
            )
            .into());
        }
        (None, None) => {}
        _ => {
            return Err(Status::invalid_argument(
                "aws_access_key_id and aws_secret_access_key must both be set or both omitted",
            )
            .into());
        }
    }

    let sdk_config = config_loader.load().await;
    let sts_config = {
        let mut builder = aws_sdk_sts::config::Builder::from(&sdk_config);
        // Endpoint overrides exist only to point tests at a local mock STS. In
        // production the endpoint is always resolved from the region so a caller
        // cannot redirect an AWS-signed AssumeRole request at an arbitrary
        // service (CWE-918). See `test_sts_endpoint_override`.
        if let Some(endpoint) = test_sts_endpoint_override(state) {
            builder = builder.endpoint_url(endpoint);
        }
        builder.build()
    };
    let client = aws_sdk_sts::Client::from_conf(sts_config);

    let max_lifetime_i64 = max_lifetime_seconds(state);
    let max_lifetime = i32::try_from(max_lifetime_i64.min(i64::from(i32::MAX))).unwrap_or(i32::MAX);
    let max_lifetime_ms = i64::from(max_lifetime).saturating_mul(1000);

    let mut req = client
        .assume_role()
        .role_arn(&role_arn)
        .role_session_name(&session_name)
        .duration_seconds(max_lifetime);

    if let Some(eid) = external_id {
        req = req.external_id(eid);
    }

    let resp = req
        .send()
        .await
        .map_err(|e| Status::internal(format!("STS AssumeRole failed: {e}")))?;

    let creds = resp
        .credentials()
        .ok_or_else(|| Status::internal("STS AssumeRole response missing credentials"))?;

    let access_key_id = creds.access_key_id().to_string();
    let secret_access_key = creds.secret_access_key().to_string();
    let session_token = creds.session_token().to_string();

    let now_ms = current_time_ms();
    let max_expires = now_ms.saturating_add(max_lifetime_ms);
    let expires_at_ms = creds.expiration().to_millis().unwrap_or(max_expires);
    let expires_at_ms = expires_at_ms.min(max_expires);

    // Map STS response fields to the env keys the profile bound to each semantic
    // output. Configure pins these from the profile's additional_outputs, so a
    // missing mapping means the state was not configured against a valid AWS STS
    // profile binding; refuse rather than guessing standard AWS names.
    let output_values = [
        ("secret_access_key", secret_access_key),
        ("session_token", session_token),
    ];
    let mut additional = HashMap::new();
    for (output_id, value) in output_values {
        let env_key = state.additional_output_keys.get(output_id).ok_or_else(|| {
            Status::failed_precondition(format!(
                "refresh state missing resolved output key for '{output_id}'; reconfigure the AWS STS refresh"
            ))
        })?;
        additional.insert(env_key.clone(), value);
    }

    Ok(MintedCredential {
        access_token: access_key_id,
        expires_at_ms,
        refresh_token: None,
        additional_credentials: additional,
    })
}

async fn request_token(
    token_url: &str,
    form: &[(String, String)],
    max_lifetime_seconds: i64,
    grant_kind: OAuthGrantKind,
) -> Result<MintedCredential, RefreshFailure> {
    let parsed = reqwest::Url::parse(token_url)
        .map_err(|_| Status::invalid_argument("token_url must be an absolute URL"))?;
    match parsed.scheme() {
        "https" => {}
        "http" if parsed.host_str().is_some_and(is_loopback_host) => {}
        _ => {
            return Err(Status::invalid_argument(
                "token_url must use https, except loopback http for local tests",
            )
            .into());
        }
    }

    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()
        .map_err(|e| Status::internal(format!("build refresh HTTP client failed: {e}")))?;
    let response = client
        .post(parsed)
        .form(form)
        .send()
        .await
        .map_err(|error| {
            let error_kind = if error.is_timeout() {
                "timeout"
            } else if error.is_connect() {
                "connect"
            } else if error.is_request() {
                "request"
            } else {
                "other"
            };
            warn!(
                error = %error.without_url(),
                error_kind,
                "OAuth token endpoint request failed"
            );
            RefreshFailure::retryable(
                Status::unavailable("token endpoint request failed"),
                "oauth_token_endpoint_unavailable",
            )
        })?;
    let status = response.status();
    if !status.is_success() {
        let body = read_bounded_oauth_error_body(response).await;
        return Err(classify_oauth_token_error(status, &body, grant_kind));
    }
    let token = response.json::<TokenResponse>().await.map_err(|_| {
        RefreshFailure::investigate(
            Status::failed_precondition("token endpoint returned invalid JSON"),
            "oauth_invalid_success_response",
        )
    })?;
    if token.access_token.trim().is_empty() {
        return Err(RefreshFailure::investigate(
            Status::failed_precondition("token endpoint returned empty access_token"),
            "oauth_empty_access_token",
        ));
    }
    let now_ms = current_time_ms();
    let lifetime_cap_seconds = if max_lifetime_seconds > 0 {
        max_lifetime_seconds
    } else {
        DEFAULT_MAX_LIFETIME_SECONDS
    };
    let lifetime_seconds = token
        .expires_in
        .filter(|value| *value > 0)
        .unwrap_or(lifetime_cap_seconds);
    let lifetime_seconds = lifetime_seconds.min(lifetime_cap_seconds);
    Ok(MintedCredential {
        access_token: token.access_token,
        expires_at_ms: now_ms.saturating_add(lifetime_seconds.saturating_mul(1000)),
        refresh_token: token
            .refresh_token
            .filter(|refresh_token| !refresh_token.trim().is_empty()),
        additional_credentials: HashMap::new(),
    })
}

async fn read_bounded_oauth_error_body(mut response: reqwest::Response) -> Vec<u8> {
    let mut body = Vec::new();
    while body.len() < MAX_OAUTH_ERROR_RESPONSE_BYTES {
        let Ok(Some(chunk)) = response.chunk().await else {
            break;
        };
        let remaining = MAX_OAUTH_ERROR_RESPONSE_BYTES.saturating_sub(body.len());
        body.extend_from_slice(&chunk[..chunk.len().min(remaining)]);
        if chunk.len() >= remaining {
            break;
        }
    }
    body
}

fn classify_oauth_token_error(
    status: reqwest::StatusCode,
    body: &[u8],
    grant_kind: OAuthGrantKind,
) -> RefreshFailure {
    if status == reqwest::StatusCode::TOO_MANY_REQUESTS || status.is_server_error() {
        return RefreshFailure::retryable(
            Status::unavailable(format!("token endpoint returned HTTP {status}")),
            "oauth_token_endpoint_retryable",
        );
    }

    let Ok(error_response) = serde_json::from_slice::<OAuthErrorResponse>(body) else {
        return RefreshFailure::investigate(
            Status::failed_precondition(format!(
                "token endpoint returned HTTP {status} without a recognized OAuth error"
            )),
            "oauth_unrecognized_error_response",
        );
    };

    match error_response.error.as_str() {
        "invalid_grant" if grant_kind == OAuthGrantKind::UserRefreshToken => {
            let subtype = error_response
                .error_subtype
                .as_deref()
                .filter(|subtype| *subtype == "invalid_rapt")
                .map(|_| "invalid_rapt");
            let message = if subtype.is_some() {
                "OAuth refresh grant requires interactive reauthorization (invalid_grant/invalid_rapt)"
            } else {
                "OAuth refresh grant is no longer usable (invalid_grant); user reauthorization is required"
            };
            RefreshFailure::reauthorize(
                Status::failed_precondition(message),
                "oauth_invalid_grant",
                subtype,
            )
        }
        "invalid_client" => RefreshFailure::fix_configuration(
            Status::failed_precondition(
                "OAuth token endpoint rejected the client configuration (invalid_client)",
            ),
            "oauth_invalid_client",
        ),
        "unauthorized_client" => RefreshFailure::fix_configuration(
            Status::failed_precondition(
                "OAuth token endpoint rejected the client grant (unauthorized_client)",
            ),
            "oauth_unauthorized_client",
        ),
        "invalid_scope" => RefreshFailure::fix_configuration(
            Status::failed_precondition(
                "OAuth token endpoint rejected the configured scopes (invalid_scope)",
            ),
            "oauth_invalid_scope",
        ),
        "unsupported_grant_type" => RefreshFailure::fix_configuration(
            Status::failed_precondition(
                "OAuth token endpoint rejected the configured grant type (unsupported_grant_type)",
            ),
            "oauth_unsupported_grant_type",
        ),
        "admin_policy_enforced" => RefreshFailure::fix_configuration(
            Status::failed_precondition(
                "OAuth access is blocked by an administrator policy (admin_policy_enforced)",
            ),
            "oauth_admin_policy_enforced",
        ),
        "access_denied"
            if error_response.error_subtype.as_deref() == Some("admin_policy_enforced") =>
        {
            RefreshFailure::fix_configuration_with_subtype(
                Status::failed_precondition(
                    "OAuth access is blocked by an administrator policy (access_denied/admin_policy_enforced)",
                ),
                "oauth_admin_policy_enforced",
                "admin_policy_enforced",
            )
        }
        "access_denied" if grant_kind == OAuthGrantKind::UserRefreshToken => {
            RefreshFailure::reauthorize(
                Status::failed_precondition(
                    "OAuth access was denied; user reauthorization is required (access_denied)",
                ),
                "oauth_access_denied",
                None,
            )
        }
        "access_denied" => RefreshFailure::fix_configuration(
            Status::failed_precondition(
                "OAuth access was denied for the non-interactive grant (access_denied)",
            ),
            "oauth_access_denied",
        ),
        "invalid_grant" => RefreshFailure::fix_configuration(
            Status::failed_precondition(
                "OAuth token endpoint rejected the non-interactive grant (invalid_grant)",
            ),
            "oauth_invalid_grant",
        ),
        "server_error" | "temporarily_unavailable" => RefreshFailure::retryable(
            Status::unavailable("OAuth token endpoint reported a temporary failure"),
            "oauth_token_endpoint_retryable",
        ),
        _ => RefreshFailure::investigate(
            Status::failed_precondition(format!(
                "token endpoint returned HTTP {status} with an unrecognized OAuth error"
            )),
            "oauth_unrecognized_error",
        ),
    }
}

pub fn refresh_scopes(state: &StoredProviderCredentialRefreshState) -> Vec<String> {
    if !state.scopes.is_empty() {
        return state.scopes.clone();
    }
    material_scopes(&state.material)
}

pub fn material_scopes(material: &HashMap<String, String>) -> Vec<String> {
    material_value(material, &["scope", "scopes"])
        .map(|raw| {
            raw.split(|ch: char| ch == ',' || ch.is_ascii_whitespace())
                .map(str::trim)
                .filter(|scope| !scope.is_empty())
                .map(ToString::to_string)
                .collect()
        })
        .unwrap_or_default()
}

pub fn parse_material_i64(
    material: &HashMap<String, String>,
    key: &str,
) -> Result<Option<i64>, Status> {
    let Some(value) = material_value(material, &[key]) else {
        return Ok(None);
    };
    value
        .parse::<i64>()
        .map(Some)
        .map_err(|_| Status::invalid_argument(format!("{key} material must be a signed integer")))
}

fn oauth2_token_url(state: &StoredProviderCredentialRefreshState) -> Result<String, Status> {
    if let Some(tenant_id) = material_value(&state.material, &["tenant_id"]) {
        return Ok(format!(
            "https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
        ));
    }
    if !state.token_url.trim().is_empty() {
        return Ok(state.token_url.clone());
    }
    Err(Status::invalid_argument(
        "oauth2_client_credentials requires token_url or tenant_id material",
    ))
}

fn google_token_url(state: &StoredProviderCredentialRefreshState) -> String {
    if state.token_url.trim().is_empty() {
        "https://oauth2.googleapis.com/token".to_string()
    } else {
        state.token_url.clone()
    }
}

fn required_material(material: &HashMap<String, String>, key: &str) -> Result<String, Status> {
    material_value(material, &[key])
        .ok_or_else(|| Status::invalid_argument(format!("{key} material is required")))
}

fn material_value(material: &HashMap<String, String>, keys: &[&str]) -> Option<String> {
    for key in keys {
        if let Some(value) = material.get(*key).map(|value| value.trim())
            && !value.is_empty()
        {
            return Some(value.to_string());
        }
    }
    None
}

fn is_loopback_host(host: &str) -> bool {
    matches!(host, "localhost" | "127.0.0.1" | "::1")
}

/// Test-only STS endpoint override. Reads the `sts_endpoint_url` material and
/// accepts it only when it is a loopback URL, so unit tests can target a local
/// mock STS. Compiled out of production builds entirely: the configure boundary
/// also rejects the `sts_endpoint_url` material key, so it can never reach a
/// stored refresh state outside tests.
#[cfg(test)]
fn test_sts_endpoint_override(state: &StoredProviderCredentialRefreshState) -> Option<String> {
    material_value(&state.material, &["sts_endpoint_url"]).filter(|endpoint| {
        reqwest::Url::parse(endpoint)
            .ok()
            .and_then(|parsed| parsed.host_str().map(is_loopback_host))
            .unwrap_or(false)
    })
}

#[cfg(not(test))]
#[allow(clippy::missing_const_for_fn)]
fn test_sts_endpoint_override(_state: &StoredProviderCredentialRefreshState) -> Option<String> {
    None
}

pub fn spawn_refresh_worker(state: std::sync::Arc<crate::ServerState>, interval: Duration) {
    info!(
        interval_seconds = interval.as_secs(),
        "provider credential refresh worker started"
    );
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(interval);
        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            ticker.tick().await;
            match Box::pin(run_refresh_worker_tick(
                state.store.as_ref(),
                Some(&state.credentials),
                Some(&state.compute),
            ))
            .await
            {
                Ok(workspaces) => {
                    for workspace in workspaces {
                        crate::config_delivery::publish_workspace_components(
                            &state,
                            &workspace,
                            crate::config_delivery::ConfigComponents::ALL,
                        );
                    }
                }
                Err(err) => {
                    warn!(error = %err, "provider credential refresh worker tick failed");
                }
            }
        }
    });
}

#[tracing::instrument(
    name = "refresh",
    skip_all,
    fields(
        otel.name = "refresh.provider_credentials",
        watched_count = tracing::field::Empty,
        due_count = tracing::field::Empty,
    )
)]
async fn run_refresh_worker_tick(
    store: &Store,
    credentials: Option<&crate::credentials::CredentialRuntime>,
    compute: Option<&crate::compute::ComputeRuntime>,
) -> Result<std::collections::HashSet<String>, Status> {
    let mut changed_workspaces = std::collections::HashSet::new();
    let now_ms = current_time_ms();
    let states = list_all_refresh_states(store).await.inspect_err(|_| {
        crate::otel_tracing::mark_error(&tracing::Span::current());
    })?;
    let watched_count = states.len();
    let due_count = states
        .iter()
        .filter(|state| state.next_refresh_at_ms <= 0 || state.next_refresh_at_ms <= now_ms)
        .count();
    let rotation_requested_count = states
        .iter()
        .filter(|state| state.status == "rotation_requested")
        .count();
    let span = tracing::Span::current();
    span.record("watched_count", watched_count);
    span.record("due_count", due_count);
    info!(
        watched_count,
        due_count, rotation_requested_count, "provider credential refresh worker sweep"
    );
    for state in states {
        if state
            .metadata
            .as_ref()
            .is_some_and(|metadata| metadata.deletion_time.is_some())
        {
            let Some(credentials) = credentials else {
                warn!(
                    provider = %state.provider_name,
                    credential_key = %state.credential_key,
                    "cannot finalize tombstoned provider refresh without credential runtime"
                );
                continue;
            };
            if let Err(err) = delete_refresh_state_with_credentials(
                store,
                credentials,
                state.object_workspace(),
                &state.provider_id,
                &state.credential_key,
            )
            .await
            {
                warn!(
                    provider = %state.provider_name,
                    credential_key = %state.credential_key,
                    error = %err,
                    "failed to finalize tombstoned provider refresh; retrying on the next sweep"
                );
            } else {
                changed_workspaces.insert(state.object_workspace().to_string());
            }
            continue;
        }
        let mut state = state;
        let expected_version = state
            .metadata
            .as_ref()
            .map_or(0, |metadata| metadata.resource_version);
        if let Err(err) =
            cleanup_pending_secret_deletions(store, credentials, &mut state, expected_version).await
        {
            warn!(
                provider = %state.provider_name,
                credential_key = %state.credential_key,
                error = %err,
                "provider refresh material cleanup failed; continuing with live refresh"
            );
        }
        let strategy = ProviderCredentialRefreshStrategy::try_from(state.strategy)
            .unwrap_or(ProviderCredentialRefreshStrategy::Unspecified);
        if !is_gateway_mintable_strategy(strategy) {
            warn!(
                provider = %state.provider_name,
                credential_key = %state.credential_key,
                strategy = %refresh_strategy_name(state.strategy),
                status = %state.status,
                "skipping non-gateway-mintable provider credential refresh state"
            );
            continue;
        }
        let due = state.next_refresh_at_ms <= 0 || state.next_refresh_at_ms <= now_ms;
        let rotation_requested = state.status == "rotation_requested";
        info!(
            provider = %state.provider_name,
            credential_key = %state.credential_key,
            strategy = %refresh_strategy_name(state.strategy),
            status = %state.status,
            expires_at_ms = state.expires_at_ms,
            seconds_until_expiry = seconds_until_ms(now_ms, state.expires_at_ms),
            next_refresh_at_ms = state.next_refresh_at_ms,
            last_refresh_at_ms = state.last_refresh_at_ms,
            seconds_until_refresh = seconds_until_ms(now_ms, state.next_refresh_at_ms),
            due,
            rotation_requested,
            "provider credential refresh watch"
        );
        if !due && !rotation_requested {
            continue;
        }
        let Some(credentials) = credentials else {
            warn!(
                provider = %state.provider_name,
                credential_key = %state.credential_key,
                "cannot refresh provider credential without credential runtime"
            );
            continue;
        };
        info!(
            provider = %state.provider_name,
            credential_key = %state.credential_key,
            strategy = %refresh_strategy_name(state.strategy),
            status = %state.status,
            "refreshing provider credential"
        );
        if let Err(err) = refresh_provider_credential(
            store,
            state.object_workspace(),
            credentials,
            compute,
            &state.provider_name,
            &state.credential_key,
        )
        .await
        {
            warn!(
                provider = %state.provider_name,
                credential_key = %state.credential_key,
                strategy = %refresh_strategy_name(state.strategy),
                status = %state.status,
                next_refresh_at_ms = state.next_refresh_at_ms,
                error = %err,
                "provider credential refresh failed"
            );
        } else {
            changed_workspaces.insert(state.object_workspace().to_string());
        }
    }
    Ok(changed_workspaces)
}

#[cfg(test)]
mod tests {
    use super::{
        MAX_OAUTH_ERROR_RESPONSE_BYTES, NewRefreshStateConfig, OAuthGrantKind, RefreshFailure,
        RefreshRetrySchedule, Status, classify_oauth_token_error,
        delete_refresh_state_with_credentials, effective_authorization_epoch,
        enqueue_pending_secret_deletion, get_refresh_state, list_all_refresh_states,
        list_refresh_states_for_provider, max_lifetime_seconds, new_refresh_state,
        next_refresh_at_ms, put_refresh_state, read_bounded_oauth_error_body,
        refresh_has_expiration, refresh_material_scope, refresh_provider_credential,
        refresh_state_name, refresh_status_from_state, refresh_strategy_name,
        run_refresh_worker_tick, seconds_until_ms, set_refresh_expiration_presence,
        validate_secret_material_references,
    };
    use crate::credentials::CredentialRuntime;
    use crate::persistence::{current_time_ms, test_store};
    use crate::storage_proto::StoredProviderCredentialRefreshStateV2 as StoredProviderCredentialRefreshState;

    fn proto_duration(seconds: i64) -> prost_types::Duration {
        prost_types::Duration { seconds, nanos: 0 }
    }

    #[test]
    fn exact_refresh_durations_preserve_fractional_values_until_protocol_boundaries() {
        let refresh_before = prost_types::Duration {
            seconds: 0,
            nanos: 500_000_000,
        };
        assert_eq!(next_refresh_at_ms(10_000, Some(&refresh_before)), 9_500);
        let submillisecond = prost_types::Duration {
            seconds: 0,
            nanos: 1,
        };
        assert_eq!(next_refresh_at_ms(10_000, Some(&submillisecond)), 9_999);

        let state = StoredProviderCredentialRefreshState {
            max_lifetime: Some(refresh_before),
            ..Default::default()
        };
        assert_eq!(max_lifetime_seconds(&state), 1);
    }
    use openshell_core::Config;
    use openshell_core::proto::datamodel::v1::ObjectMeta;
    use openshell_core::proto::{
        CredentialHandle, Provider, ProviderCredentialRefreshRecoveryAction,
        ProviderCredentialRefreshStrategy, Sandbox, SandboxSpec,
    };
    use openshell_core::{ObjectId, ObjectName, ObjectWorkspace};
    use std::collections::HashMap;
    use wiremock::matchers::{body_string_contains, method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    fn ts(milliseconds: i64) -> prost_types::Timestamp {
        openshell_core::time::timestamp_from_millis(milliseconds).unwrap()
    }

    fn test_credentials() -> CredentialRuntime {
        CredentialRuntime::from_config(&Config::new(None).with_credential_drivers(["test-static"]))
            .expect("test credential runtime")
    }

    #[test]
    fn refresh_state_name_preserves_distinct_credential_keys() {
        let provider_id = "provider-id";

        assert_ne!(
            refresh_state_name(provider_id, "API_KEY"),
            refresh_state_name(provider_id, "api_key")
        );
        assert_ne!(
            refresh_state_name(provider_id, " alex-api "),
            refresh_state_name(provider_id, " alex_api")
        );
        assert_ne!(
            refresh_state_name(provider_id, "Alex-API"),
            refresh_state_name(provider_id, "alex-api")
        );
    }

    #[tokio::test]
    async fn refresh_state_lists_hydrate_authoritative_resource_versions() {
        let store = test_store().await;
        let provider_id = "provider-id";
        let state = StoredProviderCredentialRefreshState {
            metadata: Some(ObjectMeta {
                id: "refresh-id".to_string(),
                name: refresh_state_name(provider_id, "ACCESS_TOKEN"),
                workspace: "default".to_string(),
                ..Default::default()
            }),
            provider_id: provider_id.to_string(),
            credential_key: "ACCESS_TOKEN".to_string(),
            ..Default::default()
        };
        put_refresh_state(&store, &state).await.unwrap();

        let scoped = list_refresh_states_for_provider(&store, provider_id)
            .await
            .unwrap();
        let all = list_all_refresh_states(&store).await.unwrap();
        assert_eq!(scoped[0].metadata.as_ref().unwrap().resource_version, 1);
        assert_eq!(all[0].metadata.as_ref().unwrap().resource_version, 1);
    }

    #[test]
    fn new_refresh_configuration_rotates_authorization_epoch_and_legacy_state_is_stable() {
        let provider = Provider {
            metadata: Some(ObjectMeta {
                id: "provider-id".to_string(),
                name: "provider".to_string(),
                workspace: "default".to_string(),
                ..Default::default()
            }),
            ..Default::default()
        };
        let config = || NewRefreshStateConfig {
            strategy: ProviderCredentialRefreshStrategy::Oauth2RefreshToken,
            material: HashMap::new(),
            secret_material_keys: Vec::new(),
            expires_at_ms: 0,
            token_url: "https://issuer.example/token".to_string(),
            scopes: vec!["scope".to_string()],
            refresh_before: Some(proto_duration(300)),
            max_lifetime: Some(proto_duration(3600)),
            additional_output_keys: HashMap::new(),
        };
        let first = new_refresh_state(&provider, "default", "ACCESS_TOKEN", config())
            .expect("first refresh configuration");
        let second = new_refresh_state(&provider, "default", "ACCESS_TOKEN", config())
            .expect("second refresh configuration");
        assert!(!first.authorization_epoch.is_empty());
        assert_ne!(first.authorization_epoch, second.authorization_epoch);

        let mut legacy = first;
        legacy.authorization_epoch.clear();
        assert_eq!(
            effective_authorization_epoch(&legacy).expect("legacy migration epoch"),
            legacy.metadata.as_ref().expect("metadata").id
        );
    }

    #[test]
    fn refresh_log_helpers_format_safe_operational_fields() {
        assert_eq!(seconds_until_ms(1_000, 61_000), 60);
        assert_eq!(seconds_until_ms(61_000, 1_000), 0);
        assert_eq!(seconds_until_ms(1_000, 0), 0);
        assert_eq!(
            refresh_strategy_name(ProviderCredentialRefreshStrategy::Oauth2RefreshToken as i32),
            "oauth2_refresh_token"
        );
        assert_eq!(
            refresh_strategy_name(
                ProviderCredentialRefreshStrategy::Oauth2ClientCredentials as i32
            ),
            "oauth2_client_credentials"
        );
        assert_eq!(
            refresh_strategy_name(
                ProviderCredentialRefreshStrategy::GoogleServiceAccountJwt as i32
            ),
            "google_service_account_jwt"
        );
        assert_eq!(refresh_strategy_name(i32::MAX), "unspecified");
    }

    #[test]
    fn refresh_expiration_presence_distinguishes_absent_epoch_and_legacy_values() {
        let mut state = StoredProviderCredentialRefreshState {
            metadata: Some(ObjectMeta::default()),
            ..Default::default()
        };
        assert!(!refresh_has_expiration(&state));
        assert!(refresh_status_from_state(&state).expiration_time.is_none());

        set_refresh_expiration_presence(&mut state, true);
        assert!(refresh_has_expiration(&state));
        assert_eq!(
            refresh_status_from_state(&state).expiration_time,
            Some(ts(0))
        );

        set_refresh_expiration_presence(&mut state, false);
        assert!(!refresh_has_expiration(&state));

        state.expires_at_ms = 1;
        assert!(refresh_has_expiration(&state));
        assert_eq!(
            refresh_status_from_state(&state).expiration_time,
            Some(ts(1))
        );
    }

    #[test]
    fn local_refresh_statuses_default_to_safe_recovery_actions() {
        for status in [
            Status::invalid_argument("missing material"),
            Status::failed_precondition("strategy cannot be minted"),
            Status::permission_denied("client is not allowed"),
        ] {
            let failure = RefreshFailure::from(status);
            assert_eq!(
                failure.recovery_action,
                ProviderCredentialRefreshRecoveryAction::FixConfiguration
            );
            assert_eq!(failure.failure_code, "refresh_configuration_invalid");
            assert_eq!(failure.retry_schedule, RefreshRetrySchedule::Configuration);
        }

        let retry = RefreshFailure::from(Status::unavailable("temporary backend outage"));
        assert_eq!(
            retry.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Retry
        );
        assert_eq!(retry.retry_schedule, RefreshRetrySchedule::Short);

        let investigate = RefreshFailure::from(Status::unknown("unclassified failure"));
        assert_eq!(
            investigate.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Investigate
        );
        assert_eq!(investigate.retry_schedule, RefreshRetrySchedule::Short);
    }

    #[test]
    fn pending_secret_deletions_preserve_multiple_generations_for_one_key() {
        let mut state = StoredProviderCredentialRefreshState::default();
        for handle in ["first", "second"] {
            enqueue_pending_secret_deletion(
                &mut state,
                "refresh_token",
                CredentialHandle {
                    driver: "test-static".to_string(),
                    handle: handle.to_string(),
                    metadata: HashMap::new(),
                },
            );
        }

        assert_eq!(state.pending_secret_deletions.len(), 2);
        assert_eq!(
            state.pending_secret_deletions[0]
                .handle
                .as_ref()
                .unwrap()
                .handle,
            "first"
        );
        assert_eq!(
            state.pending_secret_deletions[1]
                .handle
                .as_ref()
                .unwrap()
                .handle,
            "second"
        );
    }

    #[test]
    fn refresh_rejects_secret_material_lost_by_mixed_version_gateway() {
        let mut state = StoredProviderCredentialRefreshState {
            secret_material_keys: vec!["refresh_token".to_string()],
            ..Default::default()
        };
        let err = validate_secret_material_references(&state).unwrap_err();
        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
        assert!(err.message().contains("mixed-version gateway"));
        assert!(err.message().contains("refresh_token"));

        state.secret_material_handles.insert(
            "refresh_token".to_string(),
            CredentialHandle {
                driver: "test-static".to_string(),
                handle: "stored-token".to_string(),
                metadata: HashMap::new(),
            },
        );
        validate_secret_material_references(&state).unwrap();
    }

    #[test]
    fn oauth_invalid_grant_requires_user_reauthorization_without_exposing_description() {
        let failure = classify_oauth_token_error(
            reqwest::StatusCode::BAD_REQUEST,
            br#"{"error":"invalid_grant","error_subtype":"invalid_rapt","error_description":"sensitive provider detail"}"#,
            OAuthGrantKind::UserRefreshToken,
        );

        assert_eq!(
            failure.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Reauthorize
        );
        assert_eq!(failure.failure_code, "oauth_invalid_grant");
        assert_eq!(failure.provider_error_subtype, Some("invalid_rapt"));
        assert_eq!(failure.retry_schedule, RefreshRetrySchedule::Parked);
        assert!(
            failure
                .status
                .message()
                .contains("interactive reauthorization")
        );
        assert!(
            !failure
                .status
                .message()
                .contains("sensitive provider detail")
        );
    }

    #[test]
    fn oauth_invalid_grant_ignores_non_string_optional_subtype() {
        let failure = classify_oauth_token_error(
            reqwest::StatusCode::BAD_REQUEST,
            br#"{"error":"invalid_grant","error_subtype":{"vendor":"value"}}"#,
            OAuthGrantKind::UserRefreshToken,
        );

        assert_eq!(
            failure.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Reauthorize
        );
        assert_eq!(failure.failure_code, "oauth_invalid_grant");
        assert_eq!(failure.provider_error_subtype, None);
        assert_eq!(failure.retry_schedule, RefreshRetrySchedule::Parked);
    }

    #[test]
    fn oauth_noninteractive_invalid_grant_requires_configuration_fix() {
        let failure = classify_oauth_token_error(
            reqwest::StatusCode::BAD_REQUEST,
            br#"{"error":"invalid_grant"}"#,
            OAuthGrantKind::NonInteractive,
        );

        assert_eq!(
            failure.recovery_action,
            ProviderCredentialRefreshRecoveryAction::FixConfiguration
        );
        assert_eq!(failure.failure_code, "oauth_invalid_grant");
        assert_eq!(failure.retry_schedule, RefreshRetrySchedule::Configuration);
    }

    #[test]
    fn oauth_admin_policy_subtype_requires_configuration_fix() {
        let failure = classify_oauth_token_error(
            reqwest::StatusCode::BAD_REQUEST,
            br#"{"error":"access_denied","error_subtype":"admin_policy_enforced"}"#,
            OAuthGrantKind::UserRefreshToken,
        );

        assert_eq!(
            failure.recovery_action,
            ProviderCredentialRefreshRecoveryAction::FixConfiguration
        );
        assert_eq!(failure.failure_code, "oauth_admin_policy_enforced");
        assert_eq!(
            failure.provider_error_subtype,
            Some("admin_policy_enforced")
        );
        assert_eq!(failure.retry_schedule, RefreshRetrySchedule::Configuration);
    }

    #[test]
    fn oauth_access_denied_recovery_depends_on_grant_kind() {
        let user_failure = classify_oauth_token_error(
            reqwest::StatusCode::FORBIDDEN,
            br#"{"error":"access_denied"}"#,
            OAuthGrantKind::UserRefreshToken,
        );
        assert_eq!(
            user_failure.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Reauthorize
        );
        assert_eq!(user_failure.failure_code, "oauth_access_denied");
        assert_eq!(user_failure.retry_schedule, RefreshRetrySchedule::Parked);

        let service_failure = classify_oauth_token_error(
            reqwest::StatusCode::FORBIDDEN,
            br#"{"error":"access_denied"}"#,
            OAuthGrantKind::NonInteractive,
        );
        assert_eq!(
            service_failure.recovery_action,
            ProviderCredentialRefreshRecoveryAction::FixConfiguration
        );
        assert_eq!(service_failure.failure_code, "oauth_access_denied");
        assert_eq!(
            service_failure.retry_schedule,
            RefreshRetrySchedule::Configuration
        );
    }

    #[test]
    fn oauth_server_failure_remains_retryable() {
        let failure = classify_oauth_token_error(
            reqwest::StatusCode::SERVICE_UNAVAILABLE,
            br#"{"error":"invalid_grant"}"#,
            OAuthGrantKind::UserRefreshToken,
        );

        assert_eq!(
            failure.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Retry
        );
        assert_eq!(failure.failure_code, "oauth_token_endpoint_retryable");
        assert_eq!(failure.retry_schedule, RefreshRetrySchedule::Short);
    }

    #[test]
    fn unrecognized_oauth_error_requests_investigation_without_echoing_body() {
        let failure = classify_oauth_token_error(
            reqwest::StatusCode::BAD_REQUEST,
            br#"{"error":"vendor_secret_error","error_description":"do not expose me"}"#,
            OAuthGrantKind::UserRefreshToken,
        );

        assert_eq!(
            failure.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Investigate
        );
        assert_eq!(failure.failure_code, "oauth_unrecognized_error");
        assert_eq!(failure.retry_schedule, RefreshRetrySchedule::Short);
        assert!(!failure.status.message().contains("vendor_secret_error"));
        assert!(!failure.status.message().contains("do not expose me"));
    }

    #[test]
    fn html_and_malformed_oauth_errors_are_investigated_without_echoing_body() {
        for body in [b"<html>issuer failure</html>".as_slice(), b"{".as_slice()] {
            let failure = classify_oauth_token_error(
                reqwest::StatusCode::BAD_REQUEST,
                body,
                OAuthGrantKind::UserRefreshToken,
            );

            assert_eq!(
                failure.recovery_action,
                ProviderCredentialRefreshRecoveryAction::Investigate
            );
            assert_eq!(failure.failure_code, "oauth_unrecognized_error_response");
            assert!(!failure.status.message().contains("issuer failure"));
        }
    }

    #[tokio::test]
    async fn oversized_oauth_error_body_is_bounded_and_investigated() {
        let mock_server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/oversized"))
            .respond_with(ResponseTemplate::new(400).set_body_bytes(vec![
                b'x';
                MAX_OAUTH_ERROR_RESPONSE_BYTES
                    + 512
            ]))
            .mount(&mock_server)
            .await;

        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
        let response = reqwest::get(format!("{}/oversized", mock_server.uri()))
            .await
            .unwrap();
        let status = response.status();
        let body = read_bounded_oauth_error_body(response).await;
        assert_eq!(body.len(), MAX_OAUTH_ERROR_RESPONSE_BYTES);

        let failure = classify_oauth_token_error(status, &body, OAuthGrantKind::UserRefreshToken);
        assert_eq!(
            failure.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Investigate
        );
        assert_eq!(failure.failure_code, "oauth_unrecognized_error_response");
    }

    #[tokio::test]
    async fn invalid_local_oauth_configuration_retries_hourly() {
        let store = test_store().await;
        let provider = provider("invalid-local-config", "outlook");
        store.put_message(&provider).await.unwrap();
        let state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                strategy: ProviderCredentialRefreshStrategy::Oauth2RefreshToken,
                material: HashMap::from([
                    ("client_id".to_string(), "client-id".to_string()),
                    ("refresh_token".to_string(), "refresh-token".to_string()),
                ]),
                secret_material_keys: vec!["refresh_token".to_string()],
                expires_at_ms: 0,
                token_url: "not-an-absolute-url".to_string(),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
                additional_output_keys: HashMap::new(),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();

        let err = refresh_provider_credential(
            &store,
            "default",
            &test_credentials(),
            None,
            "invalid-local-config",
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap_err();

        assert_eq!(err.code(), tonic::Code::InvalidArgument);
        let stored = get_refresh_state(
            &store,
            "default",
            provider.object_id(),
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap()
        .unwrap();
        assert_eq!(stored.status, "configuration_required");
        assert_eq!(
            stored.recovery_action,
            ProviderCredentialRefreshRecoveryAction::FixConfiguration as i32
        );
        assert_eq!(stored.failure_code, "refresh_configuration_invalid");
        assert_eq!(
            stored.next_refresh_at_ms - stored.last_error_at_ms,
            60 * 60 * 1000
        );
    }

    #[tokio::test]
    async fn oauth_invalid_grant_persists_terminal_reauthorization_status() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .respond_with(ResponseTemplate::new(400).set_body_json(serde_json::json!({
                "error": "invalid_grant",
                "error_subtype": "invalid_rapt",
                "error_description": "provider-controlled detail"
            })))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let provider = provider("expired-grant", "outlook");
        store.put_message(&provider).await.unwrap();
        let state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                strategy: ProviderCredentialRefreshStrategy::Oauth2RefreshToken,
                material: HashMap::from([
                    ("client_id".to_string(), "client-id".to_string()),
                    (
                        "refresh_token".to_string(),
                        "expired-refresh-token".to_string(),
                    ),
                ]),
                secret_material_keys: vec!["refresh_token".to_string()],
                expires_at_ms: 0,
                token_url: format!("{}/token", mock_server.uri()),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
                additional_output_keys: HashMap::new(),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();

        let err = refresh_provider_credential(
            &store,
            "default",
            &test_credentials(),
            None,
            "expired-grant",
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap_err();

        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
        let stored = get_refresh_state(
            &store,
            "default",
            provider.object_id(),
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap()
        .unwrap();
        assert_eq!(stored.status, "reauthorization_required");
        assert_eq!(stored.next_refresh_at_ms, i64::MAX);
        assert_eq!(
            stored.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Reauthorize as i32
        );
        assert_eq!(stored.failure_code, "oauth_invalid_grant");
        assert_eq!(stored.provider_error_subtype, "invalid_rapt");
        assert!(stored.last_error_at_ms > 0);
        assert!(!stored.last_error.contains("provider-controlled detail"));
    }

    #[tokio::test]
    async fn oauth2_client_credentials_refresh_mints_and_persists_access_token() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .and(body_string_contains("grant_type=client_credentials"))
            .and(body_string_contains("client_id=client-id"))
            .and(body_string_contains(
                "scope=https%3A%2F%2Fgraph.microsoft.com%2F.default",
            ))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "minted-graph-token",
                "expires_in": 3600,
                "token_type": "Bearer"
            })))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let provider = provider("my-graph", "outlook");
        store.put_message(&provider).await.unwrap();
        let before_refresh_ms = current_time_ms();
        let mut state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                additional_output_keys: HashMap::new(),
                strategy: ProviderCredentialRefreshStrategy::Oauth2ClientCredentials,
                material: HashMap::from([
                    ("client_id".to_string(), "client-id".to_string()),
                    ("client_secret".to_string(), "client-secret".to_string()),
                ]),
                secret_material_keys: vec!["client_secret".to_string()],
                expires_at_ms: 0,
                token_url: format!("{}/token", mock_server.uri()),
                scopes: vec!["https://graph.microsoft.com/.default".to_string()],
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
            },
        )
        .unwrap();
        state.status = "investigation_required".to_string();
        state.last_error = "safe prior error".to_string();
        state.recovery_action = ProviderCredentialRefreshRecoveryAction::Investigate as i32;
        state.failure_code = "oauth_unrecognized_error".to_string();
        state.provider_error_subtype = "prior_subtype".to_string();
        state.last_error_at_ms = current_time_ms();
        put_refresh_state(&store, &state).await.unwrap();
        let authorization_epoch = state.authorization_epoch.clone();
        let credentials = test_credentials();

        let refreshed = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "my-graph",
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap();
        assert_eq!(refreshed.authorization_epoch, authorization_epoch);
        assert_eq!(refreshed.status, "refreshed");
        assert!(refreshed.expires_at_ms > 0);
        assert!(refreshed.next_refresh_at_ms > 0);
        assert!(refreshed.expires_at_ms <= before_refresh_ms + 120_000);
        assert!(refreshed.last_error.is_empty());
        assert_eq!(
            refreshed.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Unspecified as i32
        );
        assert!(refreshed.failure_code.is_empty());
        assert!(refreshed.provider_error_subtype.is_empty());
        assert_eq!(refreshed.last_error_at_ms, 0);

        let stored = store
            .get_message_by_name::<Provider>("default", "my-graph")
            .await
            .unwrap()
            .unwrap();
        let resolved = credentials
            .resolve_provider_handles(&stored, current_time_ms())
            .await
            .unwrap();
        assert_eq!(
            resolved.values.get("MS_GRAPH_ACCESS_TOKEN"),
            Some(&"minted-graph-token".to_string())
        );
        assert_eq!(
            stored
                .credential_expiration_times
                .get("MS_GRAPH_ACCESS_TOKEN"),
            Some(&ts(refreshed.expires_at_ms))
        );
    }

    #[tokio::test]
    async fn oauth2_client_credentials_refresh_stores_access_token_with_credential_runtime() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "stored-graph-token",
                "expires_in": 3600,
                "token_type": "Bearer"
            })))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let provider = provider("my-stored-graph", "outlook");
        store.put_message(&provider).await.unwrap();
        let mut state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                strategy: ProviderCredentialRefreshStrategy::Oauth2ClientCredentials,
                material: HashMap::from([
                    ("client_id".to_string(), "client-id".to_string()),
                    ("client_secret".to_string(), "client-secret".to_string()),
                ]),
                secret_material_keys: vec!["client_secret".to_string()],
                expires_at_ms: 0,
                token_url: format!("{}/token", mock_server.uri()),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
                additional_output_keys: HashMap::new(),
            },
        )
        .unwrap();
        let config = Config::new(None).with_credential_drivers(["test-static"]);
        let credentials = CredentialRuntime::from_config(&config).unwrap();
        state.secret_material_handles = credentials
            .store_refresh_material_with_object_id(
                refresh_material_scope(&state),
                "configured-client-secret",
                &HashMap::from([("client_secret".to_string(), "client-secret".to_string())]),
                &HashMap::new(),
            )
            .await
            .unwrap();
        state.material.remove("client_secret");
        put_refresh_state(&store, &state).await.unwrap();
        let authorization_epoch = state.authorization_epoch.clone();

        let refreshed = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "my-stored-graph",
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap();
        assert_eq!(refreshed.authorization_epoch, authorization_epoch);
        let stored_refresh = get_refresh_state(
            &store,
            "default",
            provider.object_id(),
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap()
        .unwrap();
        assert!(!stored_refresh.material.contains_key("client_secret"));
        assert!(
            stored_refresh
                .secret_material_handles
                .contains_key("client_secret")
        );

        let stored = store
            .get_message_by_name::<Provider>("default", "my-stored-graph")
            .await
            .unwrap()
            .unwrap();
        assert!(!stored.credentials.contains_key("MS_GRAPH_ACCESS_TOKEN"));
        let handle = stored
            .credential_handles
            .get("MS_GRAPH_ACCESS_TOKEN")
            .unwrap();
        assert_eq!(handle.driver, "test-static");
        assert_eq!(
            stored
                .credential_expiration_times
                .get("MS_GRAPH_ACCESS_TOKEN"),
            Some(&ts(refreshed.expires_at_ms))
        );

        let resolved = credentials
            .resolve_provider_handles(&stored, current_time_ms())
            .await
            .unwrap();
        assert_eq!(
            resolved.values.get("MS_GRAPH_ACCESS_TOKEN"),
            Some(&"stored-graph-token".to_string())
        );
    }

    #[tokio::test]
    async fn refresh_rejects_minted_credential_key_collision_for_attached_sandbox() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "minted-graph-token",
                "expires_in": 3600,
                "token_type": "Bearer"
            })))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let mut provider_a = provider("existing-graph", "outlook");
        provider_a.credentials.insert(
            "MS_GRAPH_ACCESS_TOKEN".to_string(),
            "existing-token".to_string(),
        );
        store.put_message(&provider_a).await.unwrap();
        let provider_b = provider("refreshing-graph", "outlook");
        store.put_message(&provider_b).await.unwrap();
        store
            .put_message(&Sandbox {
                metadata: Some(ObjectMeta {
                    id: "sandbox-collision".to_string(),
                    name: "collision".to_string(),
                    created_time: openshell_core::time::timestamp_from_millis(1).ok(),
                    labels: HashMap::new(),
                    resource_version: 0,
                    annotations: HashMap::new(),
                    workspace: "default".to_string(),
                    deletion_time: None,
                }),
                spec: Some(SandboxSpec {
                    providers: vec!["existing-graph".to_string(), "refreshing-graph".to_string()],
                    ..SandboxSpec::default()
                }),
                ..Default::default()
            })
            .await
            .unwrap();
        let state = new_refresh_state(
            &provider_b,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                additional_output_keys: HashMap::new(),
                strategy: ProviderCredentialRefreshStrategy::Oauth2ClientCredentials,
                material: HashMap::from([
                    ("client_id".to_string(), "client-id".to_string()),
                    ("client_secret".to_string(), "client-secret".to_string()),
                ]),
                secret_material_keys: vec!["client_secret".to_string()],
                expires_at_ms: 0,
                token_url: format!("{}/token", mock_server.uri()),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = test_credentials();

        let err = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "refreshing-graph",
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap_err();

        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
        assert!(err.message().contains("MS_GRAPH_ACCESS_TOKEN"));
        let stored_state = get_refresh_state(
            &store,
            "default",
            provider_b.object_id(),
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap()
        .unwrap();
        assert_eq!(stored_state.status, "configuration_required");
        assert_eq!(
            stored_state.recovery_action,
            ProviderCredentialRefreshRecoveryAction::FixConfiguration as i32
        );
        assert_eq!(stored_state.failure_code, "refresh_configuration_invalid");
        assert_eq!(
            stored_state.next_refresh_at_ms - stored_state.last_error_at_ms,
            60 * 60 * 1000
        );
        assert!(stored_state.last_error.contains("MS_GRAPH_ACCESS_TOKEN"));
        let stored_provider = store
            .get_message_by_name::<Provider>("default", "refreshing-graph")
            .await
            .unwrap()
            .unwrap();
        assert!(
            !stored_provider
                .credentials
                .contains_key("MS_GRAPH_ACCESS_TOKEN")
        );
    }

    #[tokio::test]
    async fn oauth2_refresh_token_refresh_mints_access_token_and_persists_rotated_refresh_token() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .and(body_string_contains("grant_type=refresh_token"))
            .and(body_string_contains("client_id=client-id"))
            .and(body_string_contains("refresh_token=old-refresh-token"))
            .and(body_string_contains(
                "scope=https%3A%2F%2Fgraph.microsoft.com%2F.default",
            ))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "delegated-graph-token",
                "refresh_token": "rotated-refresh-token",
                "expires_in": 3600,
                "token_type": "Bearer"
            })))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let provider = provider("my-delegated-graph", "outlook");
        store.put_message(&provider).await.unwrap();
        let state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                additional_output_keys: HashMap::new(),
                strategy: ProviderCredentialRefreshStrategy::Oauth2RefreshToken,
                material: HashMap::from([
                    ("client_id".to_string(), "client-id".to_string()),
                    ("refresh_token".to_string(), "old-refresh-token".to_string()),
                ]),
                secret_material_keys: vec!["refresh_token".to_string()],
                expires_at_ms: 0,
                token_url: format!("{}/token", mock_server.uri()),
                scopes: vec!["https://graph.microsoft.com/.default".to_string()],
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = CredentialRuntime::from_config(
            &Config::new(None).with_credential_drivers(["test-static"]),
        )
        .unwrap();

        let refreshed = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "my-delegated-graph",
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap();
        assert_eq!(refreshed.status, "refreshed");
        assert!(refreshed.expires_at_ms > 0);

        let stored_provider = store
            .get_message_by_name::<Provider>("default", "my-delegated-graph")
            .await
            .unwrap()
            .unwrap();
        assert!(
            !stored_provider
                .credentials
                .contains_key("MS_GRAPH_ACCESS_TOKEN")
        );
        assert!(
            stored_provider
                .credential_handles
                .contains_key("MS_GRAPH_ACCESS_TOKEN")
        );
        assert_eq!(
            stored_provider
                .credential_expiration_times
                .get("MS_GRAPH_ACCESS_TOKEN"),
            Some(&ts(refreshed.expires_at_ms))
        );

        let stored_state = get_refresh_state(
            &store,
            "default",
            provider.object_id(),
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap()
        .unwrap();
        assert!(!stored_state.material.contains_key("refresh_token"));
        assert!(
            stored_state
                .secret_material_handles
                .contains_key("refresh_token")
        );
        assert!(stored_state.pending_secret_deletions.is_empty());
        assert_eq!(
            credentials
                .resolve_refresh_material(
                    refresh_material_scope(&stored_state),
                    &stored_state.secret_material_handles,
                )
                .await
                .unwrap()
                .get("refresh_token"),
            Some(&"rotated-refresh-token".to_string())
        );
        assert!(
            stored_state
                .secret_material_keys
                .iter()
                .any(|key| key == "refresh_token")
        );
        assert_eq!(
            credentials.stored_credential_count(),
            Some(2),
            "only the access token and current refresh token remain"
        );
    }

    #[tokio::test]
    async fn refresh_continues_when_pending_secret_cleanup_temporarily_fails() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "minted-after-cleanup-failure",
                "expires_in": 3600,
                "token_type": "Bearer"
            })))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let provider = provider("cleanup-retry", "outlook");
        store.put_message(&provider).await.unwrap();
        let credentials = test_credentials();
        let mut state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                strategy: ProviderCredentialRefreshStrategy::Oauth2ClientCredentials,
                material: HashMap::from([
                    ("client_id".to_string(), "client-id".to_string()),
                    ("client_secret".to_string(), "client-secret".to_string()),
                ]),
                secret_material_keys: vec!["client_secret".to_string()],
                expires_at_ms: 0,
                token_url: format!("{}/token", mock_server.uri()),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
                additional_output_keys: HashMap::new(),
            },
        )
        .unwrap();
        state.secret_material_handles = credentials
            .store_refresh_material_with_object_id(
                refresh_material_scope(&state),
                "current-refresh-object",
                &HashMap::from([("client_secret".to_string(), "client-secret".to_string())]),
                &HashMap::new(),
            )
            .await
            .unwrap();
        state.material.remove("client_secret");
        let old = credentials
            .store_refresh_material_with_object_id(
                refresh_material_scope(&state),
                "old-refresh-object",
                &HashMap::from([("client_secret".to_string(), "obsolete".to_string())]),
                &HashMap::new(),
            )
            .await
            .unwrap()
            .remove("client_secret")
            .unwrap();
        enqueue_pending_secret_deletion(&mut state, "client_secret", old);
        put_refresh_state(&store, &state).await.unwrap();
        credentials.fail_next_delete();

        let refreshed = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "cleanup-retry",
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap();

        assert_eq!(refreshed.status, "refreshed");
        assert!(refreshed.pending_secret_deletions.is_empty());
        let stored = get_refresh_state(
            &store,
            "default",
            provider.object_id(),
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap()
        .unwrap();
        assert!(stored.pending_secret_deletions.is_empty());
        assert_eq!(
            credentials.stored_credential_count(),
            Some(2),
            "the current client secret and minted access token remain"
        );
    }

    #[tokio::test]
    async fn rotated_refresh_token_store_failure_requires_reauthorization() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "access-token-that-must-not-be-applied",
                "refresh_token": "replacement-refresh-token",
                "expires_in": 3600,
                "token_type": "Bearer"
            })))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let provider = provider("rotation-store-failure", "outlook");
        store.put_message(&provider).await.unwrap();
        let credentials = test_credentials();
        let mut state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                strategy: ProviderCredentialRefreshStrategy::Oauth2RefreshToken,
                material: HashMap::from([
                    ("client_id".to_string(), "client-id".to_string()),
                    ("refresh_token".to_string(), "old-refresh-token".to_string()),
                ]),
                secret_material_keys: vec!["refresh_token".to_string()],
                expires_at_ms: 0,
                token_url: format!("{}/token", mock_server.uri()),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
                additional_output_keys: HashMap::new(),
            },
        )
        .unwrap();
        state.secret_material_handles = credentials
            .store_refresh_material_with_object_id(
                refresh_material_scope(&state),
                "original-grant",
                &HashMap::from([("refresh_token".to_string(), "old-refresh-token".to_string())]),
                &HashMap::new(),
            )
            .await
            .unwrap();
        state.material.remove("refresh_token");
        put_refresh_state(&store, &state).await.unwrap();
        credentials.fail_next_store();

        let err = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "rotation-store-failure",
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap_err();

        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
        assert!(err.message().contains("must be re-authorized"));
        let stored = get_refresh_state(
            &store,
            "default",
            provider.object_id(),
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap()
        .unwrap();
        assert_eq!(stored.status, "reauthorization_required");
        assert!(stored.last_error.contains("must be re-authorized"));
        assert_eq!(stored.next_refresh_at_ms, i64::MAX);
        assert_eq!(
            stored.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Reauthorize as i32
        );
        assert_eq!(
            stored.failure_code,
            "oauth_rotated_refresh_token_store_failed"
        );
        assert_eq!(credentials.stored_credential_count(), Some(1));
        assert_eq!(
            credentials
                .resolve_refresh_material(
                    refresh_material_scope(&stored),
                    &stored.secret_material_handles,
                )
                .await
                .unwrap()
                .get("refresh_token"),
            Some(&"old-refresh-token".to_string())
        );
        let stored_provider = store
            .get_message_by_name::<Provider>("default", "rotation-store-failure")
            .await
            .unwrap()
            .unwrap();
        assert!(stored_provider.credential_handles.is_empty());
    }

    #[tokio::test]
    async fn google_service_account_refresh_mints_and_persists_access_token() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .and(body_string_contains(
                "grant_type=urn%3Aietf%3Aparams%3Aoauth%3Agrant-type%3Ajwt-bearer",
            ))
            .and(body_string_contains("assertion="))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "minted-drive-token",
                "expires_in": 1800,
                "token_type": "Bearer"
            })))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let provider = provider("my-drive", "google-drive");
        store.put_message(&provider).await.unwrap();
        let state = new_refresh_state(
            &provider,
            "default",
            "GOOGLE_DRIVE_ACCESS_TOKEN",
            NewRefreshStateConfig {
                additional_output_keys: HashMap::new(),
                strategy: ProviderCredentialRefreshStrategy::GoogleServiceAccountJwt,
                material: HashMap::from([
                    (
                        "client_email".to_string(),
                        "svc@example.iam.gserviceaccount.com".to_string(),
                    ),
                    ("private_key".to_string(), TEST_RSA_PRIVATE_KEY.to_string()),
                ]),
                secret_material_keys: vec!["private_key".to_string()],
                expires_at_ms: 0,
                token_url: format!("{}/token", mock_server.uri()),
                scopes: vec!["https://www.googleapis.com/auth/drive.readonly".to_string()],
                refresh_before: Some(proto_duration(300)),
                max_lifetime: Some(proto_duration(3600)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = test_credentials();

        let refreshed = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "my-drive",
            "GOOGLE_DRIVE_ACCESS_TOKEN",
        )
        .await
        .unwrap();
        assert_eq!(refreshed.status, "refreshed");
        assert!(refreshed.expires_at_ms > 0);

        let stored = store
            .get_message_by_name::<Provider>("default", "my-drive")
            .await
            .unwrap()
            .unwrap();
        let resolved = credentials
            .resolve_provider_handles(&stored, current_time_ms())
            .await
            .unwrap();
        assert_eq!(
            resolved.values.get("GOOGLE_DRIVE_ACCESS_TOKEN"),
            Some(&"minted-drive-token".to_string())
        );
    }

    #[tokio::test]
    async fn refresh_worker_skips_non_gateway_mintable_strategies() {
        let store = test_store().await;
        let provider = provider("my-external", "outlook");
        store.put_message(&provider).await.unwrap();
        let state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                additional_output_keys: HashMap::new(),
                strategy: ProviderCredentialRefreshStrategy::External,
                material: HashMap::new(),
                secret_material_keys: Vec::new(),
                expires_at_ms: 0,
                token_url: String::new(),
                scopes: Vec::new(),
                refresh_before: None,
                max_lifetime: None,
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();

        Box::pin(run_refresh_worker_tick(&store, None, None))
            .await
            .unwrap();

        let stored_state = get_refresh_state(
            &store,
            "default",
            provider.object_id(),
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap()
        .unwrap();
        assert_ne!(stored_state.status, "error");
        assert!(stored_state.last_error.is_empty());

        let stored_provider = store
            .get_message_by_name::<Provider>("default", "my-external")
            .await
            .unwrap()
            .unwrap();
        assert!(
            !stored_provider
                .credentials
                .contains_key("MS_GRAPH_ACCESS_TOKEN")
        );
    }

    #[tokio::test]
    async fn refresh_worker_skips_parked_reauthorization_state() {
        let store = test_store().await;
        let provider = provider("parked-refresh", "outlook");
        store.put_message(&provider).await.unwrap();
        let mut state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                strategy: ProviderCredentialRefreshStrategy::Oauth2RefreshToken,
                material: HashMap::new(),
                secret_material_keys: Vec::new(),
                expires_at_ms: 0,
                token_url: "https://issuer.example/token".to_string(),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
                additional_output_keys: HashMap::new(),
            },
        )
        .unwrap();
        state.status = "reauthorization_required".to_string();
        state.last_error = "OAuth refresh grant is no longer usable".to_string();
        state.recovery_action = ProviderCredentialRefreshRecoveryAction::Reauthorize as i32;
        state.failure_code = "oauth_invalid_grant".to_string();
        state.last_error_at_ms = current_time_ms();
        state.next_refresh_at_ms = i64::MAX;
        put_refresh_state(&store, &state).await.unwrap();

        Box::pin(run_refresh_worker_tick(
            &store,
            Some(&test_credentials()),
            None,
        ))
        .await
        .unwrap();

        let stored = get_refresh_state(
            &store,
            "default",
            provider.object_id(),
            "MS_GRAPH_ACCESS_TOKEN",
        )
        .await
        .unwrap()
        .unwrap();
        assert_eq!(stored.status, "reauthorization_required");
        assert_eq!(stored.next_refresh_at_ms, i64::MAX);
        assert_eq!(stored.failure_code, "oauth_invalid_grant");
        assert_eq!(
            stored.recovery_action,
            ProviderCredentialRefreshRecoveryAction::Reauthorize as i32
        );
    }

    #[tokio::test]
    async fn refresh_worker_finalizes_tombstoned_refresh_material() {
        let store = test_store().await;
        let provider = provider("tombstoned-refresh", "outlook");
        store.put_message(&provider).await.unwrap();
        let credentials = test_credentials();
        let mut state = new_refresh_state(
            &provider,
            "default",
            "MS_GRAPH_ACCESS_TOKEN",
            NewRefreshStateConfig {
                strategy: ProviderCredentialRefreshStrategy::Oauth2RefreshToken,
                material: HashMap::from([("refresh_token".to_string(), "delete-me".to_string())]),
                secret_material_keys: vec!["refresh_token".to_string()],
                expires_at_ms: 0,
                token_url: "https://issuer.example/token".to_string(),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(30)),
                max_lifetime: Some(proto_duration(60)),
                additional_output_keys: HashMap::new(),
            },
        )
        .unwrap();
        state.secret_material_handles = credentials
            .store_refresh_material_with_object_id(
                refresh_material_scope(&state),
                "tombstoned-grant",
                &state.material,
                &HashMap::new(),
            )
            .await
            .unwrap();
        state.material.clear();
        state.metadata.as_mut().unwrap().deletion_time =
            openshell_core::time::timestamp_from_millis(current_time_ms()).ok();
        state.status = "deleting".to_string();
        put_refresh_state(&store, &state).await.unwrap();
        assert_eq!(credentials.stored_credential_count(), Some(1));

        Box::pin(run_refresh_worker_tick(&store, Some(&credentials), None))
            .await
            .unwrap();

        assert!(
            get_refresh_state(
                &store,
                "default",
                provider.object_id(),
                "MS_GRAPH_ACCESS_TOKEN",
            )
            .await
            .unwrap()
            .is_none()
        );
        assert_eq!(credentials.stored_credential_count(), Some(0));
    }

    /// The worker ticks on a timer with no inbound request, so without a span
    /// of its own its store reads export as anonymous single-span traces.
    #[tokio::test]
    #[ignore = "flaky under concurrent test execution"]
    async fn refresh_worker_ticks_are_roots_and_store_operations_have_parents() {
        use crate::otel_tracing::test_exporter;

        let store = test_store().await;

        let traced = test_exporter::install_traced();
        Box::pin(run_refresh_worker_tick(&store, None, None))
            .await
            .unwrap();

        let spans = traced.finished_spans();
        let root = spans
            .iter()
            .find(|s| s.name == "refresh.provider_credentials")
            .unwrap_or_else(|| {
                panic!(
                    "the tick records a span of its own, got {:?}",
                    spans.iter().map(|s| &s.name).collect::<Vec<_>>()
                )
            });

        test_exporter::assert_is_root(root);
        let store_span = spans
            .iter()
            .find(|span| {
                span.name.starts_with("store.")
                    && span.span_context.trace_id() == root.span_context.trace_id()
            })
            .expect("the tick records its store operation");
        test_exporter::assert_has_parent(store_span);
    }

    #[test]
    fn refresh_strategy_name_includes_aws_sts() {
        assert_eq!(
            refresh_strategy_name(ProviderCredentialRefreshStrategy::AwsStsAssumeRole as i32),
            "aws_sts_assume_role"
        );
    }

    #[test]
    fn aws_sts_max_expires_saturates_on_saturated_clock() {
        assert_eq!(i64::MAX.saturating_add(3_600_000), i64::MAX);
        let now_ms = i64::MAX - 1_000;
        let max_lifetime_ms = 3_600_000;
        let max_expires = now_ms.saturating_add(max_lifetime_ms);
        assert_eq!(max_expires, i64::MAX);
        assert!(max_expires >= now_ms);
    }

    #[tokio::test]
    async fn aws_sts_assume_role_mints_three_credentials_from_mock_endpoint() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(body_string_contains("Action=AssumeRole"))
            .and(body_string_contains("RoleArn=arn"))
            .respond_with(ResponseTemplate::new(200).set_body_string(
                r#"<AssumeRoleResponse xmlns="https://sts.amazonaws.com/doc/2011-06-15/">
  <AssumeRoleResult>
    <AssumedRoleUser>
      <AssumedRoleId>AROA3XFRBF23:test-session</AssumedRoleId>
      <Arn>arn:aws:sts::123456789012:assumed-role/TestRole/test-session</Arn>
    </AssumedRoleUser>
    <Credentials>
      <AccessKeyId>ASIAMOCKKEY</AccessKeyId>
      <SecretAccessKey>MockSecretAccessKey123</SecretAccessKey>
      <SessionToken>MockSessionTokenXYZ</SessionToken>
      <Expiration>2099-01-01T00:00:00Z</Expiration>
    </Credentials>
  </AssumeRoleResult>
  <ResponseMetadata>
    <RequestId>01234567-89ab-cdef-0123-456789abcdef</RequestId>
  </ResponseMetadata>
</AssumeRoleResponse>"#,
            ))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let prov = provider("aws-sts-test", "aws");
        store.put_message(&prov).await.unwrap();

        let state = new_refresh_state(
            &prov,
            "default",
            "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(),
                    ),
                    ("session_name".to_string(), "test-session".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: Some(proto_duration(300)),
                max_lifetime: Some(proto_duration(3600)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = test_credentials();

        let refreshed = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "aws-sts-test",
            "AWS_ACCESS_KEY_ID",
        )
        .await
        .unwrap();
        assert_eq!(refreshed.status, "refreshed");
        assert!(refreshed.expires_at_ms > 0);

        let stored = store
            .get_message_by_name::<Provider>("default", "aws-sts-test")
            .await
            .unwrap()
            .unwrap();
        let resolved = credentials
            .resolve_provider_handles(&stored, current_time_ms())
            .await
            .unwrap();
        assert_eq!(
            resolved.values.get("AWS_ACCESS_KEY_ID"),
            Some(&"ASIAMOCKKEY".to_string())
        );
        assert_eq!(
            resolved.values.get("AWS_SECRET_ACCESS_KEY"),
            Some(&"MockSecretAccessKey123".to_string())
        );
        assert_eq!(
            resolved.values.get("AWS_SESSION_TOKEN"),
            Some(&"MockSessionTokenXYZ".to_string())
        );
    }

    #[tokio::test]
    async fn aws_sts_mint_writes_to_resolved_additional_output_keys() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(body_string_contains("Action=AssumeRole"))
            .respond_with(ResponseTemplate::new(200).set_body_string(
                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>"#,
            ))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let prov = provider("aws-sts-custom", "aws");
        store.put_message(&prov).await.unwrap();

        // The minter honors the resolved output->env-key map from state, not
        // hardcoded AWS names.
        let state = new_refresh_state(
            &prov,
            "default",
            "AWS_ACCESS_KEY_ID",
            NewRefreshStateConfig {
                additional_output_keys: HashMap::from([
                    ("secret_access_key".to_string(), "CUSTOM_SECRET".to_string()),
                    ("session_token".to_string(), "CUSTOM_SESSION".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: Some(proto_duration(300)),
                max_lifetime: Some(proto_duration(3600)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = test_credentials();

        refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "aws-sts-custom",
            "AWS_ACCESS_KEY_ID",
        )
        .await
        .unwrap();

        let stored = store
            .get_message_by_name::<Provider>("default", "aws-sts-custom")
            .await
            .unwrap()
            .unwrap();
        let resolved = credentials
            .resolve_provider_handles(&stored, current_time_ms())
            .await
            .unwrap();
        assert_eq!(
            resolved.values.get("AWS_ACCESS_KEY_ID"),
            Some(&"ASIAMOCKKEY".to_string())
        );
        assert_eq!(
            resolved.values.get("CUSTOM_SECRET"),
            Some(&"MockSecretAccessKey123".to_string())
        );
        assert_eq!(
            resolved.values.get("CUSTOM_SESSION"),
            Some(&"MockSessionTokenXYZ".to_string())
        );
        assert!(!resolved.values.contains_key("AWS_SECRET_ACCESS_KEY"));
    }

    #[tokio::test]
    async fn aws_sts_mint_rejects_partial_source_credentials() {
        let store = test_store().await;
        let prov = provider("aws-sts-partial", "aws");
        store.put_message(&prov).await.unwrap();

        // Only the access key half of the explicit source pair is present. The
        // mint must fail rather than fall back to the gateway's ambient identity.
        let state = new_refresh_state(
            &prov,
            "default",
            "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()),
                ]),
                secret_material_keys: Vec::new(),
                expires_at_ms: 0,
                token_url: String::new(),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(300)),
                max_lifetime: Some(proto_duration(3600)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = test_credentials();

        let err = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "aws-sts-partial",
            "AWS_ACCESS_KEY_ID",
        )
        .await
        .unwrap_err();
        assert_eq!(err.code(), tonic::Code::InvalidArgument);
        assert!(err.message().contains("both be set or both omitted"));

        let stored = store
            .get_message_by_name::<Provider>("default", "aws-sts-partial")
            .await
            .unwrap()
            .unwrap();
        assert!(!stored.credentials.contains_key("AWS_ACCESS_KEY_ID"));
    }

    #[tokio::test]
    async fn apply_minted_credential_writes_additional_keys() {
        use super::apply_minted_credential;

        let store = test_store().await;
        let mut prov = provider("aws-test", "aws");
        prov.credentials
            .insert("AWS_ACCESS_KEY_ID".to_string(), "old-key".to_string());
        store.put_message(&prov).await.unwrap();

        let minted = super::MintedCredential {
            access_token: "AKIAIOSFODNN7EXAMPLE".to_string(),
            expires_at_ms: 4_000_000_000_000,
            refresh_token: None,
            additional_credentials: HashMap::from([
                (
                    "AWS_SECRET_ACCESS_KEY".to_string(),
                    "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY".to_string(),
                ),
                (
                    "AWS_SESSION_TOKEN".to_string(),
                    "FwoGZXIvYXdzEBYaDH...EXAMPLETOKEN".to_string(),
                ),
            ]),
        };

        apply_minted_credential(
            &store,
            "default",
            None,
            None,
            &prov,
            "AWS_ACCESS_KEY_ID",
            &minted,
        )
        .await
        .unwrap();

        let stored = store
            .get_message_by_name::<Provider>("default", "aws-test")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            stored.credentials.get("AWS_ACCESS_KEY_ID"),
            Some(&"AKIAIOSFODNN7EXAMPLE".to_string())
        );
        assert_eq!(
            stored.credentials.get("AWS_SECRET_ACCESS_KEY"),
            Some(&"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY".to_string())
        );
        assert_eq!(
            stored.credentials.get("AWS_SESSION_TOKEN"),
            Some(&"FwoGZXIvYXdzEBYaDH...EXAMPLETOKEN".to_string())
        );
        assert_eq!(
            stored.credential_expiration_times.get("AWS_ACCESS_KEY_ID"),
            Some(&ts(4_000_000_000_000))
        );
        assert_eq!(
            stored
                .credential_expiration_times
                .get("AWS_SECRET_ACCESS_KEY"),
            Some(&ts(4_000_000_000_000))
        );
        assert_eq!(
            stored.credential_expiration_times.get("AWS_SESSION_TOKEN"),
            Some(&ts(4_000_000_000_000))
        );
    }

    #[tokio::test]
    async fn apply_minted_credential_replaces_only_refreshed_handles() {
        use super::apply_minted_credential;

        let store = test_store().await;
        let credentials = CredentialRuntime::from_config(
            &Config::new(None).with_credential_drivers(["test-static"]),
        )
        .unwrap();
        let mut prov = provider("stored-aws", "aws");
        let original_handles = credentials
            .store_provider_credentials(
                prov.object_name(),
                prov.object_workspace(),
                prov.object_id(),
                &HashMap::from([
                    ("AWS_ACCESS_KEY_ID".to_string(), "old-key".to_string()),
                    (
                        "AWS_SECRET_ACCESS_KEY".to_string(),
                        "unchanged-secret".to_string(),
                    ),
                ]),
                &HashMap::new(),
            )
            .await
            .unwrap();
        prov.credential_handles.clone_from(&original_handles);
        store.put_message(&prov).await.unwrap();

        let minted = super::MintedCredential {
            access_token: "new-key".to_string(),
            expires_at_ms: 4_000_000_000_000,
            refresh_token: None,
            additional_credentials: HashMap::new(),
        };

        apply_minted_credential(
            &store,
            "default",
            Some(&credentials),
            None,
            &prov,
            "AWS_ACCESS_KEY_ID",
            &minted,
        )
        .await
        .unwrap();

        let stored = store
            .get_message_by_name::<Provider>("default", "stored-aws")
            .await
            .unwrap()
            .unwrap();
        assert_ne!(
            stored.credential_handles.get("AWS_ACCESS_KEY_ID"),
            original_handles.get("AWS_ACCESS_KEY_ID")
        );
        assert_eq!(
            stored.credential_handles.get("AWS_SECRET_ACCESS_KEY"),
            original_handles.get("AWS_SECRET_ACCESS_KEY")
        );
        let resolved = credentials
            .resolve_provider_handles(&stored, current_time_ms())
            .await
            .unwrap();
        assert_eq!(
            resolved.values.get("AWS_ACCESS_KEY_ID"),
            Some(&"new-key".to_string())
        );
        assert_eq!(
            resolved.values.get("AWS_SECRET_ACCESS_KEY"),
            Some(&"unchanged-secret".to_string())
        );

        let old_handle_provider = Provider {
            credential_handles: HashMap::from([(
                "AWS_ACCESS_KEY_ID".to_string(),
                original_handles["AWS_ACCESS_KEY_ID"].clone(),
            )]),
            ..prov
        };
        let err = credentials
            .resolve_provider_handles(&old_handle_provider, current_time_ms())
            .await
            .unwrap_err();
        assert_eq!(err.code(), tonic::Code::NotFound);
    }

    #[tokio::test]
    async fn apply_minted_credential_validates_additional_keys_against_sandboxes() {
        use super::apply_minted_credential;

        let store = test_store().await;
        let mut existing_provider = provider("existing-aws", "aws");
        existing_provider
            .credentials
            .insert("AWS_SECRET_ACCESS_KEY".to_string(), "existing".to_string());
        store.put_message(&existing_provider).await.unwrap();

        let refreshing_provider = provider("refreshing-aws", "aws");
        store.put_message(&refreshing_provider).await.unwrap();

        store
            .put_message(&Sandbox {
                metadata: Some(ObjectMeta {
                    id: "sandbox-aws-collision".to_string(),
                    name: "aws-collision".to_string(),
                    created_time: openshell_core::time::timestamp_from_millis(1).ok(),
                    labels: HashMap::new(),
                    resource_version: 0,
                    annotations: HashMap::new(),
                    workspace: "default".to_string(),
                    deletion_time: None,
                }),
                spec: Some(SandboxSpec {
                    providers: vec!["existing-aws".to_string(), "refreshing-aws".to_string()],
                    ..SandboxSpec::default()
                }),
                ..Default::default()
            })
            .await
            .unwrap();

        let minted = super::MintedCredential {
            access_token: "AKIAIOSFODNN7EXAMPLE".to_string(),
            expires_at_ms: 4_000_000_000_000,
            refresh_token: None,
            additional_credentials: HashMap::from([
                (
                    "AWS_SECRET_ACCESS_KEY".to_string(),
                    "secret-key".to_string(),
                ),
                ("AWS_SESSION_TOKEN".to_string(), "session-token".to_string()),
            ]),
        };
        let credentials = CredentialRuntime::from_config(
            &Config::new(None).with_credential_drivers(["test-static"]),
        )
        .unwrap();

        let err = apply_minted_credential(
            &store,
            "default",
            Some(&credentials),
            None,
            &refreshing_provider,
            "AWS_ACCESS_KEY_ID",
            &minted,
        )
        .await
        .unwrap_err();
        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
        assert!(err.message().contains("AWS_SECRET_ACCESS_KEY"));
        assert_eq!(credentials.stored_credential_count(), Some(0));
    }

    // 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]
    async fn aws_sts_mint_accepts_session_token_with_source_pair() {
        let mock_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(body_string_contains("Action=AssumeRole"))
            .respond_with(ResponseTemplate::new(200).set_body_string(
                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>"#,
            ))
            .mount(&mock_server)
            .await;

        let store = test_store().await;
        let prov = provider("aws-sts-session", "aws");
        store.put_message(&prov).await.unwrap();

        // Temporary source credentials: access key + secret + session token.
        let state = new_refresh_state(
            &prov,
            "default",
            "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(), "ASIASOURCEKEY".to_string()),
                    (
                        "aws_secret_access_key".to_string(),
                        "SourceSecretKey".to_string(),
                    ),
                    (
                        "aws_session_token".to_string(),
                        "SourceSessionToken".to_string(),
                    ),
                    ("sts_endpoint_url".to_string(), mock_server.uri()),
                ]),
                secret_material_keys: vec![
                    "aws_secret_access_key".to_string(),
                    "aws_session_token".to_string(),
                ],
                expires_at_ms: 0,
                token_url: String::new(),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(300)),
                max_lifetime: Some(proto_duration(3600)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = test_credentials();

        let refreshed = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "aws-sts-session",
            "AWS_ACCESS_KEY_ID",
        )
        .await
        .unwrap();
        assert_eq!(refreshed.status, "refreshed");
        let stored = store
            .get_message_by_name::<Provider>("default", "aws-sts-session")
            .await
            .unwrap()
            .unwrap();
        let resolved = credentials
            .resolve_provider_handles(&stored, current_time_ms())
            .await
            .unwrap();
        assert_eq!(
            resolved.values.get("AWS_ACCESS_KEY_ID"),
            Some(&"ASIAMOCKKEY".to_string())
        );
    }

    #[tokio::test]
    async fn aws_sts_mint_rejects_session_token_without_source_pair() {
        let store = test_store().await;
        let prov = provider("aws-sts-lonesession", "aws");
        store.put_message(&prov).await.unwrap();

        let state = new_refresh_state(
            &prov,
            "default",
            "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_session_token".to_string(),
                        "SourceSessionToken".to_string(),
                    ),
                ]),
                secret_material_keys: vec!["aws_session_token".to_string()],
                expires_at_ms: 0,
                token_url: String::new(),
                scopes: Vec::new(),
                refresh_before: Some(proto_duration(300)),
                max_lifetime: Some(proto_duration(3600)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = test_credentials();

        let err = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "aws-sts-lonesession",
            "AWS_ACCESS_KEY_ID",
        )
        .await
        .unwrap_err();
        assert_eq!(err.code(), tonic::Code::InvalidArgument);
        assert!(err.message().contains("aws_session_token requires"));
    }

    #[tokio::test]
    async fn deletion_of_observed_refresh_preserves_same_name_replacement() {
        use super::delete_observed_refresh_state_with_credentials;
        use crate::persistence::ObjectType;
        let store = test_store().await;
        let original = StoredProviderCredentialRefreshState {
            metadata: Some(ObjectMeta {
                id: "original-id".into(),
                name: "refresh-name".into(),
                workspace: "default".into(),
                ..Default::default()
            }),
            ..Default::default()
        };
        store.put_message(&original).await.unwrap();
        let observed = store
            .get_message::<StoredProviderCredentialRefreshState>("original-id")
            .await
            .unwrap()
            .unwrap();
        store
            .delete(
                StoredProviderCredentialRefreshState::object_type(),
                "original-id",
            )
            .await
            .unwrap();
        let mut replacement = original;
        replacement.metadata.as_mut().unwrap().id = "replacement-id".into();
        store.put_message(&replacement).await.unwrap();
        let replacement = store
            .get_message::<StoredProviderCredentialRefreshState>("replacement-id")
            .await
            .unwrap()
            .unwrap();

        delete_observed_refresh_state_with_credentials(&store, &test_credentials(), observed)
            .await
            .unwrap();
        assert_eq!(
            store
                .get_message::<StoredProviderCredentialRefreshState>("replacement-id")
                .await
                .unwrap()
                .unwrap(),
            replacement
        );
    }

    #[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;
        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,
            "default",
            "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: Some(proto_duration(300)),
                max_lifetime: Some(proto_duration(3600)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = test_credentials();

        let rotate = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "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_with_credentials(
                &store,
                &credentials,
                "default",
                &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, "default", &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>("default", "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"));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn rotation_superseded_mid_flight_discards_credentials() {
        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;
        let prov = provider("aws-superseded", "aws");
        store.put_message(&prov).await.unwrap();
        let provider_id = prov.object_id().to_string();

        let state = new_refresh_state(
            &prov,
            "default",
            "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: Some(proto_duration(300)),
                max_lifetime: Some(proto_duration(3600)),
            },
        )
        .unwrap();
        put_refresh_state(&store, &state).await.unwrap();
        let credentials = test_credentials();

        let rotate = refresh_provider_credential(
            &store,
            "default",
            &credentials,
            None,
            "aws-superseded",
            "AWS_ACCESS_KEY_ID",
        );
        let interfere = async {
            if tokio::time::timeout(std::time::Duration::from_secs(15), hit_rx)
                .await
                .is_err()
            {
                return;
            }
            // Simulate a concurrent rotation or reconfigure winning the
            // generation: any write to the refresh state bumps its version, so
            // the in-flight rotation's version-matched persist will lose.
            let mut winner =
                get_refresh_state(&store, "default", &provider_id, "AWS_ACCESS_KEY_ID")
                    .await
                    .unwrap()
                    .unwrap();
            winner.last_error = "won-by-concurrent-writer".to_string();
            put_refresh_state(&store, &winner).await.unwrap();
            let _ = release_tx.send(());
        };
        let (rotate_result, ()) = tokio::join!(rotate, interfere);

        // The superseded (losing) rotation must abort rather than complete.
        assert!(
            rotate_result.is_err(),
            "a rotation whose generation was superseded must abort"
        );
        // It must not write its stale-generation credentials into the provider.
        let stored = store
            .get_message_by_name::<Provider>("default", "aws-superseded")
            .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"));
        // The concurrent writer's refresh state must survive untouched.
        let refresh = get_refresh_state(&store, "default", &provider_id, "AWS_ACCESS_KEY_ID")
            .await
            .unwrap()
            .expect("refresh state should still exist");
        assert_eq!(refresh.last_error, "won-by-concurrent-writer");
        assert_ne!(refresh.status, "refreshed");
    }

    fn provider(name: &str, provider_type: &str) -> Provider {
        Provider {
            metadata: Some(ObjectMeta {
                id: format!("{name}-id"),
                name: name.to_string(),
                created_time: openshell_core::time::timestamp_from_millis(1).ok(),
                labels: HashMap::new(),
                resource_version: 0,
                annotations: HashMap::new(),
                workspace: "default".to_string(),
                deletion_time: None,
            }),
            r#type: provider_type.to_string(),
            credentials: HashMap::new(),
            config: HashMap::new(),
            credential_expiration_times: HashMap::new(),
            profile_workspace: "default".to_string(),
            credential_handles: HashMap::new(),
        }
    }

    const TEST_RSA_PRIVATE_KEY: &str = r"-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCvCoZ0mVHpCHsF
zeeqw2caNIe/eb4BQUccFPhZfRnF7sCfyB84zTBmuwG2umRBdjFnVsfIIZRp2HcD
OESrRYYiE1RGfjBXImGVg2Wtza0HYhL1sLyX1eaEefylxoilmApAgWDh9p36h8J2
s5YHwyXPTttx4DpdWDnxju1iNmwoIB8uVE/5amWgbNvlETMBOcB1RxDHtnVy+xJz
jjjrzK4Qz9WsUTHAvngdi4Yyxvci+yKpjYTg5+UWxmAN6iW522TpLe32MDb5Ug1d
trBvvepWmdQ6CBwPhBHCt/sMoSJAYSO4RKeBnBjeLQBXFTxaOv5iTGIsRTX3K471
epHp3cT5AgMBAAECggEASQlRv/4nZN5SgsH/K8v7zb3kdHsmUly8AJYpaCGgauvr
uN/mUyueyga2uNl+MqhQBef6VWHZjO6y/gdw86v/Q2GgVQebQQhKAnpAp2w+Ceoc
siKMFqi8VkOWLU+xPbM6d97kH3TpRxt1g1T8wYFmWeF0BEiE4eUJzGaQW14M9BJ+
G0QxmP/zjX9cNpVeApKTjBWKiH4CXG3DuI3pJ93VOMpUlOsrdLXvKGTze0e01itr
MX/MHHTE+VXB4FB+/zKSA4c36egi676OSXrGC/GDmM8ntJ4CUGeD5uZsMSADiAUn
iccv5iGRWVMIKxUS5Q4k0jy8uWuK+QVP4Y6cQWYArwKBgQDhuSNORBNpIGRfsKGN
iJo/h+qinz6pEIpa3D3oVl7rpkyvgIyaTwfXvC1vfdS9V5VIel2gV2Cx0OrI8yrr
nQu1JuNV/rLmtvqX321fgBLRdoiqF3pAy1gbmdUz1elerAIYL578gXQ6jg1bbdic
kJpn0MsoDUJGwvJnXcgLqG7q3wKBgQDGhRIa4oJsj1vqICc8zt8YsCAcot3vjWLH
588X7JdBGOWJdWxfdmGXQRn5Zw9UhMQnYa3uyTBPeVcXopThlPotYeuFhLSU856T
IJzfpzCJzC4zIQayoyvJFrKe7N70iUQ986dewYy9oxQhHvFKd/qe4ylbzZJXpthX
eWEuuBSjJwKBgGkqXt6qLPj/1IQYwUw15tfOtW0LEKCoSi3HCzjidNsJ4hSqqdeD
Fr5WuDyHvcRxt+XKzTBVRYHTOnBhiw+3XasK8UQxpJyFh/+WY1jpTNs2hLnqslTZ
6LUDWSgLc+1d6qPmHAa9Ma/OWz7L0O4xGR9hUiXY95YMYe/y668yzGq1AoGBAJyU
Gsqfu7U6gYmxoKEine6QBFPx1dD7GF2KJdq93jMXGvyHZFoLOkAdtgnz0rCcI0bY
kWKUxwj4MMxQjNM8OPMQl75xBCmz2XA8Od9htDQLmqjzNKAzePabc3lMZTJFDlE6
29kuGf79IIRbLn/JECDAFT/2baW60Ep2T0OVJ5njAoGAfaCaQ4aVgjI027q7Y5qP
KfNSI8uuA8PLqmUY30I9KFWzN6VDLu00eKa90F4w3CeWRRQWXW1+007tTz3V1mNw
20A24Fi3HGQmXc7NyuLDODTJsWBICuOemCnRkvcxIlxb+ec7jp+XRmzDwKkzSnVN
pM2zFU8SeVkvHKlEuoHaP0s=
-----END PRIVATE KEY-----";
}