diff --git a/python/packages/agentsts-adk/src/agentsts/adk/_base.py b/python/packages/agentsts-adk/src/agentsts/adk/_base.py
index abae6cee9..938cb5eb0 100644
--- a/python/packages/agentsts-adk/src/agentsts/adk/_base.py
+++ b/python/packages/agentsts-adk/src/agentsts/adk/_base.py
@@ -1,5 +1,6 @@
 """Google ADK-specific STS integration."""
 
+import hashlib
 import inspect
 import logging
 import time
@@ -31,11 +32,50 @@
 
 HEADERS_KEY = "headers"
 
+# Bounds an entry whose token carries no usable expiry. The cache holds one entry
+# per (session, subject), so a token without an expiry would otherwise pin an
+# entry per caller for the lifetime of the process.
+MAX_CACHE_TTL_SECONDS = 300
+
+
+def _acting_credential(state: dict) -> Optional[str]:
+    """Return the credential this caller presented, which the cache is keyed on.
+
+    Deliberately not ``get_subject_token``: that hook receives the whole session
+    state, so an implementation reading a session-scoped field would return one
+    value for every caller and collapse the key back to a single entry per
+    session. Only the inbound Authorization header is caller-scoped by
+    construction.
+    """
+    headers = state.get(HEADERS_KEY, None)
+    if not isinstance(headers, dict):
+        return None
+    return _extract_jwt_from_headers(headers, warn=False)
+
 
 def _default_get_subject_token(state: dict) -> Optional[str]:
     """Default subject token retrieval from Authorization header in session state."""
-    headers = state.get(HEADERS_KEY, None)
-    return _extract_jwt_from_headers(headers)
+    return _acting_credential(state)
+
+
+def _subject_key(token: Optional[str]) -> str:
+    """Derive a per-principal cache discriminator from a bearer token: a hash of
+    the raw token.
+
+    A cache hit hands the caller a delegated token without performing an
+    exchange, so the key decides who receives someone else's authority. Deriving
+    it from unverified ``iss``/``sub`` claims would let a forged, unsigned token
+    select a victim's entry and never reach the STS that would have rejected it.
+    Hashing the raw token instead makes a forged token a cache miss, so it goes
+    to the STS and fails there.
+
+    The cost is a re-exchange when a principal's bearer rotates mid-session,
+    which is wanted anyway: the delegated token's lifetime tracks the subject
+    token it was exchanged from.
+    """
+    if not token:
+        return ""
+    return hashlib.sha256(token.encode()).hexdigest()
 
 
 class ADKSTSIntegration(STSIntegrationBase):
