@@ -4,7 +4,9 @@ Redis-based storage for agent threads and context.
 
 ## Main Classes
 
-- **`RedisHistoryProvider`** - Persistent chat history provider using Redis
+- **`RedisHistoryProvider`** - Persistent chat history provider using scoped Redis keys. Scoped mode requires an
+  application ID and non-empty session ID, and can additionally isolate tenants and agents. Use explicit legacy mode
+  only while deliberately migrating historical keys; scoped mode never accesses them automatically.
 - **`RedisContextProvider`** - Context provider with Redis-backed retrieval
 - **`RedisSettings`** - TypedDict connection settings for vector stores, resolved with core `load_settings` from
   explicit URL overrides, an optional .env file, or `REDIS_URL`. URLs use `SecretString` to mask credentials.
@@ -40,7 +42,9 @@ their own indexes/keys, never `FLUSHALL`. Coverage includes both formats and
 from agent_framework.redis import RedisContextProvider, RedisHistoryProvider
 
 context_provider = RedisContextProvider(redis_url="redis://localhost:6379")
-history_provider = RedisHistoryProvider(redis_url="redis://localhost:6379")
+history_provider = RedisHistoryProvider(
+    redis_url="redis://localhost:6379", application_id="my-app", agent_id="my-agent"
+)
 ```
 
 ## Import Path
 
@@ -79,6 +79,47 @@ arrays. Indexed strings cannot contain surrounding whitespace, NUL, or U+001F,
 and must fit Redis's 4096-byte TAG limit; these restrictions do not apply to
 unindexed payloads. Unsupported operations raise an error.
 
+## Isolate conversation history
+
+`RedisHistoryProvider` uses scoped keys by default. Supply a stable
+`application_id`; also supply `tenant_id` and `agent_id` whenever those
+boundaries exist in your application. The provider's `source_id` and each
+non-empty session ID are included automatically:
+
+```python
+from agent_framework.redis import RedisHistoryProvider
+
+history_provider = RedisHistoryProvider(
+    redis_url="redis://localhost:6379",
+    application_id="support-app",
+    tenant_id="contoso",
+    agent_id="triage-agent",
+)
+```
+
+Scoped mode rejects missing application or session identifiers rather than
+placing unrelated conversations under a shared fallback key. Identifiers are
+encoded independently, so they do not need to be globally unique across
+tenants, applications, agents, and provider sources.
+
+Releases that predate scoped keys used
+`{key_prefix}:{session_id or "default"}`. Existing deployments can temporarily
+retain that exact format by opting in explicitly:
+
+```python
+legacy_history_provider = RedisHistoryProvider(
+    redis_url="redis://localhost:6379",
+    key_format="legacy",
+)
+```
+
+Legacy mode does not accept scoped identifiers. Scoped mode never reads,
+rewrites, or deletes legacy keys. To migrate existing history, copy only the
+records belonging to a verified application, tenant, agent, provider source,
+and session into the corresponding scoped key using an application-owned
+migration process. After verifying the copied history, remove legacy keys
+separately according to the application's retention policy.
+
 ## Store and search documents
 
 This example uses precomputed vectors, so no embedding service is needed.
 
@@ -9,12 +9,14 @@
 from __future__ import annotations
 
 import json
+import warnings
 from collections.abc import Awaitable, Sequence
 from inspect import isawaitable
-from typing import Any, ClassVar, TypeVar, cast
+from typing import Any, ClassVar, Literal, TypeVar, cast
 
 import redis.asyncio as redis
 from agent_framework import Message
+from agent_framework._filesystem import _storage_key_segment  # pyright: ignore[reportPrivateUsage]
 from agent_framework._sessions import HistoryProvider, filter_new_messages
 from agent_framework._telemetry import mark_feature_used
 from redis.credentials import CredentialProvider
@@ -43,10 +45,17 @@ class RedisHistoryProvider(HistoryProvider):
     """Redis-backed history provider using the new HistoryProvider hooks pattern.
 
     Stores conversation history in Redis Lists, with each session isolated by a
-    unique Redis key.
+    key scoped to the application, optional tenant and agent, and provider source.
     """
 
     DEFAULT_SOURCE_ID: ClassVar[str] = "redis_memory"
+    _ENCODED_KEY_PREFIX: ClassVar[str] = "~redis-key-prefix-"
+    _ENCODED_TENANT_PREFIX: ClassVar[str] = "~redis-tenant-"
+    _ENCODED_APPLICATION_PREFIX: ClassVar[str] = "~redis-application-"
+    _ENCODED_AGENT_PREFIX: ClassVar[str] = "~redis-agent-"
+    _ENCODED_SOURCE_PREFIX: ClassVar[str] = "~redis-source-"
+    _ENCODED_SESSION_PREFIX: ClassVar[str] = "~redis-session-"
+    _ABSENT_SCOPE_SEGMENT: ClassVar[str] = "~none"
 
     def __init__(
         self,
@@ -59,6 +68,10 @@ def __init__(
         username: str | None = None,
         *,
         key_prefix: str = "chat_messages",
+        tenant_id: str | None = None,
+        application_id: str | None = None,
+        agent_id: str | None = None,
+        key_format: Literal["scoped", "legacy"] = "scoped",
         max_messages: int | None = None,
         load_messages: bool = True,
         store_outputs: bool = True,
@@ -78,7 +91,17 @@ def __init__(
             port: Redis port number. Defaults to 6380 (Azure Redis SSL port).
             ssl: Enable SSL/TLS connection. Defaults to True.
             username: Redis username.
-            key_prefix: Prefix for Redis keys. Defaults to 'chat_messages'.
+            key_prefix: Base prefix for Redis keys. Scoped mode appends independently encoded
+                tenant, application, agent, provider source, and session segments.
+                Defaults to 'chat_messages'.
+            tenant_id: Optional tenant identifier used as an independent key boundary.
+            application_id: Application identifier used as a required key boundary in scoped mode.
+            agent_id: Optional agent identifier used as an independent key boundary.
+            key_format: Redis key format. ``"scoped"`` isolates history by tenant,
+                application, agent, provider source, and session. ``"legacy"``
+                preserves the historical ``{key_prefix}:{session_id}`` format for
+                explicit migration compatibility. Scoped identifiers cannot be
+                supplied in legacy mode. Defaults to ``"scoped"``.
             max_messages: Maximum number of messages to retain per session.
                 When exceeded, oldest messages are automatically trimmed.
                 None means unlimited storage; 0 retains nothing, and no message
@@ -95,6 +118,7 @@ def __init__(
             ValueError: If both redis_url and credential_provider are provided.
             ValueError: If credential_provider is used without host parameter.
             ValueError: If max_messages is negative.
+            ValueError: If key_format or its scoped identifiers are invalid.
         """
         super().__init__(
             source_id,
@@ -113,8 +137,30 @@ def __init__(
             raise ValueError("host is required when using credential_provider")
         if max_messages is not None and max_messages < 0:
             raise ValueError("max_messages must be None (unlimited) or a non-negative integer")
+        if key_format not in ("scoped", "legacy"):
+            raise ValueError("key_format must be 'scoped' or 'legacy'")
+        if key_format == "scoped":
+            if not application_id:
+                raise ValueError("application_id must be a non-empty string when key_format='scoped'")
+            if tenant_id == "":
+                raise ValueError("tenant_id must be non-empty when supplied")
+            if agent_id == "":
+                raise ValueError("agent_id must be non-empty when supplied")
+        elif any(scope is not None for scope in (tenant_id, application_id, agent_id)):
+            raise ValueError("tenant_id, application_id, and agent_id cannot be used with key_format='legacy'")
+        if key_format == "legacy":
+            warnings.warn(
+                "key_format='legacy' is deprecated and will be removed in a future version. "
+                "Migrate persisted history to scoped keys and use key_format='scoped'.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
 
         self.key_prefix = key_prefix
+        self.tenant_id = tenant_id
+        self.application_id = application_id
+        self.agent_id = agent_id
+        self.key_format = key_format
         self.max_messages = max_messages
         self.redis_url = redis_url
 
@@ -130,9 +176,47 @@ def __init__(
         else:
             self._redis_client = redis.from_url(redis_url, decode_responses=True)  # type: ignore[no-untyped-call]
 
+    @classmethod
+    def _optional_scope_segment(cls, value: str | None, *, encoded_prefix: str) -> str:
+        """Encode an optional isolation boundary without conflating absence with a caller value."""
+        if value is None:
+            return cls._ABSENT_SCOPE_SEGMENT
+        return _storage_key_segment(value, encoded_prefix=encoded_prefix)
+
     def _redis_key(self, session_id: str | None) -> str:
-        """Get the Redis key for a given session's messages."""
-        return f"{self.key_prefix}:{session_id or 'default'}"
+        """Get a pipe-delimited scoped key or the historical colon-delimited legacy key."""
+        if self.key_format == "legacy":
+            return f"{self.key_prefix}:{session_id or 'default'}"
+        if not session_id:
+            raise ValueError("session_id must be a non-empty string when key_format='scoped'")
+
+        key_prefix_segment = _storage_key_segment(
+            self.key_prefix,
+            encoded_prefix=self._ENCODED_KEY_PREFIX,
+        )
+        tenant_segment = self._optional_scope_segment(
+            self.tenant_id,
+            encoded_prefix=self._ENCODED_TENANT_PREFIX,
+        )
+        agent_segment = self._optional_scope_segment(
+            self.agent_id,
+            encoded_prefix=self._ENCODED_AGENT_PREFIX,
+        )
+        application_segment = _storage_key_segment(
+            cast("str", self.application_id),
+            encoded_prefix=self._ENCODED_APPLICATION_PREFIX,
+        )
+        source_segment = _storage_key_segment(self.source_id, encoded_prefix=self._ENCODED_SOURCE_PREFIX)
+        session_segment = _storage_key_segment(session_id, encoded_prefix=self._ENCODED_SESSION_PREFIX)
+        return "|".join((
+            key_prefix_segment,
+            "v2",
+            tenant_segment,
+            application_segment,
+            agent_segment,
+            source_segment,
+            session_segment,
+        ))
 
     async def get_messages(
         self,
@@ -181,19 +265,17 @@ async def save_messages(
             **kwargs: Additional arguments (unused).
         """
         mark_feature_used(FeatureIndex.REDIS)
+        key = self._redis_key(session_id)
         if not messages:
             return
 
         if self.max_messages == 0:
             # Retention is disabled. Trimming cannot express this - LTRIM key 0 -1 keeps
             # the whole list - so return before serializing: no payload reaches Redis, an
-            # AOF or a replica. Stored history is deliberately left alone. _redis_key omits
-            # source_id, so the list can belong to a co-located provider, and removing
-            # stored history is what clear() is for.
+            # AOF or a replica. Stored history is deliberately left alone; removing stored
+            # history is what clear() is for.
             return
 
-        key = self._redis_key(session_id)
-
         existing_messages = await self.get_messages(session_id, state=state, **kwargs)
         new_messages = filter_new_messages(existing_messages, messages)