diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py
index 7fb10094..3b9d0988 100644
--- a/src/sap_cloud_sdk/aicore/__init__.py
+++ b/src/sap_cloud_sdk/aicore/__init__.py
@@ -13,7 +13,7 @@
 from sap_cloud_sdk.core.telemetry.metrics_decorator import record_metrics
 from sap_cloud_sdk.core.telemetry.module import Module
 from sap_cloud_sdk.core.telemetry.operation import Operation
-from .completion import acompletion, completion
+from .completion import acompletion, completion, reload_aicore_credentials
 from .filtering import (
     AzureContentFilter,
     ContentFilter,
@@ -30,6 +30,22 @@
 
 logger = logging.getLogger(__name__)
 
+# When set, the infrastructure sidecar adds the mTLS certificate transparently.
+# The SDK calls the XSUAA token endpoint over plain HTTPS with only client_id.
+# No client_secret or certificate material is required in the service binding.
+TRANSPARENT_TLS_ENV_VAR = "AICORE_TRANSPARENT_TLS"
+
+# Option 3 — transparent proxy routing.
+# Deployer injects these; agent code is identical in all environments.
+_PROXY_URL_ENV = "AICORE_PROXY_URL"
+_PROXY_API_KEY_ENV = "AICORE_PROXY_API_KEY"
+_DESTINATION_NAME_ENV = "AICORE_DESTINATION_NAME"
+
+
+def _is_transparent_tls() -> bool:
+    """Return True when transparent TLS proxy mode is active."""
+    return os.environ.get(TRANSPARENT_TLS_ENV_VAR, "").strip().lower() in ("1", "true", "yes")
+
 
 def _get_secret(
     env_var_name: str,
@@ -118,14 +134,27 @@ def _get_aicore_base_url(instance_name: str = "aicore-instance") -> str:
 def set_aicore_config(instance_name: str = "aicore-instance") -> None:
     """Load AI Core credentials and activate content filtering.
 
-    Loads secrets from files or environment variables and sets them as
-    process env vars so ``litellm`` picks them up.
+    Detects which routing mode is active based on environment variables:
+
+    - ``AICORE_PROXY_URL`` set → **proxy mode**: routes all LiteLLM calls
+      through a LiteLLM proxy; ``sap/<model>`` is aliased to
+      ``litellm_proxy/<model>`` transparently. No AI Core credentials
+      are written to the process environment.
+
+    - ``AICORE_DESTINATION_NAME`` set → **destination mode**: loads AI Core
+      credentials from a BTP Destination Service destination at startup.
+      The deployer only needs to inject Destination Service binding credentials;
+      the AI Core ``client_secret`` never needs to be in the K8s Secret.
+      Combined with the ``_clear_client_secret()`` mechanism (PR #257),
+      the secret is removed from env after the first LiteLLM call.
 
-    File mappings based on the Kubernetes secret structure:
-        clientid → AICORE_CLIENT_ID
-        clientsecret → AICORE_CLIENT_SECRET
-        url → AICORE_AUTH_URL
-        serviceurls (JSON with AI_API_URL) → AICORE_BASE_URL
+    - Neither set → **direct mode** (existing behaviour): credentials are
+      loaded from a mounted K8s secret volume or environment variables.
+      ``AICORE_TRANSPARENT_TLS=true`` suppresses ``client_secret`` and
+      relies on an mTLS sidecar.
+
+    Agent code is identical in all three modes — the deployer controls
+    routing by choosing which env vars to inject.
 
     After credentials are loaded, content filtering is activated on every
     ``sap/*`` LiteLLM call at the configured thresholds (default: severity
@@ -135,29 +164,114 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None:
     to turn filtering off at runtime, or set ``AICORE_FILTER_ENABLED=false``
     to keep it off entirely.
     """
-    # Load secrets
+    proxy_url = os.environ.get(_PROXY_URL_ENV, "")
+    destination_name = os.environ.get(_DESTINATION_NAME_ENV, "")
+
+    if proxy_url:
+        _configure_proxy_mode(proxy_url)
+    elif destination_name:
+        _configure_destination_mode(destination_name)
+    else:
+        _configure_direct_mode(instance_name)
+
+    set_filtering()
+
+
+def _configure_proxy_mode(proxy_url: str) -> None:
+    """Configure LiteLLM to route calls through an external proxy.
+
+    Sets ``litellm.api_base`` / ``litellm.api_key`` globally.
+    Model strings (e.g. ``sap/<model>``) are passed verbatim — no rewrite.
+    No AI Core credentials are written to env.
+    """
+    import litellm as _litellm
+
+    api_key = os.environ.get(_PROXY_API_KEY_ENV, "")
+    _litellm.api_base = proxy_url
+    if api_key:
+        _litellm.api_key = api_key
+    logger.info("AI Core proxy mode active — routing via %s", proxy_url)
+
+
+def _configure_destination_mode(name: str) -> None:
+    """Load AI Core credentials from a BTP Destination Service destination.
+
+    Calls the Destination Service at startup to resolve the named destination
+    and extracts ``clientId``, ``clientSecret``, ``tokenServiceURL``, and the
+    AI Core ``URL`` from the destination configuration properties. These are
+    written to the standard ``AICORE_*`` env vars so that LiteLLM can fetch
+    an OAuth token from XSUAA as usual.
+
+    Security: The deployer does NOT need to inject ``AICORE_CLIENT_SECRET``
+    directly — only Destination Service binding credentials are required in
+    the agent environment. The AI Core ``client_secret`` is fetched here
+    and removed from env after the first successful LiteLLM call
+    (PR #257 ``_clear_client_secret()`` mechanism).
+
+    Raises ``RuntimeError`` if the destination is not found or does not
+    return ``clientId`` / ``clientSecret``.
+    """
+    from sap_cloud_sdk.destination import create_client  # lazy import
+
+    client = create_client()
+    dest = client.get_destination(name)
+
+    if dest is None:
+        raise RuntimeError(
+            f"AI Core destination '{name}' not found in Destination Service. "
+            "Check that the destination exists and the binding has access."
+        )
+
+    base_url = dest.url or ""
+    if base_url and not base_url.endswith("/v2"):
+        base_url = base_url.rstrip("/") + "/v2"
+    if base_url:
+        os.environ["AICORE_BASE_URL"] = base_url
+
+    resource_group = dest.properties.get("resource_group", "default")
+    os.environ["AICORE_RESOURCE_GROUP"] = resource_group
+
+    client_id = dest.properties.get("clientId", "")
+    client_secret = dest.properties.get("clientSecret", "")
+    token_service_url = dest.properties.get("tokenServiceURL", "")
+
+    if not client_id or not client_secret:
+        raise RuntimeError(
+            f"Destination '{name}' did not return clientId/clientSecret. "
+            "Ensure the destination uses OAuth2ClientCredentials authentication "
+            "and the calling app has the Destination Service technical-user scope."
+        )
+
+    os.environ["AICORE_CLIENT_ID"] = client_id
+    os.environ["AICORE_CLIENT_SECRET"] = client_secret  # cleared after first LiteLLM call
+
+    if token_service_url:
+        if not token_service_url.endswith("/oauth/token"):
+            token_service_url = token_service_url.rstrip("/") + "/oauth/token"
+        os.environ["AICORE_AUTH_URL"] = token_service_url
+
+    logger.info("AI Core destination mode active — credentials loaded from '%s'", name)
+
+
+def _configure_direct_mode(instance_name: str) -> None:
+    """Load AI Core credentials directly from mounted secrets or env vars."""
+    transparent_tls = _is_transparent_tls()
+
     client_id = _get_secret("AICORE_CLIENT_ID", "clientid", instance_name=instance_name)
-    client_secret = _get_secret(
-        "AICORE_CLIENT_SECRET", "clientsecret", instance_name=instance_name
-    )
     auth_url = _get_secret("AICORE_AUTH_URL", "url", instance_name=instance_name)
     base_url = _get_aicore_base_url(instance_name)
     resource_group = _get_secret(
         "AICORE_RESOURCE_GROUP", default="default", instance_name=instance_name
     )
 
-    # Ensure AICORE_AUTH_URL has /oauth/token suffix
     if auth_url and not auth_url.endswith("/oauth/token"):
         auth_url = auth_url.rstrip("/") + "/oauth/token"
 
     if base_url and not base_url.endswith("/v2"):
         base_url = base_url.rstrip("/") + "/v2"
 
-    # Set environment variables for LiteLLM
     if client_id:
         os.environ["AICORE_CLIENT_ID"] = client_id
-    if client_secret:
-        os.environ["AICORE_CLIENT_SECRET"] = client_secret
     if auth_url:
         os.environ["AICORE_AUTH_URL"] = auth_url
     if base_url:
@@ -165,18 +279,22 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None:
     if resource_group:
         os.environ["AICORE_RESOURCE_GROUP"] = resource_group
 
-    # Log configuration completion (excluding sensitive information)
-    logger.info("AI Core configuration has been set successfully")
+    if transparent_tls:
+        os.environ.pop("AICORE_CLIENT_SECRET", None)
+        logger.info("AI Core transparent TLS mode active — client_secret not required")
+    else:
+        client_secret = _get_secret(
+            "AICORE_CLIENT_SECRET", "clientsecret", instance_name=instance_name
+        )
+        if client_secret:
+            os.environ["AICORE_CLIENT_SECRET"] = client_secret
 
-    # Activate content filtering for all sap/* LiteLLM model calls.
-    # AICORE_FILTER_ENABLED=false disables; AICORE_FILTER_* tune thresholds.
-    # Errors propagate — filtering misconfiguration should surface at startup
-    # rather than be swallowed silently.
-    set_filtering()
+    logger.info("AI Core configuration has been set successfully")
 
 
 __all__ = [
     "set_aicore_config",
+    "reload_aicore_credentials",
     "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..d56f29ae 100644
--- a/src/sap_cloud_sdk/aicore/completion.py
+++ b/src/sap_cloud_sdk/aicore/completion.py
@@ -14,6 +14,21 @@
 re-raising as :class:`ContentFilteredError` so callers can rely on a
 single exception type for "filter blocked you."
 
+Credential handling
+-------------------
+CLIENT_SECRET minimisation (AFSDK-4291):
+After the first successful LiteLLM call, ``AICORE_CLIENT_SECRET`` is removed
+from ``os.environ``. LiteLLM has already captured the secret inside its token
+creator closure at that point and no longer needs the env var. This minimises
+the window of exposure to child processes and container introspection.
+
+Credential rotation (reactive reload):
+When a credential is rotated while the pod is running, LiteLLM's cached token
+becomes invalid and the next token refresh raises ``litellm.AuthenticationError``.
+The wrappers intercept this, reload credentials from the mounted secret volume
+via :func:`reload_aicore_credentials`, and retry once. The secret is cleared
+again after the retry succeeds.
+
 Usage::
 
     from sap_cloud_sdk.aicore import completion, ContentFilteredError
@@ -39,12 +54,64 @@
 
 from __future__ import annotations
 
+import logging
+import os
+import threading
 from typing import Any
 
 import litellm
 
 from .filtering.filters import _parse_input_filter_error
 
+logger = logging.getLogger(__name__)
+
+# Tracks whether AICORE_CLIENT_SECRET has already been cleared after the first
+# successful LiteLLM call. Reset when credentials are reloaded so the secret
+# is cleared again after the retry succeeds.
+_secret_lock = threading.Lock()
+_secret_cleared = False
+
+
+def _clear_client_secret() -> None:
+    """Remove AICORE_CLIENT_SECRET from env after LiteLLM has cached the token.
+
+    Safe to call multiple times — subsequent calls are no-ops once cleared.
+    No-op in transparent TLS mode (secret was never written).
+    """
+    global _secret_cleared
+    with _secret_lock:
+        if not _secret_cleared:
+            if os.environ.pop("AICORE_CLIENT_SECRET", None) is not None:
+                logger.info(
+                    "AICORE_CLIENT_SECRET cleared from environment "
+                    "after token acquisition (AFSDK-4291)"
+                )
+            _secret_cleared = True
+
+
+def _reset_secret_cleared() -> None:
+    """Allow _clear_client_secret() to fire again after a credential reload."""
+    global _secret_cleared
+    with _secret_lock:
+        _secret_cleared = False
+
+
+def reload_aicore_credentials() -> None:
+    """Re-read AI Core credentials from the mounted secret volume.
+
+    Called automatically by :func:`completion` and :func:`acompletion` when
+    LiteLLM raises ``AuthenticationError`` — covers credential rotation
+    (client_secret or mTLS certificate) without requiring a pod restart.
+
+    Safe to call manually if the application needs to force a reload, e.g.
+    after a deliberate secret rotation triggered by the operator.
+    """
+    # Import here to avoid a circular import: completion ← __init__ ← completion
+    from sap_cloud_sdk.aicore import set_aicore_config
+    _reset_secret_cleared()
+    logger.info("AI Core credentials reloading after authentication failure")
+    set_aicore_config()
+
 
 def _maybe_translate_filter_error(exc: BaseException) -> BaseException:
     """Return a :class:`ContentFilteredError` if ``exc`` is a wrapped
@@ -60,19 +127,29 @@ 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.
+
+    After the first successful call, ``AICORE_CLIENT_SECRET`` is removed from
+    ``os.environ`` — LiteLLM has captured it in its token creator closure and
+    no longer needs the env var (AFSDK-4291).
 
-    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.
+    On ``AuthenticationError`` (e.g. rotated client_secret or mTLS cert),
+    reloads credentials from the mounted secret volume and retries once.
 
-    All other exceptions surface verbatim.
+    Model strings (e.g. ``sap/<model>``) are passed verbatim to LiteLLM in all
+    routing modes — proxy routing is handled by ``litellm.api_base`` configured
+    in :func:`set_aicore_config`, not by rewriting the model name.
     """
     try:
-        return litellm.completion(*args, **kwargs)
+        result = litellm.completion(*args, **kwargs)
+        _clear_client_secret()
+        return result
+    except litellm.AuthenticationError:
+        reload_aicore_credentials()
+        result = litellm.completion(*args, **kwargs)
+        _clear_client_secret()
+        return result
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
         if translated is exc:
@@ -83,10 +160,18 @@ 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 credential-minimisation and rotation semantics as :func:`completion`.
+    Model strings are passed verbatim to LiteLLM in all routing modes.
     """
     try:
-        return await litellm.acompletion(*args, **kwargs)
+        result = await litellm.acompletion(*args, **kwargs)
+        _clear_client_secret()
+        return result
+    except litellm.AuthenticationError:
+        reload_aicore_credentials()
+        result = await litellm.acompletion(*args, **kwargs)
+        _clear_client_secret()
+        return result
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
         if translated is exc:
@@ -94,4 +179,4 @@ async def acompletion(*args: Any, **kwargs: Any) -> Any:
         raise translated from exc
 
 
-__all__ = ["completion", "acompletion"]
+__all__ = ["completion", "acompletion", "reload_aicore_credentials"]
diff --git a/tests/aicore/unit/test_aicore.py b/tests/aicore/unit/test_aicore.py
index 5439329c..ac8909a6 100644
--- a/tests/aicore/unit/test_aicore.py
+++ b/tests/aicore/unit/test_aicore.py
@@ -4,12 +4,14 @@
 import os
 from unittest.mock import mock_open, patch
 
+import pytest
 
 from sap_cloud_sdk.aicore import (
     _get_aicore_base_url,
     _get_secret,
     set_aicore_config,
 )
+from sap_cloud_sdk.aicore import _is_transparent_tls
 
 
 class TestGetSecret:
@@ -710,3 +712,342 @@ def test_set_config_decorated_with_record_metrics(self):
 
             # Function should complete without errors even with decorator
             # The actual telemetry recording is tested in telemetry tests
+
+
+class TestIsTransparentTls:
+    """Test suite for _is_transparent_tls helper."""
+
+    def test_returns_true_for_value_true(self):
+        with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "true"}):
+            assert _is_transparent_tls() is True
+
+    def test_returns_true_for_value_1(self):
+        with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "1"}):
+            assert _is_transparent_tls() is True
+
+    def test_returns_true_for_value_yes(self):
+        with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "yes"}):
+            assert _is_transparent_tls() is True
+
+    def test_returns_true_case_insensitive(self):
+        with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "TRUE"}):
+            assert _is_transparent_tls() is True
+
+    def test_returns_false_when_absent(self):
+        with patch.dict("os.environ", {}, clear=True):
+            assert _is_transparent_tls() is False
+
+    def test_returns_false_for_value_false(self):
+        with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "false"}):
+            assert _is_transparent_tls() is False
+
+
+class TestSetAICoreConfigTransparentTls:
+    """Test suite for set_aicore_config in transparent TLS mode."""
+
+    def _base_secrets(self):
+        return {
+            "AICORE_CLIENT_ID": "test-client-id",
+            "AICORE_AUTH_URL": "https://auth.example.com",
+            "AICORE_RESOURCE_GROUP": "default",
+        }
+
+    def test_transparent_tls_does_not_set_client_secret(self):
+        """In transparent TLS mode, AICORE_CLIENT_SECRET must not be written to env."""
+        with (
+            patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret,
+            patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value="https://api.example.com"),
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "true"}, clear=True),
+        ):
+            mock_get_secret.side_effect = lambda name, file_name=None, default="", instance_name="aicore-instance": (
+                self._base_secrets().get(name, default)
+            )
+
+            set_aicore_config()
+
+            assert "AICORE_CLIENT_SECRET" not in os.environ
+
+    def test_transparent_tls_removes_stale_client_secret(self):
+        """Any pre-existing AICORE_CLIENT_SECRET is cleared in transparent TLS mode."""
+        with (
+            patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret,
+            patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value=""),
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch.dict(
+                "os.environ",
+                {"AICORE_TRANSPARENT_TLS": "true", "AICORE_CLIENT_SECRET": "stale-secret"},
+                clear=True,
+            ),
+        ):
+            mock_get_secret.side_effect = lambda name, file_name=None, default="", instance_name="aicore-instance": (
+                self._base_secrets().get(name, default)
+            )
+
+            set_aicore_config()
+
+            assert "AICORE_CLIENT_SECRET" not in os.environ
+
+    def test_transparent_tls_sets_other_credentials(self):
+        """Non-secret credentials are still set in transparent TLS mode."""
+        with (
+            patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret,
+            patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value="https://api.example.com"),
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "true"}, clear=True),
+        ):
+            mock_get_secret.side_effect = lambda name, file_name=None, default="", instance_name="aicore-instance": (
+                self._base_secrets().get(name, default)
+            )
+
+            set_aicore_config()
+
+            assert os.environ["AICORE_CLIENT_ID"] == "test-client-id"
+            assert os.environ["AICORE_AUTH_URL"] == "https://auth.example.com/oauth/token"
+            assert os.environ["AICORE_BASE_URL"] == "https://api.example.com/v2"
+
+    def test_standard_mode_still_sets_client_secret(self):
+        """Regression: without transparent TLS, client_secret is still written."""
+        with (
+            patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret,
+            patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value=""),
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch.dict("os.environ", {}, clear=True),
+        ):
+            mock_get_secret.side_effect = lambda name, file_name=None, default="", instance_name="aicore-instance": (
+                {**self._base_secrets(), "AICORE_CLIENT_SECRET": "my-secret"}.get(name, default)
+            )
+
+            set_aicore_config()
+
+            assert os.environ["AICORE_CLIENT_SECRET"] == "my-secret"
+
+    def test_transparent_tls_does_not_call_get_secret_for_client_secret(self):
+        """_get_secret should not be called for clientsecret in transparent TLS mode."""
+        with (
+            patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret,
+            patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value=""),
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "true"}, clear=True),
+        ):
+            mock_get_secret.return_value = ""
+
+            set_aicore_config()
+
+            called_names = [c.args[0] for c in mock_get_secret.call_args_list]
+            assert "AICORE_CLIENT_SECRET" not in called_names
+
+
+# ---------------------------------------------------------------------------
+# Proxy mode — set_aicore_config() with AICORE_PROXY_URL
+# ---------------------------------------------------------------------------
+
+
+class TestSetAICoreConfigProxyMode:
+    """set_aicore_config() routes via proxy when AICORE_PROXY_URL is present."""
+
+    def _base_proxy_env(self, **extra):
+        return {"AICORE_PROXY_URL": "https://proxy.example.com", **extra}
+
+    def test_proxy_mode_sets_litellm_api_base(self):
+        import litellm
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch.dict("os.environ", self._base_proxy_env(), clear=True),
+        ):
+            set_aicore_config()
+        assert litellm.api_base == "https://proxy.example.com"
+        litellm.api_base = None  # cleanup
+
+    def test_proxy_mode_sets_litellm_api_key_when_virtual_key_present(self):
+        import litellm
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch.dict(
+                "os.environ",
+                self._base_proxy_env(AICORE_PROXY_API_KEY="sk-virt-123"),
+                clear=True,
+            ),
+        ):
+            set_aicore_config()
+        assert litellm.api_key == "sk-virt-123"
+        litellm.api_key = None  # cleanup
+
+    def test_proxy_mode_does_not_write_aicore_credentials(self):
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch.dict("os.environ", self._base_proxy_env(), clear=True),
+        ):
+            set_aicore_config()
+            for var in ("AICORE_CLIENT_ID", "AICORE_CLIENT_SECRET", "AICORE_AUTH_URL"):
+                assert var not in os.environ, f"{var} must not be written in proxy mode"
+
+    def test_proxy_mode_takes_precedence_over_destination(self):
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch("sap_cloud_sdk.aicore._configure_destination_mode") as mock_dest,
+            patch.dict(
+                "os.environ",
+                self._base_proxy_env(AICORE_DESTINATION_NAME="aicore"),
+                clear=True,
+            ),
+        ):
+            set_aicore_config()
+        mock_dest.assert_not_called()
+
+    def test_proxy_mode_still_calls_set_filtering(self):
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering") as mock_filter,
+            patch.dict("os.environ", self._base_proxy_env(), clear=True),
+        ):
+            set_aicore_config()
+        mock_filter.assert_called_once()
+
+    def test_direct_mode_used_when_neither_proxy_nor_destination_set(self):
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch("sap_cloud_sdk.aicore._configure_direct_mode") as mock_direct,
+            patch.dict("os.environ", {}, clear=True),
+        ):
+            set_aicore_config()
+        mock_direct.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Destination mode — set_aicore_config() with AICORE_DESTINATION_NAME
+# ---------------------------------------------------------------------------
+
+
+class TestSetAICoreConfigDestinationMode:
+    """set_aicore_config() loads credentials from BTP Destination Service."""
+
+    def _mock_destination(
+        self,
+        url="https://api.ai.prod.example.com",
+        properties=None,
+        auth_tokens=None,
+    ):
+        from unittest.mock import MagicMock
+        dest = MagicMock()
+        dest.url = url
+        dest.properties = properties or {
+            "clientId": "sb-client-id",
+            "clientSecret": "client-secret-value",
+            "tokenServiceURL": "https://auth.example.com/oauth/token",
+        }
+        dest.auth_tokens = auth_tokens or []
+        return dest
+
+    def test_destination_mode_sets_base_url_with_v2_suffix(self):
+        dest = self._mock_destination(url="https://api.ai.prod.example.com")
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch(
+                "sap_cloud_sdk.destination.create_client"
+            ) as mock_create,
+            patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True),
+        ):
+            mock_create.return_value.get_destination.return_value = dest
+            set_aicore_config()
+            assert os.environ["AICORE_BASE_URL"] == "https://api.ai.prod.example.com/v2"
+
+    def test_destination_mode_does_not_double_v2(self):
+        dest = self._mock_destination(url="https://api.ai.prod.example.com/v2")
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch("sap_cloud_sdk.destination.create_client") as mock_create,
+            patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True),
+        ):
+            mock_create.return_value.get_destination.return_value = dest
+            set_aicore_config()
+            assert os.environ["AICORE_BASE_URL"] == "https://api.ai.prod.example.com/v2"
+
+    def test_destination_mode_sets_resource_group_from_properties(self):
+        dest = self._mock_destination(
+            properties={
+                "clientId": "id",
+                "clientSecret": "sec",
+                "tokenServiceURL": "https://auth.example.com/oauth/token",
+                "resource_group": "production",
+            }
+        )
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch("sap_cloud_sdk.destination.create_client") as mock_create,
+            patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True),
+        ):
+            mock_create.return_value.get_destination.return_value = dest
+            set_aicore_config()
+            assert os.environ["AICORE_RESOURCE_GROUP"] == "production"
+
+    def test_destination_mode_defaults_resource_group_to_default(self):
+        dest = self._mock_destination()  # no resource_group in properties
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch("sap_cloud_sdk.destination.create_client") as mock_create,
+            patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True),
+        ):
+            mock_create.return_value.get_destination.return_value = dest
+            set_aicore_config()
+            assert os.environ["AICORE_RESOURCE_GROUP"] == "default"
+
+    def test_destination_mode_sets_client_credentials(self):
+        dest = self._mock_destination()
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch("sap_cloud_sdk.destination.create_client") as mock_create,
+            patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True),
+        ):
+            mock_create.return_value.get_destination.return_value = dest
+            set_aicore_config()
+            assert os.environ["AICORE_CLIENT_ID"] == "sb-client-id"
+            assert os.environ["AICORE_CLIENT_SECRET"] == "client-secret-value"
+
+    def test_destination_mode_appends_oauth_token_suffix(self):
+        dest = self._mock_destination(
+            properties={
+                "clientId": "id",
+                "clientSecret": "sec",
+                "tokenServiceURL": "https://auth.example.com",  # no /oauth/token
+            }
+        )
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch("sap_cloud_sdk.destination.create_client") as mock_create,
+            patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True),
+        ):
+            mock_create.return_value.get_destination.return_value = dest
+            set_aicore_config()
+            assert os.environ["AICORE_AUTH_URL"] == "https://auth.example.com/oauth/token"
+
+    def test_destination_mode_raises_when_destination_not_found(self):
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch("sap_cloud_sdk.destination.create_client") as mock_create,
+            patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "missing"}, clear=True),
+        ):
+            mock_create.return_value.get_destination.return_value = None
+            with pytest.raises(RuntimeError, match="not found"):
+                set_aicore_config()
+
+    def test_destination_mode_raises_when_no_client_credentials(self):
+        dest = self._mock_destination(properties={"resource_group": "default"})
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering"),
+            patch("sap_cloud_sdk.destination.create_client") as mock_create,
+            patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True),
+        ):
+            mock_create.return_value.get_destination.return_value = dest
+            with pytest.raises(RuntimeError, match="clientId/clientSecret"):
+                set_aicore_config()
+
+    def test_destination_mode_still_calls_set_filtering(self):
+        dest = self._mock_destination()
+        with (
+            patch("sap_cloud_sdk.aicore.set_filtering") as mock_filter,
+            patch("sap_cloud_sdk.destination.create_client") as mock_create,
+            patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True),
+        ):
+            mock_create.return_value.get_destination.return_value = dest
+            set_aicore_config()
+        mock_filter.assert_called_once()
diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py
index 9857f1d4..61d6bb8d 100644
--- a/tests/aicore/unit/test_completion.py
+++ b/tests/aicore/unit/test_completion.py
@@ -15,17 +15,26 @@
 - :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
+import os
+from unittest.mock import MagicMock, patch
 
+import litellm
 import pytest
 
 from sap_cloud_sdk.aicore import acompletion, completion
+from sap_cloud_sdk.aicore.completion import (
+    reload_aicore_credentials,
+    _clear_client_secret,
+    _reset_secret_cleared,
+)
 from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError
 
 
@@ -187,13 +196,252 @@ 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
+
+
+# ---------------------------------------------------------------------------
+# reload_aicore_credentials()
+# ---------------------------------------------------------------------------
+
+
+class TestReloadAICoreCredentials:
+    def test_calls_set_aicore_config(self):
+        with patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_config:
+            reload_aicore_credentials()
+        mock_config.assert_called_once_with()
+
+    def test_resets_secret_cleared_flag(self):
+        """After reload, _clear_client_secret() must be able to clear the secret again."""
+        # Simulate: secret was cleared once already
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "old"}):
+            _clear_client_secret()
+        # Flag is now True — a second clear would be a no-op
+        with patch("sap_cloud_sdk.aicore.set_aicore_config"):
+            reload_aicore_credentials()
+        # After reload the flag is reset — clearing works again
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "new"}):
+            _clear_client_secret()
+            assert "AICORE_CLIENT_SECRET" not in os.environ
+
+
+# ---------------------------------------------------------------------------
+# CLIENT_SECRET minimisation — _clear_client_secret()
+# ---------------------------------------------------------------------------
+
+
+class TestClearClientSecret:
+    def setup_method(self):
+        _reset_secret_cleared()
+
+    def test_removes_secret_from_env(self):
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}):
+            _clear_client_secret()
+            assert "AICORE_CLIENT_SECRET" not in os.environ
+
+    def test_noop_when_secret_absent(self):
+        env = {}
+        with patch.dict("os.environ", env, clear=True):
+            _clear_client_secret()  # must not raise
+
+    def test_idempotent_second_call_is_noop(self):
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}):
+            _clear_client_secret()
+            os.environ["AICORE_CLIENT_SECRET"] = "restored"
+            _clear_client_secret()
+            # second call must not remove the restored value
+            assert os.environ.get("AICORE_CLIENT_SECRET") == "restored"
+
+    def test_reset_allows_clear_again(self):
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}):
+            _clear_client_secret()
+            _reset_secret_cleared()
+            os.environ["AICORE_CLIENT_SECRET"] = "new-secret"
+            _clear_client_secret()
+            assert "AICORE_CLIENT_SECRET" not in os.environ
+
+
+# ---------------------------------------------------------------------------
+# completion() clears secret on success
+# ---------------------------------------------------------------------------
+
+
+class TestCompletionClearsSecret:
+    def setup_method(self):
+        _reset_secret_cleared()
+
+    def test_secret_cleared_after_successful_call(self):
+        sentinel = object()
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel),
+            patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}),
+        ):
+            result = completion(model="sap/x", messages=[])
+        assert result is sentinel
+        assert "AICORE_CLIENT_SECRET" not in os.environ
+
+    def test_secret_not_cleared_on_filter_error(self):
+        """Filter errors are not successful calls — secret stays until next success."""
+        from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError
+        raised = ContentFilteredError(direction="input", details={}, request_id="r")
+        secret_present_after = {}
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}):
+            with pytest.raises(ContentFilteredError):
+                with patch(
+                    "sap_cloud_sdk.aicore.completion.litellm.completion",
+                    side_effect=raised,
+                ):
+                    completion(model="sap/x", messages=[])
+            secret_present_after["value"] = os.environ.get("AICORE_CLIENT_SECRET")
+        assert secret_present_after["value"] == "s3cr3t"
+
+    def test_secret_cleared_after_auth_error_and_retry(self):
+        """After reload + successful retry, secret must be cleared."""
+        sentinel = object()
+        auth_err = litellm.AuthenticationError(
+            message="401", 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"),
+            patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}),
+        ):
+            result = completion(model="sap/x", messages=[])
+
+        assert result is sentinel
+        assert "AICORE_CLIENT_SECRET" not in os.environ
+
+
+# ---------------------------------------------------------------------------
+# 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=[]))
+
+