orchestrata_ultimate.memory.cold_store

Reference › orchestrata_ultimate › memory › cold_store

GraphMemoryStore is the long-term layer of the MemoryEngine, its only importer. It persists memory entries as Memory nodes in the operational knowledge graph. The constructor resolves the URI from KG_ROUTING_URI or KG_URI and the database name through the graph factory, at call time and without a fallback, and sends no credentials. connect builds an async Bolt repository through core.graph; if the driver is not installed or the connection fails it logs a warning and leaves the store unavailable instead of raising, and available reports that state.

Cold storage is a side channel and never breaks the caller. store returns False when the graph is unavailable or the write fails. Otherwise it merges the node on the id <session_id>::<key>, because a bare key recurs across sessions and would collapse them into one node. It sets the content, tags, classification, ids and timestamps, and links the node to its Session (BELONGS_TO) and to its creating agent (CREATED_BY), and links the project to both so the nodes are reachable from the project root. The canonical labels come from knowledge_graph.canonical_schema, imported inside the method to respect the layering. The node holds no embedding. retrieve finds a node by key, search_by_scope builds a filter from the session, volk, task, date range and max_age_seconds and returns entries newest first, delete removes by key, count counts nodes (optionally per session), and _node_to_entry converts a node back to a MemoryEntry. Read and write failures are logged and return empty results.

At a glance

Class diagram

Dependency diagram

API

Classes

