orchestrata_ultimate.orchestrator.fact_store

Reference › orchestrata_ultimate › orchestrator › fact_store

RealFactStore keeps governance facts, the Prolog-style predicate and argument records the orchestrator reasons over, in two places at once: Redis as the fast store, and the graph as best-effort cold storage. No other reference page imports it; it builds on fact_graph for the graph writes, and on core.redis_factory and core.graph for its connections. The lifecycle of facts (scope, expiry, cleanup) is handled one level up by fact_manager.

The constructor creates an async Redis client, using REDIS_HOST and REDIS_PORT through the factory when no host or port is given, and an async graph repository from a URI and optional credentials. Facts are keyed in Redis under the prefix orchestrata:fact: followed by the fact id. assert_fact makes an id of the form fact_ plus twelve hex characters, writes the JSON record (id, predicate, arguments, creation time) to Redis, then writes the graph node through fact_graph.write_fact, linked to a session when a session id is given. An optional on_graph_failure callback receives the exception if the graph write fails; the Redis write is not rolled back. The remaining methods are retract_fact, has_fact, get_fact, count, clear, list_facts (paged by limit and offset) and close; the class is also an async context manager.

At a glance

Class diagram

Dependency diagram

API

Classes

RealFactStore

Decorators: traces(spec='003', story='US-003', reqs=['REQ-003-US-003-001']) · Line 37

Real fact storage backend using Redis (hot) and graph (cold).

Source

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

# mypy: ignore-errors
"""Real fact storage using Redis and the graph.

NO MOCKS. NO FAKES. NO STUBS.
Real Redis at localhost:6379
Real graph at bolt://localhost:7687
"""

# Spec: 006 - Session Management

import json
import logging
import uuid
from collections.abc import Callable
from datetime import datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.fact_graph import (
    clear_facts as graph_clear_facts,
)
from orchestrata_ultimate.orchestrator.fact_graph import (
    delete_fact as graph_delete_fact,
)
from orchestrata_ultimate.orchestrator.fact_graph import (
    get_graph_failure_count,
)
from orchestrata_ultimate.orchestrator.fact_graph import write_fact as graph_write_fact

logger = logging.getLogger(__name__)

# Re-export for backward compatibility
__all__ = ["RealFactStore", "get_graph_failure_count"]


