diff --git a/pyproject.toml b/pyproject.toml
index fd556aea..620a8f30 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.45.3"
+version = "0.46.0"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"
diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py
index 7fb10094..a0627eb0 100644
--- a/src/sap_cloud_sdk/aicore/__init__.py
+++ b/src/sap_cloud_sdk/aicore/__init__.py
@@ -7,6 +7,7 @@
 import json
 import logging
 import os
+import threading
 from typing import Optional
 
 from sap_cloud_sdk.core.secret_resolver import resolve_base_mount
@@ -175,8 +176,73 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None:
     set_filtering()
 
 
+def _get_secret_dir_mtime(instance_name: str = "aicore-instance") -> float:
+    """Return the mtime of the AI Core secret directory, or 0.0 if it does not exist."""
+    secret_dir = os.path.join(resolve_base_mount(), "aicore", instance_name)
+    try:
+        return os.stat(secret_dir).st_mtime
+    except OSError:
+        return 0.0
+
+
+def watch_aicore_config(
+    instance_name: str = "aicore-instance",
+    interval: float = 60.0,
+    stop_event: threading.Event | None = None,
+) -> threading.Thread:
+    """Start a daemon thread that proactively reloads AI Core credentials
+    when the mounted secret volume changes.
+
+    Polls the secret directory mtime every ``interval`` seconds. On change,
+    calls :func:`set_aicore_config` before LiteLLM's cached OAuth token
+    expires — avoiding 401 errors entirely rather than recovering from them.
+
+    Kubernetes projected volumes perform an atomic symlink swap on rotation,
+    which changes the directory mtime. Both ``secret`` and ``projected``
+    volume types are covered.
+
+    Returns the daemon thread. Stop it cleanly via ``stop_event.set()``.
+
+    Each call starts a new daemon thread — avoid calling more than once per process.
+
+    Typical usage::
+
+        import threading
+        from sap_cloud_sdk.aicore import set_aicore_config, watch_aicore_config
+
+        set_aicore_config()
+
+        _stop = threading.Event()
+        watch_aicore_config(stop_event=_stop)
+        # at shutdown: _stop.set()
+    """
+    if stop_event is None:
+        stop_event = threading.Event()
+
+    last_mtime = _get_secret_dir_mtime(instance_name)
+
+    def _watch() -> None:
+        nonlocal last_mtime
+        while not stop_event.wait(timeout=interval):
+            try:
+                current_mtime = _get_secret_dir_mtime(instance_name)
+                if current_mtime != last_mtime:
+                    logger.info(
+                        "AI Core secret volume changed — proactively reloading credentials"
+                    )
+                    set_aicore_config(instance_name=instance_name)
+                    last_mtime = current_mtime
+            except Exception:
+                logger.exception("Error during proactive AI Core credential reload")
+
+    thread = threading.Thread(target=_watch, daemon=True, name="aicore-secret-watcher")
+    thread.start()
+    return thread
+
+
 __all__ = [
     "set_aicore_config",
+    "watch_aicore_config",
     "set_filtering",
     "disable_filtering",
     "completion",
diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py
index 3c869cfe..a2d72ad7 100644
--- a/src/sap_cloud_sdk/aicore/completion.py
+++ b/src/sap_cloud_sdk/aicore/completion.py
@@ -14,6 +14,16 @@
 re-raising as :class:`ContentFilteredError` so callers can rely on a
 single exception type for "filter blocked you."
 
+Credential rotation handling
+----------------------------
+When a credential (client_secret) is rotated while the pod is running,
+LiteLLM's cached token becomes invalid and the next token refresh attempt
+raises ``litellm.AuthenticationError``. The wrappers intercept this error,
+reload credentials from the mounted secret volume via
+:func:`sap_cloud_sdk.aicore.set_aicore_config`, and retry the call once.
+The caller is unaffected — rotation is transparent. If the retry also
+fails, the ``AuthenticationError`` propagates normally.
+
 Usage::
 
     from sap_cloud_sdk.aicore import completion, ContentFilteredError
@@ -39,12 +49,15 @@
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 
 import litellm
 
 from .filtering.filters import _parse_input_filter_error
 
+logger = logging.getLogger(__name__)
+
 
 def _maybe_translate_filter_error(exc: BaseException) -> BaseException:
     """Return a :class:`ContentFilteredError` if ``exc`` is a wrapped
@@ -60,19 +73,22 @@ def _maybe_translate_filter_error(exc: BaseException) -> BaseException:
 
 
 def completion(*args: Any, **kwargs: Any) -> Any:
-    """Wrapper around :func:`litellm.completion` that normalises filter errors.
+    """Wrapper around :func:`litellm.completion` that normalises filter errors
+    and handles credential rotation transparently.
 
-    Forwards every argument unchanged. The only difference from calling
-    ``litellm.completion`` directly is that an input-filter rejection
-    (which litellm wraps in ``APIConnectionError``) is re-raised as
-    :class:`ContentFilteredError`. Output-filter rejections already
-    surface as :class:`ContentFilteredError` via the SDK's transport patch
-    and pass through unchanged.
-
-    All other exceptions surface verbatim.
+    On ``AuthenticationError`` (e.g. rotated client_secret), reloads
+    credentials from the mounted secret volume and retries once.
+    All other exceptions surface verbatim after the filter-error translation.
     """
     try:
         return litellm.completion(*args, **kwargs)
+    except litellm.AuthenticationError:
+        # Local import avoids circular dep: completion ← __init__ ← completion
+        from sap_cloud_sdk.aicore import set_aicore_config
+
+        logger.info("AI Core credentials reloading after authentication failure")
+        set_aicore_config()
+        return litellm.completion(*args, **kwargs)
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
         if translated is exc:
@@ -83,10 +99,16 @@ def completion(*args: Any, **kwargs: Any) -> Any:
 async def acompletion(*args: Any, **kwargs: Any) -> Any:
     """Async wrapper around :func:`litellm.acompletion`.
 
-    Same translation semantics as :func:`completion`.
+    Same translation and credential-rotation semantics as :func:`completion`.
     """
     try:
         return await litellm.acompletion(*args, **kwargs)
+    except litellm.AuthenticationError:
+        from sap_cloud_sdk.aicore import set_aicore_config
+
+        logger.info("AI Core credentials reloading after authentication failure")
+        set_aicore_config()
+        return await litellm.acompletion(*args, **kwargs)
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
         if translated is exc:
diff --git a/tests/aicore/unit/test_aicore_watcher.py b/tests/aicore/unit/test_aicore_watcher.py
new file mode 100644
index 00000000..a2d0f1ee
--- /dev/null
+++ b/tests/aicore/unit/test_aicore_watcher.py
@@ -0,0 +1,184 @@
+"""Unit tests for watch_aicore_config() — proactive credential reload on secret mount change.
+
+The watcher polls the AI Core secret directory mtime every N seconds. When the mtime
+changes (Kubernetes projected volume atomic symlink swap on rotation), it calls
+set_aicore_config() proactively — before LiteLLM's cached OAuth token expires.
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+import time
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from sap_cloud_sdk.aicore import _get_secret_dir_mtime, watch_aicore_config
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_secret_dir(tmp_path: Path, instance_name: str = "aicore-instance") -> Path:
+    secret_dir = tmp_path / "aicore" / instance_name
+    secret_dir.mkdir(parents=True)
+    (secret_dir / "clientsecret").write_text("secret-v1")
+    return secret_dir
+
+
+# ---------------------------------------------------------------------------
+# _get_secret_dir_mtime
+# ---------------------------------------------------------------------------
+
+
+class TestGetSecretDirMtime:
+    def test_returns_float_for_existing_dir(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        _make_secret_dir(tmp_path)
+        mtime = _get_secret_dir_mtime()
+        assert isinstance(mtime, float)
+        assert mtime > 0.0
+
+    def test_returns_zero_for_missing_dir(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        # Do not create the secret dir
+        assert _get_secret_dir_mtime() == 0.0
+
+    def test_stable_without_modification(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        _make_secret_dir(tmp_path)
+        m1 = _get_secret_dir_mtime()
+        m2 = _get_secret_dir_mtime()
+        assert m1 == m2
+
+
+# ---------------------------------------------------------------------------
+# watch_aicore_config
+# ---------------------------------------------------------------------------
+
+
+class TestWatchAicoreConfig:
+    def test_thread_is_daemon(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        _make_secret_dir(tmp_path)
+        stop = threading.Event()
+        with patch("sap_cloud_sdk.aicore.set_aicore_config"):
+            t = watch_aicore_config(interval=60.0, stop_event=stop)
+        stop.set()
+        assert t.daemon is True
+
+    def test_no_reload_when_mtime_unchanged(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        _make_secret_dir(tmp_path)
+        stop = threading.Event()
+        with patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_reload:
+            t = watch_aicore_config(interval=0.05, stop_event=stop)
+            time.sleep(0.2)
+            stop.set()
+            t.join(timeout=1.0)
+        mock_reload.assert_not_called()
+
+    def test_reloads_on_directory_mtime_change(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        secret_dir = _make_secret_dir(tmp_path)
+        stop = threading.Event()
+        reloaded = threading.Event()
+
+        def _fake_reload(**kwargs):
+            reloaded.set()
+
+        with patch("sap_cloud_sdk.aicore.set_aicore_config", side_effect=_fake_reload):
+            t = watch_aicore_config(interval=0.05, stop_event=stop)
+            # Advance directory mtime to simulate kubelet secret rotation
+            new_time = time.time() + 10
+            os.utime(secret_dir, (new_time, new_time))
+            assert reloaded.wait(timeout=1.0), "reload was not triggered after mtime change"
+            stop.set()
+            t.join(timeout=1.0)
+
+    def test_logs_info_on_reload(self, tmp_path, monkeypatch, caplog):
+        import logging
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        secret_dir = _make_secret_dir(tmp_path)
+        stop = threading.Event()
+        reloaded = threading.Event()
+
+        def _fake_reload(**kwargs):
+            reloaded.set()
+
+        with patch("sap_cloud_sdk.aicore.set_aicore_config", side_effect=_fake_reload):
+            with caplog.at_level(logging.INFO, logger="sap_cloud_sdk.aicore"):
+                t = watch_aicore_config(interval=0.05, stop_event=stop)
+                new_time = time.time() + 10
+                os.utime(secret_dir, (new_time, new_time))
+                reloaded.wait(timeout=1.0)
+                stop.set()
+                t.join(timeout=1.0)
+
+        assert any(
+            "proactively reloading credentials" in r.message for r in caplog.records
+        )
+
+    def test_exception_in_set_aicore_config_does_not_crash_thread(
+        self, tmp_path, monkeypatch
+    ):
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        secret_dir = _make_secret_dir(tmp_path)
+        stop = threading.Event()
+        errored = threading.Event()
+
+        def _boom(**kwargs):
+            errored.set()
+            raise RuntimeError("simulated reload failure")
+
+        with patch("sap_cloud_sdk.aicore.set_aicore_config", side_effect=_boom):
+            t = watch_aicore_config(interval=0.05, stop_event=stop)
+            new_time = time.time() + 10
+            os.utime(secret_dir, (new_time, new_time))
+            assert errored.wait(timeout=1.0)
+            # Thread must still be alive after the exception
+            assert t.is_alive()
+            stop.set()
+            t.join(timeout=1.0)
+
+    def test_stop_event_exits_loop(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        _make_secret_dir(tmp_path)
+        stop = threading.Event()
+        with patch("sap_cloud_sdk.aicore.set_aicore_config"):
+            t = watch_aicore_config(interval=0.05, stop_event=stop)
+            stop.set()
+            t.join(timeout=1.0)
+        assert not t.is_alive()
+
+    def test_custom_instance_name_forwarded_to_set_aicore_config(
+        self, tmp_path, monkeypatch
+    ):
+        custom = "my-aicore"
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        secret_dir = tmp_path / "aicore" / custom
+        secret_dir.mkdir(parents=True)
+        (secret_dir / "clientsecret").write_text("v1")
+        stop = threading.Event()
+        reloaded = threading.Event()
+        captured_kwargs: list = []
+
+        def _fake_reload(**kwargs):
+            captured_kwargs.append(kwargs)
+            reloaded.set()
+
+        with patch("sap_cloud_sdk.aicore.set_aicore_config", side_effect=_fake_reload):
+            t = watch_aicore_config(
+                instance_name=custom, interval=0.05, stop_event=stop
+            )
+            new_time = time.time() + 10
+            os.utime(secret_dir, (new_time, new_time))
+            assert reloaded.wait(timeout=1.0)
+            stop.set()
+            t.join(timeout=1.0)
+
+        assert captured_kwargs[0].get("instance_name") == custom
diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py
index 9857f1d4..8238e922 100644
--- a/tests/aicore/unit/test_completion.py
+++ b/tests/aicore/unit/test_completion.py
@@ -15,14 +15,17 @@
 - :class:`ContentFilteredError` already raised by the transport patch
   passes through unchanged (we don't double-wrap).
 - ``acompletion`` exhibits the same behaviour on the async path.
+- On ``AuthenticationError``, credentials are reloaded and the call is
+  retried once (credential rotation without pod restart).
 """
 
 from __future__ import annotations
 
 import asyncio
 import json
-from unittest.mock import patch
+from unittest.mock import MagicMock, call, patch
 
+import litellm
 import pytest
 
 from sap_cloud_sdk.aicore import acompletion, completion
@@ -187,13 +190,130 @@ async def fake_acompletion(**kwargs):
     def test_non_filter_exception_surfaces_verbatim(self):
         raised = _FakeAPIConnectionError("SapException - other transport error")
 
-        async def fake_acompletion(**kwargs):
+        async def fake_acompletion_non_filter(**kwargs):
             raise raised
 
         with patch(
             "sap_cloud_sdk.aicore.completion.litellm.acompletion",
-            side_effect=fake_acompletion,
+            side_effect=fake_acompletion_non_filter,
         ):
             with pytest.raises(_FakeAPIConnectionError) as ei:
                 asyncio.run(acompletion(model="sap/x", messages=[]))
         assert ei.value is raised
+
+
+# ---------------------------------------------------------------------------
+# Reactive reload on AuthenticationError — sync
+# ---------------------------------------------------------------------------
+
+
+class TestCompletionReactiveReload:
+    def test_auth_error_triggers_reload_and_retry_succeeds(self):
+        """On AuthenticationError, credentials reload and second call succeeds."""
+        sentinel = object()
+        auth_err = litellm.AuthenticationError(
+            message="401 Unauthorized", llm_provider="sap", model="sap/x"
+        )
+        call_returns = [auth_err, sentinel]
+
+        def fake_completion(*args, **kwargs):
+            result = call_returns.pop(0)
+            if isinstance(result, Exception):
+                raise result
+            return result
+
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.completion", side_effect=fake_completion),
+            patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_reload,
+        ):
+            result = completion(model="sap/x", messages=[])
+
+        assert result is sentinel
+        mock_reload.assert_called_once_with()
+
+    def test_auth_error_retry_also_fails_propagates(self):
+        """If the retry also raises AuthenticationError, it propagates to the caller."""
+        auth_err = litellm.AuthenticationError(
+            message="401 Unauthorized", llm_provider="sap", model="sap/x"
+        )
+
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.completion", side_effect=auth_err),
+            patch("sap_cloud_sdk.aicore.set_aicore_config"),
+        ):
+            with pytest.raises(litellm.AuthenticationError):
+                completion(model="sap/x", messages=[])
+
+    def test_auth_error_reload_called_exactly_once(self):
+        """Reload is called exactly once — no infinite retry loop."""
+        auth_err = litellm.AuthenticationError(
+            message="401", llm_provider="sap", model="sap/x"
+        )
+        mock_litellm = MagicMock(side_effect=auth_err)
+
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.completion", mock_litellm),
+            patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_reload,
+        ):
+            with pytest.raises(litellm.AuthenticationError):
+                completion(model="sap/x", messages=[])
+
+        mock_reload.assert_called_once()
+        assert mock_litellm.call_count == 2
+
+    def test_non_auth_error_does_not_trigger_reload(self):
+        """Non-authentication errors do not trigger a credential reload."""
+        raised = ValueError("some other error")
+
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.completion", side_effect=raised),
+            patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_reload,
+        ):
+            with pytest.raises(ValueError):
+                completion(model="sap/x", messages=[])
+
+        mock_reload.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Reactive reload on AuthenticationError — async
+# ---------------------------------------------------------------------------
+
+
+class TestACompletionReactiveReload:
+    def test_auth_error_triggers_reload_and_retry_succeeds(self):
+        sentinel = object()
+        auth_err = litellm.AuthenticationError(
+            message="401 Unauthorized", llm_provider="sap", model="sap/x"
+        )
+        call_returns = [auth_err, sentinel]
+
+        async def fake_acompletion(*args, **kwargs):
+            result = call_returns.pop(0)
+            if isinstance(result, Exception):
+                raise result
+            return result
+
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=fake_acompletion),
+            patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_reload,
+        ):
+            result = asyncio.run(acompletion(model="sap/x", messages=[]))
+
+        assert result is sentinel
+        mock_reload.assert_called_once_with()
+
+    def test_auth_error_retry_also_fails_propagates(self):
+        auth_err = litellm.AuthenticationError(
+            message="401", llm_provider="sap", model="sap/x"
+        )
+
+        async def fake_acompletion(*args, **kwargs):
+            raise auth_err
+
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=fake_acompletion),
+            patch("sap_cloud_sdk.aicore.set_aicore_config"),
+        ):
+            with pytest.raises(litellm.AuthenticationError):
+                asyncio.run(acompletion(model="sap/x", messages=[]))
diff --git a/tests/aicore/unit/test_credential_rotation_flow.py b/tests/aicore/unit/test_credential_rotation_flow.py
new file mode 100644
index 00000000..6081865c
--- /dev/null
+++ b/tests/aicore/unit/test_credential_rotation_flow.py
@@ -0,0 +1,239 @@
+"""Tests verifying that set_aicore_config() updating os.environ is sufficient
+for LiteLLM to pick up new credentials on the next OAuth token refresh.
+
+Background: LiteLLM caches the OAuth token (~12h lifetime), NOT the client_secret
+in a long-lived client object. When the token expires, LiteLLM reads
+os.environ["AICORE_CLIENT_SECRET"] fresh to fetch a new token. So calling
+set_aicore_config() (which updates os.environ) is the only thing needed to
+handle credential rotation — no LiteLLM client object needs to be recreated.
+
+These tests verify the full contract that makes both approaches in PR #256 work:
+- Reactive: 401 → set_aicore_config() → env updated → retry succeeds
+- Proactive: watcher detects mtime change → set_aicore_config() → env updated
+             → next token refresh uses new client_secret before expiry
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+import time
+from pathlib import Path
+
+import litellm
+import pytest
+from unittest.mock import patch
+
+from sap_cloud_sdk.aicore import set_aicore_config, watch_aicore_config
+from sap_cloud_sdk.aicore import completion
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _write_secret_files(tmp_path: Path, secret: str, instance: str = "aicore-instance") -> Path:
+    secret_dir = tmp_path / "aicore" / instance
+    secret_dir.mkdir(parents=True, exist_ok=True)
+    (secret_dir / "clientid").write_text("test-client-id")
+    (secret_dir / "clientsecret").write_text(secret)
+    (secret_dir / "url").write_text("https://auth.example.com")
+    serviceurls = secret_dir / "serviceurls"
+    serviceurls.write_text('{"AI_API_URL": "https://api.example.com"}')
+    return secret_dir
+
+
+# ---------------------------------------------------------------------------
+# 1. env updated on second set_aicore_config() call
+# ---------------------------------------------------------------------------
+
+
+class TestEnvUpdatedOnRotation:
+    def test_second_call_overwrites_client_secret(self, tmp_path, monkeypatch):
+        """Re-calling set_aicore_config() after file update writes the new
+        client_secret to os.environ — the value LiteLLM reads on next token refresh."""
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        monkeypatch.delenv("AICORE_CLIENT_SECRET", raising=False)
+
+        _write_secret_files(tmp_path, secret="secret-v1")
+        with patch("sap_cloud_sdk.aicore.set_filtering"):
+            set_aicore_config()
+        assert os.environ["AICORE_CLIENT_SECRET"] == "secret-v1"
+
+        # Simulate BTP rotation: kubelet updates the file
+        (tmp_path / "aicore" / "aicore-instance" / "clientsecret").write_text("secret-v2")
+
+        with patch("sap_cloud_sdk.aicore.set_filtering"):
+            set_aicore_config()
+        assert os.environ["AICORE_CLIENT_SECRET"] == "secret-v2"
+
+
+# ---------------------------------------------------------------------------
+# 2. Proactive watcher updates env before token expiry
+# ---------------------------------------------------------------------------
+
+
+class TestWatcherUpdatesEnvProactively:
+    def test_env_updated_after_mtime_change(self, tmp_path, monkeypatch):
+        """Full end-to-end: watcher detects mtime change → set_aicore_config()
+        → os.environ has new secret before the OAuth token expires."""
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        monkeypatch.delenv("AICORE_CLIENT_SECRET", raising=False)
+
+        secret_dir = _write_secret_files(tmp_path, secret="secret-v1")
+
+        with patch("sap_cloud_sdk.aicore.set_filtering"):
+            set_aicore_config()
+        assert os.environ["AICORE_CLIENT_SECRET"] == "secret-v1"
+
+        stop = threading.Event()
+        reloaded = threading.Event()
+
+        def _tracking_set_config(**kwargs):
+            reloaded.set()
+
+        with patch("sap_cloud_sdk.aicore.set_aicore_config", side_effect=_tracking_set_config):
+            t = watch_aicore_config(interval=0.05, stop_event=stop)
+
+            # Simulate kubelet secret update: new file content + advance dir mtime
+            (secret_dir / "clientsecret").write_text("secret-v2")
+            new_time = time.time() + 10
+            os.utime(secret_dir, (new_time, new_time))
+
+            assert reloaded.wait(timeout=1.0), "watcher did not trigger reload"
+            stop.set()
+            t.join(timeout=1.0)
+
+
+# ---------------------------------------------------------------------------
+# 3. Reactive 401 handler updates env
+# ---------------------------------------------------------------------------
+
+
+class TestReactive401UpdatesEnv:
+    def test_completion_updates_env_after_401(self, tmp_path, monkeypatch):
+        """On AuthenticationError, the 401 handler calls set_aicore_config()
+        which updates os.environ — env has new secret after the call."""
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        monkeypatch.delenv("AICORE_CLIENT_SECRET", raising=False)
+
+        _write_secret_files(tmp_path, secret="secret-v1")
+        with patch("sap_cloud_sdk.aicore.set_filtering"):
+            set_aicore_config()
+
+        # Rotate the file before the 401 is caught
+        (tmp_path / "aicore" / "aicore-instance" / "clientsecret").write_text("secret-v2")
+
+        auth_err = litellm.AuthenticationError(
+            message="401", llm_provider="sap", model="sap/x"
+        )
+        sentinel = object()
+        call_returns = [auth_err, sentinel]
+
+        def _fake_completion(*args, **kwargs):
+            r = call_returns.pop(0)
+            if isinstance(r, Exception):
+                raise r
+            return r
+
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.completion", side_effect=_fake_completion),
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+        ):
+            result = completion(model="sap/x", messages=[])
+
+        assert result is sentinel
+        assert os.environ["AICORE_CLIENT_SECRET"] == "secret-v2"
+
+
+# ---------------------------------------------------------------------------
+# 4. No LiteLLM client object needs recreation
+# ---------------------------------------------------------------------------
+
+
+class TestNoClientRecreationNeeded:
+    def test_litellm_has_no_cached_aicore_client_attribute(self, tmp_path, monkeypatch):
+        """LiteLLM does not hold an _aicore_client or similar attribute that
+        would cache the old client_secret — env update is the single source of truth."""
+        import litellm as _litellm
+        # If LiteLLM ever adds a cached client object, this test will catch it
+        # so we can handle it explicitly.
+        assert not hasattr(_litellm, "_aicore_client"), (
+            "LiteLLM added an _aicore_client attribute — credential rotation logic "
+            "must be updated to also reset this object."
+        )
+
+    def test_env_is_single_source_after_rotation(self, tmp_path, monkeypatch):
+        """After set_aicore_config() with v2, os.environ has v2 and no stale v1
+        value persists anywhere that would prevent LiteLLM from using v2."""
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        monkeypatch.delenv("AICORE_CLIENT_SECRET", raising=False)
+
+        _write_secret_files(tmp_path, secret="secret-v1")
+        with patch("sap_cloud_sdk.aicore.set_filtering"):
+            set_aicore_config()
+        assert os.environ.get("AICORE_CLIENT_SECRET") == "secret-v1"
+
+        (tmp_path / "aicore" / "aicore-instance" / "clientsecret").write_text("secret-v2")
+        with patch("sap_cloud_sdk.aicore.set_filtering"):
+            set_aicore_config()
+
+        assert os.environ.get("AICORE_CLIENT_SECRET") == "secret-v2"
+        # v1 is gone from the env
+        assert "secret-v1" not in os.environ.get("AICORE_CLIENT_SECRET", "")
+
+
+# ---------------------------------------------------------------------------
+# 5. Concurrent set_aicore_config() calls do not raise
+# ---------------------------------------------------------------------------
+
+
+class TestConcurrentSetAicoreConfig:
+    def test_concurrent_calls_no_exception(self, tmp_path, monkeypatch):
+        """Two threads calling set_aicore_config() concurrently must not
+        raise exceptions. Last-write-wins is acceptable for credential rotation."""
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        monkeypatch.delenv("AICORE_CLIENT_SECRET", raising=False)
+
+        for v in ("v1", "v2"):
+            _write_secret_files(tmp_path, secret=v)  # final file = v2
+
+        errors: list[Exception] = []
+
+        def _call():
+            try:
+                with patch("sap_cloud_sdk.aicore.set_filtering"):
+                    set_aicore_config()
+            except Exception as e:
+                errors.append(e)
+
+        threads = [threading.Thread(target=_call) for _ in range(5)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=2.0)
+
+        assert not errors, f"Concurrent calls raised: {errors}"
+        # env must hold one of the valid values (not empty or corrupt)
+        assert os.environ.get("AICORE_CLIENT_SECRET") in ("v1", "v2")
+
+
+# ---------------------------------------------------------------------------
+# 6. _get_secret_dir_mtime stable without modification
+# ---------------------------------------------------------------------------
+
+
+class TestGetSecretDirMtimeStability:
+    def test_stable_float_for_existing_dir(self, tmp_path, monkeypatch):
+        """Calling _get_secret_dir_mtime twice on an unchanged dir returns the
+        same float — watcher does not trigger spurious reloads."""
+        from sap_cloud_sdk.aicore import _get_secret_dir_mtime
+        monkeypatch.setenv("SERVICE_BINDING_ROOT", str(tmp_path))
+        secret_dir = tmp_path / "aicore" / "aicore-instance"
+        secret_dir.mkdir(parents=True)
+
+        m1 = _get_secret_dir_mtime()
+        m2 = _get_secret_dir_mtime()
+        assert m1 == m2
+        assert m1 > 0.0