Enhance Redis memory management with automatic reconnection and fallback handling

- Added resilient Redis reconnection logic and fallback to in-memory storage when Redis is unavailable.
- Implemented integration tests for Docker-backed Redis, validating TTL expiry and service reconnection.
- Updated CHANGELOG to document these enhancements and changes.

Author: Wojciech Napierała

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 All notable changes to this project will be documented in this file. The format follows [Keep a Changelog](https://keepachangelog.com/) and dates use ISO-8601 (`YYYY-MM-DD`).
 
+## [0.1.6] - 2025-11-08
+
+### Added
+- Docker-backed Redis integration tests validating TTL expiry, service reconnection, and fallback behaviour.
+
+### Changed
+- Hardened Redis working-memory store with automatic reconnection and graceful fallback to in-memory storage when the service is unavailable.
+
 ## [0.1.5] - 2025-11-07
 
 ### Changed

core/memory_manager.py

@@ -7,6 +7,7 @@
     v0.4 - 2025-11-07 - Added semantic graph utilities and relation management APIs.
     v0.5 - 2025-11-07 - Instrumented memory operations with telemetry spans and metrics.
     v0.6 - 2025-11-07 - Added retrieval helpers, tamper-evident revision hashing, and replay utilities.
+    v0.7 - 2025-11-08 - Added resilient Redis reconnection and fallback handling.
 """
 
 from __future__ import annotations
@@ -240,76 +241,85 @@ def __init__(self, config: AppConfig) -> None:
             self._logger.warning(
                 "Redis python client unavailable; using in-memory fallback store."
             )
-        else:
-            try:
-                self._client = redis_module.Redis(
-                    host=self._host,
-                    port=self._port,
-                    db=self._db,
-                    socket_timeout=2,
-                )
-                # probe the connection lazily
-                self._client.ping()
-            except Exception as exc:  # pragma: no cover - runtime check
-                self._logger.error(
-                    "Redis connection failed (%s); falling back to in-memory store.",
-                    exc,
-                )
-                self._client = None
+            return
+
+        self._initialise_client()
 
     def put(self, item: WorkingMemoryItem) -> None:
         """Store an item in working memory."""
-        try:
-            if self._client:
-                self._client.setex(
+        if self._ensure_client():
+            client = cast(Any, self._client)
+            try:
+                client.setex(
                     name=item.key,
                     time=item.ttl_seconds or self._ttl_seconds,
                     value=json.dumps(asdict(item), default=str),
                 )
-            else:
-                self._fallback[item.key] = item
-        except Exception as exc:  # pragma: no cover - client failure path
-            raise MemoryError(f"Failed to store working memory item: {exc}") from exc
+                return
+            except Exception as exc:  # pragma: no cover - client failure path
+                self._logger.warning(
+                    "Redis store failed (%s); switching to in-memory fallback.", exc
+                )
+                self._client = None
+
+        self._fallback[item.key] = item
 
     def get(self, key: str) -> Optional[WorkingMemoryItem]:
         """Retrieve an item from working memory."""
-        try:
-            if self._client:
-                payload_bytes = self._client.get(key)
+        if self._ensure_client():
+            client = cast(Any, self._client)
+            try:
+                payload_bytes = client.get(key)
                 if payload_bytes is None:
                     return None
                 data = cast(Dict[str, object], json.loads(payload_bytes.decode("utf-8")))
                 key_value = str(data.get("key", ""))
                 payload_value = cast(Dict[str, object], data.get("payload", {}))
                 ttl_value_raw = data.get("ttl_seconds", self._ttl_seconds)
-                ttl_value = int(ttl_value_raw) if isinstance(ttl_value_raw, (int, float)) else self._ttl_seconds
+                ttl_value = (
+                    int(ttl_value_raw)
+                    if isinstance(ttl_value_raw, (int, float))
+                    else self._ttl_seconds
+                )
                 return WorkingMemoryItem(
                     key=key_value,
                     payload=payload_value,
                     ttl_seconds=ttl_value,
                     created_at=self._coerce_timestamp(data.get("created_at")),
                 )
-            return self._fallback.get(key)
-        except Exception as exc:  # pragma: no cover
-            raise MemoryError(f"Failed to load working memory item: {exc}") from exc
+            except Exception as exc:  # pragma: no cover
+                self._logger.warning(
+                    "Redis retrieval failed (%s); falling back to in-memory store.",
+                    exc,
+                )
+                self._client = None
+
+        return self._fallback.get(key)
 
     def delete(self, key: str) -> None:
         """Remove an item from working memory if present."""
-        try:
-            if self._client:
-                self._client.delete(key)
-            else:
-                self._fallback.pop(key, None)
-        except Exception as exc:  # pragma: no cover
-            raise MemoryError(f"Failed to delete working memory item: {exc}") from exc
+        if self._ensure_client():
+            client = cast(Any, self._client)
+            try:
+                client.delete(key)
+                return
+            except Exception as exc:  # pragma: no cover
+                self._logger.warning(
+                    "Redis deletion failed (%s); removing item from fallback store.",
+                    exc,
+                )
+                self._client = None
+
+        self._fallback.pop(key, None)
 
     def list_items(self, pattern: str = "*") -> List[WorkingMemoryItem]:
         """Return working memory items matching the given pattern."""
-        try:
-            if self._client:
+        if self._ensure_client():
+            client = cast(Any, self._client)
+            try:
                 items: List[WorkingMemoryItem] = []
-                for key in self._client.scan_iter(match=pattern):
-                    payload_bytes = self._client.get(key)
+                for key in client.scan_iter(match=pattern):
+                    payload_bytes = client.get(key)
                     if not payload_bytes:
                         continue
                     data = cast(Dict[str, object], json.loads(payload_bytes.decode("utf-8")))
@@ -318,7 +328,11 @@ def list_items(self, pattern: str = "*") -> List[WorkingMemoryItem]:
                     )
                     payload_value = cast(Dict[str, object], data.get("payload", {}))
                     ttl_raw = data.get("ttl_seconds", self._ttl_seconds)
-                    ttl_value = int(ttl_raw) if isinstance(ttl_raw, (int, float)) else self._ttl_seconds
+                    ttl_value = (
+                        int(ttl_raw)
+                        if isinstance(ttl_raw, (int, float))
+                        else self._ttl_seconds
+                    )
                     items.append(
                         WorkingMemoryItem(
                             key=key_value,
@@ -328,9 +342,14 @@ def list_items(self, pattern: str = "*") -> List[WorkingMemoryItem]:
                         )
                     )
                 return items
-            return list(self._fallback.values())
-        except Exception as exc:  # pragma: no cover
-            raise MemoryError(f"Failed to enumerate working memory: {exc}") from exc
+            except Exception as exc:  # pragma: no cover
+                self._logger.warning(
+                    "Redis enumeration failed (%s); returning fallback store state.",
+                    exc,
+                )
+                self._client = None
+
+        return list(self._fallback.values())
 
     @staticmethod
     def _coerce_timestamp(value: Optional[object]) -> datetime:
@@ -345,6 +364,42 @@ def _coerce_timestamp(value: Optional[object]) -> datetime:
             return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
         return datetime.now(timezone.utc)
 
+    def _initialise_client(self) -> None:
+        """Initialise the Redis client if the dependency is available."""
+        try:
+            self._client = self._attempt_connect()
+        except Exception as exc:  # pragma: no cover - runtime environment dependent
+            self._logger.error(
+                "Redis connection failed during initialisation (%s); using in-memory fallback.",
+                exc,
+            )
+            self._client = None
+
+    def _attempt_connect(self) -> Optional[Any]:
+        if redis_module is None:
+            return None
+        client = redis_module.Redis(
+            host=self._host,
+            port=self._port,
+            db=self._db,
+            socket_timeout=2,
+        )
+        client.ping()
+        return client
+
+    def _ensure_client(self) -> bool:
+        """Ensure a live Redis client is available, reconnecting if needed."""
+        if redis_module is None:
+            return False
+        if self._client is None:
+            try:
+                self._client = self._attempt_connect()
+            except Exception as exc:  # pragma: no cover - runtime dependent
+                self._logger.debug("Redis reconnect attempt failed: %s", exc)
+                self._client = None
+                return False
+        return self._client is not None
+
 
 class ChromaMemoryStore:
     """Adapter around ChromaDB for episodic, semantic, and review memory."""

pyproject.toml

@@ -39,3 +39,8 @@ module = [
     "openai.*",
 ]
 ignore_missing_imports = true
+
+[tool.pytest.ini_options]
+markers = [
+    "integration: marks tests that exercise external services",
+]

tests/test_redis_integration.py

@@ -0,0 +1,165 @@
+"""Integration-style tests exercising the Redis working memory store.
+
+Updates:
+    v0.1 - 2025-11-08 - Added dockerised Redis coverage for TTL expiry,
+        reconnection, and fallback behaviour.
+"""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+import time
+from pathlib import Path
+from typing import Iterator
+
+import pytest
+import redis
+
+from config.settings import load_app_config
+from core.memory_manager import RedisMemoryStore
+from models.memory import WorkingMemoryItem
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+COMPOSE_FILE = PROJECT_ROOT / "docker-compose.yml"
+REDIS_SERVICE = "redis"
+
+
+class _RedisServiceController:
+    """Helper for managing the dockerised Redis service during tests."""
+
+    def __init__(self, compose_file: Path) -> None:
+        self._compose_file = compose_file
+
+    def run(self, *args: str) -> None:
+        command = [
+            "docker",
+            "compose",
+            "-f",
+            str(self._compose_file),
+            *args,
+        ]
+        subprocess.run(command, check=True, cwd=str(PROJECT_ROOT))
+
+    def up(self) -> None:
+        self.run("up", "-d", REDIS_SERVICE)
+
+    def stop(self) -> None:
+        self.run("stop", REDIS_SERVICE)
+
+    def start(self) -> None:
+        self.run("start", REDIS_SERVICE)
+
+    def down(self) -> None:
+        self.run("rm", "-sf", REDIS_SERVICE)
+
+    def ensure_ready(self, timeout: float = 15.0) -> None:
+        client = redis.Redis(host="localhost", port=6379, db=5, socket_timeout=1)
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            try:
+                if client.ping():
+                    return
+            except redis.exceptions.ConnectionError:
+                time.sleep(0.25)
+        raise RuntimeError("Redis service did not become ready in time")
+
+
+@pytest.fixture(scope="module")
+def redis_service() -> Iterator[_RedisServiceController]:
+    if shutil.which("docker") is None:
+        pytest.skip("Docker is required for Redis integration tests.")
+    if not COMPOSE_FILE.exists():
+        pytest.skip("docker-compose.yml not available for Redis integration tests.")
+
+    controller = _RedisServiceController(COMPOSE_FILE)
+    controller.up()
+    try:
+        try:
+            controller.ensure_ready()
+        except RuntimeError as exc:
+            pytest.skip(str(exc))
+        yield controller
+    finally:
+        controller.down()
+
+
+def _build_store(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ttl: int = 3) -> RedisMemoryStore:
+    monkeypatch.setenv("DRM_MEMORY_LOG_PATH", str(tmp_path / "revisions.jsonl"))
+    config = load_app_config()
+    config.memory.redis.ttl_seconds = ttl
+    return RedisMemoryStore(config)
+
+
+@pytest.mark.integration
+def test_redis_working_memory_ttl_expiry(
+    redis_service: _RedisServiceController,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    store = _build_store(monkeypatch, tmp_path, ttl=1)
+    item = WorkingMemoryItem(
+        key="ttl:test",
+        payload={"value": "ephemeral"},
+        ttl_seconds=1,
+    )
+
+    store.put(item)
+    assert store.get(item.key) is not None
+
+    time.sleep(2)
+    assert store.get(item.key) is None
+
+
+@pytest.mark.integration
+def test_redis_store_recovers_after_restart(
+    redis_service: _RedisServiceController,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    store = _build_store(monkeypatch, tmp_path)
+
+    first_item = WorkingMemoryItem(
+        key="reconnect:initial",
+        payload={"attempt": 1},
+        ttl_seconds=10,
+    )
+    store.put(first_item)
+    assert store.get(first_item.key) is not None
+
+    redis_service.stop()
+    time.sleep(1.0)
+    redis_service.start()
+    redis_service.ensure_ready()
+
+    second_item = WorkingMemoryItem(
+        key="reconnect:subsequent",
+        payload={"attempt": 2},
+        ttl_seconds=10,
+    )
+
+    store.put(second_item)
+    assert store.get(second_item.key) is not None
+
+
+def test_redis_fallback_store_when_unavailable(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    monkeypatch.setenv("DRM_MEMORY_LOG_PATH", str(tmp_path / "fallback.jsonl"))
+    config = load_app_config()
+    config.memory.redis.port = 6390
+    config.memory.redis.ttl_seconds = 3
+
+    store = RedisMemoryStore(config)
+
+    item = WorkingMemoryItem(
+        key="fallback:item",
+        payload={"value": "cached"},
+        ttl_seconds=3,
+    )
+    store.put(item)
+
+    retrieved = store.get(item.key)
+    assert retrieved is not None
+    assert retrieved.payload == item.payload