_GraphTemporal (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 32

Structural type for neo4j.time temporal values (DateTime/Date/Time/ Duration) read back off a node's properties.

GraphMemoryStore

Decorators: traces(spec='010', story='US-002', reqs=['REQ-010-028', 'REQ-010-029']) · Line 55

Graph-backed cold storage for long-term memory.

Functions

Source

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

# mypy: ignore-errors
"""Graph cold storage for long-term memory.

SPEC-010 FR-007: graph-backed cold storage with Memory nodes,
relationships, and temporal queries.
"""

import json
import logging
import os
from datetime import UTC, datetime, timedelta
from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_factory import resolve_kg_database, resolve_uri
from orchestrata_ultimate.core.protocols import (
    MemoryClassification,
    MemoryEntry,
    MemoryScope,
)

logger = logging.getLogger(__name__)

# The root Project slug the store()-path hub edges below MERGE onto — same
# default every other KG writer in this codebase uses (e.g.
# factory/promotion/store.py's `_project_slug()`). Read at call time via
# `_project_slug()` (RUL-003), never cached at import time.
_DEFAULT_PROJECT_SLUG = "orchestrata-ultimate"


@runtime_checkable
class _GraphTemporal(Protocol):
    """Structural type for neo4j.time temporal values (DateTime/Date/Time/
    Duration) read back off a node's properties.

    ``props.get("created_at")`` types as ``Any | None`` (``props`` comes from
    ``dict(node)`` over an untyped driver record), so pyright can't narrow
    ``hasattr(created_at, "to_native")`` the way the original duck-typed code
    relied on -- ``created_at`` stays partly typed as ``None``
    (reportOptionalMemberAccess). ``isinstance()`` against this
    ``runtime_checkable`` Protocol gives the same structural "does it have
    to_native()" check with real narrowing, without importing every concrete
    neo4j.time class.
    """

    def to_native(self) -> Any: ...


def _project_slug() -> str:
    """Resolve the root ``:Project`` slug the store() hub edges MERGE onto."""
    return os.environ.get("ORCHESTRATA_PROJECT_ID", _DEFAULT_PROJECT_SLUG)


@traces(spec="010", story="US-002", reqs=["REQ-010-028", "REQ-010-029"])
class GraphMemoryStore:
    """Graph-backed cold storage for long-term memory."""

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-031"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-031"])
    def __init__(
        self,
        uri: str | None = None,
        username: str | None = None,
        password: str | None = None,
        database: str | None = None,
        driver_kwargs: dict[str, Any] | None = None,
    ):
        # Resolve at CALL time, not import time (signature defaults freeze the
        # env at module load and break test/CLI overrides), and with no
        # fallback values — localhost never runs Neo4j in this infra, so a
        # default would mask misconfiguration with a dead endpoint. The same
        # applies to the DATABASE: a literal "neo4j" default was a Neo4j-only
        # assumption that the orchgraph cutover invalidated — that database
        # does not exist on the new engine, so every call raised
        # `Tried to retrieve an unknown database "neo4j"`.
        # resolve_uri() reads KG_ROUTING_URI/KG_URI and raises with the key
        # name when unset — same fail-loud contract as _require_env, on the
        # keys that still exist. NEO4J_URI/USERNAME/PASSWORD were retired with
        # the Neo4j containers on 2026-07-27; requiring them here took the
        # gateway down on the first restart after .env dropped them.
        self._uri = uri or resolve_uri()
        # No credentials: every remaining graph endpoint negotiates no auth.
        self._username = username or ""
        self._password = password or ""
        self._database = resolve_kg_database(database)
        # Extra driver kwargs forwarded to the Bolt driver at connect() time
        # (e.g. ``max_transaction_retry_time`` / ``connection_timeout``). Empty
        # by default — production keeps the factory's standard retry budget;
        # callers that expect a fast failure (e.g. error-path tests hitting a
        # nonexistent database) pass a short retry time so the managed
        # transaction gives up in ~1s instead of retrying the full budget.
        self._driver_kwargs: dict[str, Any] = driver_kwargs or {}
        self._repo = None

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-032"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-032"])
    async def connect(self) -> None:
        """Connect to graph."""
        try:
            from orchestrata_ultimate.core.graph import (
                GraphRepositoryConfig,
                create_async_graph_repository,
            )
            from orchestrata_ultimate.core.graph.adapters.bolt_async import (
                BoltAsyncRepository,
            )

            repo = create_async_graph_repository(
                GraphRepositoryConfig(
                    uri=self._uri,
                    auth=(self._username, self._password),
                    database=self._database,
                    driver_kwargs=self._driver_kwargs,
                )
            )
            # Force eager driver construction (mirrors the previous
            # create_async_graph() call, which validated the URI/auth
            # synchronously) instead of the seam's normal lazy creation.
            # The protocol doesn't expose `.driver` — type against the
            # concrete "bolt" backend adapter (documented escape hatch).
            assert isinstance(repo, BoltAsyncRepository)
            _ = repo.driver
            self._repo = repo
            logger.info("Connected to graph at %s", self._uri)
        except ImportError:
            logger.warning("graph driver not installed — cold storage disabled")
            self._repo = None
        except Exception as e:
            logger.warning("Failed to connect to graph: %s", e)
            self._repo = None

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-033"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-033"])
    async def close(self) -> None:
        """Close graph connection."""
        if self._repo:
            await self._repo.close()
            self._repo = None

    @property
    def available(self) -> bool:
        """Check if graph is available."""
        return self._repo is not None

    @traces(
        spec="010",
        story="US-002",
        reqs=["REQ-010-010", "REQ-010-028", "REQ-010-029"],
    )
    async def store(self, entry: MemoryEntry) -> bool:
        """Persist a memory entry as a Memory node in graph.

        Returns:
            True if the write landed, False if the graph is unavailable or
            the write itself failed. Non-fatal either way — cold storage is
            a side-channel, never a reason to break the caller.
        """
        if not self._repo:
            logger.debug("Graph not available — skipping cold store")
            return False
        # Function-scoped: knowledge_graph is a layer-3 module and memory is
        # layer-2 (architecture_checker.py LAYER_ASSIGNMENTS) — a module-level
        # import here is an upward layering violation. Deferred to call time
        # per the D7 exception (RULED, same pattern as
        # postman/postman.py's knowledge_graph.cxp_auditor import).
        from orchestrata_ultimate.knowledge_graph.canonical_schema import (
            LABEL_AGENT,
            LABEL_PROJECT,
            LABEL_SESSION,
            REL_HAS_AGENT,
            REL_HAS_SESSION,
        )

        # Memory's canonical merge key is `id` (canonical_schema
        # CANONICAL_MERGE_KEYS), built as the ratified `<session_id>::<key>`
        # derivation (ontology_mapping._memory_key) -- a bare `key` recurs
        # across sessions (two sessions legitimately reusing the same key
        # string, e.g. "summary"), so MERGEing on it collapsed every
        # session's entry for that key onto ONE node. `key` is kept as a
        # plain property (never drop data; `retrieve`/`count` still filter
        # by it).
        memory_id = f"{entry.session_id}::{entry.key}"
        # Session-254 root-connectivity ruling: this writer previously
        # MERGEd `:Session` and `:Agent` with no edge back to the root
        # `:Project` node -- the Project hub MERGEs below close that gap for
        # both (mirroring the HAS_VOLK/HAS_SESSION convention; see
        # canonical_schema.LABEL_AGENT for why `:Agent` stays a distinct
        # label from `:Volk` here rather than being renamed).
        query = f"""
        MERGE (m:Memory {{id: $id}})
        SET m.key = $key,
            m.content = $content,
            m.summary = $summary,
            m.tags = $tags,
            m.metadata = $metadata,
            m.classification = $classification,
            m.volk_id = $volk_id,
            m.session_id = $session_id,
            m.task_id = $task_id,
            m.token_count = $token_count,
            m.created_at = datetime($created_at),
            m.updated_at = datetime($updated_at)
        WITH m
        MERGE (s:{LABEL_SESSION} {{session_id: $session_id}})
        SET s.id = $session_id
        MERGE (m)-[:BELONGS_TO]->(s)
        WITH m, s
        MERGE (v:{LABEL_AGENT} {{alias: $volk_id}})
        MERGE (m)-[:CREATED_BY]->(v)
        WITH s, v
        MERGE (p:{LABEL_PROJECT} {{slug: $project}})
        MERGE (p)-[:{REL_HAS_SESSION}]->(s)
        MERGE (p)-[:{REL_HAS_AGENT}]->(v)
        """
        params = {
            "id": memory_id,
            "key": entry.key,
            "content": entry.content,
            "summary": entry.summary or "",
            "tags": 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 "",
            "token_count": entry.token_count,
            "created_at": entry.created_at.isoformat(),
            "updated_at": entry.updated_at.isoformat(),
            "project": _project_slug(),
        }
        try:
            await self._repo.execute_write(query, params)
            logger.debug("Stored memory %s in graph", entry.key)
            return True
        except Exception as e:
            logger.error("Failed to store memory in graph: %s", e)
            return False

    @traces(spec="010", story="US-002", reqs=["REQ-010-011"])
    async def retrieve(self, key: str) -> MemoryEntry | None:
        """Retrieve a memory entry from graph by key."""
        if not self._repo:
            return None
        query = "MATCH (m:Memory {key: $key}) RETURN m"
        try:
            rows = await self._repo.execute_read(query, {"key": key})
            if rows:
                return self._node_to_entry(rows[0]["m"])
        except Exception as e:
            logger.error("Failed to retrieve memory from graph: %s", e)
        return None

    @traces(spec="010", story="US-002", reqs=["REQ-010-013", "REQ-010-030"])
    async def search_by_scope(
        self,
        scope: MemoryScope,
        limit: int = 100,
        max_age_seconds: int | None = None,
    ) -> list[MemoryEntry]:
        """Search memories by scope filters.

        Args:
            scope: Scope filter (session/volk/task/date range).
            limit: Maximum entries to return.
            max_age_seconds: If set, additionally constrain ``m.created_at`` to
                be within this many seconds of now.
        """
        if not self._repo:
            return []
        conditions = []
        params: dict[str, Any] = {"limit": limit}
        if scope.session_id:
            conditions.append("m.session_id = $session_id")
            params["session_id"] = scope.session_id
        if scope.volk_id:
            conditions.append("m.volk_id = $volk_id")
            params["volk_id"] = scope.volk_id
        if scope.task_id:
            conditions.append("m.task_id = $task_id")
            params["task_id"] = scope.task_id
        if scope.date_from:
            conditions.append("m.created_at >= datetime($date_from)")
            params["date_from"] = scope.date_from.isoformat()
        if scope.date_to:
            conditions.append("m.created_at <= datetime($date_to)")
            params["date_to"] = scope.date_to.isoformat()
        if max_age_seconds is not None:
            cutoff = (
                datetime.now(UTC) - timedelta(seconds=max_age_seconds)
            ).isoformat()
            conditions.append("m.created_at >= datetime($max_age_cutoff)")
            params["max_age_cutoff"] = cutoff
        where = " AND ".join(conditions) if conditions else "true"
        query = f"""
        MATCH (m:Memory) WHERE {where}
        RETURN m ORDER BY m.created_at DESC LIMIT $limit
        """
        try:
            rows = await self._repo.execute_read(query, params)
            return [self._node_to_entry(r["m"]) for r in rows]
        except Exception as e:
            logger.error("Failed to search memories in graph: %s", e)
            return []

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-034"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-037"])
    async def delete(self, key: str) -> bool:
        """Delete a memory entry from graph."""
        if not self._repo:
            return False
        query = """
        MATCH (m:Memory {key: $key})
        DETACH DELETE m
        RETURN count(m) as deleted
        """
        try:
            rows = await self._repo.execute_write(query, {"key": key})
            return bool(rows) and rows[0]["deleted"] > 0
        except Exception as e:
            logger.error("Failed to delete memory from graph: %s", e)
            return False

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-035"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-038"])
    async def count(self, session_id: str | None = None) -> int:
        """Count memories in graph."""
        if not self._repo:
            return 0
        if session_id:
            query = "MATCH (m:Memory {session_id: $sid}) RETURN count(m) as c"
            params = {"sid": session_id}
        else:
            query = "MATCH (m:Memory) RETURN count(m) as c"
            params = {}
        try:
            rows = await self._repo.execute_read(query, params)
            return rows[0]["c"] if rows else 0
        except Exception as e:
            logger.error("Failed to count memories in graph: %s", e)
            return 0

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-036"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-039"])
    def _node_to_entry(self, node: Any) -> MemoryEntry:
        """Convert a graph node to MemoryEntry."""
        props = dict(node) if hasattr(node, "__iter__") else node
        tags = props.get("tags", [])
        if isinstance(tags, str):
            tags = json.loads(tags)
        metadata = props.get("metadata", "{}")
        if isinstance(metadata, str):
            metadata = json.loads(metadata)
        created_at = props.get("created_at")
        if isinstance(created_at, _GraphTemporal):
            created_at = created_at.to_native()
        elif isinstance(created_at, str):
            created_at = datetime.fromisoformat(created_at)
        else:
            created_at = datetime.now(UTC)
        updated_at = props.get("updated_at")
        if isinstance(updated_at, _GraphTemporal):
            updated_at = updated_at.to_native()
        elif isinstance(updated_at, str):
            updated_at = datetime.fromisoformat(updated_at)
        else:
            updated_at = datetime.now(UTC)
        classification = props.get("classification", "private")
        if isinstance(classification, str):
            classification = MemoryClassification(classification)
        volk_id = props.get("volk_id", "")
        return MemoryEntry(
            key=props.get("key", ""),
            content=props.get("content", ""),
            summary=props.get("summary") or None,
            tags=tags,
            metadata=metadata,
            classification=classification,
            created_at=created_at,
            updated_at=updated_at,
            volk_id=volk_id,  # pyright: ignore[reportCallIssue]  # pydantic populate_by_name keyword not modeled by pyright
            session_id=props.get("session_id", ""),
            task_id=props.get("task_id") or None,
            token_count=int(props.get("token_count", 0)),
        )