orchestrata_ultimate.orchestrator.fact_manager

Reference › orchestrata_ultimate › orchestrator › fact_manager

FactManager governs how long governance facts live. It is imported by engine, which creates one for the policy engine, and uses the FactScope enum from core.enums. It does not store facts itself: the constructor takes an assert_fn that stores a fact and returns its id, a retract_fn that removes it (the durable fact storage itself is in fact_store), and a cleanup interval of 60 seconds.

register_fact holds a lock while it calls assert_fn, so registration is atomic, and raises RuntimeError if the assertion fails. It records a FactMetadata (predicate, arguments, scope, session id, creation time and optional time-to-live) in a map, in a per-session set or the per-request set, and pushes the expiry onto a min-heap so expired facts can be found cheaply. FactMetadata.expires_at and is_expired compute expiry from the TTL. Facts leave in four ways: unregister_fact for one fact; on_session_end, which retracts every fact of a session and returns the count; on_request_end, which clears request-scoped facts (after on_request_start); and cleanup_expired, run by a background daemon thread started with start and stopped with stop. Inspection methods are get_fact, get_session_facts, get_facts_by_predicate, get_stats (a FactStats record), export_facts and clear_all.

At a glance

Class diagram

Dependency diagram

API

Classes

FactMetadata (dataclass)

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

Metadata for tracking facts.

FactStats (dataclass)

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

Statistics about facts in the system.

FactManager

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

Manages fact lifecycle across the governance system.

Functions

Source

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

# mypy: ignore-errors
"""Fact lifecycle management for Orchestrata.

Manages Prolog fact creation, cleanup, and synchronization
across sessions and requests.

Based on DESIGN_DECISIONS.md specifications.
"""

# Spec: 006 - Session Management

import heapq
import logging
import threading
import time
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.enums import FactScope

# Module-level logger
_logger = logging.getLogger(__name__)


@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
def _utc_now() -> datetime:
    """Get current UTC time with timezone info."""
    return datetime.now(UTC)


@dataclass
@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
class FactMetadata:
    """Metadata for tracking facts."""

    fact_id: str
    predicate: str
    arguments: tuple
    scope: FactScope
    session_id: str | None
    created_at: datetime
    ttl: timedelta | None = None

    @property
    def expires_at(self) -> datetime | None:
        """Calculate expiration time."""
        if self.ttl:
            return self.created_at + self.ttl
        return None

    def is_expired(self) -> bool:
        """Check if fact has expired."""
        if not self.ttl:
            return False
        # Compute directly rather than through `expires_at`: `self.ttl` is
        # confirmed truthy here, so the sum is a `datetime`, but that
        # guarantee doesn't carry through the property call for pyright.
        return _utc_now() > self.created_at + self.ttl


@dataclass
@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
class FactStats:
    """Statistics about facts in the system."""

    total: int = 0
    by_scope: dict[str, int] = field(default_factory=dict)
    by_session: dict[str, int] = field(default_factory=dict)
    expired_cleaned: int = 0
    session_cleaned: int = 0
    request_cleaned: int = 0


