orchestrata_ultimate.governance.interfaces

Reference › orchestrata_ultimate › governance › interfaces

The protocol contracts of the governance package, one Protocol per collaborator, so that callers depend on a shape and not on a class. The package initialiser re-exports them, and each concrete class implements one:

• IPolicyEngine (evaluate, evaluate_batch, is_allowed) is implemented by policy_engine.
• IConstitutionalEnforcer (check_article, check_all, get_articles) is implemented by constitutional.
• IEthicalBoundary (check, detect_harm, detect_deception, detect_privacy_violation) is implemented by ethical.
• IDynamicPolicyManager (add_rule, remove_rule, reload, get_version) is implemented by dynamic_policy.
• IResourceAccessController (file, network, API and database checks) is implemented by resource_access.

The argument and result types come from types. The module holds declarations only.

At a glance

Class diagram

Dependency diagram

API

Classes

IPolicyEngine (protocol)

Bases: Protocol · Decorators: traces(spec='007', story='US-PGOV-001', reqs=['REQ-007-US-001-003']) · Line 26

Central governance policy engine.

IConstitutionalEnforcer (protocol)

Bases: Protocol · Line 51

Enforces the 12 constitutional articles.

IEthicalBoundary (protocol)

Bases: Protocol · Decorators: traces(spec='007', story='US-PGOV-005', reqs=['REQ-007-US-005-001', 'REQ-007-US-005-002', 'REQ-007-US-005-003']) · Line 85

Enforces ethical boundaries on actions.

IDynamicPolicyManager (protocol)

Bases: Protocol · Line 109

Manages dynamic policy rules with hot-reload.

IResourceAccessController (protocol)

Bases: Protocol · Decorators: traces(spec='007', story='US-PGOV-004', reqs=['REQ-007-US-004-005', 'REQ-007-US-004-006']) · Line 146

Controls access to files, network, APIs, and databases.

Source

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

"""Governance interfaces for Orchestrata.

Protocol classes defining the governance service contracts.
"""

# Spec: 007a - Prolog Governance

from typing import Any, Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.governance.types import (
    AccessResult,
    Article,
    ComplianceResult,
    EthicalResult,
    HarmAssessment,
    PolicyDecision,
)


@traces(
    spec="007",
    story="US-PGOV-001",
    reqs=["REQ-007-US-001-003"],
)
class IPolicyEngine(Protocol):
    """Central governance policy engine."""

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=["REQ-007-US-001-001"],
    )
    async def evaluate(self, action: str, context: dict[str, Any]) -> PolicyDecision:
        """Evaluate a single action against all governance policies."""
        ...

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-004"])
    async def evaluate_batch(
        self, actions: list[dict[str, Any]]
    ) -> list[PolicyDecision]:
        """Evaluate multiple actions in batch."""
        ...

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-002"])
    async def is_allowed(self, action: str, context: dict[str, Any]) -> bool:
        """Quick check if an action is allowed."""
        ...


class IConstitutionalEnforcer(Protocol):
    """Enforces the 12 constitutional articles."""

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=["REQ-007-FR2-012", "REQ-007-US-002-003"],
    )
    async def check_article(
        self, article_num: int, action: str, context: dict[str, Any]
    ) -> ComplianceResult:
        """Check compliance with a specific article."""
        ...

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=["REQ-007-FR10-017", "REQ-007-US-002-002", "REQ-007-US-002-008"],
    )
    async def check_all(self, action: str, context: dict[str, Any]) -> ComplianceResult:
        """Check compliance with all articles."""
        ...

    @traces(spec="007", story="US-PGOV-002", reqs=["REQ-007-US-002-001"])
    def get_articles(self) -> list[Article]:
        """Return all constitutional article definitions."""
        ...


@traces(
    spec="007",
    story="US-PGOV-005",
    reqs=["REQ-007-US-005-001", "REQ-007-US-005-002", "REQ-007-US-005-003"],
)
class IEthicalBoundary(Protocol):
    """Enforces ethical boundaries on actions."""

    @traces(spec="007", story="FR-PGOV-001", reqs=["REQ-007-FR1-006"])
    async def check(self, action: str, context: dict[str, Any]) -> EthicalResult:
        """Run all ethical checks on an action."""
        ...

    @traces(spec="007", story="FR-PGOV-005", reqs=["REQ-007-FR5-001"])
    async def detect_harm(self, action: str) -> HarmAssessment:
        """Detect potential harm from an action."""
        ...

    @traces(spec="007", story="FR-PGOV-005", reqs=["REQ-007-FR5-002"])
    async def detect_deception(self, action: str) -> bool:
        """Detect if an action involves deception."""
        ...

    @traces(spec="007", story="FR-PGOV-005", reqs=["REQ-007-FR5-003"])
    async def detect_privacy_violation(self, action: str) -> bool:
        """Detect if an action violates privacy."""
        ...


class IDynamicPolicyManager(Protocol):
    """Manages dynamic policy rules with hot-reload."""

    @traces(
        spec="007",
        story="US-PGOV-003",
        reqs=["REQ-007-FR10-051", "REQ-007-US-003-002"],
    )
    async def add_rule(self, rule: dict[str, Any]) -> str:
        """Add a new policy rule. Returns rule ID."""
        ...

    @traces(spec="007", story="US-PGOV-003", reqs=["REQ-007-FR10-051"])
    async def remove_rule(self, rule_id: str) -> None:
        """Remove a policy rule by ID."""
        ...

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

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


@traces(
    spec="007",
    story="US-PGOV-004",
    reqs=["REQ-007-US-004-005", "REQ-007-US-004-006"],
)
class IResourceAccessController(Protocol):
    """Controls access to files, network, APIs, and databases."""

    @traces(spec="007", story="US-PGOV-004", reqs=["REQ-007-US-004-001"])
    async def check_file_access(self, actor: str, path: str, mode: str) -> AccessResult:
        """Check if actor can access a file path with given mode."""
        ...

    @traces(spec="007", story="US-PGOV-004", reqs=["REQ-007-US-004-002"])
    async def check_network_access(
        self, actor: str, domain: str, port: int
    ) -> AccessResult:
        """Check if actor can access a network domain:port."""
        ...

    @traces(spec="007", story="US-PGOV-004", reqs=["REQ-007-US-004-003"])
    async def check_api_access(
        self, actor: str, endpoint: str, method: str
    ) -> AccessResult:
        """Check if actor can access an API endpoint with given method."""
        ...

    @traces(spec="007", story="US-PGOV-004", reqs=["REQ-007-US-004-004"])
    async def check_db_access(
        self, actor: str, schema: str, operation: str
    ) -> AccessResult:
        """Check if actor can perform a database operation."""
        ...