orchestrata_ultimate.memory.hot_store

Reference › orchestrata_ultimate › memory › hot_store

RedisMemoryStore is the working-memory layer of the MemoryEngine, its only importer. It holds recent entries in Redis with a TTL. The constructor takes the host and port as arguments or from REDIS_HOST and REDIS_PORT read at call time; if neither is available it raises ValueError, with no hardcoded fallback. connect creates the client through the core Redis factory and close releases it; the other methods assert that the store is connected.

Each entry is written in one pipeline under keys prefixed orchestrata_ultimate:memory:: a hash per (session, volk, key) holding the content, tags, metadata, classification, timestamps and token count; a sorted index per session and volk scored by creation time; a set per tag; and, when the entry has an embedding, a JSON embedding under a flat embed namespace. Every key expires after the TTL. Reads are retrieve (exact key), retrieve_by_key_global (any session), retrieve_by_tag, get_embeddings, get_all_keys and get_all_entries for a session. retrieve_by_scope scans the index keys matching the session and volk of a MemoryScope, uses the sorted-set score range to apply max_age_seconds cheaply, and filters the rest in Python through _matches_scope. delete removes one entry; delete_session deletes the session's embeddings first, because the embed namespace is shared across sessions, and then its other keys, and returns the count.

At a glance

Class diagram

Dependency diagram

API

Classes

RedisMemoryStore

Decorators: traces(spec='010', story='US-001', reqs=['REQ-010-024']) · Line 49

Redis-backed hot storage for working memory.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/memory/hot_store.py

# mypy: ignore-errors
"""Redis hot storage for working memory.

SPEC-010 FR-006: Redis-backed hot storage with TTL, hash structures,
and embedding vector storage for fast similarity search.
"""

import json
import logging
import os
from collections.abc import Awaitable
from datetime import UTC, datetime, timedelta
from typing import Any, cast

import redis.asyncio as redis

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import MemoryEntry, MemoryScope

logger = logging.getLogger(__name__)


def _redis_result[T](value: Awaitable[T] | T) -> Awaitable[T]:
    """Narrow a redis-py command result back to its true async type.

    redis-py types command methods (``hgetall``, ``smembers``, ...) as
    ``Awaitable[T] | T`` because the same command mixin backs both the sync
    and async clients (see ``redis.typing.ResponseT``). ``RedisMemoryStore``
    only ever holds an async client, which always returns an awaitable at
    runtime -- this documents that fact for pyright instead of suppressing
    the check.
    """
    return cast(Awaitable[T], value)


# Redis key patterns
_ENTRY_KEY = "orchestrata_ultimate:memory:{session_id}:{volk_id}:{key}"
_INDEX_KEY = "orchestrata_ultimate:memory:{session_id}:{volk_id}:index"
_TAG_KEY = "orchestrata_ultimate:memory:{session_id}:{volk_id}:tags:{tag}"
_EMBED_KEY = "orchestrata_ultimate:memory:embed:{key}"

# Default TTL: 24 hours
DEFAULT_TTL = 86400
# Extended TTL on migration failure: 7 days
EXTENDED_TTL = 604800