@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
class FactManager:
    """Manages fact lifecycle across the governance system.

    Provides:
    - Fact registration and tracking
    - Scope-based cleanup (session, request)
    - TTL-based expiration
    - Statistics and monitoring
    """

    def __init__(
        self,
        assert_fn: Callable[[str, tuple], str],
        retract_fn: Callable[[str], bool],
        cleanup_interval_seconds: int = 60,
    ):
        """Initialize fact manager.

        Args:
            assert_fn: Function to assert facts (predicate, args) -> fact_id
            retract_fn: Function to retract facts (fact_id) -> success
            cleanup_interval_seconds: Interval for background cleanup.
        """
        self._assert_fn = assert_fn
        self._retract_fn = retract_fn
        self._cleanup_interval = cleanup_interval_seconds

        # Fact tracking
        self._facts: dict[str, FactMetadata] = {}
        self._session_facts: dict[str, set[str]] = {}
        self._request_facts: set[str] = set()

        # Expiration heap for O(log n) expired fact lookup
        # Each entry is (expiration_datetime, fact_id)
        self._expiration_heap: list[tuple[datetime, str]] = []

        # Statistics
        self._stats = FactStats()

        # Threading
        self._lock = threading.Lock()
        self._cleanup_thread: threading.Thread | None = None
        self._running = False

    # ===== Lifecycle Management =====

    def start(self) -> None:
        """Start background cleanup thread."""
        if self._running:
            return
        self._running = True
        self._cleanup_thread = threading.Thread(target=self._cleanup_loop, daemon=True)
        self._cleanup_thread.start()

    def stop(self) -> None:
        """Stop background cleanup thread."""
        self._running = False
        if self._cleanup_thread:
            self._cleanup_thread.join(timeout=5.0)
            self._cleanup_thread = None

    def _cleanup_loop(self) -> None:
        """Background loop for periodic cleanup."""
        while self._running:
            time.sleep(self._cleanup_interval)
            if self._running:
                self.cleanup_expired()

    # ===== Fact Registration =====

    @traces(
        spec="006",
        story="US-004",
        reqs=["REQ-006-US-004-001", "REQ-006-FR8-001", "REQ-006-FR3-001"],
    )
    def register_fact(
        self,
        predicate: str,
        arguments: tuple,
        scope: FactScope = FactScope.SESSION,
        session_id: str | None = None,
        ttl_seconds: int | None = None,
    ) -> str:
        """Register a new fact.

        Thread-safe: Acquires lock before assert to ensure atomic registration.

        Args:
            predicate: Prolog predicate name.
            arguments: Fact arguments as tuple.
            scope: Fact lifecycle scope.
            session_id: Session ID for session-scoped facts.
            ttl_seconds: Optional time-to-live.

        Returns:
            Fact ID.

        Raises:
            RuntimeError: If fact assertion fails.
        """
        with self._lock:
            # Assert the fact via provided function INSIDE lock to prevent race condition
            try:
                fact_id = self._assert_fn(predicate, arguments)
            except Exception as e:
                _logger.error(f"Failed to assert fact {predicate}: {e}")
                raise RuntimeError(f"Fact assertion failed: {e}") from e

            created_at = _utc_now()

            # Create metadata
            metadata = FactMetadata(
                fact_id=fact_id,
                predicate=predicate,
                arguments=arguments,
                scope=scope,
                session_id=session_id,
                created_at=created_at,
                ttl=timedelta(seconds=ttl_seconds) if ttl_seconds else None,
            )

            # Track the fact
            self._facts[fact_id] = metadata

            # Add to expiration heap if TTL is set (O(log n) insertion)
            if ttl_seconds and metadata.expires_at:
                heapq.heappush(self._expiration_heap, (metadata.expires_at, fact_id))

            # Update scope tracking
            if scope == FactScope.SESSION and session_id:
                if session_id not in self._session_facts:
                    self._session_facts[session_id] = set()
                self._session_facts[session_id].add(fact_id)
            elif scope == FactScope.REQUEST:
                self._request_facts.add(fact_id)

            # Update stats
            self._stats.total += 1
            scope_key = scope.value
            self._stats.by_scope[scope_key] = self._stats.by_scope.get(scope_key, 0) + 1
            if session_id:
                self._stats.by_session[session_id] = (
                    self._stats.by_session.get(session_id, 0) + 1
                )

        return fact_id

    @traces(spec="006", story="US-002", reqs=["REQ-006-US-002-003"])
    def unregister_fact(self, fact_id: str) -> bool:
        """Unregister and retract a fact.

        Thread-safe: Acquires lock to update tracking, calls retract outside lock
        to prevent potential deadlock.

        Args:
            fact_id: Fact identifier.

        Returns:
            True if fact was unregistered.
        """
        # First, get metadata and remove from tracking under lock
        with self._lock:
            if fact_id not in self._facts:
                return False

            metadata = self._facts[fact_id]

            # Remove from tracking BEFORE retract to prevent re-use during retract
            del self._facts[fact_id]

            if metadata.scope == FactScope.SESSION and metadata.session_id:
                self._session_facts.get(metadata.session_id, set()).discard(fact_id)
            elif metadata.scope == FactScope.REQUEST:
                self._request_facts.discard(fact_id)

            # Update stats
            self._stats.total -= 1
            scope_key = metadata.scope.value
            self._stats.by_scope[scope_key] = max(
                0, self._stats.by_scope.get(scope_key, 1) - 1
            )
            if metadata.session_id:
                self._stats.by_session[metadata.session_id] = max(
                    0, self._stats.by_session.get(metadata.session_id, 1) - 1
                )

        # Retract OUTSIDE lock to prevent deadlock
        try:
            success = self._retract_fn(fact_id)
            if not success:
                _logger.warning(f"Failed to retract fact {fact_id} from Prolog")
                # Note: We've already removed from tracking, so this is a partial failure
                # The fact is effectively unregistered from the manager
            return True  # Return True because we unregistered from manager
        except Exception as e:
            _logger.error(f"Error retracting fact {fact_id}: {e}")
            return True  # Still return True - fact is unregistered from manager

    # ===== Session Lifecycle =====

    def on_session_start(self, session_id: str) -> None:
        """Initialize tracking for a new session.

        Args:
            session_id: Session identifier.
        """
        with self._lock:
            if session_id not in self._session_facts:
                self._session_facts[session_id] = set()

    @traces(spec="006", story="US-005", reqs=["REQ-006-FR5-001"])
    def on_session_end(self, session_id: str) -> int:
        """Clean up all facts for a session.

        Args:
            session_id: Session identifier.

        Returns:
            Number of facts cleaned up.
        """
        with self._lock:
            fact_ids = self._session_facts.get(session_id, set()).copy()

        count = 0
        for fact_id in fact_ids:
            if self.unregister_fact(fact_id):
                count += 1

        with self._lock:
            if session_id in self._session_facts:
                del self._session_facts[session_id]
            if session_id in self._stats.by_session:
                del self._stats.by_session[session_id]
            self._stats.session_cleaned += count

        return count

    # ===== Request Lifecycle =====

    def on_request_start(self) -> None:
        """Initialize tracking for a new request."""
        # Request facts are tracked automatically via scope

    def on_request_end(self) -> int:
        """Clean up all request-scoped facts.

        Returns:
            Number of facts cleaned up.
        """
        with self._lock:
            fact_ids = self._request_facts.copy()

        count = 0
        for fact_id in fact_ids:
            if self.unregister_fact(fact_id):
                count += 1

        with self._lock:
            self._stats.request_cleaned += count

        return count

    # ===== TTL Cleanup =====

    def cleanup_expired(self) -> int:
        """Clean up expired facts using O(log n) heap-based lookup.

        Returns:
            Number of facts cleaned up.
        """
        now = _utc_now()
        expired_fact_ids: list[str] = []

        with self._lock:
            # Pop expired entries from heap (O(k log n) where k is expired count)
            while self._expiration_heap:
                expires_at, fact_id = self._expiration_heap[0]

                if expires_at > now:
                    # No more expired facts (heap is sorted by expiration time)
                    break

                # Pop the expired entry
                heapq.heappop(self._expiration_heap)

                # Verify fact still exists and is actually expired
                # (fact might have been unregistered already)
                if fact_id in self._facts and self._facts[fact_id].is_expired():
                    expired_fact_ids.append(fact_id)

        # Unregister expired facts (outside main lock iteration)
        count = 0
        for fact_id in expired_fact_ids:
            if self.unregister_fact(fact_id):
                count += 1

        with self._lock:
            self._stats.expired_cleaned += count

        return count

    # ===== Queries =====

    def get_fact(self, fact_id: str) -> FactMetadata | None:
        """Get fact metadata.

        Args:
            fact_id: Fact identifier.

        Returns:
            FactMetadata or None.
        """
        with self._lock:
            return self._facts.get(fact_id)

    @traces(
        spec="006",
        story="US-003",
        reqs=["REQ-006-US-003-002", "REQ-006-FR3-004"],
    )
    def get_session_facts(self, session_id: str) -> list[FactMetadata]:
        """Get all facts for a session.

        Args:
            session_id: Session identifier.

        Returns:
            List of FactMetadata.
        """
        with self._lock:
            fact_ids = self._session_facts.get(session_id, set())
            return [self._facts[fid] for fid in fact_ids if fid in self._facts]

    @traces(
        spec="006",
        story="US-009",
        reqs=["REQ-006-FR9-002"],
    )
    def get_facts_by_predicate(self, predicate: str) -> list[FactMetadata]:
        """Get all facts with a given predicate.

        Args:
            predicate: Predicate name.

        Returns:
            List of FactMetadata.
        """
        with self._lock:
            return [
                metadata
                for metadata in self._facts.values()
                if metadata.predicate == predicate
            ]

    @traces(spec="006", story="US-012", reqs=["REQ-006-FR12-003"])
    def get_stats(self) -> FactStats:
        """Get current statistics.

        Returns:
            Copy of current stats.
        """
        with self._lock:
            return FactStats(
                total=self._stats.total,
                by_scope=dict(self._stats.by_scope),
                by_session=dict(self._stats.by_session),
                expired_cleaned=self._stats.expired_cleaned,
                session_cleaned=self._stats.session_cleaned,
                request_cleaned=self._stats.request_cleaned,
            )

    # ===== Bulk Operations =====

    def clear_all(self) -> int:
        """Clear all facts (use with caution).

        Returns:
            Number of facts cleared.
        """
        with self._lock:
            all_facts = list(self._facts.keys())

        count = 0
        for fact_id in all_facts:
            if self.unregister_fact(fact_id):
                count += 1

        return count

    def export_facts(self) -> list[dict[str, Any]]:
        """Export all facts for debugging/persistence.

        Returns:
            List of fact dictionaries.
        """
        with self._lock:
            return [
                {
                    "fact_id": m.fact_id,
                    "predicate": m.predicate,
                    "arguments": m.arguments,
                    "scope": m.scope.value,
                    "session_id": m.session_id,
                    "created_at": m.created_at.isoformat(),
                    "ttl_seconds": m.ttl.total_seconds() if m.ttl else None,
                }
                for m in self._facts.values()
            ]