orchestrata_ultimate.governance.cache

Reference › orchestrata_ultimate › governance › cache

The decision cache of the governance engine, SPEC-007 (REQ-007-FR10-015, -121, REQ-007-FR3-006, REQ-007-US-001-002, -009). GovernanceCache saves PolicyDecision values so that a repeat of the same action and context is answered without re-running the Prolog pipeline. policy_engine is the importer: it reads the cache after the safe-mode check and stores allowed decisions after the Prolog step. The local tier uses the shared core.lru_cache; the shared tier uses the Redis state manager from infra.redis_state.

The cache has two tiers. get(key) looks in the local least-recently-used cache first (default 1000 entries, 60 seconds), then in Redis (default 300 seconds), and copies a Redis hit into the local tier; Redis errors are logged and count as a miss. set(key, decision) writes to both tiers. invalidate(pattern) removes matching entries and returns the count, clear empties both tiers, and get_hit_rate and get_stats report local hits, Redis hits and misses. _serialize_decision and _deserialize_decision convert decisions to and from JSON-safe dictionaries. The module function generate_cache_key(action, context) builds a deterministic key by hashing the action with the sorted context items (SHA-256, first 32 hex characters).

At a glance

Class diagram

Dependency diagram

API

Classes

GovernanceCache

Decorators: traces(spec='007', story='US-PGOV-001', reqs=['REQ-007-US-001-005', 'REQ-007-US-001-009', 'REQ-007-FR3-006', 'REQ-007-F… · Line 47

Two-tier governance decision cache.

Functions

Source

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

# mypy: ignore-errors
"""Multi-tier governance cache for Orchestrata.

Provides a two-level cache (local LRU + Redis) for governance
policy decisions. Cached queries target < 5ms latency.
"""

# Spec: 007a - Prolog Governance

from __future__ import annotations

import hashlib
import logging
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any, cast

if TYPE_CHECKING:
    from orchestrata_ultimate.infra.redis_state import RedisStateManager

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.lru_cache import LRUCache
from orchestrata_ultimate.governance.constants import (
    CACHE_LOCAL_MAX_SIZE,
    CACHE_LOCAL_TTL_SECONDS,
    CACHE_REDIS_PREFIX,
    CACHE_REDIS_TTL_SECONDS,
    REDIS_SCAN_BATCH_SIZE,
)
from orchestrata_ultimate.governance.types import DecisionType, PolicyDecision

_logger = logging.getLogger(__name__)


@traces(
    spec="007",
    story="US-PGOV-001",
    reqs=[
        "REQ-007-US-001-005",
        "REQ-007-US-001-009",
        "REQ-007-FR3-006",
        "REQ-007-FR10-110",
        "REQ-007-US-004-011",
        "REQ-007-US-006-002",
        "REQ-007-US-006-009",
    ],
)
class GovernanceCache:
    """Two-tier governance decision cache.

    Level 1: In-process LRU cache (microsecond access)
    Level 2: Redis cache (millisecond access, shared across processes)

    Cache keys are SHA-256 hashes of action + context.
    """

    def __init__(
        self,
        redis: RedisStateManager | None = None,
        local_max_size: int = CACHE_LOCAL_MAX_SIZE,
        local_ttl: float = CACHE_LOCAL_TTL_SECONDS,
        redis_ttl: int = CACHE_REDIS_TTL_SECONDS,
    ) -> None:
        self._local: LRUCache[dict[str, Any]] = LRUCache(
            max_size=local_max_size, default_ttl=local_ttl
        )
        self._redis = redis
        self._redis_ttl = redis_ttl
        self._hits_local = 0
        self._hits_redis = 0
        self._misses = 0

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-009"])
    def _redis_key(self, key: str) -> str:
        return f"{CACHE_REDIS_PREFIX}:{key}"

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-009"])
    def _serialize_decision(self, decision: PolicyDecision) -> dict[str, Any]:
        """Serialize a PolicyDecision to a JSON-safe dict."""
        return {
            "decision_type": decision.decision_type.value,
            "action": decision.action,
            "actor": decision.actor,
            "allowed": decision.allowed,
            "reason": decision.reason,
            "articles_checked": decision.articles_checked,
            "duration_ms": decision.duration_ms,
            "timestamp": decision.timestamp.isoformat(),
        }

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-009"])
    def _deserialize_decision(self, data: dict[str, Any]) -> PolicyDecision:
        """Deserialize a dict back to a PolicyDecision."""

        ts = data.get("timestamp")
        if isinstance(ts, str):
            timestamp = datetime.fromisoformat(ts)
        else:
            timestamp = datetime.now(UTC)

        return PolicyDecision(
            decision_type=DecisionType(data["decision_type"]),
            action=data["action"],
            actor=data["actor"],
            allowed=data["allowed"],
            reason=data["reason"],
            articles_checked=data.get("articles_checked", []),
            cache_hit=True,
            duration_ms=data.get("duration_ms", 0.0),
            timestamp=timestamp,
        )

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR10-015",
            "REQ-007-FR10-121",
            "REQ-007-US-001-002",
        ],
    )
    def get(self, key: str) -> PolicyDecision | None:
        """Get a cached decision. Checks local first, then Redis.

        Returns None on cache miss.
        """
        # Level 1: local LRU
        local_data = self._local.get(key)
        if local_data is not None:
            self._hits_local += 1
            return self._deserialize_decision(local_data)

        # Level 2: Redis
        if self._redis is not None:
            try:
                redis_data = self._redis.cache_get(self._redis_key(key))
                if redis_data is not None:
                    self._hits_redis += 1
                    # Promote to local cache
                    self._local.put(key, redis_data)
                    return self._deserialize_decision(redis_data)
            except Exception as exc:
                _logger.warning("Redis cache get failed: %s", exc)

        self._misses += 1
        return None

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR3-006",
            "REQ-007-US-001-009",
        ],
    )
    def set(self, key: str, decision: PolicyDecision) -> None:
        """Store a decision in both cache tiers."""
        data = self._serialize_decision(decision)

        # Level 1: local
        self._local.put(key, data)

        # Level 2: Redis
        if self._redis is not None:
            try:
                self._redis.cache_set(
                    self._redis_key(key),
                    data,
                    ttl_seconds=self._redis_ttl,
                )
            except Exception as exc:
                _logger.warning("Redis cache set failed: %s", exc)

    def invalidate(self, pattern: str) -> int:
        """Invalidate cache entries matching a pattern.

        For local cache, clears everything (LRU doesn't support pattern match).
        For Redis, deletes keys matching the pattern.
        Returns approximate count of invalidated entries.
        """
        count = 0

        # Local: clear all (no pattern support in LRU)
        stats = self._local.get_stats()
        local_size = stats.get("size", 0)
        self._local.clear()
        count += local_size

        # Redis: scan and delete matching keys
        if self._redis is not None:
            try:
                # Lazy import (RULED D7): governance (layer 3) must not
                # top-level-import orchestrator (layer 4).
                from orchestrata_ultimate.infra.redis_state import (  # noqa: PLC0415
                    RedisNamespace,
                )

                # cache_set stores under "<RedisNamespace.CACHE>:<redis_key>",
                # so the scan pattern must carry that namespace too — without
                # it the MATCH never hits a stored key and the delete no-ops.
                redis_pattern = (
                    f"{RedisNamespace.CACHE.value}:{CACHE_REDIS_PREFIX}:{pattern}*"
                )
                client = self._redis.client
                cursor = 0
                while True:
                    # cast: redis-py types command methods as
                    # ``ResponseT | Awaitable[ResponseT]`` because the mixin
                    # backs both sync and async clients; ``client`` here is
                    # RedisStateManager.client, the SYNC client (precedent:
                    # infra/redis_state.py's per-call `cast(...)`).
                    cursor, keys = cast(
                        "tuple[int, list[bytes]]",
                        client.scan(
                            cursor=cursor,
                            match=redis_pattern,
                            count=REDIS_SCAN_BATCH_SIZE,
                        ),
                    )
                    if keys:
                        client.delete(*keys)
                        count += len(keys)
                    if cursor == 0:
                        break
            except Exception as exc:
                _logger.warning("Redis cache invalidate failed: %s", exc)

        return count

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=["REQ-007-US-006-002"],
    )
    def get_hit_rate(self) -> float:
        """Return overall cache hit rate as a percentage."""
        total = self._hits_local + self._hits_redis + self._misses
        if total == 0:
            return 0.0
        return ((self._hits_local + self._hits_redis) / total) * 100.0

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=["REQ-007-US-006-009"],
    )
    def get_stats(self) -> dict[str, Any]:
        """Return cache statistics."""
        local_stats = self._local.get_stats()
        return {
            "local_size": local_stats.get("size", 0),
            "local_max_size": local_stats.get("max_size", 0),
            "local_hit_rate": local_stats.get("hit_rate_percent", 0.0),
            "hits_local": self._hits_local,
            "hits_redis": self._hits_redis,
            "misses": self._misses,
            "overall_hit_rate": self.get_hit_rate(),
        }

    def clear(self) -> None:
        """Clear both cache tiers."""
        self._local.clear()
        if self._redis is not None:
            try:
                self.invalidate("")
            except Exception as exc:
                _logger.warning("Redis cache clear failed: %s", exc)
        self._hits_local = 0
        self._hits_redis = 0
        self._misses = 0


@traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-009"])
def generate_cache_key(action: str, context: dict[str, Any]) -> str:
    """Generate a deterministic cache key from action and context."""
    key_data = f"{action}:{sorted(context.items())}"
    return hashlib.sha256(key_data.encode()).hexdigest()[:32]