@@ -125,6 +165,8 @@ def __init__(
         self.audience = audience
         self.token_cache: Dict[str, _TokenCacheEntry] = {}
         self.actor_token_cache: Optional[_TokenCacheEntry] = None
+        # Earliest expiry across token_cache; None when no cached token expires.
+        self._earliest_expiry: Optional[int] = None
 
     def add_to_agent(self, agent: BaseAgent):
         """
@@ -149,9 +191,17 @@ def add_to_agent(self, agent: BaseAgent):
                 logger.debug(f"add_to_agent: updated MCP tool's header provider for agent {agent_name}")
 
     def header_provider(self, readonly_context: Optional[ReadonlyContext]) -> Dict[str, str]:
-        # access saved token
-        cache_entry = self.token_cache.get(self.cache_key(readonly_context._invocation_context))
+        # Runs on every tool call, so it fails closed rather than raising into the
+        # invocation: without a context there is no acting subject to key on.
+        invocation_context = getattr(readonly_context, "_invocation_context", None)
+        if invocation_context is None:
+            logger.debug("no invocation context for tool call, leaving existing headers in place")
+            return {}
+
+        cache_key = self.cache_key(invocation_context)
+        cache_entry = self.token_cache.get(cache_key) if cache_key else None
         if not cache_entry:
+            logger.debug("no cached access token for this caller, leaving existing headers in place")
             return {}
 
         logger.debug("Using cached access token for tool invocation")
@@ -166,7 +216,18 @@ async def before_run_callback(
         invocation_context: InvocationContext,
     ) -> Optional[dict]:
         """Propagate token to model before execution."""
-        cache_key = self.cache_key(invocation_context)
+        # Resolve the acting caller before the cache lookup: the cache is keyed by
+        # subject, and a session carrying messages from several subjects would
+        # otherwise reuse whichever caller arrived first.
+        state = invocation_context.session.state
+        credential = _acting_credential(state)
+        # The exchange payload comes from get_subject_token, which on its own is
+        # not caller-scoped: see _acting_credential.
+        subject_token = self._read_subject_token(state)
+        cache_key = self._cache_key_for(invocation_context.session.id, credential, subject_token)
+        if cache_key is None:
+            logger.debug("subject token not found in session state for token propagation")
+            return None
 
         # Check if we have a valid cached subject token
         cached_entry = self.token_cache.get(cache_key)
@@ -178,17 +239,6 @@ async def before_run_callback(
                 logger.debug("Using cached subject token (no expiry)")
             return None
 
-        # No valid cached token, need to get/exchange subject token
-        get_subject_token = (
-            self.sts_integration.get_subject_token
-            if self.sts_integration and self.sts_integration.get_subject_token
-            else _default_get_subject_token
-        )
-        subject_token = get_subject_token(invocation_context.session.state)
-        if not subject_token:
-            logger.debug("subject token not found in session state for token propagation")
-            return None
-
         if self.sts_integration:
             # Get actor token (from cache or fetch dynamically)
             actor_token = await self._get_actor_token()
@@ -209,20 +259,87 @@ async def before_run_callback(
                 logger.warning(f"STS token exchange failed: {e}")
                 return None
 
-        # Extract expiry from the token
+        # Extract expiry from the token, bounding tokens that carry none so every
+        # entry stays evictable.
         expiry = _extract_jwt_expiry(subject_token)
+        if expiry is None:
+            expiry = int(time.time()) + MAX_CACHE_TTL_SECONDS
+        # The entry is keyed by the caller's credential, so it must not outlive
+        # it: replaying an expired bearer would otherwise keep hitting a cached
+        # delegated token instead of reaching the STS.
+        expiry = _earlier_expiry(expiry, _extract_jwt_expiry(credential))
 
         # Cache the token with metadata
         self.token_cache[cache_key] = _TokenCacheEntry(
             token=subject_token,
             expiry=expiry,
         )
+        self._earliest_expiry = _earlier_expiry(self._earliest_expiry, expiry)
         logger.debug("Cached new subject token")
         return None
 
-    def cache_key(self, invocation_context: InvocationContext) -> str:
-        """Generate a cache key based on the session ID."""
-        return invocation_context.session.id
+    def _read_subject_token(self, state: dict) -> Optional[str]:
+        """Resolve the acting caller's subject token from session state.
+
+        get_subject_token is caller-supplied, so a raising implementation fails
+        closed (no token propagated) instead of aborting the agent run.
+        """
+        get_subject_token = (
+            self.sts_integration.get_subject_token
+            if self.sts_integration and self.sts_integration.get_subject_token
+            else _default_get_subject_token
+        )
+        try:
+            return get_subject_token(state)
+        except Exception as e:
+            logger.warning(f"Failed to read subject token from session state: {e}")
+            return None
+
+    def _cache_key_for(
+        self,
+        session_id: str,
+        credential: Optional[str],
+        subject_token: Optional[str],
+    ) -> Optional[str]:
+        """Build the cache key for one caller in one session.
+
+        The key names the caller's own credential, so an entry can never be
+        handed to a caller other than the one it was minted for, whatever
+        get_subject_token returns.
+
+        Without an inbound credential there is nothing caller-scoped to key on,
+        so the hook's output stands in for it: that mode has no per-caller
+        identity to preserve, and one entry per session is correct for it.
+        """
+        if not session_id:
+            # Nothing identifies the conversation, so entries could only be
+            # shared between unrelated ones.
+            return None
+        if not subject_token:
+            # No token to exchange, so there is nothing to cache.
+            return None
+        caller = _subject_key(credential) or _subject_key(subject_token)
+        if not caller:
+            # An empty subject identifies no principal: an entry stored under it
+            # would be shared by every credential-less caller in the session.
+            return None
+        return f"{session_id}\0{caller}"
+
+    def cache_key(self, invocation_context: InvocationContext) -> Optional[str]:
+        """Key the cache on the session and the acting subject, so a session
+        carrying messages from several subjects keeps one token per subject
+        instead of collapsing onto whichever arrived first.
+
+        The caller's credential is read straight from the run's own state, so
+        the key is derivable on every tool call. get_subject_token is consulted
+        only when no inbound credential identifies the caller.
+        """
+        session = getattr(invocation_context, "session", None)
+        if session is None:
+            return None
+        credential = _acting_credential(session.state)
+        subject_token = credential or self._read_subject_token(session.state)
+        return self._cache_key_for(session.id, credential, subject_token)
 
     async def _get_actor_token(self) -> Optional[str]:
         """Get actor token from cache or fetch dynamically.
@@ -276,13 +393,7 @@ async def after_run_callback(
         invocation_context: InvocationContext,
     ) -> Optional[dict]:
         """Clean up expired tokens after run, preserving valid tokens."""
-        cache_key = self.cache_key(invocation_context)
-        cache_entry = self.token_cache.get(cache_key)
-
-        # Clean up subject token cache - only remove if expired
-        if cache_entry and _has_token_expired(cache_entry.expiry):
-            logger.debug("Removing expired subject token from cache")
-            self.token_cache.pop(cache_key, None)
+        self._sweep_expired_subject_tokens()
 
         # Clean up expired actor token cache
         if self.actor_token_cache and _has_token_expired(self.actor_token_cache.expiry):
@@ -291,6 +402,37 @@ async def after_run_callback(
 
         return None
 
+    def _sweep_expired_subject_tokens(self) -> None:
+        """Drop every expired subject token from the cache.
+
+        A session holds one entry per subject and only the acting subject's key
+        is derivable here, so entries belonging to other subjects and other
+        sessions are swept too; scoping the sweep to the current session would
+        keep the entries of sessions that never run again forever. The earliest
+        expiry gates the scan, so a growing cache is only walked when there is
+        something to evict.
+        """
+        if self._earliest_expiry is None or not _has_token_expired(self._earliest_expiry):
+            return
+
+        earliest_expiry: Optional[int] = None
+        for key, entry in list(self.token_cache.items()):
+            if _has_token_expired(entry.expiry):
+                logger.debug("Removing expired subject token from cache")
+                self.token_cache.pop(key, None)
+                continue
+            earliest_expiry = _earlier_expiry(earliest_expiry, entry.expiry)
+        self._earliest_expiry = earliest_expiry
+
+
+def _earlier_expiry(current: Optional[int], candidate: Optional[int]) -> Optional[int]:
+    """Return the earlier of two expiries; None means "does not expire"."""
+    if candidate is None:
+        return current
+    if current is None:
+        return candidate
+    return min(current, candidate)
+
 
 def _has_token_expired(expiry: Optional[int], buffer_seconds: int = 5) -> bool:
     """Check if a token has expired or will expire soon.
@@ -311,31 +453,38 @@ def _has_token_expired(expiry: Optional[int], buffer_seconds: int = 5) -> bool:
     return expiry <= (current_time + buffer_seconds)
 
 
-def _extract_jwt_from_headers(headers: dict[str, str]) -> Optional[str]:
+def _extract_jwt_from_headers(headers: dict[str, str], warn: bool = True) -> Optional[str]:
     """Extract JWT from request headers for STS token exchange.
 
     Args:
         headers: Dictionary of request headers
+        warn: Whether a missing or malformed header is worth a warning. The
+            cache-key probe reads the same header as the subject-token lookup,
+            so only one of the two reports it.
 
     Returns:
         JWT token string if found in Authorization header, None otherwise
     """
     if not headers:
-        logger.warning("No headers provided for JWT extraction")
+        if warn:
+            logger.warning("No headers provided for JWT extraction")
         return None
 
     auth_header = headers.get("Authorization") or headers.get("authorization")
     if not auth_header:
-        logger.warning("No Authorization header found in request")
+        if warn:
+            logger.warning("No Authorization header found in request")
         return None
 
     if not auth_header.startswith("Bearer "):
-        logger.warning("Authorization header must start with Bearer")
+        if warn:
+            logger.warning("Authorization header must start with Bearer")
         return None
 
     jwt_token = auth_header.removeprefix("Bearer ").strip()
     if not jwt_token:
-        logger.warning("Empty JWT token found in Authorization header")
+        if warn:
+            logger.warning("Empty JWT token found in Authorization header")
         return None
 
     logger.debug(f"Successfully extracted JWT token (length: {len(jwt_token)})")
diff --git a/python/packages/agentsts-adk/tests/test_adk_integration.py b/python/packages/agentsts-adk/tests/test_adk_integration.py
index f6802384a..deced6c82 100644
--- a/python/packages/agentsts-adk/tests/test_adk_integration.py
+++ b/python/packages/agentsts-adk/tests/test_adk_integration.py
@@ -1,5 +1,6 @@
 """Tests for ADK integration classes (STS + token propagation)."""
 
+import time
 from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
@@ -9,10 +10,33 @@
 from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
 
 from agentsts.adk import ADKSTSIntegration, ADKTokenPropagationPlugin
-from agentsts.adk._base import HEADERS_KEY
+from agentsts.adk._base import HEADERS_KEY, MAX_CACHE_TTL_SECONDS
 from agentsts.adk._base import _extract_jwt_expiry as extract_jwt_expiry
 from agentsts.adk._base import _extract_jwt_from_headers as extract_jwt_from_headers
 from agentsts.adk._base import _has_token_expired as has_token_expired
+from agentsts.adk._base import _subject_key as subject_key
+
+
+class _ScanCountingDict(dict):
+    """A dict that counts full traversals, however they are spelled."""
+
+    scans = 0
+
+    def items(self):
+        self.scans += 1
+        return super().items()
+
+    def keys(self):
+        self.scans += 1
+        return super().keys()
+
+    def values(self):
+        self.scans += 1
+        return super().values()
+
+    def __iter__(self):
+        self.scans += 1
+        return super().__iter__()
 
 
 class TestADKTokenPropagationPlugin:
@@ -78,8 +102,8 @@ async def test_subject_token_from_callback(self):
             resource=None,
             audience=None,
         )
-        assert "sess-key-1" in plugin.token_cache
-        assert plugin.token_cache["sess-key-1"].token == "exchanged-token"
+        assert plugin.cache_key(ic) in plugin.token_cache
+        assert plugin.token_cache[plugin.cache_key(ic)].token == "exchanged-token"
 
     @pytest.mark.asyncio
     async def test_resource_and_audience_passed_to_exchange(self):
@@ -181,7 +205,7 @@ async def test_default_callback_extracts_from_headers(self):
             resource=None,
             audience=None,
         )
-        assert plugin.token_cache["sess-key-4"].token == "exchanged-via-headers"
+        assert plugin.token_cache[plugin.cache_key(ic)].token == "exchanged-via-headers"
 
     @pytest.mark.asyncio
     async def test_downstream_token_propagation_without_sts(self):
@@ -190,8 +214,8 @@ async def test_downstream_token_propagation_without_sts(self):
         ic = self._make_invocation_context("sess-2", headers={"Authorization": "Bearer subj-token-123"})
         result = await plugin.before_run_callback(invocation_context=ic)
         assert result is None
-        assert "sess-2" in plugin.token_cache
-        assert plugin.token_cache["sess-2"].token == "subj-token-123"
+        assert plugin.cache_key(ic) in plugin.token_cache
+        assert plugin.token_cache[plugin.cache_key(ic)].token == "subj-token-123"
 
         # propagate toolset
         mcp_toolset = Mock(spec=MCPToolset)
@@ -209,7 +233,7 @@ async def test_downstream_token_propagation_without_sts(self):
         # cleanup - token should still be cached if not expired
         await plugin.after_run_callback(invocation_context=ic)
         # Token has no expiry, so it's preserved
-        assert "sess-2" in plugin.token_cache
+        assert plugin.cache_key(ic) in plugin.token_cache
 
     @pytest.mark.asyncio
     async def test_sts_token_exchange_success(self):
@@ -234,8 +258,8 @@ async def test_sts_token_exchange_success(self):
             )
             # optional debug log length check
             mock_logger.debug.assert_called()  # at least one debug log
-        assert "sess-3" in plugin.token_cache
-        assert plugin.token_cache["sess-3"].token == "access-token-XYZ"
+        assert plugin.cache_key(ic) in plugin.token_cache
+        assert plugin.token_cache[plugin.cache_key(ic)].token == "access-token-XYZ"
 
         ro_ctx = self._make_readonly_context(ic)
         headers = plugin.header_provider(ro_ctx)
@@ -244,7 +268,7 @@ async def test_sts_token_exchange_success(self):
         # cleanup - token should still be cached if not expired
         await plugin.after_run_callback(invocation_context=ic)
         # Token has no expiry, so it's preserved
-        assert "sess-3" in plugin.token_cache
+        assert plugin.cache_key(ic) in plugin.token_cache
 
     @pytest.mark.asyncio
     async def test_sts_token_exchange_failure(self):
@@ -260,7 +284,7 @@ async def test_sts_token_exchange_failure(self):
             result = await plugin.before_run_callback(invocation_context=ic)
             assert result is None
             mock_logger.warning.assert_called_once()
-        assert "sess-4" not in plugin.token_cache
+        assert plugin.cache_key(ic) not in plugin.token_cache
         # header provider should yield empty dict
         ro_ctx = self._make_readonly_context(ic)
         assert plugin.header_provider(ro_ctx) == {}
@@ -286,11 +310,11 @@ async def test_after_run_callback_removes_expired_token(self):
         # Mock expiry to return expired timestamp
         with patch("agentsts.adk._base._extract_jwt_expiry", return_value=past_expiry):
             await plugin.before_run_callback(invocation_context=ic)
-            assert "sess-6" in plugin.token_cache
+            assert plugin.cache_key(ic) in plugin.token_cache
 
         # Token is expired, should be removed
         await plugin.after_run_callback(invocation_context=ic)
-        assert "sess-6" not in plugin.token_cache
+        assert plugin.cache_key(ic) not in plugin.token_cache
 
     @pytest.mark.asyncio
     async def test_dynamic_token_fetch_success_sync(self):
@@ -327,8 +351,8 @@ async def test_dynamic_token_fetch_success_sync(self):
             assert any("Fetched and cached new actor token" in call for call in debug_calls)
 
         # Verify token is cached
-        assert "sess-7" in plugin.token_cache
-        cache_entry = plugin.token_cache["sess-7"]
+        assert plugin.cache_key(ic) in plugin.token_cache
+        cache_entry = plugin.token_cache[plugin.cache_key(ic)]
         assert cache_entry.token == "access-token-dynamic"
 
     @pytest.mark.asyncio
@@ -366,8 +390,8 @@ async def async_fetch_token():
             assert any("Fetched and cached new actor token" in call for call in debug_calls)
 
         # Verify token is cached
-        assert "sess-7a" in plugin.token_cache
-        cache_entry = plugin.token_cache["sess-7a"]
+        assert plugin.cache_key(ic) in plugin.token_cache
+        cache_entry = plugin.token_cache[plugin.cache_key(ic)]
         assert cache_entry.token == "access-token-dynamic-async"
 
     @pytest.mark.asyncio
@@ -395,7 +419,7 @@ async def test_dynamic_token_fetch_failure_sync(self):
             assert "Failed to fetch actor token dynamically" in warning_msg
 
         # No token should be cached
-        assert "sess-8" not in plugin.token_cache
+        assert plugin.cache_key(ic) not in plugin.token_cache
 
     @pytest.mark.asyncio
     async def test_dynamic_token_fetch_failure_async(self):
@@ -422,7 +446,7 @@ async def async_fetch_token_failing():
             assert "Failed to fetch actor token dynamically" in warning_msg
 
         # No token should be cached
-        assert "sess-8a" not in plugin.token_cache
+        assert plugin.cache_key(ic) not in plugin.token_cache
 
     @pytest.mark.asyncio
     async def test_dynamic_token_preserved_when_not_expired(self):
@@ -446,15 +470,15 @@ async def test_dynamic_token_preserved_when_not_expired(self):
             await plugin.before_run_callback(invocation_context=ic)
 
         # Verify token is cached with expiry
-        assert "sess-9" in plugin.token_cache
-        cache_entry = plugin.token_cache["sess-9"]
+        assert plugin.cache_key(ic) in plugin.token_cache
+        cache_entry = plugin.token_cache[plugin.cache_key(ic)]
         assert cache_entry.expiry == future_expiry
 
         # Call after_run_callback - token should be preserved
         await plugin.after_run_callback(invocation_context=ic)
 
         # Verify token is still cached (not expired)
-        assert "sess-9" in plugin.token_cache
+        assert plugin.cache_key(ic) in plugin.token_cache
 
     @pytest.mark.asyncio
     async def test_dynamic_token_removed_when_expired(self):
@@ -478,13 +502,13 @@ async def test_dynamic_token_removed_when_expired(self):
             await plugin.before_run_callback(invocation_context=ic)
 
         # Verify token is cached
-        assert "sess-10" in plugin.token_cache
+        assert plugin.cache_key(ic) in plugin.token_cache
 
         # Call after_run_callback - token should be removed (expired)
         await plugin.after_run_callback(invocation_context=ic)
 
         # Verify token is removed
-        assert "sess-10" not in plugin.token_cache
+        assert plugin.cache_key(ic) not in plugin.token_cache
 
     @pytest.mark.asyncio
     async def test_valid_token_preserved_in_cache(self):
@@ -501,15 +525,15 @@ async def test_valid_token_preserved_in_cache(self):
         await plugin.before_run_callback(invocation_context=ic)
 
         # Verify token is cached with expected value
-        assert "sess-11" in plugin.token_cache
-        cache_entry = plugin.token_cache["sess-11"]
+        assert plugin.cache_key(ic) in plugin.token_cache
+        cache_entry = plugin.token_cache[plugin.cache_key(ic)]
         assert cache_entry.token == "access-token-static"
 
         # Call after_run_callback - token should still be in cache if not expired
         await plugin.after_run_callback(invocation_context=ic)
 
         # Verify the same cache entry is still present
-        assert plugin.token_cache.get("sess-11") is cache_entry
+        assert plugin.token_cache.get(plugin.cache_key(ic)) is cache_entry
 
     @pytest.mark.asyncio
     async def test_actor_token_cached_and_reused(self):
@@ -717,8 +741,8 @@ async def test_subject_token_cached_and_reused(self):
             assert sts.exchange_token.call_count == 1
 
             # Verify token is cached
-            assert "sess-18" in plugin.token_cache
-            assert plugin.token_cache["sess-18"].token == "exchanged-token"
+            assert plugin.cache_key(ic) in plugin.token_cache
+            assert plugin.token_cache[plugin.cache_key(ic)].token == "exchanged-token"
 
             # Second call with same session - should use cached token
             await plugin.before_run_callback(invocation_context=ic)
@@ -747,12 +771,12 @@ async def test_subject_token_reexchanged_after_expiry(self):
         with patch("agentsts.adk._base._extract_jwt_expiry", return_value=past_expiry):
             await plugin.before_run_callback(invocation_context=ic)
             assert sts.exchange_token.call_count == 1
-            assert plugin.token_cache["sess-19"].token == "token-1"
-            assert plugin.token_cache["sess-19"].expiry == past_expiry
+            assert plugin.token_cache[plugin.cache_key(ic)].token == "token-1"
+            assert plugin.token_cache[plugin.cache_key(ic)].expiry == past_expiry
 
             # Cleanup expired token
             await plugin.after_run_callback(invocation_context=ic)
-            assert "sess-19" not in plugin.token_cache
+            assert plugin.cache_key(ic) not in plugin.token_cache
 
         # Second call - should detect missing cache and re-exchange
         with patch("agentsts.adk._base._extract_jwt_expiry", return_value=future_expiry):
@@ -760,12 +784,16 @@ async def test_subject_token_reexchanged_after_expiry(self):
 
             # Verify exchange was called again
             assert sts.exchange_token.call_count == 2
-            assert plugin.token_cache["sess-19"].token == "token-2"
-            assert plugin.token_cache["sess-19"].expiry == future_expiry
+            assert plugin.token_cache[plugin.cache_key(ic)].token == "token-2"
+            assert plugin.token_cache[plugin.cache_key(ic)].expiry == future_expiry
 
     @pytest.mark.asyncio
     async def test_subject_token_cache_no_expiry(self):
-        """Case: subject token without expiry is cached indefinitely and reused."""
+        """Case: subject token without expiry is given a bounded lifetime and reused.
+
+        The cache holds one entry per (session, subject), so an entry that never
+        expires would pin a slot per caller for the lifetime of the process.
+        """
         sts = Mock(spec=ADKSTSIntegration)
         sts.get_subject_token = None
         sts.fetch_actor_token = None
@@ -780,11 +808,13 @@ async def test_subject_token_cache_no_expiry(self):
             # First call
             await plugin.before_run_callback(invocation_context=ic)
             assert sts.exchange_token.call_count == 1
-            assert plugin.token_cache["sess-20"].expiry is None
+            expiry = plugin.token_cache[plugin.cache_key(ic)].expiry
+            assert expiry is not None
+            assert expiry <= int(time.time()) + MAX_CACHE_TTL_SECONDS
 
-            # after_run_callback should preserve it (no expiry)
+            # after_run_callback preserves it until the bounded lifetime elapses
             await plugin.after_run_callback(invocation_context=ic)
-            assert "sess-20" in plugin.token_cache
+            assert plugin.cache_key(ic) in plugin.token_cache
 
             # Second call - should reuse cached token
             await plugin.before_run_callback(invocation_context=ic)
@@ -829,6 +859,69 @@ async def async_fetch_token():
             # Fetch should not be called again
             assert fetch_count == 1
 
+    @pytest.mark.asyncio
+    async def test_subject_token_callback_failure_does_not_abort_run(self):
+        """Case: a raising get_subject_token fails closed instead of breaking the run."""
+
+        def raising_get_subject_token(state):
+            raise RuntimeError("callback exploded")
+
+        sts = Mock(spec=ADKSTSIntegration)
+        sts.get_subject_token = raising_get_subject_token
+        sts.fetch_actor_token = None
+        sts._actor_token = "static-actor"
+        sts.exchange_token = AsyncMock()
+
+        plugin = ADKTokenPropagationPlugin(sts)
+        ic = self._make_invocation_context("sess-raise", headers={"Authorization": "Bearer subject-token"})
+
+        assert await plugin.before_run_callback(invocation_context=ic) is None
+        assert plugin.token_cache == {}
+        sts.exchange_token.assert_not_awaited()
+        assert plugin.header_provider(self._make_readonly_context(ic)) == {}
+
+    @pytest.mark.asyncio
+    async def test_after_run_callback_sweeps_expired_tokens_of_other_sessions(self):
+        """Case: the sweep is not scoped to the running session, so entries of
+        sessions that never run again cannot pile up."""
+        past_expiry = int(time.time()) - 100
+        future_expiry = int(time.time()) + 3600
+
+        plugin = ADKTokenPropagationPlugin()
+        expired_ic = self._make_invocation_context("sess-expired", headers={"Authorization": "Bearer AAA"})
+        live_ic = self._make_invocation_context("sess-live", headers={"Authorization": "Bearer BBB"})
+        running_ic = self._make_invocation_context("sess-running", headers={"Authorization": "Bearer CCC"})
+
+        with patch("agentsts.adk._base._extract_jwt_expiry", return_value=past_expiry):
+            await plugin.before_run_callback(invocation_context=expired_ic)
+        with patch("agentsts.adk._base._extract_jwt_expiry", return_value=future_expiry):
+            await plugin.before_run_callback(invocation_context=live_ic)
+            await plugin.before_run_callback(invocation_context=running_ic)
+
+        await plugin.after_run_callback(invocation_context=running_ic)
+
+        assert plugin.cache_key(expired_ic) not in plugin.token_cache
+        assert plugin.cache_key(live_ic) in plugin.token_cache
+        assert plugin.cache_key(running_ic) in plugin.token_cache
+
+    @pytest.mark.asyncio
+    async def test_after_run_callback_skips_scan_until_something_expires(self):
+        """Case: the cache is only walked once the earliest expiry is reached."""
+        plugin = ADKTokenPropagationPlugin()
+        ic = self._make_invocation_context("sess-gate", headers={"Authorization": "Bearer AAA"})
+
+        with patch("agentsts.adk._base._extract_jwt_expiry", return_value=int(time.time()) + 3600):
+            await plugin.before_run_callback(invocation_context=ic)
+
+        # Counts every way the cache can be walked, so the assertion still holds
+        # if the sweep stops going through items().
+        counting_cache = _ScanCountingDict(plugin.token_cache)
+        plugin.token_cache = counting_cache
+        await plugin.after_run_callback(invocation_context=ic)
+
+        assert counting_cache.scans == 0
+        assert plugin.cache_key(ic) in plugin.token_cache
+
     def test_extract_jwt_from_headers_success(self):
         """Test successful JWT extraction from headers."""
         headers = {"Authorization": "Bearer jwt-token-123"}
@@ -1040,6 +1133,263 @@ def test_expiry_end_to_end_with_real_jwt(self):
         assert extract_jwt_expiry(no_exp_token) is None
         assert has_token_expired(extract_jwt_expiry(no_exp_token)) is False
 
+    @staticmethod
+    def _jwt(
+        iss: str,
+        sub: str,
+        signing_key: str = "test-signing-key-of-at-least-32-bytes!",
+        expiry: int | None = None,
+    ) -> str:
+        import jwt as pyjwt
+
+        claims = {"iss": iss, "sub": sub}
+        if expiry is not None:
+            claims["exp"] = expiry
+        return pyjwt.encode(claims, signing_key, algorithm="HS256")
+
+    @pytest.mark.asyncio
+    async def test_two_subjects_in_one_session_keep_separate_tokens(self):
+        """Case: one session, two callers -> each keeps and reuses its own exchanged token."""
+        alice = self._jwt("https://dex.example", "alice")
+        bob = self._jwt("https://dex.example", "bob")
+
+        sts = Mock(spec=ADKSTSIntegration)
+        sts.get_subject_token = None
+        sts.fetch_actor_token = None
+        sts._actor_token = "actor-token"
+        sts.exchange_token = AsyncMock(side_effect=lambda subject_token, **_: f"exchanged-for-{subject_token[-5:]}")
+        plugin = ADKTokenPropagationPlugin(sts)
+
+        ic_alice = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {alice}"})
+        ic_bob = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {bob}"})
+
+        await plugin.before_run_callback(invocation_context=ic_alice)
+        await plugin.before_run_callback(invocation_context=ic_bob)
+
+        # Both callers share a session id, so a session-only key would have
+        # collapsed them onto whichever exchanged first.
+        assert plugin.cache_key(ic_alice) != plugin.cache_key(ic_bob)
+        assert len(plugin.token_cache) == 2
+        assert sts.exchange_token.await_count == 2
+
+        # Each caller's tool invocation gets its own token back.
+        alice_headers = plugin.header_provider(self._make_readonly_context(ic_alice))
+        bob_headers = plugin.header_provider(self._make_readonly_context(ic_bob))
+        assert alice_headers["Authorization"] == f"Bearer {plugin.token_cache[plugin.cache_key(ic_alice)].token}"
+        assert alice_headers != bob_headers
+
+    @pytest.mark.asyncio
+    async def test_same_subject_reuses_cached_token(self):
+        """Case: same caller twice in one session -> exchanged once, second call is a cache hit."""
+        alice = self._jwt("https://dex.example", "alice")
+
+        sts = Mock(spec=ADKSTSIntegration)
+        sts.get_subject_token = None
+        sts.fetch_actor_token = None
+        sts._actor_token = "actor-token"
+        sts.exchange_token = AsyncMock(return_value="exchanged-once")
+        plugin = ADKTokenPropagationPlugin(sts)
+
+        ic = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {alice}"})
+        await plugin.before_run_callback(invocation_context=ic)
+        await plugin.before_run_callback(invocation_context=ic)
+
+        assert sts.exchange_token.await_count == 1
+        assert len(plugin.token_cache) == 1
+
+    def test_subject_key_ignores_unverified_claims(self):
+        """Case: a cache hit skips the exchange, so a token merely claiming the
+        victim's iss/sub must not select the victim's entry."""
+        genuine = self._jwt("https://dex.example", "alice", signing_key="genuine-signing-key-of-32-bytes-plus")
+        forged = self._jwt("https://dex.example", "alice", signing_key="attacker-signing-key-of-32-bytes-ok")
+
+        assert subject_key(genuine) != subject_key(forged)
+
+    def test_subject_key_partitions_opaque_tokens(self):
+        """Case: distinct credentials partition, and no credential yields no key."""
+        assert subject_key("opaque-a") != subject_key("opaque-b")
+        assert subject_key("opaque-a") == subject_key("opaque-a")
+        assert subject_key(None) == ""
+
+    @pytest.mark.asyncio
+    async def test_forged_subject_claims_do_not_reuse_cached_token(self):
+        """Case: the forged-token path end to end -> no cached entry is handed back."""
+        genuine = self._jwt("https://dex.example", "alice", signing_key="genuine-signing-key-of-32-bytes-plus")
+        forged = self._jwt("https://dex.example", "alice", signing_key="attacker-signing-key-of-32-bytes-ok")
+
+        plugin = ADKTokenPropagationPlugin(sts_integration=None)
+        ic_genuine = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {genuine}"})
+        await plugin.before_run_callback(invocation_context=ic_genuine)
+        assert len(plugin.token_cache) == 1
+
+        ic_forged = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {forged}"})
+        assert plugin.header_provider(self._make_readonly_context(ic_forged)) == {}
+
+    @pytest.mark.asyncio
+    async def test_token_without_expiry_stays_evictable(self):
+        """Case: a token carrying no exp is given a bounded lifetime, not an immortal entry."""
+        plugin = ADKTokenPropagationPlugin(sts_integration=None)
+        ic = self._make_invocation_context("sess-ttl", headers={"Authorization": "Bearer opaque-token"})
+
+        with patch("agentsts.adk._base._extract_jwt_expiry", return_value=None):
+            await plugin.before_run_callback(invocation_context=ic)
+
+        entry = plugin.token_cache[plugin.cache_key(ic)]
+        assert entry.expiry is not None
+        assert entry.expiry <= int(time.time()) + MAX_CACHE_TTL_SECONDS
+        # The bounded expiry also arms the sweep gate, which a None expiry leaves unset.
+        assert plugin._earliest_expiry == entry.expiry
+
+    def test_header_provider_without_context_fails_closed(self):
+        """Case: a tool call with no invocation context gets no header instead of raising."""
+        plugin = ADKTokenPropagationPlugin(sts_integration=None)
+
+        assert plugin.header_provider(None) == {}
+        assert plugin.header_provider(self._make_readonly_context(None)) == {}
+
+    @pytest.mark.asyncio
+    async def test_after_run_sweeps_other_subjects_expired_entries(self):
+        """Case: expired entry for a subject other than the acting one is still evicted."""
+        alice = self._jwt("https://dex.example", "alice")
+        bob = self._jwt("https://dex.example", "bob")
+
+        plugin = ADKTokenPropagationPlugin(sts_integration=None)
+        ic_alice = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {alice}"})
+        ic_bob = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {bob}"})
+
+        past_expiry = int(time.time()) - 100
+        with patch("agentsts.adk._base._extract_jwt_expiry", return_value=past_expiry):
+            await plugin.before_run_callback(invocation_context=ic_alice)
+        assert plugin.cache_key(ic_alice) in plugin.token_cache
+
+        # bob runs; alice's entry has expired and must not survive the sweep.
+        await plugin.after_run_callback(invocation_context=ic_bob)
+        assert plugin.cache_key(ic_alice) not in plugin.token_cache
+
+    @pytest.mark.asyncio
+    async def test_tokenless_caller_does_not_reuse_cached_session_token(self):
+        """Case: session holds a cached token, but a caller with no subject token gets nothing back."""
+        alice = self._jwt("https://dex.example", "alice")
+
+        plugin = ADKTokenPropagationPlugin(sts_integration=None)
+        ic_alice = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {alice}"})
+        await plugin.before_run_callback(invocation_context=ic_alice)
+        assert len(plugin.token_cache) == 1
+
+        ic_anon = self._make_invocation_context("shared-sess", headers=None)
+        await plugin.before_run_callback(invocation_context=ic_anon)
+        assert len(plugin.token_cache) == 1
+
+        assert plugin.header_provider(self._make_readonly_context(ic_anon)) == {}
+
+    @pytest.mark.asyncio
+    async def test_session_scoped_get_subject_token_does_not_collapse_the_key(self):
+        """Case: a get_subject_token reading session state (not the caller's header)
+        still keeps one entry per caller, since the key comes from the credential."""
+        alice = self._jwt("https://dex.example", "alice")
+        bob = self._jwt("https://dex.example", "bob")
+
+        sts = Mock(spec=ADKSTSIntegration)
+        # Reads a session-scoped field, so it returns the same token for every caller.
+        sts.get_subject_token = lambda state: state.get("session_token")
+        sts.fetch_actor_token = None
+        sts._actor_token = "actor-token"
+        sts.exchange_token = AsyncMock(side_effect=["exchanged-alice", "exchanged-bob"])
+        plugin = ADKTokenPropagationPlugin(sts)
+
+        ic_alice = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {alice}"})
+        ic_bob = self._make_invocation_context("shared-sess", headers={"Authorization": f"Bearer {bob}"})
+        for ic in (ic_alice, ic_bob):
+            ic.session.state["session_token"] = "one-token-for-the-whole-session"
+
+        await plugin.before_run_callback(invocation_context=ic_alice)
+        await plugin.before_run_callback(invocation_context=ic_bob)
+
+        assert len(plugin.token_cache) == 2
+        assert plugin.token_cache[plugin.cache_key(ic_alice)].token == "exchanged-alice"
+        assert plugin.token_cache[plugin.cache_key(ic_bob)].token == "exchanged-bob"
+
+    @pytest.mark.asyncio
+    async def test_get_subject_token_is_not_called_per_tool_call(self):
+        """Case: header_provider runs on every tool call, so it must reuse the key
+        resolved for the run rather than re-invoking the caller-supplied hook."""
+        alice = self._jwt("https://dex.example", "alice")
+
+        calls = []
+
+        def counting_hook(state):
+            calls.append(state)
+            return state.get(HEADERS_KEY, {}).get("Authorization", "").removeprefix("Bearer ")
+
+        sts = Mock(spec=ADKSTSIntegration)
+        sts.get_subject_token = counting_hook
+        sts.fetch_actor_token = None
+        sts._actor_token = "actor-token"
+        sts.exchange_token = AsyncMock(return_value="exchanged-alice")
+        plugin = ADKTokenPropagationPlugin(sts)
+
+        ic = self._make_invocation_context("sess-hook", headers={"Authorization": f"Bearer {alice}"})
+        await plugin.before_run_callback(invocation_context=ic)
+        after_run = len(calls)
+
+        ro_ctx = self._make_readonly_context(ic)
+        for _ in range(3):
+            assert plugin.header_provider(ro_ctx) == {"Authorization": "Bearer exchanged-alice"}
+
+        assert len(calls) == after_run
+
+    @pytest.mark.asyncio
+    async def test_cached_entry_does_not_outlive_the_caller_credential(self):
+        """Case: the entry is keyed by the caller's credential, so a caller replaying
+        an expired bearer must miss the cache and reach the STS."""
+        alice = self._jwt("https://dex.example", "alice", expiry=int(time.time()) + 30)
+        long_lived = self._jwt("https://dex.example", "alice", expiry=int(time.time()) + 3600)
+
+        sts = Mock(spec=ADKSTSIntegration)
+        sts.get_subject_token = None
+        sts.fetch_actor_token = None
+        sts._actor_token = "actor-token"
+        sts.exchange_token = AsyncMock(return_value=long_lived)
+        plugin = ADKTokenPropagationPlugin(sts)
+
+        ic = self._make_invocation_context("sess-ttl", headers={"Authorization": f"Bearer {alice}"})
+        await plugin.before_run_callback(invocation_context=ic)
+
+        entry = plugin.token_cache[plugin.cache_key(ic)]
+        assert entry.expiry == extract_jwt_expiry(alice)
+
+    def test_empty_subject_is_not_cacheable(self):
+        """Case: a caller with no credential yields no cache key, so credential-less
+        callers cannot come to share one entry."""
+        plugin = ADKTokenPropagationPlugin(sts_integration=None)
+
+        assert plugin._cache_key_for("sess-x", None, None) is None
+        assert plugin._cache_key_for("sess-x", "", "") is None
+        # No credential but a subject token from the hook still keys per session.
+        assert plugin._cache_key_for("sess-x", None, "hook-token") is not None
+
+    def test_unidentified_session_is_not_cacheable(self):
+        """Case: without a session id, entries could only be shared between
+        unrelated conversations."""
+        plugin = ADKTokenPropagationPlugin(sts_integration=None)
+
+        assert plugin._cache_key_for("", "a-credential", "a-credential") is None
+
+    @pytest.mark.asyncio
+    async def test_header_provider_still_resolves_after_the_run_ends(self):
+        """Case: the key comes from the run's own state, not from state held between
+        callbacks, so a tool call outliving after_run_callback still authenticates."""
+        alice = self._jwt("https://dex.example", "alice")
+
+        plugin = ADKTokenPropagationPlugin(sts_integration=None)
+        ic = self._make_invocation_context("sess-late", headers={"Authorization": f"Bearer {alice}"})
+        await plugin.before_run_callback(invocation_context=ic)
+
+        await plugin.after_run_callback(invocation_context=ic)
+        # The entry is unexpired, so the sweep kept it.
+        assert plugin.cache_key(ic) in plugin.token_cache
+        assert plugin.header_provider(self._make_readonly_context(ic)) == {"Authorization": f"Bearer {alice}"}
+
 
 class TestADKSTSIntegration:
     """Test cases for ADKSTSIntegration."""