Blocking I/O in async completion and Starlette middleware can stall concurrent requests

diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py
index 9fd0f85..940569e 100644
--- a/src/sap_cloud_sdk/aicore/__init__.py
+++ b/src/sap_cloud_sdk/aicore/__init__.py
@@ -4,6 +4,7 @@ This module provides utilities to load AI Core credentials from mounted secrets
 variables and configure them for use with LiteLLM.
 """
 
+import asyncio
 import json
 import logging
 import os
@@ -430,7 +431,8 @@ def patch_litellm_for_credential_rotation() -> None:
         try:
             return await _orig_acompletion(*args, **kwargs)
         except _litellm.AuthenticationError:
-            _reload_reactive()
+            # Secret loading and Destination resolution have no async APIs.
+            await asyncio.to_thread(_reload_reactive)
             return await _orig_acompletion(*args, **kwargs)
 
     _litellm.completion = _completion
diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py
index d97dc63..2e30952 100644
--- a/src/sap_cloud_sdk/aicore/completion.py
+++ b/src/sap_cloud_sdk/aicore/completion.py
@@ -64,6 +64,7 @@ adoption of the SDK.
 
 from __future__ import annotations
 
+import asyncio
 import logging
 from collections.abc import AsyncIterator, Iterator
 from typing import Any
@@ -198,7 +199,8 @@ async def acompletion(*args: Any, **kwargs: Any) -> Any:
     try:
         result = await litellm.acompletion(*args, **kwargs)
     except litellm.AuthenticationError:
-        _reload_reactive()
+        # Secret loading and Destination resolution only expose synchronous APIs.
+        await asyncio.to_thread(_reload_reactive)
         result = await litellm.acompletion(*args, **kwargs)
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
diff --git a/src/sap_cloud_sdk/core/runtime_context/starlette.py b/src/sap_cloud_sdk/core/runtime_context/starlette.py
index eb0fcb0..b191b71 100644
--- a/src/sap_cloud_sdk/core/runtime_context/starlette.py
+++ b/src/sap_cloud_sdk/core/runtime_context/starlette.py
@@ -10,6 +10,7 @@ from sap_cloud_sdk.core.runtime_context._envelope import RequestEnvelope
 from sap_cloud_sdk.core.runtime_context._protocol import ContextProvider
 
 try:
+    from starlette.concurrency import run_in_threadpool
     from starlette.middleware.base import BaseHTTPMiddleware
     from starlette.requests import Request
     from starlette.responses import Response
@@ -47,6 +48,12 @@ class StarletteContextMiddleware(BaseHTTPMiddleware):
 
     async def dispatch(self, request: Request, call_next: Any) -> Response:
         envelope = RequestEnvelope(headers=dict(request.headers))
-        ctx = _merge([p.extract(envelope) for p in self._providers])
+        # The provider API is synchronous; IAS verification uses PyJWKClient,
+        # which can fetch JWKS over the network. Keep extraction in one worker
+        # call to preserve provider order without a handoff for each provider.
+        ctx = await run_in_threadpool(self._extract_context, envelope)
         async with async_sdk_context(ctx):
             return await call_next(request)
+
+    def _extract_context(self, envelope: RequestEnvelope) -> RuntimeContext:
+        return _merge([p.extract(envelope) for p in self._providers])
diff --git a/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py b/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py
index 14162ea..f7bf565 100644
--- a/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py
+++ b/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py
@@ -19,6 +19,7 @@ from sap_cloud_sdk.ias import set_auth_context
 from sap_cloud_sdk.ias._token import IASClaims
 
 try:
+    from starlette.concurrency import run_in_threadpool
     from starlette.middleware.base import BaseHTTPMiddleware
     from starlette.requests import Request
     from starlette.responses import Response
@@ -43,7 +44,10 @@ class _IASMiddleware(BaseHTTPMiddleware):
         self._token_verifier = token_verifier
 
     async def dispatch(self, request: Request, call_next: Any) -> Response:
-        verified_claims, attrs = _verify_and_extract(request, self._token_verifier)
+        # PyJWKClient has no async API and may fetch signing keys over the network.
+        verified_claims, attrs = await run_in_threadpool(
+            _verify_and_extract, request, self._token_verifier
+        )
         set_auth_context(verified_claims)
         token = self._attrs_var.set(attrs)
         try:
diff --git a/tests/core/unit/test_async_blocking.py b/tests/core/unit/test_async_blocking.py
new file mode 100644
index 0000000..54eb111
--- /dev/null
+++ b/tests/core/unit/test_async_blocking.py
@@ -0,0 +1,122 @@
+"""Blocking SDK dependencies must not stall unrelated async work."""
+
+import asyncio
+from contextvars import ContextVar
+import threading
+from unittest.mock import AsyncMock, MagicMock
+
+import litellm
+import pytest
+
+from sap_cloud_sdk import aicore
+from sap_cloud_sdk.core.runtime_context import IASContextProvider, get_context
+from sap_cloud_sdk.core.runtime_context.providers._ias import GLOBAL_TENANT_ID
+from sap_cloud_sdk.core.runtime_context.starlette import StarletteContextMiddleware
+from sap_cloud_sdk.core.telemetry.constants import ATTR_SAP_TENANT_ID
+from sap_cloud_sdk.core.telemetry.middleware.starlette_a2a import _IASMiddleware
+from sap_cloud_sdk.ias import IASClaims, VerifiedIASClaims, get_auth_context
+
+
+class BlockingDependency:
+    """Hold worker I/O until the event loop explicitly releases it."""
+
+    def __init__(self):
+        self.loop = asyncio.get_running_loop()
+        self.started = asyncio.Event()
+        self.release = threading.Event()
+
+    def __call__(self):
+        self.loop.call_soon_threadsafe(self.started.set)
+        if not self.release.wait(timeout=5):
+            raise RuntimeError("Event loop could not release blocking I/O")
+
+    async def run(self, operation):
+        task = asyncio.create_task(operation)
+        try:
+            await asyncio.wait_for(self.started.wait(), timeout=2)
+            # This point is reached while the synchronous dependency is blocked.
+            # A direct call on the event loop would prevent reaching it in time.
+            assert not task.done()
+        finally:
+            self.release.set()
+            await task
+        return task.result()
+
+
[email protected]
[email protected]("middleware_kind", ["telemetry", "runtime_context"])
+async def test_slow_verification_yields_and_preserves_downstream_context(
+    middleware_kind,
+):
+    blocker = BlockingDependency()
+    caller_value = ContextVar("caller_value", default=None)
+    caller_value.set("request-local")
+    claims = IASClaims(sap_gtid="tenant-1")
+
+    def verify(authorization):
+        assert authorization == "Bearer token"
+        assert caller_value.get() == "request-local"
+        blocker()
+        return VerifiedIASClaims(claims=claims)
+
+    request = MagicMock()
+    request.headers = {"authorization": "Bearer token"}
+    attrs = ContextVar("test_attrs", default={})
+    if middleware_kind == "telemetry":
+        middleware = _IASMiddleware(MagicMock(), attrs, verify)
+    else:
+        middleware = StarletteContextMiddleware(
+            MagicMock(), providers=[IASContextProvider(token_verifier=verify)]
+        )
+
+    response = object()
+
+    async def call_next(req):
+        assert req is request
+        assert caller_value.get() == "request-local"
+        if middleware_kind == "telemetry":
+            assert get_auth_context() is claims
+            assert attrs.get() == {ATTR_SAP_TENANT_ID: "tenant-1"}
+        else:
+            assert get_context().get(GLOBAL_TENANT_ID) == "tenant-1"
+        return response
+
+    assert await blocker.run(middleware.dispatch(request, call_next)) is response
+    assert attrs.get() == {}
+    assert get_context().get(GLOBAL_TENANT_ID) is None
+
+
[email protected]
[email protected]("patched", [False, True])
[email protected]("retry_fails", [False, True])
+async def test_slow_credential_reload_yields_and_retries_once(
+    monkeypatch, patched, retry_fails
+):
+    blocker = BlockingDependency()
+    error = litellm.AuthenticationError(
+        message="401", llm_provider="sap", model="sap/x"
+    )
+    response = object()
+    completion = AsyncMock(side_effect=[error, error if retry_fails else response])
+    reload_config = MagicMock(side_effect=blocker)
+    monkeypatch.setattr(litellm, "acompletion", completion)
+    monkeypatch.setattr(aicore, "set_aicore_config", reload_config)
+    if patched:
+        monkeypatch.setattr(litellm, "_sap_aicore_patched", False, raising=False)
+        # Restore both functions after the patch installs its wrappers.
+        monkeypatch.setattr(litellm, "completion", litellm.completion)
+        aicore.patch_litellm_for_credential_rotation()
+        call = litellm.acompletion
+    else:
+        call = aicore.acompletion
+
+    if retry_fails:
+        with pytest.raises(litellm.AuthenticationError) as caught:
+            await blocker.run(call(model="sap/x", messages=[]))
+        assert caught.value is error
+    else:
+        assert await blocker.run(call(model="sap/x", messages=[])) is response
+    reload_config.assert_called_once_with()
+    assert completion.await_count == 2
+    for invocation in completion.await_args_list:
+        assert invocation.kwargs == {"model": "sap/x", "messages": []}