orchestrata_ultimate.governance.dynamic_policy

Reference › orchestrata_ultimate › governance › dynamic_policy

Runtime management of policy rules, SPEC-007 story US-PGOV-003 (REQ-007-US-003-001 to -009, REQ-007-FR6-002). DynamicPolicyManager implements IDynamicPolicyManager from interfaces and lets rules be added, removed, reloaded and rolled back while the system runs, by asserting them as Prolog facts through the GovernancePrologClient. No other module imports it yet.

add_rule validates the rule with validate_before_activate, rejects it with PolicyValidationError if invalid or if the maximum rule count is reached, saves the current rule set on a rollback stack, assigns an id and asserts a persistent fact named by the rule's predicate (default custom_policy) with the id, action and decision. validate_before_activate returns a ValidationResult: a rule needs an action or predicate, a rule whose action matches an existing rule with a different decision produces a warning, and a failed Prolog valid_policy check is an error. remove_rule removes one, reload asserts every stored rule again, logs if it exceeds a time limit and bumps the version, rollback(steps) restores earlier rule sets from the stack and reloads, and get_version, get_rules and get_rule_count read the state.

At a glance

Class diagram

Dependency diagram

API

Classes

DynamicPolicyManager

Bases: IDynamicPolicyManager · Decorators: traces(spec='007', story='US-PGOV-003', reqs=['REQ-007-US-003-001', 'REQ-007-US-003-002', 'REQ-007-US-003-003', 'REQ-00… · Line 41

Manages dynamic policy rules with hot-reload capability.

Source

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

# mypy: ignore-errors
"""Dynamic policy management for Orchestrata governance.

Supports hot-reload of policy rules without downtime.
"""

from __future__ import annotations

import logging
import time
import uuid
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.governance.constants import (
    MAX_POLICY_RULES,
    POLICY_RELOAD_TIMEOUT_SECONDS,
    REDIS_POLICY_VERSION_KEY,
)
from orchestrata_ultimate.governance.exceptions import PolicyValidationError
from orchestrata_ultimate.governance.interfaces import IDynamicPolicyManager
from orchestrata_ultimate.governance.prolog_client import GovernancePrologClient
from orchestrata_ultimate.governance.types import ValidationResult

_logger = logging.getLogger(__name__)


@traces(
    spec="007",
    story="US-PGOV-003",
    reqs=[
        "REQ-007-US-003-001",
        "REQ-007-US-003-002",
        "REQ-007-US-003-003",
        "REQ-007-US-003-004",
        "REQ-007-US-003-008",
        "REQ-007-US-003-009",
        "REQ-007-US-003-010",
    ],
)
class DynamicPolicyManager(IDynamicPolicyManager):
    """Manages dynamic policy rules with hot-reload capability.

    Rules are stored as Prolog facts and can be added/removed
    at runtime without downtime.
    """

    def __init__(
        self,
        prolog_client: GovernancePrologClient,
        redis: Any | None = None,
    ) -> None:
        self._prolog_client = prolog_client
        self._redis = redis
        self._rules: dict[str, dict[str, Any]] = {}
        self._version = "1.0.0"
        self._version_counter = 0
        self._rollback_stack: list[dict[str, dict[str, Any]]] = []

    async def add_rule(self, rule: dict[str, Any]) -> str:
        """Add a new policy rule.

        Validates the rule before activation.
        Returns the rule ID.
        """
        # Validate
        validation = await self.validate_before_activate(rule)
        if not validation.valid:
            raise PolicyValidationError(
                f"Rule validation failed: {', '.join(validation.errors)}"
            )

        # Check limit
        if len(self._rules) >= MAX_POLICY_RULES:
            raise PolicyValidationError(
                f"Maximum policy rules ({MAX_POLICY_RULES}) reached"
            )

        # Generate rule ID
        rule_id = rule.get("id", str(uuid.uuid4()))
        rule["id"] = rule_id

        # Save rollback point
        self._rollback_stack.append(dict(self._rules))

        # Assert rule as Prolog fact
        predicate = rule.get("predicate", "custom_policy")
        args = [rule_id, rule.get("action", "any"), rule.get("decision", "allow")]
        try:
            from orchestrata_ultimate.core.enums import FactScope

            fact_id = self._prolog_client.bridge.assert_fact(
                predicate, args, FactScope.PERSISTENT
            )
            rule["fact_id"] = fact_id
        except Exception as exc:
            _logger.error("Failed to assert policy rule: %s", exc)
            raise PolicyValidationError(f"Cannot assert rule: {exc}") from exc

        self._rules[rule_id] = rule
        self._bump_version()
        _logger.info("Policy rule added: %s", rule_id)
        return rule_id

    async def remove_rule(self, rule_id: str) -> None:
        """Remove a policy rule by ID."""
        rule = self._rules.get(rule_id)
        if rule is None:
            _logger.warning("Rule not found: %s", rule_id)
            return

        # Save rollback point
        self._rollback_stack.append(dict(self._rules))

        # Retract from Prolog
        fact_id = rule.get("fact_id")
        if fact_id:
            try:
                self._prolog_client.bridge.retract_fact(fact_id)
            except Exception as exc:
                _logger.warning("Failed to retract rule fact: %s", exc)

        del self._rules[rule_id]
        self._bump_version()
        _logger.info("Policy rule removed: %s", rule_id)

    @traces(spec="007", story="US-PGOV-003", reqs=["REQ-007-US-003-001"])
    async def reload(self) -> None:
        """Reload all policy rules from the current rule set.

        Re-asserts all rules in Prolog.
        """
        start = time.monotonic()
        from orchestrata_ultimate.core.enums import FactScope

        for rule_id, rule in self._rules.items():
            predicate = rule.get("predicate", "custom_policy")
            args = [rule_id, rule.get("action", "any"), rule.get("decision", "allow")]
            try:
                fact_id = self._prolog_client.bridge.assert_fact(
                    predicate, args, FactScope.PERSISTENT
                )
                rule["fact_id"] = fact_id
            except Exception as exc:
                _logger.error("Failed to reload rule %s: %s", rule_id, exc)

        elapsed = time.monotonic() - start
        if elapsed > POLICY_RELOAD_TIMEOUT_SECONDS:
            _logger.warning(
                "Policy reload took %.1fs (timeout=%ds)",
                elapsed,
                POLICY_RELOAD_TIMEOUT_SECONDS,
            )

        self._bump_version()
        _logger.info(
            "Policy reload complete: %d rules in %.1fs", len(self._rules), elapsed
        )

    @traces(
        spec="007",
        story="US-PGOV-003",
        reqs=[
            "REQ-007-US-003-004",
            "REQ-007-US-003-010",
        ],
    )
    def get_version(self) -> str:
        """Get current policy version string."""
        return self._version

    @traces(
        spec="007",
        story="US-PGOV-003",
        reqs=["REQ-007-US-003-002", "REQ-007-US-003-008", "REQ-007-FR6-002"],
    )
    async def validate_before_activate(self, rule: dict[str, Any]) -> ValidationResult:
        """Validate a policy rule before activation."""
        errors: list[str] = []
        warnings: list[str] = []

        if not isinstance(rule, dict):
            errors.append("Rule must be a dictionary")
            return ValidationResult(valid=False, errors=errors)

        # Required fields
        if "action" not in rule and "predicate" not in rule:
            errors.append("Rule must have 'action' or 'predicate' field")

        # Check for conflicting rules
        action = rule.get("action")
        if action:
            for existing_id, existing in self._rules.items():
                if existing.get("action") == action:
                    decision = rule.get("decision", "allow")
                    existing_decision = existing.get("decision", "allow")
                    if decision != existing_decision:
                        warnings.append(
                            f"Conflicts with existing rule {existing_id}: "
                            f"'{existing_decision}' vs '{decision}'"
                        )

        # Validate via Prolog if possible
        try:
            if not self._prolog_client.valid_policy(rule):
                errors.append("Prolog validation failed")
        except Exception as exc:
            _logger.warning(
                "Prolog policy validation unavailable: %s", exc, exc_info=True
            )

        return ValidationResult(
            valid=len(errors) == 0,
            errors=errors,
            warnings=warnings,
        )

    @traces(
        spec="007",
        story="US-PGOV-003",
        reqs=[
            "REQ-007-US-003-003",
            "REQ-007-US-003-009",
        ],
    )
    async def rollback(self, steps: int = 1) -> None:
        """Rollback to a previous policy state."""
        for _ in range(min(steps, len(self._rollback_stack))):
            if self._rollback_stack:
                self._rules = self._rollback_stack.pop()
        await self.reload()
        _logger.info("Policy rolled back %d steps", steps)

    def get_rules(self) -> dict[str, dict[str, Any]]:
        """Get all current policy rules."""
        return dict(self._rules)

    def get_rule_count(self) -> int:
        """Get count of active policy rules."""
        return len(self._rules)

    @traces(spec="007", story="US-PGOV-003", reqs=["REQ-007-US-003-004"])
    def _bump_version(self) -> None:
        """Increment policy version."""
        self._version_counter += 1
        self._version = f"1.0.{self._version_counter}"
        if self._redis is not None:
            try:
                self._redis.client.set(REDIS_POLICY_VERSION_KEY, self._version)
            except Exception as exc:
                _logger.warning(
                    "Redis policy version update failed: %s", exc, exc_info=True
                )