orchestrata_ultimate.orchestrator.policy_registry

Reference › orchestrata_ultimate › orchestrator › policy_registry

A versioned store for governance policies, where each policy is a named bundle of Prolog rules. No other reference page imports it. A Policy has an id, name, PolicyType, PolicyStatus, the rule text (prolog_rules), a version number, timestamps and metadata; PolicyVersion records one historical revision.

PolicyRegistry keeps policies, their version history and a list of watcher callbacks in memory. register adds a policy; update replaces its rules, increments the version, appends a PolicyVersion and notifies watchers; rollback returns to an earlier version; deactivate and activate change status; get, get_by_type, get_history and export_all read. add_watcher registers a callback called with the policy id and the event name. If a storage directory is given the constructor creates it, and save_to_disk writes policies.json and history.json there while load_from_disk reads them back and returns the count; without a directory save_to_disk logs a warning and returns false. Methods report failure by returning false (for example for an unknown policy id) rather than raising.

At a glance

Class diagram

Dependency diagram

API

Classes

PolicyType (enum)

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

Types of policies managed by the registry.

PolicyStatus (enum)

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

Policy lifecycle status.

Policy (dataclass)

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

Represents a governance policy with versioning.

PolicyVersion (dataclass)

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

Historical version of a policy.

PolicyRegistry

Decorators: traces(spec='003', story='US-003', reqs=['REQ-003-US-010-025', 'REQ-003-US-003-025', 'REQ-003-US-004-025']) · Line 76

Registry for managing policies with versioning and watchers.

Source

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

"""Policy registry for governance and compliance.

Manages policy lifecycle, versioning, and watchers for runtime policy updates.
"""

# Spec: 007a - Prolog Governance

import json
import logging
from collections.abc import Callable
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime
from enum import Enum
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
class PolicyType(Enum):
    """Types of policies managed by the registry."""

    SECURITY = "security"
    ACCESS = "access"
    COMPLIANCE = "compliance"
    TOOL = "tool"
    SESSION = "session"
    WORKFLOW = "workflow"


@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
class PolicyStatus(Enum):
    """Policy lifecycle status."""

    ACTIVE = "active"
    INACTIVE = "inactive"
    DEPRECATED = "deprecated"


@dataclass
@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
class Policy:
    """Represents a governance policy with versioning."""

    policy_id: str
    name: str
    policy_type: PolicyType
    status: PolicyStatus
    prolog_rules: str
    description: str = ""
    version: int = 1
    created_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
class PolicyVersion:
    """Historical version of a policy."""

    version: int
    prolog_rules: str
    timestamp: str
    updated_by: str = "system"


