diff --git a/docs/GUIDELINES.md b/docs/GUIDELINES.md
index d17198bd..a12ca925 100644
--- a/docs/GUIDELINES.md
+++ b/docs/GUIDELINES.md
@@ -106,3 +106,87 @@
 - Never log or expose sensitive information (passwords, tokens, etc.)
 - Validate all inputs, especially from external sources
 - Follow principle of least privilege in code design
+
+## Credential Binding Rotation
+
+BTP service bindings can rotate at runtime. Long-lived clients holding stale credentials will fail with auth errors once the old secrets expire. Every SDK module that reads credentials from mounts or env vars must support rotation.
+
+### Two resilience layers
+
+**Proactive** — check `ConfigFactory.has_changed()` before every credential use. If the mtime of the secret directory changed, re-read the binding and discard any cached tokens or session objects.
+
+**Reactive** — on a credential-rejection error (e.g. HTTP 401 for OAuth modules, S3 `InvalidAccessKeyId`/`SignatureDoesNotMatch` for object store), refresh credentials and retry the operation once.
+
+### Pattern for OAuth2 modules
+
+Token providers and auth classes must accept either a fixed config object **or** a callable (factory) returning the config:
+
+```python
+if callable(config) and not isinstance(config, MyConfig):
+    self._config_factory: Callable[[], MyConfig] = config
+    self._config = config()
+else:
+    self._config_factory = lambda: config  # static; no rotation tracking
+    self._config = config
+```
+
+Before serving a cached token, call `_refresh_if_rotated()`:
+
+```python
+def _refresh_if_rotated(self) -> None:
+    has_changed = getattr(self._config_factory, "has_changed", None)
+    if callable(has_changed) and has_changed():
+        self._config = self._config_factory()
+        self._cached_token = None          # discard stale token
+        # rebuild session/client if needed
+```
+
+### Pattern for key-based clients (e.g. object store / MinIO)
+
+Credentials are baked into the client at construction time, so the client itself must be rebuilt on rotation. Wrap every public operation in `_execute_with_retry`:
+
+```python
+_CREDENTIAL_ERROR_CODES = frozenset({"InvalidAccessKeyId", "SignatureDoesNotMatch"})
+
+def _execute_with_retry(self, fn):
+    self._refresh_if_rotated()
+    try:
+        return fn()
+    except S3Error as e:
+        if e.code in _CREDENTIAL_ERROR_CODES:
+            with self._lock:
+                self._creds = self._config_factory()
+                self._client = self._create_client()
+            return fn()
+        raise
+```
+
+Use a `threading.Lock` when rebuilding the client to avoid races under concurrent calls.
+
+### Public API factory functions (`create_client`)
+
+- **Auto-detection path** (no explicit `config=`): pass a `ConfigFactory` instance directly to the token provider / auth class. `ConfigFactory` carries `has_changed()` automatically.
+- **Explicit `config=` path**: wrap in a static lambda to preserve the factory interface without rotation tracking.
+
+```python
+def create_client(*, instance=None, config=None):
+    if config is not None:
+        credentials = lambda: config  # static, no rotation
+    else:
+        credentials = _make_config_factory(instance)
+    return MyClient(credentials)
+```
+
+### `ConfigFactory` contract
+
+`ConfigFactory[C]` (from `sap_cloud_sdk.core.secret_resolver`) reads bindings on every `__call__()` and tracks secret directory mtime via `has_changed()`. To use it, the binding dataclass must:
+- Have all fields defaulting to `""` so `binding_cls()` can be called with no args.
+- Implement `validate()` raising on missing required fields.
+
+### Testing rotation
+
+Every module that supports rotation must have tests for:
+1. **Proactive**: `has_changed()` returns `True` → token/session/client is rebuilt before the next operation.
+2. **Reactive** (key-based clients): credential-rejection error on first call → retried once with fresh credentials.
+3. **No rebuild**: `has_changed()` returns `False` → existing session/client reused.
+4. **Static config**: plain config object (no `has_changed`) → no error, no rotation check.
diff --git a/pyproject.toml b/pyproject.toml
index b375e35b..fce3e853 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.53.3"
+version = "0.54.0"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"
diff --git a/src/sap_cloud_sdk/adms/_ias_fetcher.py b/src/sap_cloud_sdk/adms/_ias_fetcher.py
index 783edeea..3d48af46 100644
--- a/src/sap_cloud_sdk/adms/_ias_fetcher.py
+++ b/src/sap_cloud_sdk/adms/_ias_fetcher.py
@@ -14,10 +14,13 @@
 
 from __future__ import annotations
 
-from typing import Optional
+import logging
+from typing import Callable, Optional
 
 import requests
 
+logger = logging.getLogger(__name__)
+
 from sap_cloud_sdk.adms._token_cache import InMemoryTokenCache, TokenCache
 from sap_cloud_sdk.adms.config import AdmsConfig
 from sap_cloud_sdk.adms.exceptions import AuthError
@@ -76,17 +79,35 @@ class IasTokenFetcher:
 
     def __init__(
         self,
-        config: AdmsConfig,
+        config: AdmsConfig | Callable[[], AdmsConfig],
         session: Optional[requests.Session] = None,
         cache: Optional[TokenCache] = None,
     ) -> None:
-        self._ias_url = config.ias_url.rstrip("/")
-        self._client_id = config.client_id
-        self._client_secret = config.client_secret
+        if callable(config) and not isinstance(config, AdmsConfig):
+            self._config_factory: Callable[[], AdmsConfig] = config
+            self._config = config()
+        else:
+            self._config_factory = lambda: config  # type: ignore[arg-type]
+            self._config = config  # type: ignore[assignment]
         self._session = session or requests.Session()
-        self._token_url = self._ias_url + "/oauth2/token"
         self._cache: TokenCache = cache or InMemoryTokenCache()
-        self._resource: Optional[str] = config.resource
+        self._apply_config()
+
+    def _apply_config(self) -> None:
+        """Sync derived attributes from the current ``_config``."""
+        self._ias_url = self._config.ias_url.rstrip("/")
+        self._client_id = self._config.client_id
+        self._client_secret = self._config.client_secret
+        self._token_url = self._ias_url + "/oauth2/token"
+        self._resource: Optional[str] = self._config.resource
+
+    def _refresh_if_rotated(self) -> None:
+        """Proactively refresh credentials and clear the token cache if the binding changed."""
+        has_changed = getattr(self._config_factory, "has_changed", None)
+        if callable(has_changed) and has_changed():
+            self._config = self._config_factory()
+            self._apply_config()
+            self._cache = InMemoryTokenCache()
 
     # ------------------------------------------------------------------
     # Public API
@@ -105,6 +126,7 @@ def get_token(self) -> str:
             AuthError: If the IAS token endpoint returns an error or the
                 response is missing ``access_token``.
         """
+        self._refresh_if_rotated()
         cached = self._cache.get(_CC_CACHE_KEY)
         if cached:
             return cached
diff --git a/src/sap_cloud_sdk/adms/client.py b/src/sap_cloud_sdk/adms/client.py
index 22f3f759..671ff396 100644
--- a/src/sap_cloud_sdk/adms/client.py
+++ b/src/sap_cloud_sdk/adms/client.py
@@ -57,7 +57,8 @@
     _DocumentRelationApi,
 )
 from sap_cloud_sdk.adms._token_cache import TokenCache
-from sap_cloud_sdk.adms.config import AdmsConfig, load_from_env_or_mount
+from sap_cloud_sdk.adms.config import AdmsConfig, _make_config_factory
+from sap_cloud_sdk.adms.exceptions import ConfigError
 
 
 # ---------------------------------------------------------------------------
@@ -171,9 +172,18 @@ def create_client(
         raise ValueError(
             "instance must not be an empty string; omit it to use 'default'"
         )
-    binding = config or load_from_env_or_mount(instance)
-    token_fetcher = IasTokenFetcher(config=binding, cache=token_cache)
-    http = AdmsHttp(config=binding, token_fetcher=token_fetcher, user_jwt=user_jwt)
+    try:
+        if config is not None:
+            token_fetcher = IasTokenFetcher(config=config, cache=token_cache)
+        else:
+            token_fetcher = IasTokenFetcher(
+                config=_make_config_factory(instance), cache=token_cache
+            )
+    except RuntimeError as exc:
+        raise ConfigError(str(exc)) from exc
+    http = AdmsHttp(
+        config=token_fetcher._config, token_fetcher=token_fetcher, user_jwt=user_jwt
+    )
     return AdmsClient(http)
 
 
@@ -206,10 +216,17 @@ def create_async_client(
         raise ValueError(
             "instance must not be an empty string; omit it to use 'default'"
         )
-    binding = config or load_from_env_or_mount(instance)
-    token_fetcher = IasTokenFetcher(config=binding, cache=token_cache)
+    try:
+        if config is not None:
+            token_fetcher = IasTokenFetcher(config=config, cache=token_cache)
+        else:
+            token_fetcher = IasTokenFetcher(
+                config=_make_config_factory(instance), cache=token_cache
+            )
+    except RuntimeError as exc:
+        raise ConfigError(str(exc)) from exc
     http = AsyncAdmsHttp(
-        config=binding,
+        config=token_fetcher._config,
         token_fetcher=token_fetcher,
         client=http_client,
         user_jwt=user_jwt,
diff --git a/src/sap_cloud_sdk/adms/config.py b/src/sap_cloud_sdk/adms/config.py
index f0108bd9..4b00eb1e 100644
--- a/src/sap_cloud_sdk/adms/config.py
+++ b/src/sap_cloud_sdk/adms/config.py
@@ -17,12 +17,16 @@
 """
 
 from dataclasses import dataclass
+from typing import TYPE_CHECKING
 
 from sap_cloud_sdk.core.secret_resolver.resolver import (
     read_from_mount_and_fallback_to_env_var,
 )
 from sap_cloud_sdk.adms.exceptions import ConfigError
 
+if TYPE_CHECKING:
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
+
 _DEFAULT_INSTANCE = "default"
 _SECRET_MOUNT_BASE = "/etc/secrets/appfnd"
 _ENV_VAR_BASE = "CLOUD_SDK_CFG"
@@ -125,3 +129,35 @@ def load_from_env_or_mount(instance: str | None = None) -> AdmsConfig:
 
     raw.validate()
     return raw.to_config()
+
+
+def _make_config_factory(instance: str | None = None) -> "ConfigFactory[AdmsConfig]":
+    """Return a :class:`~sap_cloud_sdk.core.secret_resolver.ConfigFactory` for the given instance.
+
+    The factory re-reads the binding on every call and tracks the secret
+    directory mtime for proactive rotation detection.
+
+    Args:
+        instance: Binding instance name. Defaults to ``"default"``.
+
+    Returns:
+        A callable that produces a fresh :class:`AdmsConfig`.
+    """
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
+
+    inst = instance or _DEFAULT_INSTANCE
+
+    def _extract(binding: _BindingData) -> AdmsConfig:
+        try:
+            return binding.to_config()
+        except Exception as exc:
+            raise ConfigError(
+                f"failed to load ADMS configuration for instance '{inst}': {exc}"
+            ) from exc
+
+    return ConfigFactory(
+        module="adms",
+        instance=inst,
+        binding_cls=_BindingData,
+        extract=_extract,
+    )
diff --git a/src/sap_cloud_sdk/aicore/user-guide.md b/src/sap_cloud_sdk/aicore/user-guide.md
index ad37e300..095a9204 100644
--- a/src/sap_cloud_sdk/aicore/user-guide.md
+++ b/src/sap_cloud_sdk/aicore/user-guide.md
@@ -131,7 +131,7 @@ before the cached OAuth token expires — so agents never see a 401 at all.
 ```python
 from sap_cloud_sdk.aicore import set_aicore_config, watch_aicore_config
 
-set_aicore_config()    # load credentials at startup
+set_aicore_config()  # load credentials at startup
 watch_aicore_config()  # proactive reload on secret rotation
 ```
 
@@ -158,7 +158,7 @@ directly, bypassing the SDK's reactive handler. Two options:
 from sap_cloud_sdk.aicore import set_aicore_config, watch_aicore_config
 
 set_aicore_config()
-watch_aicore_config()   # ADD THIS — no other changes needed
+watch_aicore_config()  # ADD THIS — no other changes needed
 ```
 
 **Option B — also add reactive reload for ChatLiteLLM:**
@@ -171,7 +171,7 @@ from sap_cloud_sdk.aicore import (
 )
 
 set_aicore_config()
-patch_litellm_for_credential_rotation()   # patches litellm.completion globally
+patch_litellm_for_credential_rotation()  # patches litellm.completion globally
 watch_aicore_config()
 ```
 
diff --git a/src/sap_cloud_sdk/core/auditlog/__init__.py b/src/sap_cloud_sdk/core/auditlog/__init__.py
index 34bd8b6e..c1b800c0 100644
--- a/src/sap_cloud_sdk/core/auditlog/__init__.py
+++ b/src/sap_cloud_sdk/core/auditlog/__init__.py
@@ -33,7 +33,7 @@
     ChangeAttribute,
     DeletedAttribute,
 )
