After 3 Years of Redis, We Were Testing AI Agent Memory Expiry Wrong

I was woken up at 2 AM by an on-call alert. A user complained that the AI Agent carried over shopping preferences from the previous session into a new one. My first instinct: did the Redis cache fail to isolate? I opened redis-cli and found memory:agent:123:session:456 had existed for 12 hours, with TTL of -1 — this key should have expired after 10 minutes. That's when I realized: after years of using Redis for AI Agent memory storage, our manual tests never truly covered the expiration boundary.

Problem breakdown

AI Agent memory storage has two hard requirements: isolation and expiration policy. Isolation means memories from different sessions of the same agent must be separated, never leaking across. Expiration means session memories should be automatically cleaned up by TTL to avoid contaminating later inference. How did we test before? After development, we manually ran redis-cli set/get to check a few entries, then waited a few minutes to see if they disappeared. The root cause: manual tests only verified that reads/writes “looked like they worked”, but didn't verify concurrent overwrites, TTL boundaries, or key naming rules. On top of that, Redis expiration has two mechanisms — lazy deletion and active deletion — so a manual spot check might see the key still present right after expiry, producing a "false green". Conventional unit tests that mock Redis can't simulate real expiration behavior at all.

Solution design

The choice was straightforward: use Pytest for automated regression, a real Redis 7 instance for testing, switching databases via environment variables. We rejected fakeredis because its expiration policy doesn't fully match real Redis, especially active expiration and TTL-overwriting behavior with SET. Testing against it would be pointless. We also rejected unittest because Pytest's fixtures and scope management give each test case a clean Redis and allow cleanup via yield. Architecturally, we wrapped the storage layer in AgentMemoryStore, and all write operations must go through save_memory. Tests target only this entry point, keeping them aligned with the production path.

Core implementation

This code defines the storage layer, unifies key naming, and uses SET ... EX to ensure every write carries a TTL, avoiding clearing expiration on overwrite.

# memory_store.py
import json
from typing import Any, Optional
import redis

class AgentMemoryStore:
    """AI Agent 会话记忆存储，基于 Redis String 实现。

    key 规则：memory:{agent_id}:{session_id}
    为什么这样设计：Redis 的 key 自带命名空间隔离，能防止不同 agent/session 串数据。
    """
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def save_memory(self, agent_id: str, session_id: str, memory: dict[str, Any], ttl_seconds: int = 600) -> None:
        key = f"memory:{agent_id}:{session_id}"
        payload = json.dumps(memory, ensure_ascii=False)
        # 关键：必须用 SET key value EX ttl，而不是先 SET 再 EXPIRE。
        # 否则如果两条命令之间 Redis 重启/故障，可能留下没有 TTL 的 key。
        self.redis.set(key, payload, ex=ttl_seconds)

    def get_memory(self, agent_id: str, session_id: str) -> Optional[dict[str, Any]]:
        key = f"memory:{agent_id}:{session_id}"
        raw = self.redis.get(key)
        if raw is None:
            return None
        return json.loads(raw)

    def get_ttl(self, agent_id: str, session_id: str) -> int:
        key = f"memory:{agent_id}:{session_id}"
        return self.redis.ttl(key)

This code provides pytest fixtures: each test case gets an isolated database, with cleanup after to avoid cross-test contamination.

# tests/conftest.py
import os
import pytest
import redis
from memory_store import AgentMemoryStore

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/15")

@pytest.fixture()
def redis_client():
    # 用 db=15 作为测试专用，避免误删本地开发缓存
    client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
    client.flushdb()
    yield client
    client.flushdb()
    client.close()

@pytest.fixture()
def store(redis_client):
    return AgentMemoryStore(redis_client)

This code contains the core test cases, directly verifying isolation and expiration policy.

# tests/test_memory_store.py
import time
import pytest

def test_memory_isolated_between_sessions(store):
    agent_id = "agent-1"
    store.save_memory(agent_id, "session-a", {"intent": "buy_shoes"}, ttl_seconds=600)
    store.save_memory(agent_id, "session-b", {"intent": "compare_prices"}, ttl_seconds=600)

    # 同一个 agent 的两个 session 应该读到各自记忆，不能串
    assert store.get_memory(agent_id, "session-a")["intent"] == "buy_shoes"
    assert stor