@traces(
    spec="003",
    story="US-003",
    reqs=["REQ-003-US-010-025", "REQ-003-US-003-025", "REQ-003-US-004-025"],
)
class PolicyRegistry:
    """Registry for managing policies with versioning and watchers."""

    @traces(spec="003", story="US-003", reqs=["REQ-003-US-003-025"])
    def __init__(self, storage_dir: Path | None = None) -> None:
        """Initialize policy registry.

        Args:
            storage_dir: Directory for persisting policies to disk
        """
        self._policies: dict[str, Policy] = {}
        self._history: dict[str, list[PolicyVersion]] = {}
        self._watchers: list[Callable[[str, str], None]] = []
        self._storage_dir = Path(storage_dir) if storage_dir else None

        if self._storage_dir:
            self._storage_dir.mkdir(parents=True, exist_ok=True)

    @traces(spec="003", story="US-003", reqs=["REQ-003-US-003-025"])
    def register(self, policy: Policy) -> bool:
        """Register a new policy.

        Args:
            policy: Policy to register

        Returns:
            True if successful, False if policy already exists
        """
        if policy.policy_id in self._policies:
            logger.warning(f"Policy {policy.policy_id} already exists")
            return False

        self._policies[policy.policy_id] = policy

        # Initialize version history
        self._history[policy.policy_id] = [
            PolicyVersion(
                version=policy.version,
                prolog_rules=policy.prolog_rules,
                timestamp=policy.created_at,
            )
        ]

        self._notify_watchers(policy.policy_id, "registered")
        logger.info(f"Registered policy: {policy.policy_id}")
        return True

    @traces(
        spec="003",
        story="US-003",
        reqs=["REQ-003-US-010-025", "REQ-003-US-003-025", "REQ-003-US-004-025"],
    )
    def update(self, policy_id: str, new_rules: str) -> bool:
        """Update policy rules and increment version.

        Args:
            policy_id: ID of policy to update
            new_rules: New Prolog rules

        Returns:
            True if successful, False if policy not found
        """
        policy = self._policies.get(policy_id)
        if not policy:
            logger.warning(f"Policy {policy_id} not found")
            return False

        # Store current version in history
        policy.version += 1
        policy.prolog_rules = new_rules
        policy.updated_at = datetime.now(UTC).isoformat()

        # Add to version history
        self._history[policy_id].append(
            PolicyVersion(
                version=policy.version,
                prolog_rules=new_rules,
                timestamp=policy.updated_at,
            )
        )

        self._notify_watchers(policy_id, "updated")
        logger.info(f"Updated policy {policy_id} to version {policy.version}")
        return True

    def get(self, policy_id: str) -> Policy | None:
        """Retrieve a policy by ID.

        Args:
            policy_id: Policy identifier

        Returns:
            Policy if found, None otherwise
        """
        return self._policies.get(policy_id)

    def get_by_type(self, policy_type: PolicyType) -> list[Policy]:
        """Get all policies of a specific type.

        Args:
            policy_type: Type of policies to retrieve

        Returns:
            List of policies matching the type
        """
        return [
            policy
            for policy in self._policies.values()
            if policy.policy_type == policy_type
        ]

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-006"])
    def deactivate(self, policy_id: str) -> bool:
        """Deactivate a policy.

        Args:
            policy_id: Policy to deactivate

        Returns:
            True if successful, False if not found
        """
        policy = self._policies.get(policy_id)
        if not policy:
            return False

        policy.status = PolicyStatus.INACTIVE
        policy.updated_at = datetime.now(UTC).isoformat()
        self._notify_watchers(policy_id, "deactivated")
        logger.info(f"Deactivated policy: {policy_id}")
        return True

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-006"])
    def activate(self, policy_id: str) -> bool:
        """Activate a policy.

        Args:
            policy_id: Policy to activate

        Returns:
            True if successful, False if not found
        """
        policy = self._policies.get(policy_id)
        if not policy:
            return False

        policy.status = PolicyStatus.ACTIVE
        policy.updated_at = datetime.now(UTC).isoformat()
        self._notify_watchers(policy_id, "activated")
        logger.info(f"Activated policy: {policy_id}")
        return True

    @traces(spec="007", story="US-PGOV-003", reqs=["REQ-007-US-003-004"])
    def get_history(self, policy_id: str) -> list[PolicyVersion]:
        """Get version history for a policy.

        Args:
            policy_id: Policy identifier

        Returns:
            List of policy versions, empty if not found
        """
        return self._history.get(policy_id, [])

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-006"])
    def rollback(self, policy_id: str, target_version: int) -> bool:
        """Rollback policy to a previous version.

        Args:
            policy_id: Policy to rollback
            target_version: Version number to rollback to

        Returns:
            True if successful, False otherwise
        """
        policy = self._policies.get(policy_id)
        history = self._history.get(policy_id, [])

        if not policy or not history:
            logger.warning(f"Cannot rollback policy {policy_id}: not found")
            return False

        # Find target version in history
        target = next((v for v in history if v.version == target_version), None)
        if not target:
            logger.warning(f"Version {target_version} not found for policy {policy_id}")
            return False

        # Rollback to target version
        policy.prolog_rules = target.prolog_rules
        policy.version = len(history) + 1  # Rollback creates new version
        policy.updated_at = datetime.now(UTC).isoformat()

        # Add rollback to history
        self._history[policy_id].append(
            PolicyVersion(
                version=policy.version,
                prolog_rules=target.prolog_rules,
                timestamp=policy.updated_at,
                updated_by=f"rollback_to_v{target_version}",
            )
        )

        self._notify_watchers(policy_id, "rolled_back")
        logger.info(f"Rolled back policy {policy_id} to version {target_version}")
        return True

    def add_watcher(self, callback: Callable[[str, str], None]) -> None:
        """Add a watcher for policy changes.

        Args:
            callback: Function called with (policy_id, event_type)
        """
        self._watchers.append(callback)
        logger.debug(f"Added policy watcher: {callback.__name__}")

    def _notify_watchers(self, policy_id: str, event: str) -> None:
        """Notify all watchers of a policy event.

        Args:
            policy_id: Policy that changed
            event: Event type (registered, updated, etc.)
        """
        for watcher in self._watchers:
            try:
                watcher(policy_id, event)
            except Exception as e:
                logger.error(f"Watcher {watcher.__name__} failed: {e}")

    def export_all(self) -> list[dict[str, Any]]:
        """Export all policies as dictionaries.

        Returns:
            List of policy dictionaries
        """
        exported = []
        for policy in self._policies.values():
            policy_dict = asdict(policy)
            # Convert enums to strings for JSON serialization
            policy_dict["policy_type"] = policy.policy_type.value
            policy_dict["status"] = policy.status.value
            exported.append(policy_dict)
        return exported

    @traces(spec="003", story="US-003", reqs=["REQ-003-US-003-009"])
    def save_to_disk(self) -> bool:
        """Persist all policies to disk.

        Returns:
            True if successful, False otherwise
        """
        if not self._storage_dir:
            logger.warning("No storage directory configured")
            return False

        try:
            policies_file = self._storage_dir / "policies.json"
            history_file = self._storage_dir / "history.json"

            # Save policies
            exported = self.export_all()
            with open(policies_file, "w") as f:
                json.dump(exported, f, indent=2)

            # Save history
            history_data = {}
            for policy_id, versions in self._history.items():
                history_data[policy_id] = [
                    {
                        "version": v.version,
                        "prolog_rules": v.prolog_rules,
                        "timestamp": v.timestamp,
                        "updated_by": v.updated_by,
                    }
                    for v in versions
                ]

            with open(history_file, "w") as f:
                json.dump(history_data, f, indent=2)

            logger.info(f"Saved {len(self._policies)} policies to disk")
            return True

        except (OSError, ValueError, TypeError) as e:
            logger.error(f"Failed to save policies: {e}")
            return False

    @traces(spec="003", story="US-003", reqs=["REQ-003-US-003-025"])
    def load_from_disk(self) -> int:
        """Load policies from disk.

        Returns:
            Number of policies loaded
        """
        if not self._storage_dir:
            logger.warning("No storage directory configured")
            return 0

        try:
            policies_file = self._storage_dir / "policies.json"
            history_file = self._storage_dir / "history.json"

            if not policies_file.exists():
                logger.info("No policies file found")
                return 0

            # Load policies
            with open(policies_file) as f:
                policies_data = json.load(f)

            for policy_dict in policies_data:
                # Convert string enums back to enums
                policy_dict["policy_type"] = PolicyType(policy_dict["policy_type"])
                policy_dict["status"] = PolicyStatus(policy_dict["status"])

                policy = Policy(**policy_dict)
                self._policies[policy.policy_id] = policy

            # Load history if available
            if history_file.exists():
                with open(history_file) as f:
                    history_data = json.load(f)

                for policy_id, versions_data in history_data.items():
                    self._history[policy_id] = [
                        PolicyVersion(**version_dict) for version_dict in versions_data
                    ]

            logger.info(f"Loaded {len(self._policies)} policies from disk")
            return len(self._policies)

        except (OSError, json.JSONDecodeError, KeyError, ValueError) as e:
            logger.error(f"Failed to load policies: {e}")
            return 0