@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
class RealFactStore:
    """Real fact storage backend using Redis (hot) and graph (cold).

    Redis: Fast fact lookup and existence checks
    Graph: Persistent storage with graph relationships
    """

    def __init__(
        self,
        redis_host: str | None = None,
        redis_port: int | None = None,
        redis_db: int = 0,
        graph_uri: str | None = None,
        # Empty, not an env read: NEO4J_USERNAME was retired with the Neo4j
        # containers on 2026-07-27 and every remaining graph endpoint
        # negotiates no auth. It was also a DEFAULT ARGUMENT, sampled from the
        # env once at import, so no later override could reach it.
        graph_user: str = "",
        graph_password: str | None = None,
        key_prefix: str = "orchestrata:fact:",
    ):
        """Initialize with real connections.

        Args:
            redis_host: Redis server host; resolved from REDIS_HOST when None.
            redis_port: Redis server port; resolved from REDIS_PORT when None.
            redis_db: Redis database number.
            graph_uri: Graph connection URI.
            graph_user: Graph username.
            graph_password: Graph password.
            key_prefix: Redis key prefix for fact storage.
        """
        _graph_password = graph_password or ""

        from orchestrata_ultimate.core.graph import (
            GraphRepositoryConfig,
            create_async_graph_repository,
        )
        from orchestrata_ultimate.core.redis_factory import (
            create_async_redis,
            resolve_redis_host,
            resolve_redis_port,
        )

        self._redis = create_async_redis(
            host=redis_host if redis_host is not None else resolve_redis_host(),
            port=redis_port if redis_port is not None else resolve_redis_port(),
            db=redis_db,
        )

        self._graph_repo = create_async_graph_repository(
            GraphRepositoryConfig(
                uri=graph_uri,
                auth=(graph_user, _graph_password),
            )
        )

        self._prefix = key_prefix

    @traces(
        spec="006",
        story="US-004",
        reqs=["REQ-006-US-004-001", "REQ-006-FR8-001"],
    )
    async def assert_fact(
        self,
        predicate: str,
        arguments: tuple,
        session_id: str = "",
        on_graph_failure: Callable[[Exception], None] | None = None,
    ) -> str:
        """Assert a fact to both Redis and graph.

        Args:
            predicate: Prolog predicate name.
            arguments: Fact arguments.
            session_id: Optional session ID to link the fact to.
            on_graph_failure: Optional callback invoked with the
                exception when graph cold storage write fails.

        Returns:
            Unique fact ID.
        """
        fact_id = f"fact_{uuid.uuid4().hex[:12]}"
        fact_data = {
            "fact_id": fact_id,
            "predicate": predicate,
            "arguments": list(arguments),
            "created_at": datetime.now().isoformat(),
        }

        redis_key = f"{self._prefix}{fact_id}"
        await self._redis.set(
            redis_key,
            json.dumps(fact_data),
        )
        await graph_write_fact(
            self._graph_repo,
            fact_id,
            predicate,
            arguments,
            fact_data["created_at"],
            session_id,
            on_graph_failure,
        )

        return fact_id

    @traces(spec="006", story="US-002", reqs=["REQ-006-US-002-003"])
    async def retract_fact(self, fact_id: str) -> bool:
        """Retract a fact from both Redis and graph.

        Args:
            fact_id: Fact identifier.

        Returns:
            True if fact was retracted.
        """
        redis_key = f"{self._prefix}{fact_id}"

        if not await self._redis.exists(redis_key):
            return False

        await self._redis.delete(redis_key)
        await graph_delete_fact(self._graph_repo, fact_id)

        return True

    async def has_fact(self, fact_id: str) -> bool:
        """Check if fact exists in Redis.

        Args:
            fact_id: Fact identifier.

        Returns:
            True if fact exists.
        """
        redis_key = f"{self._prefix}{fact_id}"
        return await self._redis.exists(redis_key) > 0

    async def get_fact(
        self,
        fact_id: str,
    ) -> dict[str, Any] | None:
        """Get fact data from Redis.

        Args:
            fact_id: Fact identifier.

        Returns:
            Fact data or None.
        """
        redis_key = f"{self._prefix}{fact_id}"
        data = await self._redis.get(redis_key)
        if data:
            return json.loads(data)
        return None

    async def count(self) -> int:
        """Count facts in Redis.

        Returns:
            Number of facts.
        """
        pattern = f"{self._prefix}*"
        keys = [key async for key in self._redis.scan_iter(pattern)]
        return len(keys)

    async def clear(self) -> int:
        """Clear all facts from Redis and graph.

        Returns:
            Number of facts cleared.
        """
        pattern = f"{self._prefix}*"
        keys = [key async for key in self._redis.scan_iter(pattern)]
        cleared = len(keys)

        if keys:
            await self._redis.delete(*keys)

        await graph_clear_facts(self._graph_repo)

        return cleared

    async def list_facts(
        self,
        limit: int = 1000,
        offset: int = 0,
    ) -> list[dict[str, Any]]:
        """List facts from Redis with pagination.

        Args:
            limit: Maximum number of facts to return.
            offset: Number of facts to skip.

        Returns:
            List of fact data dictionaries.
        """
        pattern = f"{self._prefix}*"
        facts: list[dict[str, Any]] = []
        skipped = 0
        async for key in self._redis.scan_iter(pattern):
            if skipped < offset:
                skipped += 1
                continue
            if len(facts) >= limit:
                break
            data = await self._redis.get(key)
            if data:
                facts.append(json.loads(data))
        return facts

    async def close(self) -> None:
        """Close connections."""
        await self._redis.aclose()
        await self._graph_repo.close()

    async def __aenter__(self) -> RealFactStore:
        """Async context manager entry."""
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: Any,
    ) -> bool:
        """Async context manager exit."""
        await self.close()
        return False