Python: RedisHistoryProvider ignores source_id in its key, so two providers on one session share and overwrite each other's history

[chinmayv095]

Describe the bug

RedisHistoryProvider stores every session's messages under

def _redis_key(self, session_id: str | None) -> str:
    return f"{self.key_prefix}:{session_id or 'default'}"

source_id is not part of the key. Two RedisHistoryProvider instances with different
source_ids but the same key_prefix — which is the default, "chat_messages" — therefore share
a single Redis list for a given session. Running more than one history provider on an agent is a
supported configuration: Agent.history_providers returns all providers that are instances of
HistoryProvider, and the per-service-call middleware iterates them, so the collision is reachable
from ordinary use rather than from a contrived setup.

Two consequences, both silent:

1. History contamination. A provider configured as a write-only audit or evaluation sink
   (load_messages=False) writes into the same list the primary provider reads, so its copies are
   loaded back into the agent's context on the next turn.
2. Cross-provider data destruction. clear() deletes the shared key, so clearing the audit
   provider destroys the primary provider's conversation history.

CosmosHistoryProvider, the other history backend in the repo, does not have this problem — it
carries source_id in the document and filters on it in get_messages, clear and
list_sessions:

query = (
    "SELECT c.message FROM c "
    "WHERE c.session_id = @session_id AND c.source_id = @source_id "
    "ORDER BY c.sort_key ASC"
)

This also looks like an unfinished part of #3995 ([BREAKING] Scope provider state by source_id and
standardize source IDs), which made source_id the provider scoping key — hooks now receive
session.state.setdefault(provider.source_id, {}) — and standardized the default source IDs of the
built-in providers, RedisHistoryProvider among them. That change scoped provider state by
source_id; the Redis provider's own storage key was not scoped with it.

To reproduce

import asyncio
from unittest.mock import patch

from agent_framework import Message
from agent_framework_redis._history_provider import RedisHistoryProvider


class FakeRedis:
    """Minimal in-memory stand-in for the list commands the provider uses."""

    def __init__(self):
        self.store: dict[str, list[str]] = {}

    async def rpush(self, key, val):
        self.store.setdefault(key, []).append(val)

    async def lrange(self, key, s, e):
        return self.store.get(key, [])

    async def llen(self, key):
        return len(self.store.get(key, []))

    async def delete(self, key):
        self.store.pop(key, None)

    def pipeline(self, transaction=True):
        outer = self

        class P:
            async def __aenter__(s):
                return s

            async def __aexit__(s, *a):
                return False

            async def rpush(s, k, v):
                await outer.rpush(k, v)

            async def execute(s):
                pass

        return P()


async def main():
    fake = FakeRedis()
    with patch("agent_framework_redis._history_provider.redis.from_url", return_value=fake):
        primary = RedisHistoryProvider("memory", redis_url="redis://x")
        audit = RedisHistoryProvider("audit", redis_url="redis://x", load_messages=False)

    await primary.save_messages("s1", [Message(role="user", contents=["hello from the user"])])
    await audit.save_messages("s1", [Message(role="assistant", contents=["AUDIT COPY"])])

    print("keys written:", list(fake.store))
    got = await primary.get_messages("s1")
    print("primary sees:", [m.text for m in got])

    await audit.clear("s1")
    got = await primary.get_messages("s1")
    print("primary sees after audit.clear():", [m.text for m in got])


asyncio.run(main())

Output:

keys written: ['chat_messages:s1']
primary sees: ['hello from the user', 'AUDIT COPY']
primary sees after audit.clear(): []

Expected behavior

Each provider's history is isolated by source_id, the way CosmosHistoryProvider isolates it, so
that a second provider on the same session can neither be read by nor destroy the first.

Notes on a fix

I am raising this as an issue rather than a PR because the obvious fix is a storage-format change
and the migration call is yours, not mine. Putting source_id into the key — for example
{key_prefix}:{source_id}:{session_id} — orphans the history of every existing deployment that
did not set key_prefix explicitly, which is a bigger decision than the bug warrants me making.

Options as I see them, roughly in increasing order of disruption:

1. Document key_prefix as the isolation knob and leave the key alone — cheapest, but it
   leaves the default configuration broken and diverging from the Cosmos backend.
2. Default key_prefix to include the source_id while leaving an explicitly-passed
   key_prefix untouched, so only users on the default are affected.
3. Add source_id to the key with a release note, matching CosmosHistoryProvider outright.

Happy to send a PR for whichever you prefer, including the documentation-only option.

Platform

• Language: Python
• Package: agent-framework-redis 1.0.0b260730
• Source: python/packages/redis/agent_framework_redis/_history_provider.py