-from sap_cloud_sdk.core.auditlog.config import AuditLogConfig, _load_config_from_env
+from sap_cloud_sdk.core.auditlog.config import AuditLogConfig, _make_config_factory
 from sap_cloud_sdk.core.auditlog.exceptions import (
     AuditLogError,
     ClientCreationError,
@@ -67,7 +67,7 @@ def create_client(
             transport = HttpTransport(config)
             return AuditLogClient(transport, _telemetry_source=_telemetry_source)
 
-        transport = HttpTransport(_load_config_from_env())
+        transport = HttpTransport(_make_config_factory())
         return AuditLogClient(transport, _telemetry_source=_telemetry_source)
 
     except Exception as e:
diff --git a/src/sap_cloud_sdk/core/auditlog/_http_transport.py b/src/sap_cloud_sdk/core/auditlog/_http_transport.py
index 0b10b752..0cae3b89 100644
--- a/src/sap_cloud_sdk/core/auditlog/_http_transport.py
+++ b/src/sap_cloud_sdk/core/auditlog/_http_transport.py
@@ -1,6 +1,9 @@
 """HTTP transport implementation for cloud mode."""
 
+from typing import Callable, Optional
+
 import requests
+
 from oauthlib.oauth2 import BackendApplicationClient
 from requests_oauthlib import OAuth2Session
 
@@ -18,29 +21,52 @@
 
 
 class HttpTransport(Transport):
-    """HTTP-based transport for cloud mode with OAuth2 authentication."""
+    """HTTP-based transport for cloud mode with OAuth2 authentication.
+
+    Accepts either a fixed :class:`AuditLogConfig` or a config factory (any callable
+    returning ``AuditLogConfig`` with an optional ``has_changed() -> bool`` method).
+    When a factory is supplied, credentials are re-read on every token refresh and
+    the factory's ``has_changed()`` method is checked proactively before each request
+    so that rotated secrets are picked up automatically.
+    """
 
-    def __init__(self, config: AuditLogConfig):
+    def __init__(self, config: AuditLogConfig | Callable[[], AuditLogConfig]):
         """Initialize HTTP transport with provided configuration.
 
         Args:
-            config: AuditLogConfig with OAuth2 credentials and service URL
+            config: AuditLogConfig (or a factory returning one) with OAuth2 credentials
+                and service URL.
         """
-        self.config = config
-
-        token_url = f"{config.oauth_url.rstrip('/')}/oauth/token"
-
-        client = BackendApplicationClient(client_id=config.client_id)
-        self.oauth = OAuth2Session(client=client)
+        if callable(config) and not isinstance(config, AuditLogConfig):
+            self._config_factory: Callable[[], AuditLogConfig] = config
+            self.config = config()
+        else:
+            self._config_factory = lambda: config  # type: ignore[arg-type]
+            self.config = config  # type: ignore[assignment]
+        self.oauth: Optional[OAuth2Session] = None
+
+    def _ensure_session(self) -> OAuth2Session:
+        """Return a valid OAuth2 session, refreshing credentials if the binding changed."""
+        has_changed = getattr(self._config_factory, "has_changed", None)
+        if callable(has_changed) and has_changed():
+            self.config = self._config_factory()
+            self.oauth = None
+
+        if self.oauth is None:
+            token_url = f"{self.config.oauth_url.rstrip('/')}/oauth/token"
+            client = BackendApplicationClient(client_id=self.config.client_id)
+            oauth = OAuth2Session(client=client)
+            try:
+                oauth.fetch_token(
+                    token_url=token_url,
+                    client_id=self.config.client_id,
+                    client_secret=self.config.client_secret,
+                )
+            except Exception as e:
+                raise AuthenticationError(f"Failed to obtain OAuth2 token: {e}")
+            self.oauth = oauth
 
-        try:
-            _token = self.oauth.fetch_token(
-                token_url=token_url,
-                client_id=config.client_id,
-                client_secret=config.client_secret,
-            )
-        except Exception as e:
-            raise AuthenticationError(f"Failed to obtain OAuth2 token: {e}")
+        return self.oauth
 
     def send(self, event: AuditMessage) -> None:
         """Send audit event via HTTP.
@@ -58,7 +84,8 @@ def send(self, event: AuditMessage) -> None:
             path_prefix = "/audit-log/oauth2/v2"
             url = f"{self.config.service_url.rstrip('/')}{path_prefix}{endpoint}"
 
-            response = self.oauth.post(
+            oauth = self._ensure_session()
+            response = oauth.post(
                 url,
                 json=event_dict,
                 headers={"Content-Type": "application/json"},
diff --git a/src/sap_cloud_sdk/core/auditlog/config.py b/src/sap_cloud_sdk/core/auditlog/config.py
index 745a497b..d3368367 100644
--- a/src/sap_cloud_sdk/core/auditlog/config.py
+++ b/src/sap_cloud_sdk/core/auditlog/config.py
@@ -6,9 +6,13 @@
 
 import json
 from dataclasses import dataclass
+from typing import TYPE_CHECKING
 
 from sap_cloud_sdk.core.auditlog.exceptions import ClientCreationError
 
+if TYPE_CHECKING:
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
+
 
 @dataclass
 class AuditLogConfig:
@@ -43,8 +47,8 @@ class BindingData:
     and returns a flat AuditLogConfig.
     """
 
-    url: str
-    uaa: str
+    url: str = ""
+    uaa: str = ""
 
     def validate(self) -> None:
         """Validate that all required fields are set."""
@@ -121,3 +125,32 @@ def _load_config_from_env() -> AuditLogConfig:
 
     except Exception as e:
         raise ClientCreationError(f"Failed to load configuration: {e}")
+
+
+def _make_config_factory() -> "ConfigFactory[AuditLogConfig]":
+    """Return a :class:`~sap_cloud_sdk.core.secret_resolver.ConfigFactory` for auditlog.
+
+    The factory re-reads the binding on every call and tracks the secret
+    directory mtime for proactive rotation detection.
+
+    Returns:
+        A callable that produces a fresh :class:`AuditLogConfig`.
+    """
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
+
+    def _extract(binding: BindingData) -> AuditLogConfig:
+        try:
+            return binding.extract_config()
+        except ClientCreationError:
+            raise
+        except Exception as exc:
+            raise ClientCreationError(
+                f"Failed to load auditlog configuration: {exc}"
+            ) from exc
+
+    return ConfigFactory(
+        module="auditlog",
+        instance="default",
+        binding_cls=BindingData,
+        extract=_extract,
+    )
diff --git a/src/sap_cloud_sdk/core/runtime_context/user-guide.md b/src/sap_cloud_sdk/core/runtime_context/user-guide.md
index de1b41fd..152f4a92 100644
--- a/src/sap_cloud_sdk/core/runtime_context/user-guide.md
+++ b/src/sap_cloud_sdk/core/runtime_context/user-guide.md
@@ -117,6 +117,7 @@ to check a toggle for the current request:
 ```python
 from sap_cloud_sdk.core.runtime_context import is_feature_enabled
 
+
 @app.route("/")
 async def handler(request):
     if is_feature_enabled("my-feature"):
diff --git a/src/sap_cloud_sdk/dms/__init__.py b/src/sap_cloud_sdk/dms/__init__.py
index 33fdf2fc..738e57a3 100644
--- a/src/sap_cloud_sdk/dms/__init__.py
+++ b/src/sap_cloud_sdk/dms/__init__.py
@@ -42,7 +42,7 @@
     UserClaim,
 )
 from sap_cloud_sdk.dms.client import DMSClient
-from sap_cloud_sdk.dms.config import load_sdm_config_from_env_or_mount
+from sap_cloud_sdk.dms.config import _make_config_factory
 from sap_cloud_sdk.dms.exceptions import DMSError
 
 
@@ -70,7 +70,10 @@ def create_client(
         DMSError: If client creation fails due to configuration or initialization issues.
     """
     try:
-        credentials = dms_cred or load_sdm_config_from_env_or_mount(instance)
+        if dms_cred is not None:
+            credentials = dms_cred
+        else:
+            credentials = _make_config_factory(instance)
         client = DMSClient(
             credentials,
             connect_timeout=connect_timeout,
diff --git a/src/sap_cloud_sdk/dms/_auth.py b/src/sap_cloud_sdk/dms/_auth.py
deleted file mode 100644
index cedb032d..00000000
--- a/src/sap_cloud_sdk/dms/_auth.py
+++ /dev/null
@@ -1,112 +0,0 @@
-import logging
-import time
-import requests
-from collections import OrderedDict
-from requests.exceptions import RequestException
-from typing import Optional, TypedDict
-from sap_cloud_sdk.dms.exceptions import (
-    DMSError,
-    DMSConnectionError,
-    DMSPermissionDeniedException,
-)
-from sap_cloud_sdk.dms.model import DMSCredentials
-from sap_cloud_sdk.core._tenant import _validate_tenant_subdomain
-
-logger = logging.getLogger(__name__)
-
-
-class _TokenResponse(TypedDict):
-    access_token: str
-    expires_in: int
-
-
-class _CachedToken:
-    def __init__(self, token: str, expires_at: float) -> None:
-        self.token = token
-        self.expires_at = expires_at
-
-    def is_valid(self) -> bool:
-        return time.monotonic() < self.expires_at - 30
-
-
-_MAX_CACHE_SIZE = 10
-
-
-class Auth:
-    """Fetches and caches OAuth2 access tokens for DMS service requests."""
-
-    def __init__(self, credentials: DMSCredentials) -> None:
-        self._credentials = credentials
-        self._cache: OrderedDict[str, _CachedToken] = OrderedDict()
-
-    def get_token(self, tenant_subdomain: Optional[str] = None) -> str:
-        cache_key = tenant_subdomain or "technical"
-
-        cached = self._cache.get(cache_key)
-        if cached and cached.is_valid():
-            self._cache.move_to_end(cache_key)  # Mark as recently used by moving to end
-            logger.debug("Using cached token for key '%s'", cache_key)
-            return cached.token
-
-        logger.debug("Fetching new token for key '%s'", cache_key)
-        token_url = self._resolve_token_url(tenant_subdomain)
-        token = self._fetch_token(token_url)
-
-        if len(self._cache) >= _MAX_CACHE_SIZE:
-            evicted, _ = self._cache.popitem(last=False)
-            logger.debug("Cache full — evicted token for key '%s'", evicted)
-
-        self._cache[cache_key] = _CachedToken(
-            token=token["access_token"],
-            expires_at=time.monotonic() + token.get("expires_in", 3600),
-        )
-        logger.debug("Token cached for key '%s'", cache_key)
-        return self._cache[cache_key].token
-
-    def _resolve_token_url(self, tenant_subdomain: Optional[str]) -> str:
-        if not tenant_subdomain:
-            return self._credentials.token_url
-        _validate_tenant_subdomain(tenant_subdomain)
-
-        logger.debug("Resolving token URL for tenant '%s'", tenant_subdomain)
-        return self._credentials.token_url.replace(
-            self._credentials.identityzone,
-            tenant_subdomain,
-        )
-
-    def _fetch_token(self, token_url: str) -> _TokenResponse:
-        try:
-            response = requests.post(
-                f"{token_url}/oauth/token",
-                data={
-                    "grant_type": "client_credentials",
-                    "client_id": self._credentials.client_id,
-                    "client_secret": self._credentials.client_secret,
-                },
-                headers={"Content-Type": "application/x-www-form-urlencoded"},
-                timeout=10,
-            )
-            response.raise_for_status()
-        except requests.exceptions.ConnectionError as e:
-            logger.error("Failed to connect to token endpoint")
-            raise DMSConnectionError(
-                "Failed to connect to the authentication server"
-            ) from e
-        except requests.exceptions.HTTPError as e:
-            status = e.response.status_code if e.response is not None else None
-            logger.error("Token request failed with status %s", status)
-            if status in (401, 403):
-                raise DMSPermissionDeniedException(
-                    "Authentication failed — invalid client credentials", status
-                ) from e
-            raise DMSError("Failed to obtain access token", status) from e
-        except RequestException as e:
-            logger.error("Unexpected error during token fetch")
-            raise DMSConnectionError("Unexpected error during authentication") from e
-
-        payload: _TokenResponse = response.json()
-        if not payload.get("access_token"):
-            raise DMSError("Token response missing access_token")
-
-        logger.debug("Token fetched successfully")
-        return payload
diff --git a/src/sap_cloud_sdk/dms/_http.py b/src/sap_cloud_sdk/dms/_http.py
index 9b4f877f..f0bec0ca 100644
--- a/src/sap_cloud_sdk/dms/_http.py
+++ b/src/sap_cloud_sdk/dms/_http.py
@@ -1,9 +1,8 @@
 import logging
 from typing import Any, Optional
 from requests import Response
-import requests
 from requests.exceptions import RequestException
-from sap_cloud_sdk.dms._auth import Auth
+from sap_cloud_sdk.core.protocol.http import HttpClient, HttpMethod, XsuaaAuthProvider
 from sap_cloud_sdk.dms.exceptions import (
     DMSError,
     DMSConflictException,
@@ -19,17 +18,17 @@
 
 
 class HttpInvoker:
-    """Low-level HTTP layer. Injects auth headers and enforces timeouts."""
+    """Low-level HTTP layer for DMS. Wraps HttpClient with DMS error mapping."""
 
     def __init__(
         self,
-        auth: Auth,
+        auth_provider: XsuaaAuthProvider,
         base_url: str,
         connect_timeout: int | None = None,
         read_timeout: int | None = None,
     ) -> None:
-        self._auth = auth
-        self._base_url = base_url.rstrip("/")
+        timeout = float(read_timeout or 30)
+        self._http = HttpClient(base_url, auth_provider, timeout=timeout)
         self._connect_timeout = connect_timeout or 10
         self._read_timeout = read_timeout or 30
 
@@ -44,11 +43,12 @@ def get(
         logger.debug("GET %s", path)
         return self._handle(
             self._execute(
-                lambda: requests.get(
-                    f"{self._base_url}{path}",
+                lambda: self._http.request(
+                    HttpMethod.GET,
+                    path,
+                    tenant_subdomain=tenant_subdomain,
                     headers=self._merged_headers(tenant_subdomain, headers, user_claim),
                     params=params,
-                    timeout=(self._connect_timeout, self._read_timeout),
                 )
             )
         )
@@ -64,11 +64,12 @@ def post(
         logger.debug("POST %s", path)
         return self._handle(
             self._execute(
-                lambda: requests.post(
-                    f"{self._base_url}{path}",
+                lambda: self._http.request(
+                    HttpMethod.POST,
+                    path,
+                    tenant_subdomain=tenant_subdomain,
                     headers=self._merged_headers(tenant_subdomain, headers, user_claim),
                     json=payload,
-                    timeout=(self._connect_timeout, self._read_timeout),
                 )
             )
         )
@@ -84,11 +85,12 @@ def put(
         logger.debug("PUT %s", path)
         return self._handle(
             self._execute(
-                lambda: requests.put(
-                    f"{self._base_url}{path}",
+                lambda: self._http.request(
+                    HttpMethod.PUT,
+                    path,
+                    tenant_subdomain=tenant_subdomain,
                     headers=self._merged_headers(tenant_subdomain, headers, user_claim),
                     json=payload,
-                    timeout=(self._connect_timeout, self._read_timeout),
                 )
             )
         )
@@ -103,10 +105,11 @@ def delete(
         logger.debug("DELETE %s", path)
         return self._handle(
             self._execute(
-                lambda: requests.delete(
-                    f"{self._base_url}{path}",
+                lambda: self._http.request(
+                    HttpMethod.DELETE,
+                    path,
+                    tenant_subdomain=tenant_subdomain,
                     headers=self._merged_headers(tenant_subdomain, headers, user_claim),
-                    timeout=(self._connect_timeout, self._read_timeout),
                 )
             )
         )
@@ -128,12 +131,13 @@ def post_form(
         logger.debug("POST_FORM %s", path)
         return self._handle(
             self._execute(
-                lambda: requests.post(
-                    f"{self._base_url}{path}",
+                lambda: self._http.request(
+                    HttpMethod.POST,
+                    path,
+                    tenant_subdomain=tenant_subdomain,
                     headers=self._auth_header(tenant_subdomain, user_claim),
                     data=data,
                     files=files,
-                    timeout=(self._connect_timeout, self._read_timeout),
                 )
             )
         )
@@ -154,38 +158,30 @@ def get_stream(
         logger.debug("GET_STREAM %s", path)
         return self._handle(
             self._execute(
-                lambda: requests.get(
-                    f"{self._base_url}{path}",
+                lambda: self._http.request(
+                    HttpMethod.GET,
+                    path,
+                    tenant_subdomain=tenant_subdomain,
                     headers=self._merged_headers(tenant_subdomain, None, user_claim),
                     params=params,
                     stream=True,
-                    timeout=(self._connect_timeout, self._read_timeout),
                 )
             )
         )
 
     def _execute(self, fn: Any) -> Response:
-        """Execute an HTTP call, wrapping network errors into DMSConnectionError."""
         try:
             return fn()
-        except requests.exceptions.ConnectionError as e:
+        except RequestException as e:
             logger.error("Connection error during HTTP request")
             raise DMSConnectionError("Failed to connect to the DMS service") from e
-        except requests.exceptions.Timeout as e:
-            logger.error("Request timed out")
-            raise DMSConnectionError("Request to DMS service timed out") from e
-        except RequestException as e:
-            logger.error("Unexpected network error")
-            raise DMSConnectionError("Unexpected network error") from e
 
     def _auth_header(
         self,
         tenant_subdomain: Optional[str] = None,
         user_claim: Optional[UserClaim] = None,
     ) -> dict[str, str]:
-        """Auth-only headers (no Content-Type). Used by post_form."""
         return {
-            "Authorization": f"Bearer {self._auth.get_token(tenant_subdomain)}",
             **self._user_claim_headers(user_claim),
         }
 
@@ -193,7 +189,6 @@ def _default_headers(
         self, tenant_subdomain: Optional[str] = None
     ) -> dict[str, str]:
         return {
-            "Authorization": f"Bearer {self._auth.get_token(tenant_subdomain)}",
             "Content-Type": "application/json",
             "Accept": "application/json",
         }
@@ -225,11 +220,9 @@ def _handle(self, response: Response) -> Response:
         if response.status_code in (200, 201, 204):
             return response
 
-        # error_content kept for debugging but not surfaced in the exception message
         error_content = response.text
         logger.warning("Request failed with status %s", response.status_code)
 
-        # Try to extract the server's error message from the JSON body
         try:
             body = response.json()
             server_message = body.get("message", "") if isinstance(body, dict) else ""
diff --git a/src/sap_cloud_sdk/dms/client.py b/src/sap_cloud_sdk/dms/client.py
index 61f7b3ec..5bcaa785 100644
--- a/src/sap_cloud_sdk/dms/client.py
+++ b/src/sap_cloud_sdk/dms/client.py
@@ -21,9 +21,10 @@
     QueryResultPage,
     _prop_val,
 )
-from sap_cloud_sdk.dms._auth import Auth
 from sap_cloud_sdk.dms._http import HttpInvoker
+from sap_cloud_sdk.core.protocol.http import XsuaaAuthProvider
 from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics
+from typing import Callable
 
 logger = logging.getLogger(__name__)
 
@@ -86,7 +87,7 @@ class DMSClient:
 
     def __init__(
         self,
-        credentials: DMSCredentials,
+        credentials: Union[DMSCredentials, Callable[[], DMSCredentials]],
         connect_timeout: Optional[int] = None,
         read_timeout: Optional[int] = None,
     ) -> None:
@@ -98,14 +99,20 @@ def __init__(
             authentication and handles environment detection.
 
         Args:
-            credentials: OAuth2 credentials and service URI for the DMS instance.
+            credentials: OAuth2 credentials (or a factory returning them) for the DMS instance.
             connect_timeout: TCP connection timeout in seconds. Defaults to 10.
             read_timeout: Response read timeout in seconds. Defaults to 30.
         """
-        auth = Auth(credentials)
+        factory: Callable[[], DMSCredentials]
+        if callable(credentials) and not isinstance(credentials, DMSCredentials):
+            factory = credentials
+        else:
+            factory = lambda: credentials  # type: ignore[return-value]
+        auth_provider = XsuaaAuthProvider(factory)
+        base_url = factory().uri
         self._http: HttpInvoker = HttpInvoker(
-            auth=auth,
-            base_url=credentials.uri,
+            auth_provider=auth_provider,
+            base_url=base_url,
             connect_timeout=connect_timeout,
             read_timeout=read_timeout,
         )
diff --git a/src/sap_cloud_sdk/dms/config.py b/src/sap_cloud_sdk/dms/config.py
index 8a396e16..7940fe56 100644
--- a/src/sap_cloud_sdk/dms/config.py
+++ b/src/sap_cloud_sdk/dms/config.py
@@ -1,6 +1,6 @@
 import json
 from dataclasses import dataclass
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, TYPE_CHECKING
 from urllib.parse import urlparse
 
 from sap_cloud_sdk.core.secret_resolver.resolver import (
@@ -9,6 +9,9 @@
 from sap_cloud_sdk.destination.exceptions import ConfigError
 from sap_cloud_sdk.dms.model import DMSCredentials
 
+if TYPE_CHECKING:
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
+
 
 @dataclass
 class BindingData:
@@ -19,8 +22,8 @@ class BindingData:
         uaa: JSON string containing XSUAA authentication credentials
     """
 
-    uri: str
-    uaa: str
+    uri: str = ""
+    uaa: str = ""
 
     def validate(self) -> None:
         """Validate the binding data.
@@ -134,3 +137,37 @@ def load_sdm_config_from_env_or_mount(instance: Optional[str] = None) -> DMSCred
         raise ConfigError(
             f"failed to load sdm configuration for instance='{inst}': {e}"
         )
+
+
+def _make_config_factory(
+    instance: Optional[str] = None,
+) -> "ConfigFactory[DMSCredentials]":
+    """Return a :class:`~sap_cloud_sdk.core.secret_resolver.ConfigFactory` for the given instance.
+
+    The factory re-reads the binding on every call and tracks the secret
+    directory mtime for proactive rotation detection.
+
+    Args:
+        instance: Binding instance name. Defaults to ``"default"``.
+
+    Returns:
+        A callable that produces fresh :class:`DMSCredentials`.
+    """
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
+
+    inst = instance or "default"
+
+    def _extract(binding: BindingData) -> DMSCredentials:
+        try:
+            return binding.to_credentials()
+        except Exception as exc:
+            raise ConfigError(
+                f"failed to load DMS configuration for instance '{inst}': {exc}"
+            ) from exc
+
+    return ConfigFactory(
+        module="sdm",
+        instance=inst,
+        binding_cls=BindingData,
+        extract=_extract,
+    )
diff --git a/src/sap_cloud_sdk/objectstore/__init__.py b/src/sap_cloud_sdk/objectstore/__init__.py
index 770c1ad8..ea2f815a 100644
--- a/src/sap_cloud_sdk/objectstore/__init__.py
+++ b/src/sap_cloud_sdk/objectstore/__init__.py
@@ -1,6 +1,7 @@
 """SAP Cloud SDK for Python - Object Store module
 
-The create_client() uses secret resolver to load credentials from mounts/env vars
+The create_client() uses ConfigFactory to load credentials from mounts/env vars
+with proactive rotation detection via mtime tracking.
 
 Usage:
     from sap_cloud_sdk.objectstore import create_client
@@ -19,7 +20,15 @@
 )
 from sap_cloud_sdk.objectstore._models import ObjectStoreBindingData, ObjectMetadata
 from sap_cloud_sdk.objectstore._s3 import ObjectStoreClient
-from sap_cloud_sdk.core.secret_resolver import read_from_mount_and_fallback_to_env_var
+
+
+def _make_static_factory(config: ObjectStoreBindingData):
+    """Wrap a fixed config in a no-op factory (no rotation tracking)."""
+
+    def _factory() -> ObjectStoreBindingData:
+        return config
+
+    return _factory
 
 
 def create_client(
@@ -28,13 +37,15 @@ def create_client(
     config: Optional[ObjectStoreBindingData] = None,
     disable_ssl: bool = False,
 ) -> ObjectStoreClient:
-    """Creates an ObjectStoreClient with automatic local/cloud detection.
-    Uses secret resolver to load credentials from mounted secrets or environment variables
+    """Create an ObjectStoreClient with automatic credential detection.
+
+    Credentials are loaded from a mounted volume or environment variables and
+    tracked for secret rotation via :class:`~sap_cloud_sdk.core.secret_resolver.ConfigFactory`.
 
     Args:
-        instance: Instance name for cloud mode secret resolution. Must be a non-empty string.
-        config: Optional explicit configuration. If provided, auto-detection is skipped
-                and this configuration is used directly.
+        instance: Instance name for secret resolution. Must be a non-empty string.
+        config: Optional explicit configuration. When provided, binding
+                discovery is skipped and rotation tracking is disabled.
         disable_ssl: Whether to disable SSL/TLS connections. Defaults to False.
 
     Returns:
@@ -47,20 +58,18 @@ def create_client(
     if not instance or not instance.strip():
         raise ValueError("instance parameter must be a non-empty string")
 
-    # Cloud mode: with explicit configuration
     if config is not None:
-        return ObjectStoreClient(config, disable_ssl=disable_ssl)
+        return ObjectStoreClient(_make_static_factory(config), disable_ssl=disable_ssl)
+
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
 
-    # Cloud mode: use secret resolver to load configuration
-    config = ObjectStoreBindingData()
-    read_from_mount_and_fallback_to_env_var(
-        base_volume_mount="/etc/secrets/appfnd",
-        base_var_name="CLOUD_SDK_CFG",
+    factory: ConfigFactory[ObjectStoreBindingData] = ConfigFactory(
         module="objectstore",
         instance=instance,
-        target=config,
+        binding_cls=ObjectStoreBindingData,
+        extract=lambda b: b,
     )
-    return ObjectStoreClient(config, disable_ssl=disable_ssl)
+    return ObjectStoreClient(factory, disable_ssl=disable_ssl)
 
 
 __all__ = [
diff --git a/src/sap_cloud_sdk/objectstore/_models.py b/src/sap_cloud_sdk/objectstore/_models.py
index 84902461..a6d6083e 100644
--- a/src/sap_cloud_sdk/objectstore/_models.py
+++ b/src/sap_cloud_sdk/objectstore/_models.py
@@ -18,6 +18,25 @@ class ObjectStoreBindingData:
     bucket: str = ""
     host: str = ""
 
+    def validate(self) -> None:
+        """Raise ClientCreationError if any required field is empty."""
+        from sap_cloud_sdk.objectstore.exceptions import ClientCreationError
+
+        missing = [
+            name
+            for name, value in [
+                ("access_key_id", self.access_key_id),
+                ("secret_access_key", self.secret_access_key),
+                ("bucket", self.bucket),
+                ("host", self.host),
+            ]
+            if not value
+        ]
+        if missing:
+            raise ClientCreationError(
+                f"Object Store binding missing required fields: {', '.join(missing)}"
+            )
+
 
 @dataclass(frozen=True)
 class ObjectMetadata:
diff --git a/src/sap_cloud_sdk/objectstore/_s3.py b/src/sap_cloud_sdk/objectstore/_s3.py
index cfc8168c..7131a794 100644
--- a/src/sap_cloud_sdk/objectstore/_s3.py
+++ b/src/sap_cloud_sdk/objectstore/_s3.py
@@ -1,10 +1,14 @@
 """S3 backend implementation for object store operations using MinIO client."""
 
 import io
+import logging
 import os
+import threading
+
+logger = logging.getLogger(__name__)
 from datetime import datetime
 from http.client import HTTPResponse
-from typing import BinaryIO, List, cast
+from typing import TYPE_CHECKING, Any, BinaryIO, Callable, List, TypeVar, cast
 
 import minio.datatypes
 from minio import Minio
@@ -13,13 +17,16 @@
 from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics
 from sap_cloud_sdk.objectstore.exceptions import (
     ClientCreationError,
-    ObjectOperationError,
-    ObjectNotFoundError,
     ListObjectsError,
+    ObjectNotFoundError,
+    ObjectOperationError,
 )
 from sap_cloud_sdk.objectstore._models import ObjectStoreBindingData, ObjectMetadata
 from sap_cloud_sdk.objectstore.utils import _normalize_host
 
+if TYPE_CHECKING:
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
+
 # Validation error message constants
 EMPTY_NAME_ERROR = "name must be a non-empty string"
 EMPTY_CONTENT_TYPE_ERROR = "content_type must be a non-empty string"
@@ -29,33 +36,51 @@
 NEGATIVE_SIZE_ERROR = "size must be non-negative"
 INVALID_PREFIX_TYPE_ERROR = "prefix must be a string"
 
+# S3 error codes that indicate credential rejection (trigger reactive refresh)
+_CREDENTIAL_ERROR_CODES = frozenset({"InvalidAccessKeyId", "SignatureDoesNotMatch"})
+
+_T = TypeVar("_T")
+
 
 class ObjectStoreClient:
-    """S3-compatible object storage client.
+    """S3-compatible object storage client with binding-rotation support.
 
     Provides a unified interface for object storage operations using the MinIO client library.
     Supports upload, download, delete, list, and metadata operations on S3-compatible storage.
+
+    Rotation resilience is handled in two layers:
+    - **Proactive**: checks the secret-directory mtime via the config factory's
+      ``has_changed()`` method before every operation and rebuilds the MinIO client
+      when a change is detected.
+    - **Reactive**: on ``InvalidAccessKeyId`` or ``SignatureDoesNotMatch`` S3 errors,
+      refreshes credentials and retries the operation exactly once.
     """
 
     def __init__(
-        self, creds_config: ObjectStoreBindingData, *, disable_ssl: bool = False
+        self,
+        config_factory: "ConfigFactory[ObjectStoreBindingData]",
+        *,
+        disable_ssl: bool = False,
     ) -> None:
         """Initialize the object storage client.
 
         Args:
-            creds_config: Connection credentials and endpoint configuration.
+            config_factory: Factory that re-reads S3 credentials on every call.
+                Must implement the :class:`~sap_cloud_sdk.core.secret_resolver.ConfigFactory`
+                protocol (callable + optional ``has_changed()``).
             disable_ssl: Whether to disable SSL/TLS connections. Defaults to False.
 
         Raises:
             ClientCreationError: If client initialization fails.
         """
-
-        self._creds_config = creds_config
+        self._config_factory = config_factory
         self._disable_ssl = disable_ssl
+        self._lock = threading.Lock()
+        self._creds_config = config_factory()
         self._minio_client = self._create_minio_client()
 
     def _create_minio_client(self) -> Minio:
-        """Create MinIO client with proper configuration."""
+        """Create MinIO client from the current credentials config."""
         try:
             return Minio(
                 endpoint=_normalize_host(self._creds_config.host),
@@ -63,10 +88,38 @@ def _create_minio_client(self) -> Minio:
                 secret_key=self._creds_config.secret_access_key,
                 secure=not self._disable_ssl,
             )
-
         except Exception as e:
             raise ClientCreationError(f"Failed to create MinIO client: {e}") from e
 
+    def _refresh_credentials(self) -> None:
+        """Re-read credentials and rebuild the MinIO client. Caller must hold ``_lock``."""
+        self._creds_config = self._config_factory()
+        self._minio_client = self._create_minio_client()
+
+    def _refresh_if_rotated(self) -> None:
+        """Proactively refresh if the secret directory mtime has changed."""
+        has_changed: Any = getattr(self._config_factory, "has_changed", None)
+        if callable(has_changed) and has_changed():
+            with self._lock:
+                self._refresh_credentials()
+
+    def _execute_with_retry(self, fn: Callable[[], _T]) -> _T:
+        """Run *fn* against the current MinIO client, retrying once on credential errors.
+
+        Calls ``_refresh_if_rotated()`` first (proactive), then executes *fn*.
+        On ``InvalidAccessKeyId`` or ``SignatureDoesNotMatch``, refreshes credentials
+        and retries exactly once (reactive).
+        """
+        self._refresh_if_rotated()
+        try:
+            return fn()
+        except S3Error as e:
+            if e.code in _CREDENTIAL_ERROR_CODES:
+                with self._lock:
+                    self._refresh_credentials()
+                return fn()
+            raise
+
     @record_metrics(Module.OBJECTSTORE, Operation.OBJECTSTORE_PUT_OBJECT_FROM_BYTES)
     def put_object_from_bytes(self, name: str, data: bytes, content_type: str) -> None:
         """Upload an object from bytes.
@@ -88,12 +141,14 @@ def put_object_from_bytes(self, name: str, data: bytes, content_type: str) -> No
             raise ValueError(EMPTY_CONTENT_TYPE_ERROR)
 
         try:
-            self._minio_client.put_object(
-                bucket_name=self._creds_config.bucket,
-                object_name=name,
-                data=io.BytesIO(data),
-                length=len(data),
-                content_type=content_type,
+            self._execute_with_retry(
+                lambda: self._minio_client.put_object(
+                    bucket_name=self._creds_config.bucket,
+                    object_name=name,
+                    data=io.BytesIO(data),
+                    length=len(data),
+                    content_type=content_type,
+                )
             )
         except S3Error as e:
             raise ObjectOperationError(
@@ -128,12 +183,14 @@ def put_object(
             raise ValueError(EMPTY_CONTENT_TYPE_ERROR)
 
         try:
-            self._minio_client.put_object(
-                bucket_name=self._creds_config.bucket,
-                object_name=name,
-                data=stream,
-                length=size,
-                content_type=content_type,
+            self._execute_with_retry(
+                lambda: self._minio_client.put_object(
+                    bucket_name=self._creds_config.bucket,
+                    object_name=name,
+                    data=stream,
+                    length=size,
+                    content_type=content_type,
+                )
             )
         except S3Error as e:
             raise ObjectOperationError(
@@ -165,19 +222,20 @@ def put_object_from_file(
             raise ValueError(EMPTY_CONTENT_TYPE_ERROR)
 
         try:
-            # Check if file exists and get size
             if not os.path.isfile(file_path):
                 raise ObjectOperationError(f"File not found: {file_path}")
 
             file_size = os.path.getsize(file_path)
 
             with open(file_path, "rb") as file_stream:
-                self._minio_client.put_object(
-                    bucket_name=self._creds_config.bucket,
-                    object_name=name,
-                    data=file_stream,
-                    length=file_size,
-                    content_type=content_type,
+                self._execute_with_retry(
+                    lambda: self._minio_client.put_object(
+                        bucket_name=self._creds_config.bucket,
+                        object_name=name,
+                        data=file_stream,
+                        length=file_size,
+                        content_type=content_type,
+                    )
                 )
         except S3Error as e:
             raise ObjectOperationError(
@@ -207,8 +265,10 @@ def get_object(self, name: str) -> HTTPResponse:
         try:
             response = cast(
                 HTTPResponse,
-                self._minio_client.get_object(
-                    bucket_name=self._creds_config.bucket, object_name=name
+                self._execute_with_retry(
+                    lambda: self._minio_client.get_object(
+                        bucket_name=self._creds_config.bucket, object_name=name
+                    )
                 ),
             )
             return response
@@ -238,15 +298,17 @@ def delete_object(self, name: str) -> None:
             raise ValueError(EMPTY_NAME_ERROR)
 
         try:
-            self._minio_client.remove_object(
-                bucket_name=self._creds_config.bucket, object_name=name
+            self._execute_with_retry(
+                lambda: self._minio_client.remove_object(
+                    bucket_name=self._creds_config.bucket, object_name=name
+                )
             )
         except S3Error as e:
             if e.code != "NoSuchKey":
                 raise ObjectOperationError(
                     f"Failed to delete object '{name}': {e.code} - {e.message}"
                 ) from e
-            # For NoSuchKey, we still consider it successful (idempotent delete)
+            # NoSuchKey is treated as a successful idempotent delete
         except Exception as e:
             raise ObjectOperationError(f"Failed to delete object '{name}': {e}") from e
 
@@ -269,8 +331,10 @@ def list_objects(self, prefix: str) -> List[ObjectMetadata]:
 
         result = []
         try:
-            objects = self._minio_client.list_objects(
-                bucket_name=self._creds_config.bucket, prefix=prefix
+            objects = self._execute_with_retry(
+                lambda: self._minio_client.list_objects(
+                    bucket_name=self._creds_config.bucket, prefix=prefix
+                )
             )
 
             for obj in objects:
@@ -313,17 +377,19 @@ def head_object(self, name: str) -> ObjectMetadata:
             raise ValueError(EMPTY_NAME_ERROR)
 
         try:
-            stat: minio.datatypes.Object = self._minio_client.stat_object(
-                bucket_name=self._creds_config.bucket, object_name=name
+            stat: minio.datatypes.Object = self._execute_with_retry(
+                lambda: self._minio_client.stat_object(
+                    bucket_name=self._creds_config.bucket, object_name=name
+                )
             )
 
             return ObjectMetadata(
                 key=name,
                 last_modified=stat.last_modified or datetime.min,
-                etag=(stat.etag or "").strip('"'),  # Remove quotes from etag
+                etag=(stat.etag or "").strip('"'),
                 size=stat.size or 0,
-                storage_class=None,  # stat_object doesn't provide storage class
-                owner=None,  # stat_object doesn't provide owner
+                storage_class=None,
+                owner=None,
             )
         except S3Error as e:
             if e.code == "NoSuchKey":
diff --git a/src/sap_cloud_sdk/print/__init__.py b/src/sap_cloud_sdk/print/__init__.py
index 69397096..0d6b7a10 100644
--- a/src/sap_cloud_sdk/print/__init__.py
+++ b/src/sap_cloud_sdk/print/__init__.py
@@ -34,7 +34,7 @@
     PrintTask,
     PrintTaskMetadata,
 )
-from sap_cloud_sdk.print.config import load_from_env_or_mount, PrintConfig
+from sap_cloud_sdk.print.config import PrintConfig, _make_config_factory
 from sap_cloud_sdk.print._http import PrintHttp, TokenProvider
 from sap_cloud_sdk.print.client import PrintClient
 from sap_cloud_sdk.print.exceptions import (
@@ -72,9 +72,12 @@ def create_client(
         ClientCreationError: If client creation fails.
     """
     try:
-        binding = config or load_from_env_or_mount(instance)
-        tp = TokenProvider(binding)
-        http = PrintHttp(config=binding, token_provider=tp)
+        if config is not None:
+            tp = TokenProvider(config)
+        else:
+            factory = _make_config_factory(instance)
+            tp = TokenProvider(factory)
+        http = PrintHttp(config=tp._config, token_provider=tp)
         return PrintClient(http, _telemetry_source=_telemetry_source)
     except Exception as e:
         _record_error_metric(
diff --git a/src/sap_cloud_sdk/print/_http.py b/src/sap_cloud_sdk/print/_http.py
index ce12da7d..2a8715f9 100644
--- a/src/sap_cloud_sdk/print/_http.py
+++ b/src/sap_cloud_sdk/print/_http.py
@@ -5,7 +5,7 @@
 import base64
 import json
 import logging
-from typing import Any, Dict, Optional, Protocol
+from typing import Any, Callable, Dict, Optional, Protocol
 
 import requests
 from requests import Response
@@ -27,14 +27,34 @@ def resolve_username(self) -> str: ...
 
 
 class TokenProvider:
-    """Provides OAuth2 access tokens via client credentials flow."""
-
-    def __init__(self, config: PrintConfig) -> None:
-        self._config = config
-        client = BackendApplicationClient(client_id=config.client_id)
+    """Provides OAuth2 access tokens via client credentials flow.
+
+    Accepts either a fixed :class:`PrintConfig` or a config factory (any callable
+    returning ``PrintConfig`` with an optional ``has_changed() -> bool`` method).
+    When a factory is supplied, credentials are re-read on every token fetch and
+    the factory's ``has_changed()`` method is checked before serving a cached token
+    so that rotated secrets are picked up proactively.
+    """
+
+    def __init__(self, config: PrintConfig | Callable[[], PrintConfig]) -> None:
+        if callable(config) and not isinstance(config, PrintConfig):
+            self._config_factory: Callable[[], PrintConfig] = config
+            self._config = config()
+        else:
+            self._config_factory = lambda: config  # type: ignore[arg-type]
+            self._config = config  # type: ignore[assignment]
+        client = BackendApplicationClient(client_id=self._config.client_id)
         self._session = OAuth2Session(client=client)
         self._cached_token: Optional[str] = None
 
+    def _refresh_if_rotated(self) -> None:
+        has_changed = getattr(self._config_factory, "has_changed", None)
+        if callable(has_changed) and has_changed():
+            self._config = self._config_factory()
+            self._cached_token = None
+            client = BackendApplicationClient(client_id=self._config.client_id)
+            self._session = OAuth2Session(client=client)
+
     def get_token(self) -> str:
         """Return a valid bearer token for the Print Service.
 
@@ -45,6 +65,7 @@ def get_token(self) -> str:
             HttpError: If the token response is missing an access_token or
                 token acquisition fails.
         """
+        self._refresh_if_rotated()
 
         try:
             token: Dict[str, Any] = self._session.fetch_token(
diff --git a/src/sap_cloud_sdk/print/config.py b/src/sap_cloud_sdk/print/config.py
index f3a5e996..7b28e7c7 100644
--- a/src/sap_cloud_sdk/print/config.py
+++ b/src/sap_cloud_sdk/print/config.py
@@ -15,7 +15,7 @@
 """
 
 from dataclasses import dataclass
-from typing import Optional
+from typing import Optional, TYPE_CHECKING
 import json
 import logging
 
@@ -24,6 +24,9 @@
 )
 from sap_cloud_sdk.print.exceptions import ConfigError
 
+if TYPE_CHECKING:
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
+
 logger = logging.getLogger(__name__)
 
 
@@ -118,3 +121,37 @@ def load_from_env_or_mount(instance: Optional[str] = None) -> PrintConfig:
         raise ConfigError(
             f"failed to load print configuration for instance='{inst}': {e}"
         ) from e
+
+
+def _make_config_factory(
+    instance: Optional[str] = None,
+) -> "ConfigFactory[PrintConfig]":
+    """Return a :class:`~sap_cloud_sdk.core.secret_resolver.ConfigFactory` for the given instance.
+
+    The factory re-reads the binding on every call and tracks the secret
+    directory mtime for proactive rotation detection.
+
+    Args:
+        instance: Binding instance name. Defaults to ``"default"``.
+
+    Returns:
+        A callable that produces a fresh :class:`PrintConfig`.
+    """
+    from sap_cloud_sdk.core.secret_resolver import ConfigFactory
+
+    inst = instance or "default"
+
+    def _extract(binding: _BindingData) -> PrintConfig:
+        try:
+            return binding.to_config()
+        except Exception as exc:
+            raise ConfigError(
+                f"failed to load print configuration for instance '{inst}': {exc}"
+            ) from exc
+
+    return ConfigFactory(
+        module="print",
+        instance=inst,
+        binding_cls=_BindingData,
+        extract=_extract,
+    )
diff --git a/tests/adms/integration/conftest.py b/tests/adms/integration/conftest.py
index c691f19d..66368fae 100644
--- a/tests/adms/integration/conftest.py
+++ b/tests/adms/integration/conftest.py
@@ -24,42 +24,30 @@
     AsyncAdmsClient,
     create_async_client,
 )
-from sap_cloud_sdk.adms.config import AdmsConfig, load_from_env_or_mount
 from sap_cloud_sdk.adms.exceptions import ConfigError
 
 
 # ---------------------------------------------------------------------------
-# Configuration fixture
+# Client fixtures
 # ---------------------------------------------------------------------------
 
 
 @pytest.fixture(scope="session")
-def adms_config() -> AdmsConfig:
-    """Resolve AdmsConfig from env/secret-mount.
-
-    Skips the entire integration suite when required credentials are missing.
-    """
+def adms_client() -> AdmsClient:
+    """Sync AdmsClient wired to the real ADM instance via ConfigFactory."""
     try:
-        return load_from_env_or_mount("default")
+        return create_client()
     except ConfigError as exc:
         pytest.skip(f"ADMS integration tests skipped — missing config: {exc}")
 
 
-# ---------------------------------------------------------------------------
-# Client fixtures
-# ---------------------------------------------------------------------------
-
-
-@pytest.fixture(scope="session")
-def adms_client(adms_config: AdmsConfig) -> AdmsClient:
-    """Sync AdmsClient wired to the real ADM instance."""
-    return create_client(config=adms_config)
-
-
 @pytest.fixture(scope="function")
-def async_adms_client(adms_config: AdmsConfig) -> AsyncAdmsClient:
-    """Async AdmsClient wired to the real ADM instance."""
-    return create_async_client(config=adms_config)
+def async_adms_client() -> AsyncAdmsClient:
+    """Async AdmsClient wired to the real ADM instance via ConfigFactory."""
+    try:
+        return create_async_client()
+    except ConfigError as exc:
+        pytest.skip(f"ADMS integration tests skipped — missing config: {exc}")
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/adms/unit/test_client.py b/tests/adms/unit/test_client.py
index f789428a..fe5e23d2 100644
--- a/tests/adms/unit/test_client.py
+++ b/tests/adms/unit/test_client.py
@@ -143,26 +143,36 @@ def test_with_user_jwt_uses_new_http(self, mock_http):
 
 class TestCreateClientFactory:
     def test_raises_config_error_on_missing_binding(self):
+        factory = MagicMock(side_effect=ConfigError("missing fields"))
         with patch(
-            "sap_cloud_sdk.adms.client.load_from_env_or_mount",
-            side_effect=ConfigError("missing fields"),
+            "sap_cloud_sdk.adms.client._make_config_factory",
+            return_value=factory,
         ):
             with pytest.raises(ConfigError, match="missing fields"):
                 create_client(instance="nonexistent-instance")
 
     def test_unexpected_exception_propagates_as_is(self):
-        """Real bugs (e.g. ``RuntimeError`` from internal logic) must surface
-        as themselves rather than being silently wrapped — wrapping makes
-        debugging harder and previously masked SDK programming errors as
-        "client creation failed".
-        """
+        """Exceptions other than RuntimeError (e.g. programming errors) must
+        surface as themselves rather than being silently swallowed."""
+        factory = MagicMock(side_effect=ValueError("unexpected"))
         with patch(
-            "sap_cloud_sdk.adms.client.load_from_env_or_mount",
-            side_effect=RuntimeError("unexpected"),
+            "sap_cloud_sdk.adms.client._make_config_factory",
+            return_value=factory,
         ):
-            with pytest.raises(RuntimeError, match="unexpected"):
+            with pytest.raises(ValueError, match="unexpected"):
                 create_client(instance="bad-instance")
 
+    def test_runtime_error_from_secret_resolver_becomes_config_error(self):
+        """RuntimeError from ConfigFactory (missing secrets) must be wrapped as
+        ConfigError so callers only need to handle one exception type."""
+        factory = MagicMock(side_effect=RuntimeError("env var not found: CLOUD_SDK_CFG_ADMS_DEFAULT_CLIENTID"))
+        with patch(
+            "sap_cloud_sdk.adms.client._make_config_factory",
+            return_value=factory,
+        ):
+            with pytest.raises(ConfigError):
+                create_client(instance="missing")
+
     def test_returns_adms_client_on_success(self):
         mock_config = AdmsConfig(
             service_url="https://adm.example.com",
@@ -170,9 +180,11 @@ def test_returns_adms_client_on_success(self):
             client_id="cid",
             client_secret="cs",
         )
+        factory = MagicMock(return_value=mock_config)
+        factory.has_changed = MagicMock(return_value=False)
         with patch(
-            "sap_cloud_sdk.adms.client.load_from_env_or_mount",
-            return_value=mock_config,
+            "sap_cloud_sdk.adms.client._make_config_factory",
+            return_value=factory,
         ):
             client = create_client()
 
@@ -185,10 +197,10 @@ def test_accepts_explicit_config(self):
             client_id="cid",
             client_secret="cs",
         )
-        with patch("sap_cloud_sdk.adms.client.load_from_env_or_mount") as mock_load:
+        with patch("sap_cloud_sdk.adms.client._make_config_factory") as mock_factory_fn:
             client = create_client(config=mock_config)
 
-        mock_load.assert_not_called()
+        mock_factory_fn.assert_not_called()
         assert isinstance(client, AdmsClient)
 
     def test_user_jwt_forwarded_to_http(self):
@@ -198,9 +210,11 @@ def test_user_jwt_forwarded_to_http(self):
             client_id="cid",
             client_secret="cs",
         )
+        factory = MagicMock(return_value=mock_config)
+        factory.has_changed = MagicMock(return_value=False)
         with patch(
-            "sap_cloud_sdk.adms.client.load_from_env_or_mount",
-            return_value=mock_config,
+            "sap_cloud_sdk.adms.client._make_config_factory",
+            return_value=factory,
         ):
             client = create_client(user_jwt="user-jwt-123")
 
@@ -444,25 +458,28 @@ async def test_context_manager(self, config):
 
 class TestCreateAsyncClient:
     def test_raises_config_error_when_no_binding(self):
+        mock_factory = MagicMock(side_effect=ConfigError("no binding"))
         with patch(
-            "sap_cloud_sdk.adms.client.load_from_env_or_mount",
-            side_effect=ConfigError("no binding"),
+            "sap_cloud_sdk.adms.client._make_config_factory",
+            return_value=mock_factory,
         ):
             with pytest.raises(ConfigError):
                 create_async_client(instance="missing")
 
     def test_returns_async_client(self, config):
+        mock_factory = MagicMock(return_value=config)
+        mock_factory.has_changed = MagicMock(return_value=False)
         with patch(
-            "sap_cloud_sdk.adms.client.load_from_env_or_mount",
-            return_value=config,
+            "sap_cloud_sdk.adms.client._make_config_factory",
+            return_value=mock_factory,
         ):
             client = create_async_client()
         assert isinstance(client, AsyncAdmsClient)
 
     def test_accepts_explicit_config(self, config):
-        with patch("sap_cloud_sdk.adms.client.load_from_env_or_mount") as mock_load:
+        with patch("sap_cloud_sdk.adms.client._make_config_factory") as mock_make:
             client = create_async_client(config=config)
-        mock_load.assert_not_called()
+        mock_make.assert_not_called()
         assert isinstance(client, AsyncAdmsClient)
 
 
diff --git a/tests/adms/unit/test_ias_fetcher.py b/tests/adms/unit/test_ias_fetcher.py
index bb25b67d..d4c87a8c 100644
--- a/tests/adms/unit/test_ias_fetcher.py
+++ b/tests/adms/unit/test_ias_fetcher.py
@@ -197,3 +197,51 @@ def test_obo_and_cc_caches_are_isolated(self, fetcher, mock_session):
             if call[1]["data"]["grant_type"] == "client_credentials"
         ]
         assert len(cc_grant_calls) == 1
+
+
+class TestIasTokenFetcherRotation:
+
+    def test_proactive_rotation_clears_cache_when_binding_changed(self, mock_session):
+        original_config = _make_config()
+        new_config = _make_config(client_id="new-client-id", client_secret="new-secret")
+
+        mock_session.post.return_value = _make_token_response("new-token")
+
+        mock_factory = MagicMock(return_value=original_config)
+        mock_factory.has_changed = MagicMock(side_effect=[False, True])
+
+        fetcher = IasTokenFetcher(config=mock_factory, session=mock_session)
+        # Seed the cache
+        mock_session.post.return_value = _make_token_response("old-token")
+        fetcher.get_token()
+        assert fetcher._cache.get(_CC_CACHE_KEY) == "old-token"
+
+        # Next call: has_changed() returns True → cache is cleared → new token fetched
+        mock_factory.return_value = new_config
+        mock_session.post.return_value = _make_token_response("new-token")
+        token = fetcher.get_token()
+
+        assert token == "new-token"
+        assert fetcher._config is new_config
+
+    def test_no_cache_clear_when_binding_unchanged(self, mock_session):
+        config = _make_config()
+        mock_session.post.return_value = _make_token_response("cached-token")
+
+        mock_factory = MagicMock(return_value=config)
+        mock_factory.has_changed = MagicMock(return_value=False)
+
+        fetcher = IasTokenFetcher(config=mock_factory, session=mock_session)
+        fetcher.get_token()  # populates cache
+
+        mock_session.post.reset_mock()
+        fetcher.get_token()  # second call: has_changed() False → cache hit
+
+        mock_session.post.assert_not_called()
+
+    def test_static_config_skips_rotation_check(self, config, mock_session):
+        mock_session.post.return_value = _make_token_response("tok")
+        fetcher = IasTokenFetcher(config=config, session=mock_session)
+        fetcher.get_token()
+        # No has_changed attribute on a plain AdmsConfig — should not raise
+        assert fetcher._cache.get(_CC_CACHE_KEY) == "tok"
diff --git a/tests/core/unit/auditlog/unit/test_create_client.py b/tests/core/unit/auditlog/unit/test_create_client.py
index f3b86a20..c68740dd 100644
--- a/tests/core/unit/auditlog/unit/test_create_client.py
+++ b/tests/core/unit/auditlog/unit/test_create_client.py
@@ -11,16 +11,11 @@
 
 class TestCreateClient:
 
-    @patch('sap_cloud_sdk.core.auditlog._load_config_from_env')
+    @patch('sap_cloud_sdk.core.auditlog._make_config_factory')
     @patch('sap_cloud_sdk.core.auditlog.HttpTransport')
-    def test_create_client_cloud_mode(self, mock_http_transport, mock_load_config):
-        mock_config = AuditLogConfig(
-            client_id="test_client",
-            client_secret="test_secret",
-            oauth_url="https://oauth.example.com",
-            service_url="https://service.example.com"
-        )
-        mock_load_config.return_value = mock_config
+    def test_create_client_cloud_mode(self, mock_http_transport, mock_make_factory):
+        mock_factory = MagicMock()
+        mock_make_factory.return_value = mock_factory
 
         mock_transport = MagicMock()
         mock_http_transport.return_value = mock_transport
@@ -28,8 +23,8 @@ def test_create_client_cloud_mode(self, mock_http_transport, mock_load_config):
         client = create_client()
 
         assert isinstance(client, AuditLogClient)
-        mock_load_config.assert_called_once()
-        mock_http_transport.assert_called_once_with(mock_config)
+        mock_make_factory.assert_called_once()
+        mock_http_transport.assert_called_once_with(mock_factory)
         assert client._transport == mock_transport
 
     @patch('sap_cloud_sdk.core.auditlog.HttpTransport')
@@ -50,23 +45,18 @@ def test_create_client_with_custom_config(self, mock_http_transport):
         mock_http_transport.assert_called_once_with(custom_config)
         assert client._transport == mock_transport
 
-    @patch('sap_cloud_sdk.core.auditlog._load_config_from_env')
-    def test_create_client_config_loading_exception(self, mock_load_config):
-        mock_load_config.side_effect = Exception("Config loading failed")
+    @patch('sap_cloud_sdk.core.auditlog._make_config_factory')
+    def test_create_client_config_loading_exception(self, mock_make_factory):
+        mock_make_factory.side_effect = Exception("Config loading failed")
 
         with pytest.raises(ClientCreationError, match="Failed to create audit log client"):
             create_client()
 
-    @patch('sap_cloud_sdk.core.auditlog._load_config_from_env')
+    @patch('sap_cloud_sdk.core.auditlog._make_config_factory')
     @patch('sap_cloud_sdk.core.auditlog.HttpTransport')
-    def test_create_client_http_transport_exception(self, mock_http_transport, mock_load_config):
-        mock_config = AuditLogConfig(
-            client_id="test_client",
-            client_secret="test_secret",
-            oauth_url="https://oauth.example.com",
-            service_url="https://service.example.com"
-        )
-        mock_load_config.return_value = mock_config
+    def test_create_client_http_transport_exception(self, mock_http_transport, mock_make_factory):
+        mock_factory = MagicMock()
+        mock_make_factory.return_value = mock_factory
 
         mock_http_transport.side_effect = Exception("HTTP transport failed")
 
@@ -87,17 +77,12 @@ def test_create_client_custom_config_transport_exception(self, mock_http_transpo
         with pytest.raises(ClientCreationError, match="Failed to create audit log client"):
             create_client(config=custom_config)
 
-    @patch('sap_cloud_sdk.core.auditlog._load_config_from_env')
+    @patch('sap_cloud_sdk.core.auditlog._make_config_factory')
     @patch('sap_cloud_sdk.core.auditlog.HttpTransport')
     @patch('sap_cloud_sdk.core.auditlog.AuditLogClient')
-    def test_create_client_client_creation_exception(self, mock_client_class, mock_http_transport, mock_load_config):
-        mock_config = AuditLogConfig(
-            client_id="test_client",
-            client_secret="test_secret",
-            oauth_url="https://oauth.example.com",
-            service_url="https://service.example.com"
-        )
-        mock_load_config.return_value = mock_config
+    def test_create_client_client_creation_exception(self, mock_client_class, mock_http_transport, mock_make_factory):
+        mock_factory = MagicMock()
+        mock_make_factory.return_value = mock_factory
 
         mock_transport = MagicMock()
         mock_http_transport.return_value = mock_transport
diff --git a/tests/core/unit/auditlog/unit/test_http_transport.py b/tests/core/unit/auditlog/unit/test_http_transport.py
index 467f5ab0..288c4504 100644
--- a/tests/core/unit/auditlog/unit/test_http_transport.py
+++ b/tests/core/unit/auditlog/unit/test_http_transport.py
@@ -29,6 +29,9 @@ def test_initialization_success(self, mock_oauth_session):
         mock_session = MagicMock()
         mock_oauth_session.return_value = mock_session
         mock_session.fetch_token.return_value = {"access_token": "test_token"}
+        mock_response = MagicMock()
+        mock_response.status_code = 201
+        mock_session.post.return_value = mock_response
 
         config = AuditLogConfig(
             client_id="test_client",
@@ -38,8 +41,10 @@ def test_initialization_success(self, mock_oauth_session):
         )
 
         transport = HttpTransport(config)
-
         assert transport.config == config
+        assert transport.oauth is None  # lazy — no session yet
+
+        transport.send(SecurityEvent(data="init test"))
         mock_session.fetch_token.assert_called_once_with(
             token_url="https://oauth.example.com/oauth/token",
             client_id="test_client",
@@ -51,6 +56,9 @@ def test_initialization_oauth_url_with_trailing_slash(self, mock_oauth_session):
         mock_session = MagicMock()
         mock_oauth_session.return_value = mock_session
         mock_session.fetch_token.return_value = {"access_token": "test_token"}
+        mock_response = MagicMock()
+        mock_response.status_code = 201
+        mock_session.post.return_value = mock_response
 
         config = AuditLogConfig(
             client_id="test_client",
@@ -60,6 +68,7 @@ def test_initialization_oauth_url_with_trailing_slash(self, mock_oauth_session):
         )
 
         transport = HttpTransport(config)
+        transport.send(SecurityEvent(data="trailing slash test"))
 
         mock_session.fetch_token.assert_called_once_with(
             token_url="https://oauth.example.com/oauth/token",
@@ -80,8 +89,9 @@ def test_initialization_auth_failure(self, mock_oauth_session):
             service_url="https://service.example.com"
         )
 
-        with pytest.raises(AuthenticationError, match="Failed to obtain OAuth2 token"):
-            HttpTransport(config)
+        transport = HttpTransport(config)
+        with pytest.raises(TransportError):
+            transport.send(SecurityEvent(data="auth fail test"))
 
     def test_get_endpoint_security_event(self):
         with patch('sap_cloud_sdk.core.auditlog._http_transport.OAuth2Session') as mock_oauth:
@@ -368,3 +378,88 @@ def test_send_unexpected_error(self, mock_oauth_session):
 
         with pytest.raises(TransportError, match="Unexpected error sending audit event"):
             transport.send(event)
+
+
+class TestHttpTransportRotation:
+
+    @patch('sap_cloud_sdk.core.auditlog._http_transport.OAuth2Session')
+    def test_proactive_rotation_rebuilds_session_when_binding_changed(self, mock_oauth):
+        original_config = AuditLogConfig(
+            client_id="old-client",
+            client_secret="old-secret",
+            oauth_url="https://old-oauth.example.com",
+            service_url="https://service.example.com",
+        )
+        new_config = AuditLogConfig(
+            client_id="new-client",
+            client_secret="new-secret",
+            oauth_url="https://new-oauth.example.com",
+            service_url="https://service.example.com",
+        )
+
+        mock_session = MagicMock()
+        mock_oauth.return_value = mock_session
+        mock_session.fetch_token.return_value = {"access_token": "new-token"}
+        mock_response = MagicMock()
+        mock_response.status_code = 201
+        mock_session.post.return_value = mock_response
+
+        mock_factory = MagicMock(return_value=original_config)
+        mock_factory.has_changed = MagicMock(return_value=True)
+        mock_factory.return_value = new_config
+
+        transport = HttpTransport(mock_factory)
+        transport.send(SecurityEvent(data="rotation test"))
+
+        assert transport.config is new_config
+        # factory called once at init, once on rotation
+        assert mock_factory.call_count == 2
+
+    @patch('sap_cloud_sdk.core.auditlog._http_transport.OAuth2Session')
+    def test_no_rebuild_when_binding_unchanged(self, mock_oauth):
+        config = AuditLogConfig(
+            client_id="client",
+            client_secret="secret",
+            oauth_url="https://oauth.example.com",
+            service_url="https://service.example.com",
+        )
+        mock_session = MagicMock()
+        mock_oauth.return_value = mock_session
+        mock_session.fetch_token.return_value = {"access_token": "tok"}
+        mock_response = MagicMock()
+        mock_response.status_code = 201
+        mock_session.post.return_value = mock_response
+
+        mock_factory = MagicMock(return_value=config)
+        mock_factory.has_changed = MagicMock(return_value=False)
+
+        transport = HttpTransport(mock_factory)
+        transport.send(SecurityEvent(data="no rotation"))
+        oauth_after_first = transport.oauth
+
+        transport.send(SecurityEvent(data="second call"))
+
+        mock_factory.has_changed.assert_called()
+        assert mock_factory.call_count == 1  # no extra factory call
+        assert transport.oauth is oauth_after_first  # same session
+
+    @patch('sap_cloud_sdk.core.auditlog._http_transport.OAuth2Session')
+    def test_static_config_skips_rotation_check(self, mock_oauth):
+        config = AuditLogConfig(
+            client_id="client",
+            client_secret="secret",
+            oauth_url="https://oauth.example.com",
+            service_url="https://service.example.com",
+        )
+        mock_session = MagicMock()
+        mock_oauth.return_value = mock_session
+        mock_session.fetch_token.return_value = {"access_token": "tok"}
+        mock_response = MagicMock()
+        mock_response.status_code = 201
+        mock_session.post.return_value = mock_response
+
+        transport = HttpTransport(config)
+        transport.send(SecurityEvent(data="static config"))
+
+        # No has_changed on plain AuditLogConfig — no error, session created once
+        assert transport.oauth is mock_session
diff --git a/tests/dms/integration/conftest.py b/tests/dms/integration/conftest.py
index 09efd78b..11bfccaf 100644
--- a/tests/dms/integration/conftest.py
+++ b/tests/dms/integration/conftest.py
@@ -52,17 +52,22 @@ def _setup_test_repositories(dms_client):
         )
         created_repos.append(repo.id)
     except DMSError as e:
+        for repo_id in created_repos:
+            try:
+                dms_client.delete_repository(repo_id)
+            except Exception:
+                logger.warning("Failed to clean up partially-created repository %s", repo_id)
         pytest.skip(f"DMS ECM repository connection not available — skipping DMS integration tests: {e}")
 
-    yield
-
-    # Cleanup: delete repositories we created
-    for repo_id in created_repos:
-        try:
-            dms_client.delete_repository(repo_id)
-            logger.info("Cleaned up test repository %s", repo_id)
-        except Exception as e:
-            logger.warning("Failed to clean up test repository %s: %s", repo_id, e)
+    try:
+        yield
+    finally:
+        for repo_id in created_repos:
+            try:
+                dms_client.delete_repository(repo_id)
+                logger.info("Cleaned up test repository %s", repo_id)
+            except Exception as e:
+                logger.warning("Failed to clean up test repository %s: %s", repo_id, e)
 
 
 def _setup_cloud_mode():
diff --git a/tests/dms/unit/test_auth.py b/tests/dms/unit/test_auth.py
deleted file mode 100644
index 49b6becb..00000000
--- a/tests/dms/unit/test_auth.py
+++ /dev/null
@@ -1,129 +0,0 @@
-"""Unit tests for sap_cloud_sdk.dms._auth.Auth."""
-
-import pytest
-from unittest.mock import patch
-
-from sap_cloud_sdk.dms._auth import Auth, _MAX_CACHE_SIZE
-from sap_cloud_sdk.dms.model import DMSCredentials
-
-
-def _make_credentials(identityzone: str = "provider-zone") -> DMSCredentials:
-    return DMSCredentials(
-        uri="https://dms.example.com",
-        token_url=f"https://{identityzone}.authentication.region",
-        client_id="cid",
-        client_secret="csecret",
-        identityzone=identityzone,
-    )
-
-
-class TestResolveTokenUrl:
-    def test_no_subdomain_returns_provider_url(self):
-        creds = _make_credentials()
-        auth = Auth(creds)
-        assert auth._resolve_token_url(None) == creds.token_url
-        assert auth._resolve_token_url("") == creds.token_url
-
-    def test_valid_subdomain_replaces_identityzone(self):
-        creds = _make_credentials()
-        auth = Auth(creds)
-        result = auth._resolve_token_url("tenant-123")
-        assert result == "https://tenant-123.authentication.region"
-
-    def test_invalid_subdomain_raises_value_error(self):
-        creds = _make_credentials()
-        auth = Auth(creds)
-        with pytest.raises(ValueError, match="Invalid tenant_subdomain"):
-            auth._resolve_token_url("-bad")
-        with pytest.raises(ValueError, match="Invalid tenant_subdomain"):
-            auth._resolve_token_url("has.dot")
-
-    @patch("sap_cloud_sdk.dms._auth._validate_tenant_subdomain")
-    def test_resolve_token_url_calls_validator(self, mock_validate):
-        creds = _make_credentials()
-        auth = Auth(creds)
-
-        auth._resolve_token_url("tenant-abc")
-        mock_validate.assert_called_once_with("tenant-abc")
-
-    @patch("sap_cloud_sdk.dms._auth._validate_tenant_subdomain")
-    def test_validator_not_called_when_no_subdomain(self, mock_validate):
-        creds = _make_credentials()
-        auth = Auth(creds)
-
-        auth._resolve_token_url(None)
-        auth._resolve_token_url("")
-        mock_validate.assert_not_called()
-
-
-class TestGetToken:
-    def test_returns_token(self):
-        creds = _make_credentials()
-        auth = Auth(creds)
-        with patch.object(
-            auth,
-            "_fetch_token",
-            return_value={"access_token": "tok-1", "expires_in": 3600},
-        ):
-            assert auth.get_token() == "tok-1"
-
-    def test_caches_token_on_second_call(self):
-        creds = _make_credentials()
-        auth = Auth(creds)
-        with patch.object(
-            auth,
-            "_fetch_token",
-            return_value={"access_token": "tok-1", "expires_in": 3600},
-        ) as mock_fetch:
-            auth.get_token()
-            auth.get_token()
-            mock_fetch.assert_called_once()
-
-    def test_subscriber_and_provider_cached_separately(self):
-        creds = _make_credentials()
-        auth = Auth(creds)
-        with patch.object(
-            auth,
-            "_fetch_token",
-            side_effect=[
-                {"access_token": "prov-tok", "expires_in": 3600},
-                {"access_token": "sub-tok", "expires_in": 3600},
-            ],
-        ) as mock_fetch:
-            prov = auth.get_token()
-            sub = auth.get_token(tenant_subdomain="tenant-x")
-            assert prov == "prov-tok"
-            assert sub == "sub-tok"
-            assert mock_fetch.call_count == 2
-
-    def test_invalid_subdomain_raises_before_fetch(self):
-        creds = _make_credentials()
-        auth = Auth(creds)
-        with patch.object(
-            auth,
-            "_fetch_token",
-            return_value={"access_token": "tok", "expires_in": 3600},
-        ) as mock_fetch:
-            with pytest.raises(ValueError, match="Invalid tenant_subdomain"):
-                auth.get_token(tenant_subdomain="-invalid")
-            mock_fetch.assert_not_called()
-
-    def test_cache_evicts_oldest_when_full(self):
-        creds = _make_credentials()
-        auth = Auth(creds)
-        side_effects = [
-            {"access_token": f"tok-{i}", "expires_in": 3600}
-            for i in range(_MAX_CACHE_SIZE + 1)
-        ]
-        with patch.object(auth, "_fetch_token", side_effect=side_effects):
-            for i in range(_MAX_CACHE_SIZE):
-                auth.get_token(tenant_subdomain=f"tenant-{i:02d}")
-
-            assert len(auth._cache) == _MAX_CACHE_SIZE
-            assert "tenant-00" in auth._cache
-
-            # One more entry should evict the oldest (tenant-00)
-            auth.get_token(tenant_subdomain="tenant-99")
-            assert len(auth._cache) == _MAX_CACHE_SIZE
-            assert "tenant-00" not in auth._cache
-            assert "tenant-99" in auth._cache
diff --git a/tests/dms/unit/test_client_admin.py b/tests/dms/unit/test_client_admin.py
index fcdb8009..5a74b576 100644
--- a/tests/dms/unit/test_client_admin.py
+++ b/tests/dms/unit/test_client_admin.py
@@ -67,7 +67,7 @@ def _mock_response(data, status_code=200):
 
 @pytest.fixture
 def client():
-    with patch("sap_cloud_sdk.dms.client.Auth"):
+    with patch("sap_cloud_sdk.dms.client.XsuaaAuthProvider"):
         with patch("sap_cloud_sdk.dms.client.HttpInvoker") as MockHttp:
             mock_http = Mock()
             MockHttp.return_value = mock_http
diff --git a/tests/dms/unit/test_client_cmis.py b/tests/dms/unit/test_client_cmis.py
index 80a08931..d93687d1 100644
--- a/tests/dms/unit/test_client_cmis.py
+++ b/tests/dms/unit/test_client_cmis.py
@@ -107,7 +107,7 @@ def client():
         identityzone="test-zone",
     )
     with (
-        patch("sap_cloud_sdk.dms.client.Auth"),
+        patch("sap_cloud_sdk.dms.client.XsuaaAuthProvider"),
         patch("sap_cloud_sdk.dms.client.HttpInvoker") as mock_http_cls,
     ):
         mock_http = Mock()
diff --git a/tests/dms/unit/test_http_invoker.py b/tests/dms/unit/test_http_invoker.py
index 96203001..38c71edc 100644
--- a/tests/dms/unit/test_http_invoker.py
+++ b/tests/dms/unit/test_http_invoker.py
@@ -1,11 +1,12 @@
-"""Unit tests for HttpInvoker (get, post_form, get_stream, header methods)."""
+"""Unit tests for HttpInvoker (get, post, put, delete, post_form, get_stream)."""
 
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, patch, MagicMock
 
 import pytest
 import requests
 
 from sap_cloud_sdk.dms._http import HttpInvoker
+from sap_cloud_sdk.core.protocol.http import HttpMethod, XsuaaAuthProvider
 from sap_cloud_sdk.dms.exceptions import (
     DMSConflictException,
     DMSConnectionError,
@@ -16,47 +17,37 @@
 )
 
 
-@pytest.fixture
-def mock_auth():
-    auth = Mock()
-    auth.get_token.return_value = "test-token-123"
-    return auth
+def _make_response(status_code=200, json_data=None, text=""):
+    resp = Mock()
+    resp.status_code = status_code
+    resp.text = text
+    if json_data is not None:
+        resp.json.return_value = json_data
+    else:
+        resp.json.side_effect = ValueError("No JSON")
+    return resp
 
 
 @pytest.fixture
-def invoker(mock_auth):
-    return HttpInvoker(
-        auth=mock_auth,
-        base_url="https://api.example.com",
-        connect_timeout=5,
-        read_timeout=15,
-    )
-
-
-# ---------------------------------------------------------------
-# Header helpers
-# ---------------------------------------------------------------
-
+def mock_auth_provider():
+    return Mock(spec=XsuaaAuthProvider)
 
-class TestHeaders:
-    def test_auth_header(self, invoker):
-        headers = invoker._auth_header()
-        assert headers == {"Authorization": "Bearer test-token-123"}
 
-    def test_auth_header_with_tenant(self, invoker, mock_auth):
-        invoker._auth_header("tenant-sub")
-        mock_auth.get_token.assert_called_with("tenant-sub")
+@pytest.fixture
+def mock_http_client():
+    return Mock()
 
-    def test_default_headers(self, invoker):
-        headers = invoker._default_headers()
-        assert headers["Authorization"] == "Bearer test-token-123"
-        assert headers["Content-Type"] == "application/json"
-        assert headers["Accept"] == "application/json"
 
-    def test_merged_headers_applies_overrides(self, invoker):
-        merged = invoker._merged_headers(None, {"Accept": "text/xml"})
-        assert merged["Accept"] == "text/xml"
-        assert merged["Authorization"] == "Bearer test-token-123"
+@pytest.fixture
+def invoker(mock_auth_provider, mock_http_client):
+    with patch("sap_cloud_sdk.dms._http.HttpClient", return_value=mock_http_client):
+        inv = HttpInvoker(
+            auth_provider=mock_auth_provider,
+            base_url="https://api.example.com",
+            connect_timeout=5,
+            read_timeout=15,
+        )
+    return inv, mock_http_client
 
 
 # ---------------------------------------------------------------
@@ -65,125 +56,81 @@ def test_merged_headers_applies_overrides(self, invoker):
 
 
 class TestGet:
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_basic(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 200
-        mock_resp.content = b'{"key": "val"}'
-        mock_resp.json.return_value = {"key": "val"}
-        mock_get.return_value = mock_resp
-
-        result = invoker.get("/rest/v2/repos")
-
-        mock_get.assert_called_once_with(
-            "https://api.example.com/rest/v2/repos",
-            headers={
-                "Authorization": "Bearer test-token-123",
-                "Content-Type": "application/json",
-                "Accept": "application/json",
-            },
-            params=None,
-            timeout=(5, 15),
-        )
-        assert result is mock_resp
+    def test_get_basic(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(200, {"key": "val"})
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_with_params(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 200
-        mock_get.return_value = mock_resp
+        result = inv.get("/rest/v2/repos")
 
-        result = invoker.get("/path", params={"objectId": "abc", "cmisselector": "acl"})
+        http.request.assert_called_once()
+        call_args = http.request.call_args
+        assert call_args[0][0] == HttpMethod.GET
+        assert call_args[0][1] == "/rest/v2/repos"
+        assert result.status_code == 200
 
-        call_kwargs = mock_get.call_args[1]
-        assert call_kwargs["params"] == {"objectId": "abc", "cmisselector": "acl"}
-        assert result is mock_resp
+    def test_get_with_params(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(200)
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_with_custom_headers(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 200
-        mock_get.return_value = mock_resp
-
-        invoker.get("/repos", headers={"Accept": "application/vnd.sap.sdm+json"})
+        inv.get("/path", params={"objectId": "abc", "cmisselector": "acl"})
 
-        call_kwargs = mock_get.call_args[1]
-        # Custom Accept should override default
-        assert call_kwargs["headers"]["Accept"] == "application/vnd.sap.sdm+json"
-        # Auth should still be present
-        assert call_kwargs["headers"]["Authorization"] == "Bearer test-token-123"
+        call_kwargs = http.request.call_args[1]
+        assert call_kwargs["params"] == {"objectId": "abc", "cmisselector": "acl"}
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_with_tenant(self, mock_get, invoker, mock_auth):
-        mock_resp = Mock()
-        mock_resp.status_code = 200
-        mock_get.return_value = mock_resp
+    def test_get_with_tenant(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(200)
 
-        invoker.get("/path", tenant_subdomain="sub1")
+        inv.get("/path", tenant_subdomain="sub1")
 
-        mock_auth.get_token.assert_called_with("sub1")
+        call_kwargs = http.request.call_args[1]
+        assert call_kwargs["tenant_subdomain"] == "sub1"
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_404_raises_not_found(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 404
-        mock_resp.text = "Not Found"
-        mock_resp.json.side_effect = ValueError("No JSON")
-        mock_get.return_value = mock_resp
+    def test_get_404_raises_not_found(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(404, text="Not Found")
 
         with pytest.raises(DMSObjectNotFoundException) as exc_info:
-            invoker.get("/missing")
+            inv.get("/missing")
         assert exc_info.value.status_code == 404
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_400_raises_invalid_argument(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 400
-        mock_resp.text = "Bad Request"
-        mock_resp.json.side_effect = ValueError("No JSON")
-        mock_get.return_value = mock_resp
+    def test_get_400_raises_invalid_argument(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(400, text="Bad Request")
 
         with pytest.raises(DMSInvalidArgumentException) as exc_info:
-            invoker.get("/bad")
+            inv.get("/bad")
         assert exc_info.value.status_code == 400
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_401_raises_permission_denied(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 401
-        mock_resp.text = "Unauthorized"
-        mock_resp.json.side_effect = ValueError("No JSON")
-        mock_get.return_value = mock_resp
+    def test_get_401_raises_permission_denied(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(401, text="Unauthorized")
 
         with pytest.raises(DMSPermissionDeniedException) as exc_info:
-            invoker.get("/unauthorized")
+            inv.get("/unauthorized")
         assert exc_info.value.status_code == 401
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_500_raises_runtime(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 500
-        mock_resp.text = "Internal Server Error"
-        mock_resp.json.side_effect = ValueError("No JSON")
-        mock_get.return_value = mock_resp
+    def test_get_500_raises_runtime(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(500, text="Internal Server Error")
 
         with pytest.raises(DMSRuntimeException) as exc_info:
-            invoker.get("/error")
+            inv.get("/error")
         assert exc_info.value.status_code == 500
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_connection_error(self, mock_get, invoker):
-        mock_get.side_effect = requests.exceptions.ConnectionError("refused")
+    def test_get_connection_error(self, invoker):
+        inv, http = invoker
+        http.request.side_effect = requests.exceptions.ConnectionError("refused")
 
         with pytest.raises(DMSConnectionError):
-            invoker.get("/unreachable")
+            inv.get("/unreachable")
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_get_timeout_error(self, mock_get, invoker):
-        mock_get.side_effect = requests.exceptions.Timeout("timed out")
+    def test_get_timeout_error(self, invoker):
+        inv, http = invoker
+        http.request.side_effect = requests.exceptions.Timeout("timed out")
 
         with pytest.raises(DMSConnectionError):
-            invoker.get("/slow")
+            inv.get("/slow")
 
 
 # ---------------------------------------------------------------
@@ -192,73 +139,57 @@ def test_get_timeout_error(self, mock_get, invoker):
 
 
 class TestErrorMessageExtraction:
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_400_extracts_json_message(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 400
-        mock_resp.text = '{"exception": "versioning", "message": "The object is not the latest version"}'
-        mock_resp.json.return_value = {
-            "exception": "versioning",
-            "message": "The object is not the latest version",
-        }
-        mock_get.return_value = mock_resp
+    def test_400_extracts_json_message(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(
+            400,
+            json_data={"exception": "versioning", "message": "The object is not the latest version"},
+            text='{"message": "The object is not the latest version"}',
+        )
 
         with pytest.raises(DMSInvalidArgumentException) as exc_info:
-            invoker.get("/bad")
+            inv.get("/bad")
         assert "The object is not the latest version" in str(exc_info.value)
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_400_fallback_when_no_json(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 400
-        mock_resp.text = "Bad Request"
-        mock_resp.json.side_effect = ValueError("No JSON")
-        mock_get.return_value = mock_resp
+    def test_400_fallback_when_no_json(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(400, text="Bad Request")
 
         with pytest.raises(DMSInvalidArgumentException) as exc_info:
-            invoker.get("/bad")
-        assert "Request contains invalid or disallowed parameters" in str(
-            exc_info.value
+            inv.get("/bad")
+        assert "Request contains invalid or disallowed parameters" in str(exc_info.value)
+
+    def test_404_extracts_json_message(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(
+            404,
+            json_data={"message": "Document abc-123 not found"},
+            text='{"message": "Document abc-123 not found"}',
         )
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_404_extracts_json_message(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 404
-        mock_resp.text = '{"message": "Document abc-123 not found"}'
-        mock_resp.json.return_value = {"message": "Document abc-123 not found"}
-        mock_get.return_value = mock_resp
-
         with pytest.raises(DMSObjectNotFoundException) as exc_info:
-            invoker.get("/missing")
+            inv.get("/missing")
         assert "Document abc-123 not found" in str(exc_info.value)
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_409_raises_conflict(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 409
-        mock_resp.text = '{"exception": "versioning", "message": "Object already exists with name test.txt"}'
-        mock_resp.json.return_value = {
-            "exception": "versioning",
-            "message": "Object already exists with name test.txt",
-        }
-        mock_get.return_value = mock_resp
+    def test_409_raises_conflict(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(
+            409,
+            json_data={"message": "Object already exists with name test.txt"},
+            text='{"message": "Object already exists with name test.txt"}',
+        )
 
         with pytest.raises(DMSConflictException) as exc_info:
-            invoker.get("/conflict")
+            inv.get("/conflict")
         assert exc_info.value.status_code == 409
         assert "Object already exists with name test.txt" in str(exc_info.value)
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_409_fallback_when_no_json(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 409
-        mock_resp.text = "Conflict"
-        mock_resp.json.side_effect = ValueError("No JSON")
-        mock_get.return_value = mock_resp
+    def test_409_fallback_when_no_json(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(409, text="Conflict")
 
         with pytest.raises(DMSConflictException) as exc_info:
-            invoker.get("/conflict")
+            inv.get("/conflict")
         assert "conflicts with the current state" in str(exc_info.value)
 
 
@@ -268,96 +199,65 @@ def test_409_fallback_when_no_json(self, mock_get, invoker):
 
 
 class TestPostForm:
-    @patch("sap_cloud_sdk.dms._http.requests.post")
-    def test_post_form_basic(self, mock_post, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 201
-        mock_resp.content = b'{"succinctProperties": {}}'
-        mock_resp.json.return_value = {"succinctProperties": {}}
-        mock_post.return_value = mock_resp
+    def test_post_form_basic(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(201, {"succinctProperties": {}})
 
         form = {"cmisaction": "createFolder", "objectId": "root-id"}
-        result = invoker.post_form("/browser/repo1/root", data=form)
-
-        mock_post.assert_called_once_with(
-            "https://api.example.com/browser/repo1/root",
-            headers={"Authorization": "Bearer test-token-123"},
-            data=form,
-            files=None,
-            timeout=(5, 15),
-        )
-        assert result is mock_resp
+        result = inv.post_form("/browser/repo1/root", data=form)
+
+        http.request.assert_called_once()
+        call_args = http.request.call_args
+        assert call_args[0][0] == HttpMethod.POST
+        assert call_args[0][1] == "/browser/repo1/root"
+        call_kwargs = http.request.call_args[1]
+        assert call_kwargs["data"] == form
+        assert result.status_code == 201
 
-    @patch("sap_cloud_sdk.dms._http.requests.post")
-    def test_post_form_no_content_type_header(self, mock_post, invoker):
+    def test_post_form_no_content_type_header(self, invoker):
         """post_form must NOT set Content-Type — let requests handle it."""
-        mock_resp = Mock()
-        mock_resp.status_code = 201
-        mock_post.return_value = mock_resp
+        inv, http = invoker
+        http.request.return_value = _make_response(201)
 
-        invoker.post_form("/path", data={"key": "val"})
+        inv.post_form("/path", data={"key": "val"})
 
-        headers_sent = mock_post.call_args[1]["headers"]
+        headers_sent = http.request.call_args[1]["headers"]
         assert "Content-Type" not in headers_sent
 
-    @patch("sap_cloud_sdk.dms._http.requests.post")
-    def test_post_form_with_files(self, mock_post, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 201
-        mock_post.return_value = mock_resp
+    def test_post_form_with_files(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(201)
 
         files = {"media": ("test.pdf", b"content", "application/pdf")}
-        invoker.post_form("/path", data={"cmisaction": "createDocument"}, files=files)
+        inv.post_form("/path", data={"cmisaction": "createDocument"}, files=files)
 
-        call_kwargs = mock_post.call_args[1]
+        call_kwargs = http.request.call_args[1]
         assert call_kwargs["files"] == files
         assert call_kwargs["data"] == {"cmisaction": "createDocument"}
 
-    @patch("sap_cloud_sdk.dms._http.requests.post")
-    def test_post_form_with_tenant(self, mock_post, invoker, mock_auth):
-        mock_resp = Mock()
-        mock_resp.status_code = 201
-        mock_post.return_value = mock_resp
+    def test_post_form_with_tenant(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(201)
 
-        invoker.post_form("/path", data={"a": "b"}, tenant_subdomain="tenant-x")
+        inv.post_form("/path", data={"a": "b"}, tenant_subdomain="tenant-x")
 
-        mock_auth.get_token.assert_called_with("tenant-x")
+        call_kwargs = http.request.call_args[1]
+        assert call_kwargs["tenant_subdomain"] == "tenant-x"
 
-    @patch("sap_cloud_sdk.dms._http.requests.post")
-    def test_post_form_500_raises_runtime(self, mock_post, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 500
-        mock_resp.text = "Internal Server Error"
-        mock_resp.json.side_effect = ValueError("No JSON")
-        mock_post.return_value = mock_resp
+    def test_post_form_500_raises_runtime(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(500, text="Internal Server Error")
 
         with pytest.raises(DMSRuntimeException) as exc_info:
-            invoker.post_form("/path", data={})
+            inv.post_form("/path", data={})
         assert exc_info.value.status_code == 500
 
-    @patch("sap_cloud_sdk.dms._http.requests.post")
-    def test_post_form_204_returns_response(self, mock_post, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 204
-        mock_resp.content = b""
-        mock_post.return_value = mock_resp
-
-        result = invoker.post_form("/path", data={})
-        assert result is mock_resp
-
-
-# ---------------------------------------------------------------
-# Base URL stripping
-# ---------------------------------------------------------------
-
+    def test_post_form_204_returns_response(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(204)
 
-class TestBaseUrl:
-    def test_trailing_slash_stripped(self, mock_auth):
-        inv = HttpInvoker(
-            auth=mock_auth,
-            base_url="https://api.example.com/",
-        )
-        assert inv._base_url == "https://api.example.com"
+        result = inv.post_form("/path", data={})
+        assert result.status_code == 204
 
 
 # ---------------------------------------------------------------
@@ -366,47 +266,38 @@ def test_trailing_slash_stripped(self, mock_auth):
 
 
 class TestGetStream:
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_returns_raw_response(self, mock_get, invoker):
+    def test_returns_raw_response(self, invoker):
+        inv, http = invoker
         mock_resp = Mock()
         mock_resp.status_code = 200
         mock_resp.content = b"binary content"
-        mock_get.return_value = mock_resp
+        http.request.return_value = mock_resp
 
-        result = invoker.get_stream(
+        result = inv.get_stream(
             "/browser/repo1/root", params={"objectId": "d1", "cmisselector": "content"}
         )
 
         assert result is mock_resp
-        mock_get.assert_called_once()
-        call_kwargs = mock_get.call_args
-        assert call_kwargs[1]["stream"] is True
-        assert call_kwargs[1]["params"] == {"objectId": "d1", "cmisselector": "content"}
+        http.request.assert_called_once()
+        call_kwargs = http.request.call_args[1]
+        assert call_kwargs["stream"] is True
+        assert call_kwargs["params"] == {"objectId": "d1", "cmisselector": "content"}
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_raises_on_error(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 404
-        mock_resp.text = "Not found"
-        mock_resp.json.side_effect = ValueError("No JSON")
-        mock_get.return_value = mock_resp
+    def test_raises_on_error(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(404, text="Not found")
 
         with pytest.raises(DMSObjectNotFoundException) as exc_info:
-            invoker.get_stream(
+            inv.get_stream(
                 "/browser/repo1/root",
                 params={"objectId": "d1", "cmisselector": "content"},
             )
         assert exc_info.value.status_code == 404
 
-    @patch("sap_cloud_sdk.dms._http.requests.get")
-    def test_uses_auth_headers(self, mock_get, invoker):
-        mock_resp = Mock()
-        mock_resp.status_code = 200
-        mock_resp.content = b"data"
-        mock_get.return_value = mock_resp
+    def test_passes_tenant_subdomain(self, invoker):
+        inv, http = invoker
+        http.request.return_value = _make_response(200)
 
-        invoker.get_stream("/path")
+        inv.get_stream("/path", tenant_subdomain="sub1")
 
-        headers = mock_get.call_args[1]["headers"]
-        assert "Authorization" in headers
-        assert headers["Authorization"] == "Bearer test-token-123"
+        assert http.request.call_args[1]["tenant_subdomain"] == "sub1"
diff --git a/tests/objectstore/integration/conftest.py b/tests/objectstore/integration/conftest.py
index d9971227..bb4b2ba8 100644
--- a/tests/objectstore/integration/conftest.py
+++ b/tests/objectstore/integration/conftest.py
@@ -48,7 +48,9 @@ def integration_env() -> Dict[str, str]:
             missing_vars.append(var)
 
     if missing_vars:
-        pytest.skip(f"Missing required environment variables for cloud integration tests: {missing_vars}")
+        pytest.skip(
+            f"Missing required environment variables for cloud integration tests: {missing_vars}"
+        )
 
     # Ensure SSL is enabled for cloud services
     env_vars["CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_SSL_ENABLED"] = os.getenv(
@@ -66,19 +68,17 @@ def integration_env() -> Dict[str, str]:
 
 @pytest.fixture(scope="session")
 def objectstore_client(integration_env):
-    """Create an ObjectStore client for cloud testing using explicit configuration."""
+    """Create an ObjectStore client via ConfigFactory (reads CLOUD_SDK_CFG_* env vars)."""
     try:
-        config = ObjectStoreBindingData(
-            host=integration_env["CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_HOST"],
-            access_key_id=integration_env["CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_ACCESS_KEY_ID"],
-            secret_access_key=integration_env["CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_SECRET_ACCESS_KEY"],
-            bucket=integration_env["CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_BUCKET"],
-        )
-        disable_ssl = integration_env.get("CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_SSL_ENABLED", "true").lower() in ("false", "0")
-        client = create_client("default", config=config, disable_ssl=disable_ssl)
+        disable_ssl = integration_env.get(
+            "CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_SSL_ENABLED", "true"
+        ).lower() in ("false", "0")
+        client = create_client("default", disable_ssl=disable_ssl)
         return client
     except Exception as e:
-        pytest.fail(f"Failed to create ObjectStore client for cloud integration tests: {e}")
+        pytest.fail(
+            f"Failed to create ObjectStore client for cloud integration tests: {e}"
+        )
 
 
 @pytest.fixture
@@ -89,6 +89,7 @@ def test_prefix() -> str:
 
 # ===== CLEANUP INFRASTRUCTURE =====
 
+
 def cleanup_by_prefix(client, prefix: str, timeout: float = 10.0) -> bool:
     """Timeout-controlled cleanup with eventual consistency handling."""
     start_time = time.time()
@@ -103,7 +104,9 @@ def cleanup_by_prefix(client, prefix: str, timeout: float = 10.0) -> bool:
 
             # Check timeout
             if time.time() - start_time > timeout:
-                logger.warning(f"Cleanup timeout reached after {timeout}s, cleaned {cleaned_count} objects")
+                logger.warning(
+                    f"Cleanup timeout reached after {timeout}s, cleaned {cleaned_count} objects"
+                )
                 break
 
         if cleaned_count > 0:
@@ -126,8 +129,12 @@ def cleanup_all_test_objects():
         try:
             objects = objectstore_client.list_objects("sdk-python-integration-tests/")
             if objects:
-                logger.info(f"Found {len(objects)} leftover integration test objects, cleaning up...")
-                cleanup_by_prefix(objectstore_client, "sdk-python-integration-tests/", timeout=30.0)
+                logger.info(
+                    f"Found {len(objects)} leftover integration test objects, cleaning up..."
+                )
+                cleanup_by_prefix(
+                    objectstore_client, "sdk-python-integration-tests/", timeout=30.0
+                )
                 logger.info("Session cleanup completed")
         except Exception as e:
             logger.warning(f"Session cleanup failed: {e}")
@@ -166,7 +173,9 @@ def register_object(object_name: str):
 
                 # Respect timeout
                 if time.time() - start_time > 10.0:
-                    logger.warning(f"Object cleanup timeout reached, cleaned {cleaned_count}/{len(created_objects)} objects")
+                    logger.warning(
+                        f"Object cleanup timeout reached, cleaned {cleaned_count}/{len(created_objects)} objects"
+                    )
                     break
 
             except Exception as e:
@@ -182,11 +191,17 @@ def failure_simulation(integration_env):
     """Utilities for simulating various failure conditions using explicit configuration."""
     base_config = ObjectStoreBindingData(
         host=integration_env["CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_HOST"],
-        access_key_id=integration_env["CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_ACCESS_KEY_ID"],
-        secret_access_key=integration_env["CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_SECRET_ACCESS_KEY"],
+        access_key_id=integration_env[
+            "CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_ACCESS_KEY_ID"
+        ],
+        secret_access_key=integration_env[
+            "CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_SECRET_ACCESS_KEY"
+        ],
         bucket=integration_env["CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_BUCKET"],
     )
-    disable_ssl = integration_env.get("CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_SSL_ENABLED", "true").lower() in ("false", "0")
+    disable_ssl = integration_env.get(
+        "CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_SSL_ENABLED", "true"
+    ).lower() in ("false", "0")
 
     class FailureSimulator:
         def create_client_with_network_failure(self):
@@ -219,10 +234,7 @@ def setup_intermittent_failure(self):
 # Configure pytest markers for integration tests
 def pytest_configure(config):
     """Configure pytest markers."""
-    config.addinivalue_line(
-        "markers",
-        "integration: mark test as integration test"
-    )
+    config.addinivalue_line("markers", "integration: mark test as integration test")
 
 
 def pytest_collection_modifyitems(config, items):
diff --git a/tests/objectstore/unit/test_create_client.py b/tests/objectstore/unit/test_create_client.py
index 979abbc5..264ee321 100644
--- a/tests/objectstore/unit/test_create_client.py
+++ b/tests/objectstore/unit/test_create_client.py
@@ -1,55 +1,84 @@
 """Tests for create_client factory function."""
 
-from unittest.mock import Mock, patch
+from unittest.mock import MagicMock, Mock, patch
 import pytest
 
 from sap_cloud_sdk.objectstore import create_client
 from sap_cloud_sdk.objectstore._models import ObjectStoreBindingData
 
 
+def _make_factory(creds: ObjectStoreBindingData) -> Mock:
+    """Return a mock config factory that yields *creds* and has ``has_changed()``."""
+    factory = Mock(return_value=creds)
+    factory.has_changed = Mock(return_value=False)
+    return factory
+
+
 class TestCreateClient:
 
-    @patch('sap_cloud_sdk.objectstore.read_from_mount_and_fallback_to_env_var')
-    @patch('sap_cloud_sdk.objectstore.ObjectStoreClient')
-    def test_create_client_cloud_mode(self, mock_client_class, mock_resolver):
+    @patch("sap_cloud_sdk.core.secret_resolver.ConfigFactory")
+    @patch("sap_cloud_sdk.objectstore.ObjectStoreClient")
+    def test_create_client_cloud_mode(self, mock_client_class, mock_factory_class):
+        mock_creds = ObjectStoreBindingData(
+            access_key_id="k", secret_access_key="s", bucket="b", host="h"
+        )
+        mock_factory = _make_factory(mock_creds)
+        mock_factory_class.return_value = mock_factory
+
         mock_client = Mock()
         mock_client_class.return_value = mock_client
 
         result = create_client("production", disable_ssl=True)
 
-        mock_resolver.assert_called_once()
-        call_args = mock_resolver.call_args
-        assert call_args[1]["module"] == "objectstore"
-        assert call_args[1]["instance"] == "production"
-        assert isinstance(call_args[1]["target"], ObjectStoreBindingData)
-        mock_client_class.assert_called_once_with(call_args[1]["target"], disable_ssl=True)
+        mock_factory_class.assert_called_once_with(
+            module="objectstore",
+            instance="production",
+            binding_cls=ObjectStoreBindingData,
+            extract=mock_factory_class.call_args.kwargs["extract"],
+        )
+        mock_client_class.assert_called_once_with(mock_factory, disable_ssl=True)
         assert result == mock_client
 
     def test_create_client_empty_instance_raises_error(self):
-        """Test that create_client raises ValueError for empty instance."""
         with pytest.raises(ValueError, match="instance parameter must be a non-empty string"):
             create_client("")
 
         with pytest.raises(ValueError, match="instance parameter must be a non-empty string"):
-            create_client("   ")  # whitespace only
+            create_client("   ")
 
         with pytest.raises(ValueError, match="instance parameter must be a non-empty string"):
             create_client(None)  # type: ignore
 
-
-    @patch('sap_cloud_sdk.objectstore.ObjectStoreClient')
+    @patch("sap_cloud_sdk.objectstore.ObjectStoreClient")
     def test_create_client_with_explicit_config(self, mock_client_class):
-        """Test that create_client uses explicit config when provided."""
         mock_config = ObjectStoreBindingData(
             access_key_id="explicit_key",
             secret_access_key="explicit_secret",
             bucket="explicit-bucket",
-            host="explicit.host.com"
+            host="explicit.host.com",
         )
         mock_client = Mock()
         mock_client_class.return_value = mock_client
 
         result = create_client("ignored-instance", config=mock_config, disable_ssl=True)
 
-        mock_client_class.assert_called_once_with(mock_config, disable_ssl=True)
+        # ObjectStoreClient receives a static factory wrapping the explicit config
+        mock_client_class.assert_called_once()
+        call_args = mock_client_class.call_args
+        factory = call_args.args[0]
+        assert call_args.kwargs["disable_ssl"] is True
+        assert factory() is mock_config
         assert result == mock_client
+
+    @patch("sap_cloud_sdk.objectstore.ObjectStoreClient")
+    def test_create_client_explicit_config_no_has_changed(self, mock_client_class):
+        """Static factory for explicit config should not have has_changed (no rotation tracking)."""
+        mock_config = ObjectStoreBindingData(
+            access_key_id="k", secret_access_key="s", bucket="b", host="h"
+        )
+        mock_client_class.return_value = Mock()
+
+        create_client("instance", config=mock_config)
+
+        factory = mock_client_class.call_args.args[0]
+        assert not hasattr(factory, "has_changed")
diff --git a/tests/objectstore/unit/test_s3_client.py b/tests/objectstore/unit/test_s3_client.py
index fec111f1..3976ab5e 100644
--- a/tests/objectstore/unit/test_s3_client.py
+++ b/tests/objectstore/unit/test_s3_client.py
@@ -4,7 +4,7 @@
 import os
 from datetime import datetime
 from http.client import HTTPResponse
-from unittest.mock import Mock, patch, mock_open
+from unittest.mock import Mock, call, patch, mock_open
 import pytest
 from minio.error import S3Error
 
@@ -15,74 +15,163 @@
 )
 
 
-class TestObjectStoreClient:
+def _make_creds(
+    *,
+    access_key_id: str = "test_key",
+    secret_access_key: str = "test_secret",
+    bucket: str = "test-bucket",
+    host: str = "s3.amazonaws.com",
+) -> ObjectStoreBindingData:
+    return ObjectStoreBindingData(
+        access_key_id=access_key_id,
+        secret_access_key=secret_access_key,
+        bucket=bucket,
+        host=host,
+    )
 
-    def setup_method(self):
-        self.creds = ObjectStoreBindingData(
-            access_key_id="test_key",
-            secret_access_key="test_secret",
-            bucket="test-bucket",
-            host="s3.amazonaws.com"
-        )
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
-    def test_client_creation_ssl_enabled(self, mock_minio_class):
-        mock_minio = Mock()
-        mock_minio_class.return_value = mock_minio
+def _make_factory(
+    creds: ObjectStoreBindingData | None = None,
+    *,
+    has_changed: bool = False,
+) -> Mock:
+    """Return a mock config factory with controllable ``has_changed()``."""
+    if creds is None:
+        creds = _make_creds()
+    factory = Mock(return_value=creds)
+    factory.has_changed = Mock(return_value=has_changed)
+    return factory
+
+
+def _make_s3_error(code: str) -> S3Error:
+    return S3Error(code, "error message", "resource", "request-id", "host-id", Mock())
 
-        client = ObjectStoreClient(self.creds, disable_ssl=False)
 
+class TestObjectStoreClient:
+
+    # ------------------------------------------------------------------
+    # Construction
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    def test_client_creation_ssl_enabled(self, mock_minio_class):
+        mock_minio_class.return_value = Mock()
+        client = ObjectStoreClient(_make_factory(), disable_ssl=False)
         mock_minio_class.assert_called_once_with(
             endpoint="s3.amazonaws.com",
             access_key="test_key",
             secret_key="test_secret",
-            secure=True
+            secure=True,
         )
-        assert client._minio_client == mock_minio
+        assert client._minio_client == mock_minio_class.return_value
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_client_creation_ssl_disabled(self, mock_minio_class):
-        mock_minio = Mock()
-        mock_minio_class.return_value = mock_minio
-
-        client = ObjectStoreClient(self.creds, disable_ssl=True)
-
+        mock_minio_class.return_value = Mock()
+        ObjectStoreClient(_make_factory(), disable_ssl=True)
         mock_minio_class.assert_called_once_with(
             endpoint="s3.amazonaws.com",
             access_key="test_key",
             secret_key="test_secret",
-            secure=False
+            secure=False,
         )
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_client_creation_failure(self, mock_minio_class):
         mock_minio_class.side_effect = Exception("Connection failed")
-
         with pytest.raises(ClientCreationError, match="Failed to create MinIO client"):
-            ObjectStoreClient(self.creds)
+            ObjectStoreClient(_make_factory())
+
+    # ------------------------------------------------------------------
+    # Proactive rotation — has_changed() returns True
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    def test_proactive_rotation_rebuilds_minio_client(self, mock_minio_class):
+        original_minio = Mock()
+        rotated_minio = Mock()
+        mock_minio_class.side_effect = [original_minio, rotated_minio]
+
+        rotated_creds = _make_creds(access_key_id="new_key", secret_access_key="new_secret")
+        # factory() is called once in __init__ (initial creds) and once in _refresh_credentials
+        factory = Mock(side_effect=[_make_creds(), rotated_creds])
+        # has_changed() is NOT called during __init__; it's first called by _refresh_if_rotated
+        # inside put_object_from_bytes. Returning True on the first call triggers the refresh.
+        factory.has_changed = Mock(return_value=True)
+
+        client = ObjectStoreClient(factory)
+        assert client._minio_client is original_minio
+
+        client.put_object_from_bytes("f.txt", b"data", "text/plain")
+
+        assert client._minio_client is rotated_minio
+        assert client._creds_config is rotated_creds
+        rotated_minio.put_object.assert_called_once()
+
+    # ------------------------------------------------------------------
+    # Reactive rotation — credential S3 errors trigger one retry
+    # ------------------------------------------------------------------
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @pytest.mark.parametrize("error_code", ["InvalidAccessKeyId", "SignatureDoesNotMatch"])
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    def test_reactive_rotation_retries_on_credential_error(self, mock_minio_class, error_code):
+        original_minio = Mock()
+        rotated_minio = Mock()
+        mock_minio_class.side_effect = [original_minio, rotated_minio]
+
+        rotated_creds = _make_creds(access_key_id="new_key")
+        factory = Mock(side_effect=[_make_creds(), rotated_creds])
+        factory.has_changed = Mock(return_value=False)
+
+        original_minio.put_object.side_effect = _make_s3_error(error_code)
+        rotated_minio.put_object.return_value = None
+
+        client = ObjectStoreClient(factory)
+        client.put_object_from_bytes("f.txt", b"data", "text/plain")
+
+        assert original_minio.put_object.call_count == 1
+        assert rotated_minio.put_object.call_count == 1
+        assert client._creds_config is rotated_creds
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    def test_non_credential_s3_error_is_not_retried(self, mock_minio_class):
+        mock_minio = Mock()
+        mock_minio_class.return_value = mock_minio
+        mock_minio.put_object.side_effect = _make_s3_error("AccessDenied")
+
+        client = ObjectStoreClient(_make_factory())
+
+        with pytest.raises(ObjectOperationError, match="Failed to upload object"):
+            client.put_object_from_bytes("f.txt", b"data", "text/plain")
+
+        assert mock_minio.put_object.call_count == 1
+
+    # ------------------------------------------------------------------
+    # put_object_from_bytes
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_put_object_from_bytes_success(self, mock_minio_class):
         mock_minio = Mock()
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
         test_data = b"Hello, World!"
 
         client.put_object_from_bytes("test.txt", test_data, "text/plain")
 
         mock_minio.put_object.assert_called_once()
         call_args = mock_minio.put_object.call_args
-        assert call_args.kwargs['bucket_name'] == 'test-bucket'
-        assert call_args.kwargs['object_name'] == 'test.txt'
-        assert call_args.kwargs['length'] == len(test_data)
-        assert call_args.kwargs['content_type'] == 'text/plain'
-        assert isinstance(call_args.kwargs['data'], io.BytesIO)
+        assert call_args.kwargs["bucket_name"] == "test-bucket"
+        assert call_args.kwargs["object_name"] == "test.txt"
+        assert call_args.kwargs["length"] == len(test_data)
+        assert call_args.kwargs["content_type"] == "text/plain"
+        assert isinstance(call_args.kwargs["data"], io.BytesIO)
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_put_object_from_bytes_validation(self, mock_minio_class):
         mock_minio_class.return_value = Mock()
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
 
         with pytest.raises(ValueError, match="name must be a non-empty string"):
             client.put_object_from_bytes("", b"data", "text/plain")
@@ -93,121 +182,151 @@ def test_put_object_from_bytes_validation(self, mock_minio_class):
         with pytest.raises(ValueError, match="content_type must be a non-empty string"):
             client.put_object_from_bytes("test.txt", b"data", "")
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_put_object_from_bytes_s3_error(self, mock_minio_class):
         mock_minio = Mock()
-        s3_error = S3Error("AccessDenied", "Access denied", "test.txt", "123", "456", Mock())
-        mock_minio.put_object.side_effect = s3_error
+        mock_minio.put_object.side_effect = _make_s3_error("AccessDenied")
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
 
         with pytest.raises(ObjectOperationError, match="Failed to upload object"):
             client.put_object_from_bytes("test.txt", b"data", "text/plain")
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    # ------------------------------------------------------------------
+    # put_object (stream)
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_put_object_from_stream_success(self, mock_minio_class):
         mock_minio = Mock()
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
         stream = io.BytesIO(b"stream data")
 
         client.put_object("test.txt", stream, 11, "text/plain")
 
-        # Note: The implementation reads the stream and creates a new BytesIO
         mock_minio.put_object.assert_called_once()
         call_args = mock_minio.put_object.call_args
-        assert call_args.kwargs['bucket_name'] == 'test-bucket'
-        assert call_args.kwargs['object_name'] == 'test.txt'
-        assert call_args.kwargs['length'] == 11
-        assert call_args.kwargs['content_type'] == 'text/plain'
+        assert call_args.kwargs["bucket_name"] == "test-bucket"
+        assert call_args.kwargs["object_name"] == "test.txt"
+        assert call_args.kwargs["length"] == 11
+        assert call_args.kwargs["content_type"] == "text/plain"
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_put_object_validation(self, mock_minio_class):
         mock_minio_class.return_value = Mock()
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
 
         with pytest.raises(ValueError, match="size must be non-negative"):
             client.put_object("test.txt", io.BytesIO(b"data"), -1, "text/plain")
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
-    @patch('builtins.open', new_callable=mock_open, read_data=b"file content")
-    @patch('os.path.isfile', return_value=True)
-    @patch('os.path.getsize', return_value=12)
+    # ------------------------------------------------------------------
+    # put_object_from_file
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    @patch("builtins.open", new_callable=mock_open, read_data=b"file content")
+    @patch("os.path.isfile", return_value=True)
+    @patch("os.path.getsize", return_value=12)
     def test_put_object_from_file_success(self, mock_getsize, mock_isfile, mock_file, mock_minio_class):
         mock_minio = Mock()
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
         client.put_object_from_file("test.txt", "/path/to/file.txt", "text/plain")
 
         mock_isfile.assert_called_once_with("/path/to/file.txt")
         mock_getsize.assert_called_once_with("/path/to/file.txt")
         mock_minio.put_object.assert_called_once()
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
-    @patch('builtins.open', new_callable=mock_open, read_data=b"file content")
-    @patch('os.path.isfile', return_value=False)
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    @patch("builtins.open", new_callable=mock_open, read_data=b"file content")
+    @patch("os.path.isfile", return_value=False)
     def test_put_object_from_file_not_found(self, mock_isfile, mock_file, mock_minio_class):
         mock_minio_class.return_value = Mock()
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
 
         with pytest.raises(ObjectOperationError, match="File not found"):
             client.put_object_from_file("test.txt", "/nonexistent.txt", "text/plain")
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    # ------------------------------------------------------------------
+    # get_object
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_get_object_success(self, mock_minio_class):
         mock_minio = Mock()
         mock_response = Mock(spec=HTTPResponse)
         mock_minio.get_object.return_value = mock_response
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
         result = client.get_object("test.txt")
 
         mock_minio.get_object.assert_called_once_with(
-            bucket_name='test-bucket',
-            object_name='test.txt'
+            bucket_name="test-bucket", object_name="test.txt"
         )
         assert result == mock_response
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_get_object_not_found(self, mock_minio_class):
         mock_minio = Mock()
-        s3_error = S3Error("NoSuchKey", "Key not found", "test.txt", "123", "456", Mock())
-        mock_minio.get_object.side_effect = s3_error
+        mock_minio.get_object.side_effect = _make_s3_error("NoSuchKey")
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
 
         with pytest.raises(ObjectNotFoundError, match="Object 'test.txt' not found"):
             client.get_object("test.txt")
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    def test_get_object_empty_name_validation(self, mock_minio_class):
+        mock_minio_class.return_value = Mock()
+        client = ObjectStoreClient(_make_factory())
+
+        with pytest.raises(ValueError, match="name must be a non-empty string"):
+            client.get_object("")
+
+    # ------------------------------------------------------------------
+    # delete_object
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_delete_object_success(self, mock_minio_class):
         mock_minio = Mock()
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
         client.delete_object("test.txt")
 
         mock_minio.remove_object.assert_called_once_with(
-            bucket_name='test-bucket',
-            object_name='test.txt'
+            bucket_name="test-bucket", object_name="test.txt"
         )
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_delete_object_not_found_ignored(self, mock_minio_class):
         mock_minio = Mock()
-        s3_error = S3Error("NoSuchKey", "Key not found", "test.txt", "123", "456", Mock())
-        mock_minio.remove_object.side_effect = s3_error
+        mock_minio.remove_object.side_effect = _make_s3_error("NoSuchKey")
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
-        client.delete_object("test.txt")
+        client = ObjectStoreClient(_make_factory())
+        client.delete_object("test.txt")  # should not raise
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    def test_delete_object_empty_name_validation(self, mock_minio_class):
+        mock_minio_class.return_value = Mock()
+        client = ObjectStoreClient(_make_factory())
+
+        with pytest.raises(ValueError, match="name must be a non-empty string"):
+            client.delete_object("")
+
+    # ------------------------------------------------------------------
+    # list_objects
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_list_objects_success(self, mock_minio_class):
         mock_minio = Mock()
 
@@ -222,31 +341,40 @@ def test_list_objects_success(self, mock_minio_class):
         mock_minio.list_objects.return_value = [mock_obj1]
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
         result = client.list_objects("prefix/")
 
         mock_minio.list_objects.assert_called_once_with(
-            bucket_name='test-bucket',
-            prefix='prefix/'
+            bucket_name="test-bucket", prefix="prefix/"
         )
-
         assert len(result) == 1
         assert result[0].key == "prefix/file1.txt"
         assert result[0].etag == '"abc123"'
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_list_objects_s3_error(self, mock_minio_class):
         mock_minio = Mock()
-        s3_error = S3Error("AccessDenied", "Access denied", "", "123", "456", Mock())
-        mock_minio.list_objects.side_effect = s3_error
+        mock_minio.list_objects.side_effect = _make_s3_error("AccessDenied")
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
 
         with pytest.raises(ListObjectsError, match="Failed to list objects"):
             client.list_objects("prefix/")
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    def test_list_objects_prefix_validation(self, mock_minio_class):
+        mock_minio_class.return_value = Mock()
+        client = ObjectStoreClient(_make_factory())
+
+        with pytest.raises(ValueError, match="prefix must be a string"):
+            client.list_objects(123)  # ty: ignore[invalid-argument-type]
+
+    # ------------------------------------------------------------------
+    # head_object
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_head_object_success(self, mock_minio_class):
         mock_minio = Mock()
 
@@ -258,89 +386,61 @@ def test_head_object_success(self, mock_minio_class):
         mock_minio.stat_object.return_value = mock_stat
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
         result = client.head_object("test.txt")
 
         mock_minio.stat_object.assert_called_once_with(
-            bucket_name='test-bucket',
-            object_name='test.txt'
+            bucket_name="test-bucket", object_name="test.txt"
         )
-
         assert result.key == "test.txt"
         assert result.etag == "abc123"
         assert result.size == 100
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_head_object_not_found(self, mock_minio_class):
         mock_minio = Mock()
-        s3_error = S3Error("NoSuchKey", "Key not found", "test.txt", "123", "456", Mock())
-        mock_minio.stat_object.side_effect = s3_error
+        mock_minio.stat_object.side_effect = _make_s3_error("NoSuchKey")
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
 
         with pytest.raises(ObjectNotFoundError, match="Object 'test.txt' not found"):
             client.head_object("test.txt")
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
+    def test_head_object_empty_name_validation(self, mock_minio_class):
+        mock_minio_class.return_value = Mock()
+        client = ObjectStoreClient(_make_factory())
+
+        with pytest.raises(ValueError, match="name must be a non-empty string"):
+            client.head_object("")
+
+    # ------------------------------------------------------------------
+    # object_exists
+    # ------------------------------------------------------------------
+
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_object_exists_true(self, mock_minio_class):
         mock_minio = Mock()
         mock_minio.stat_object.return_value = Mock()
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
-        result = client.object_exists("test.txt")
-
-        assert result is True
+        client = ObjectStoreClient(_make_factory())
+        assert client.object_exists("test.txt") is True
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_object_exists_false(self, mock_minio_class):
         mock_minio = Mock()
-        s3_error = S3Error("NoSuchKey", "Key not found", "test.txt", "123", "456", Mock())
-        mock_minio.stat_object.side_effect = s3_error
+        mock_minio.stat_object.side_effect = _make_s3_error("NoSuchKey")
         mock_minio_class.return_value = mock_minio
 
-        client = ObjectStoreClient(self.creds)
-        result = client.object_exists("test.txt")
-
-        assert result is False
+        client = ObjectStoreClient(_make_factory())
+        assert client.object_exists("test.txt") is False
 
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
-    def test_get_object_empty_name_validation(self, mock_minio_class):
-        mock_minio_class.return_value = Mock()
-        client = ObjectStoreClient(self.creds)
-
-        with pytest.raises(ValueError, match="name must be a non-empty string"):
-            client.get_object("")
-
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
-    def test_delete_object_empty_name_validation(self, mock_minio_class):
-        mock_minio_class.return_value = Mock()
-        client = ObjectStoreClient(self.creds)
-
-        with pytest.raises(ValueError, match="name must be a non-empty string"):
-            client.delete_object("")
-
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
-    def test_head_object_empty_name_validation(self, mock_minio_class):
-        mock_minio_class.return_value = Mock()
-        client = ObjectStoreClient(self.creds)
-
-        with pytest.raises(ValueError, match="name must be a non-empty string"):
-            client.head_object("")
-
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
+    @patch("sap_cloud_sdk.objectstore._s3.Minio")
     def test_object_exists_empty_name_validation(self, mock_minio_class):
         mock_minio_class.return_value = Mock()
-        client = ObjectStoreClient(self.creds)
+        client = ObjectStoreClient(_make_factory())
 
         with pytest.raises(ValueError, match="name must be a non-empty string"):
             client.object_exists("")
-
-    @patch('sap_cloud_sdk.objectstore._s3.Minio')
-    def test_list_objects_prefix_validation(self, mock_minio_class):
-        mock_minio_class.return_value = Mock()
-        client = ObjectStoreClient(self.creds)
-
-        with pytest.raises(ValueError, match="prefix must be a string"):
-            client.list_objects(123)  # ty: ignore[invalid-argument-type]
diff --git a/tests/print/unit/test_http.py b/tests/print/unit/test_http.py
index c10c706f..8c0225b0 100644
--- a/tests/print/unit/test_http.py
+++ b/tests/print/unit/test_http.py
@@ -221,3 +221,62 @@ def test_fetch_token_exception_raises_http_error(self, mock_oauth):
         provider = TokenProvider(_config())
         with pytest.raises(HttpError, match="failed to acquire token"):
             provider.get_token()
+
+
+class TestTokenProviderRotation:
+
+    @patch("sap_cloud_sdk.print._http.OAuth2Session")
+    def test_proactive_rotation_rebuilds_session_when_binding_changed(self, mock_oauth):
+        new_config = PrintConfig(
+            url="https://api.eu10.print.services.sap",
+            token_url="https://new-tenant.authentication.eu10.hana.ondemand.com/oauth/token",
+            client_id="new-client-id",
+            client_secret="new-client-secret",
+        )
+        mock_session = MagicMock()
+        mock_oauth.return_value = mock_session
+        mock_session.fetch_token.return_value = {"access_token": "tok-v2"}
+
+        mock_factory = MagicMock(return_value=new_config)
+        mock_factory.has_changed = MagicMock(return_value=True)
+
+        provider = TokenProvider(mock_factory)
+        # has_changed() is True: provider must re-read config before fetching
+        token = provider.get_token()
+
+        assert token == "tok-v2"
+        assert provider._config is new_config
+        # factory called once at init, once on rotation
+        assert mock_factory.call_count == 2
+
+    @patch("sap_cloud_sdk.print._http.OAuth2Session")
+    def test_no_rebuild_when_binding_unchanged(self, mock_oauth):
+        mock_session = MagicMock()
+        mock_oauth.return_value = mock_session
+        mock_session.fetch_token.return_value = {"access_token": "tok-same"}
+
+        mock_factory = MagicMock(return_value=_config())
+        mock_factory.has_changed = MagicMock(return_value=False)
+
+        provider = TokenProvider(mock_factory)
+        init_session = provider._session
+
+        provider.get_token()
+
+        mock_factory.has_changed.assert_called_once()
+        assert provider._session is init_session  # no rebuild
+        assert mock_factory.call_count == 1  # no extra factory call
+
+    @patch("sap_cloud_sdk.print._http.OAuth2Session")
+    def test_static_config_has_no_has_changed_check(self, mock_oauth):
+        mock_session = MagicMock()
+        mock_oauth.return_value = mock_session
+        mock_session.fetch_token.return_value = {"access_token": "tok-static"}
+
+        provider = TokenProvider(_config())
+        init_session = provider._session
+
+        provider.get_token()
+
+        # no has_changed() — session stays the same
+        assert provider._session is init_session
diff --git a/uv.lock b/uv.lock
index 05a51d19..c88ff3e2 100644
--- a/uv.lock
+++ b/uv.lock
@@ -4300,7 +4300,7 @@ wheels = [
 
 [[package]]
 name = "sap-cloud-sdk"
-version = "0.53.3"
+version = "0.54.0"
 source = { editable = "." }
 dependencies = [
     { name = "cryptography" },