@traces(spec="010", story="US-001", reqs=["REQ-010-024"])
class RedisMemoryStore:
    """Redis-backed hot storage for working memory."""

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-047"])
    def __init__(
        self,
        redis_host: str | None = None,
        redis_port: int | None = None,
        redis_db: int = 0,
        redis_password: str | None = None,
        ttl: int = DEFAULT_TTL,
    ):
        """Initialize the store, resolving host/port from env at call time.

        Args:
            redis_host: Redis host. Defaults to the ``REDIS_HOST`` env var
                (read now, not at import/def time, so test/CLI overrides
                apply). No hardcoded fallback.
            redis_port: Redis port. Defaults to the ``REDIS_PORT`` env var.
                No hardcoded fallback.
            redis_db: Redis logical DB index.
            redis_password: Optional Redis auth password.
            ttl: Entry TTL in seconds.

        Raises:
            ValueError: If ``redis_host``/``redis_port`` is not supplied and
                the corresponding env var is unset.
        """
        if redis_host is not None:
            self._host = redis_host
        else:
            env_host = os.environ.get("REDIS_HOST")
            if env_host is None:
                raise ValueError("redis_host not provided and REDIS_HOST is not set")
            self._host = env_host

        if redis_port is not None:
            self._port = redis_port
        else:
            env_port = os.environ.get("REDIS_PORT")
            if env_port is None:
                raise ValueError("redis_port not provided and REDIS_PORT is not set")
            self._port = int(env_port)

        self._db = redis_db
        self._password = redis_password
        self._ttl = ttl
        self._client: redis.Redis | None = None

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-048"])
    async def connect(self) -> None:
        """Connect to Redis."""
        from orchestrata_ultimate.core.redis_factory import create_async_redis

        self._client = create_async_redis(
            host=self._host,
            port=self._port,
            db=self._db,
            password=self._password,
        )

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-049"])
    async def close(self) -> None:
        """Close Redis connection."""
        if self._client:
            await self._client.aclose()
            self._client = None

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-050"])
    def _entry_key(self, session_id: str, volk_id: str, key: str) -> str:
        return _ENTRY_KEY.format(session_id=session_id, volk_id=volk_id, key=key)

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-051"])
    def _index_key(self, session_id: str, volk_id: str) -> str:
        return _INDEX_KEY.format(session_id=session_id, volk_id=volk_id)

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-052"])
    def _tag_key(self, session_id: str, volk_id: str, tag: str) -> str:
        return _TAG_KEY.format(session_id=session_id, volk_id=volk_id, tag=tag)

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-053"])
    def _embed_key(self, key: str) -> str:
        return _EMBED_KEY.format(key=key)

    @traces(
        spec="010",
        story="US-001",
        reqs=["REQ-010-024", "REQ-010-025", "REQ-010-027"],
    )
    async def store(self, entry: MemoryEntry) -> None:
        """Store a memory entry in Redis."""
        assert self._client is not None, "Not connected to Redis"
        rkey = self._entry_key(entry.session_id, entry.volk_id, entry.key)
        data = {
            "key": entry.key,
            "content": entry.content,
            "summary": entry.summary or "",
            "tags": json.dumps(entry.tags),
            "metadata": json.dumps(entry.metadata),
            "classification": entry.classification.value,
            "volk_id": entry.volk_id,
            "session_id": entry.session_id,
            "task_id": entry.task_id or "",
            "created_at": entry.created_at.isoformat(),
            "updated_at": entry.updated_at.isoformat(),
            "token_count": str(entry.token_count),
        }
        pipe = self._client.pipeline()
        pipe.hset(rkey, mapping=data)
        pipe.expire(rkey, self._ttl)
        # Add to sorted index (score = timestamp)
        idx_key = self._index_key(entry.session_id, entry.volk_id)
        pipe.zadd(idx_key, {entry.key: entry.created_at.timestamp()})
        pipe.expire(idx_key, self._ttl)
        # Add to tag sets
        for tag in entry.tags:
            tkey = self._tag_key(entry.session_id, entry.volk_id, tag)
            pipe.sadd(tkey, entry.key)
            pipe.expire(tkey, self._ttl)
        # Store embedding if present
        if entry.embedding:
            ekey = self._embed_key(entry.key)
            pipe.set(ekey, json.dumps(entry.embedding))
            pipe.expire(ekey, self._ttl)
        await pipe.execute()

    @traces(spec="010", story="US-001", reqs=["REQ-010-002"])
    async def retrieve(
        self, session_id: str, volk_id: str, key: str
    ) -> MemoryEntry | None:
        """Retrieve a memory entry by key."""
        assert self._client is not None, "Not connected to Redis"
        rkey = self._entry_key(session_id, volk_id, key)
        data = await _redis_result(self._client.hgetall(rkey))
        if not data:
            return None
        return self._deserialize(data)

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-054"])
    async def retrieve_by_key_global(self, key: str) -> MemoryEntry | None:
        """Retrieve by key, scanning across sessions. Used for key-only lookups."""
        assert self._client is not None, "Not connected to Redis"
        pattern = f"orchestrata_ultimate:memory:*:*:{key}"
        async for rkey in self._client.scan_iter(match=pattern, count=100):
            data = await _redis_result(self._client.hgetall(rkey))
            if data:
                return self._deserialize(data)
        return None

    @traces(spec="010", story="US-001", reqs=["REQ-010-002"])
    async def retrieve_by_scope(
        self,
        scope: MemoryScope,
        limit: int = 100,
        max_age_seconds: int | None = None,
    ) -> list[MemoryEntry]:
        """Retrieve entries matching a scope filter.

        Args:
            scope: Scope filter (session/volk/task/date range).
            limit: Maximum entries to retrieve before any in-Python filtering.
            max_age_seconds: If set, only retrieve entries created within this
                many seconds of now. Applied efficiently via Redis sorted-set
                score range (score = entry timestamp).
        """
        assert self._client is not None, "Not connected to Redis"
        entries: list[MemoryEntry] = []
        session_id = scope.session_id or "*"
        volk_id = scope.volk_id or "*"
        pattern = f"orchestrata_ultimate:memory:{session_id}:{volk_id}:index"

        min_score: float | None = None
        if max_age_seconds is not None:
            min_score = (
                datetime.now(UTC) - timedelta(seconds=max_age_seconds)
            ).timestamp()

        async for idx_key in self._client.scan_iter(match=pattern, count=100):
            # Key format: orchestrata_ultimate:memory:{session_id}:{volk_id}:index
            # volk_id may contain colons (e.g. U:Cd), so rejoin middle parts
            parts = idx_key.split(":")
            s_id = parts[2]
            v_id = ":".join(parts[3:-1])

            if min_score is not None:
                keys = await self._client.zrevrangebyscore(
                    idx_key, "+inf", min_score, start=0, num=limit
                )
            else:
                keys = await self._client.zrevrange(idx_key, 0, limit - 1)

            for key in keys:
                entry = await self.retrieve(s_id, v_id, key)
                if entry and self._matches_scope(entry, scope):
                    entries.append(entry)
                    if len(entries) >= limit:
                        return entries
        return entries

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-055", "REQ-010-012"])
    async def retrieve_by_tag(
        self, session_id: str, volk_id: str, tag: str
    ) -> list[MemoryEntry]:
        """Retrieve all entries with a specific tag."""
        assert self._client is not None, "Not connected to Redis"
        tkey = self._tag_key(session_id, volk_id, tag)
        keys = await _redis_result(self._client.smembers(tkey))
        entries = []
        for key in keys:
            entry = await self.retrieve(session_id, volk_id, key)
            if entry:
                entries.append(entry)
        return entries

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-027"])
    async def get_embeddings(self, session_id: str, volk_id: str) -> list[tuple]:
        """Get all embeddings for a session/volk pair.

        Returns list of (key, embedding_vector) tuples.
        """
        assert self._client is not None, "Not connected to Redis"
        idx_key = self._index_key(session_id, volk_id)
        keys = await self._client.zrevrange(idx_key, 0, -1)
        results = []
        for key in keys:
            ekey = self._embed_key(key)
            raw = await self._client.get(ekey)
            if raw:
                results.append((key, json.loads(raw)))
        return results

    async def delete(self, session_id: str, volk_id: str, key: str) -> bool:
        """Delete a memory entry."""
        assert self._client is not None, "Not connected to Redis"
        rkey = self._entry_key(session_id, volk_id, key)
        data = await _redis_result(self._client.hgetall(rkey))
        if not data:
            return False
        pipe = self._client.pipeline()
        pipe.delete(rkey)
        idx_key = self._index_key(session_id, volk_id)
        pipe.zrem(idx_key, key)
        # Remove from tag sets
        tags_raw = data.get("tags", "[]")
        tags = json.loads(tags_raw) if tags_raw else []
        for tag in tags:
            tkey = self._tag_key(session_id, volk_id, tag)
            pipe.srem(tkey, key)
        # Remove embedding
        pipe.delete(self._embed_key(key))
        await pipe.execute()
        return True

    async def get_all_keys(self, session_id: str, volk_id: str) -> list[str]:
        """Get all memory keys for a session/volk pair."""
        assert self._client is not None, "Not connected to Redis"
        idx_key = self._index_key(session_id, volk_id)
        return await self._client.zrevrange(idx_key, 0, -1)

    async def get_all_entries(self, session_id: str) -> list[MemoryEntry]:
        """Get all entries for a session (all volks)."""
        assert self._client is not None, "Not connected to Redis"
        pattern = f"orchestrata_ultimate:memory:{session_id}:*:index"
        entries = []
        async for idx_key in self._client.scan_iter(match=pattern, count=100):
            # Key format: orchestrata_ultimate:memory:{session_id}:{volk_id}:index
            # volk_id may contain colons (e.g. U:Cd), so rejoin middle parts
            parts = idx_key.split(":")
            volk_id = ":".join(parts[3:-1])
            keys = await self._client.zrevrange(idx_key, 0, -1)
            for key in keys:
                entry = await self.retrieve(session_id, volk_id, key)
                if entry:
                    entries.append(entry)
        return entries

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-023"])
    async def delete_session(self, session_id: str) -> int:
        """Delete all entries for a session. Returns count deleted."""
        assert self._client is not None, "Not connected to Redis"
        count = 0
        # Embeddings live in a flat "...:memory:embed:{key}" namespace shared
        # across ALL sessions, so we must delete ONLY this session's embeddings.
        # The per-(session,volk) index zsets hold this session's entry keys.
        index_pattern = f"orchestrata_ultimate:memory:{session_id}:*:index"
        async for idx_key in self._client.scan_iter(match=index_pattern, count=100):
            if ":tags:" in idx_key:  # a tag literally named "index" -- skip
                continue
            for entry_key in await self._client.zrange(idx_key, 0, -1):
                await self._client.delete(self._embed_key(entry_key))
        # Then delete this session's entry/index/tag keys.
        pattern = f"orchestrata_ultimate:memory:{session_id}:*"
        async for rkey in self._client.scan_iter(match=pattern, count=100):
            await self._client.delete(rkey)
            count += 1
        return count

    def _deserialize(self, data: dict[str, Any]) -> MemoryEntry:
        """Deserialize Redis hash to MemoryEntry."""
        return MemoryEntry.from_dict(data)

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-017"])
    def _matches_scope(self, entry: MemoryEntry, scope: MemoryScope) -> bool:
        """Check if an entry matches scope filters."""
        if scope.session_id and entry.session_id != scope.session_id:
            return False
        if scope.volk_id and entry.volk_id != scope.volk_id:
            return False
        if scope.task_id and entry.task_id != scope.task_id:
            return False
        if scope.date_from and entry.created_at < scope.date_from:
            return False
        if scope.date_to and entry.created_at > scope.date_to:
            return False
        return True