orchestrata_ultimate.governance.constitutional

Reference › orchestrata_ultimate › governance › constitutional

The constitutional enforcer, SPEC-007 story US-PGOV-002 (REQ-007-FR2-012, -013, REQ-007-FR10-017, -119, REQ-007-US-002-002, -003, -008). The constitution is a numbered set of articles written as Prolog rules, and ConstitutionalEnforcer implements IConstitutionalEnforcer from interfaces to ask whether an action complies with them. policy_engine builds one by default and calls check_critical first; the kernel dispatcher calls check_article for Article XII (plan fidelity) in audit-only mode; and the gateway governance handler imports it. The queries go through the GovernancePrologClient. The curated governance page describes the articles as eleven; the code's check_all docstring says twelve, and the article count is a module constant.

check_article(n, action, context) returns a ComplianceResult from types: an article number outside the valid range or a raised error is reported as non-compliant, otherwise it asks constitutional_check/3 and _article_result builds the result with the elapsed time. check_all runs every article in turn and merges the passed and failed lists with per-article details and the total duration. check_critical runs only the critical safety articles (I, III and VII). get_articles and get_article return the article definitions. human_override(action, actor, reason) records a human override by asserting two ground facts, human_override_active and human_consent_given, at session scope; the docstring notes the Prolog rule no longer asserts them itself (ADR-0005 and ADR-0025). requires_human_consent(action) asks requires_human_consent/1, and on a Prolog error returns True (consent required) when strict mode is on.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ConstitutionalEnforcer

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

Enforces the 12 constitutional articles via Prolog queries.

Source

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

"""Constitutional enforcer for Orchestrata governance.

Implements all 12 constitutional articles via Prolog queries.
"""

import logging
import time
from collections.abc import Mapping
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.governance._strict import governance_strict
from orchestrata_ultimate.governance.constants import (
    ARTICLE_COUNT,
    CRITICAL_ARTICLES,
)
from orchestrata_ultimate.governance.interfaces import IConstitutionalEnforcer
from orchestrata_ultimate.governance.prolog_client import GovernancePrologClient
from orchestrata_ultimate.governance.types import (
    Article,
    ArticlePriority,
    ComplianceResult,
)

_logger = logging.getLogger(__name__)

# Article definitions with priority phases
ARTICLES: list[Article] = [
    Article(
        1,
        "Human Autonomy",
        "Humans retain ultimate control and override authority",
        ArticlePriority.PHASE_1,
        ["human_override", "requires_human_consent"],
    ),
    Article(
        2,
        "Transparency",
        "All operations must be explainable and auditable",
        ArticlePriority.PHASE_2,
        ["operation_transparent"],
    ),
    Article(
        3,
        "Harm Prevention",
        "System must not cause harm to users or third parties",
        ArticlePriority.PHASE_1,
        ["may_not_cause_harm", "assess_harm"],
    ),
    Article(
        4,
        "Privacy",
        "Personal data must be protected and handled lawfully",
        ArticlePriority.PHASE_2,
        ["respects_privacy", "handle_pii"],
    ),
    Article(
        5,
        "Fairness",
        "System must operate without bias or discrimination",
        ArticlePriority.PHASE_3,
        ["operates_fairly", "detect_bias"],
    ),
    Article(
        6,
        "Accountability",
        "All actions must have clear accountability trails",
        ArticlePriority.PHASE_2,
        ["accountable_operation"],
    ),
    Article(
        7,
        "Security",
        "System integrity and security must be maintained",
        ArticlePriority.PHASE_1,
        ["secure_operation", "verify_integrity"],
    ),
    Article(
        8,
        "Sustainability",
        "Resources must be used efficiently",
        ArticlePriority.PHASE_4,
        ["resource_efficient"],
    ),
    Article(
        9,
        "Dignity",
        "System must respect human dignity",
        ArticlePriority.PHASE_3,
        ["respects_dignity"],
    ),
    Article(
        10,
        "Rights",
        "System must protect fundamental rights",
        ArticlePriority.PHASE_3,
        ["protects_rights"],
    ),
    Article(
        11,
        "Continuous Improvement",
        "System must support valid improvements",
        ArticlePriority.PHASE_4,
        ["valid_improvement"],
    ),
    Article(
        12,
        "Plan Fidelity",
        "An active ratified plan is executed as written; "
        "divergence only via a human-raised issue",
        ArticlePriority.PHASE_1,
        ["plan_fidelity"],
    ),
]


@traces(
    spec="007",
    story="US-PGOV-002",
    reqs=[
        "REQ-007-US-002-001",
        "REQ-007-US-002-002",
        "REQ-007-US-002-003",
        "REQ-007-US-002-008",
        "REQ-007-FR2-012",
        "REQ-007-FR10-119",
    ],
)
class ConstitutionalEnforcer(IConstitutionalEnforcer):
    """Enforces the 12 constitutional articles via Prolog queries.

    Each article check delegates to constitutional.pl predicates.
    """

    def __init__(self, prolog_client: GovernancePrologClient) -> None:
        self._prolog_client = prolog_client
        self._articles = {a.number: a for a in 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 constitutional article."""
        if article_num < 1 or article_num > ARTICLE_COUNT:
            return ComplianceResult(
                compliant=False,
                articles_failed=[article_num],
                details={"error": f"Invalid article number: {article_num}"},
            )

        start = time.monotonic()
        try:
            compliant = self._prolog_client.constitutional_check(
                article_num, action, context
            )
        except Exception as exc:
            _logger.error(
                "Constitutional check failed for article %d: %s",
                article_num,
                exc,
            )
            return ComplianceResult(
                compliant=False,
                articles_failed=[article_num],
                details={"error": str(exc)},
            )

        return self._article_result(
            article_num, compliant, (time.monotonic() - start) * 1000
        )

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=[
            "REQ-007-FR2-012",
            "REQ-007-US-002-003",
        ],
    )
    def _article_result(
        self, article_num: int, compliant: bool, elapsed_ms: float
    ) -> ComplianceResult:
        """Build the ComplianceResult for a completed per-article check."""
        if compliant:
            return ComplianceResult(
                compliant=True,
                articles_passed=[article_num],
                details={"duration_ms": elapsed_ms},
            )
        return ComplianceResult(
            compliant=False,
            articles_failed=[article_num],
            details={
                "duration_ms": elapsed_ms,
                "article": self._articles.get(
                    article_num, Article(article_num, "Unknown", "")
                ),
            },
        )

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=[
            "REQ-007-FR10-017",
            "REQ-007-FR10-119",
            "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 12 articles."""
        passed: list[int] = []
        failed: list[int] = []
        details: dict[str, Any] = {}

        start = time.monotonic()
        for article_num in range(1, ARTICLE_COUNT + 1):
            result = await self.check_article(article_num, action, context)
            if result.compliant:
                passed.extend(result.articles_passed)
            else:
                failed.extend(result.articles_failed)
                details[f"article_{article_num}"] = result.details

        elapsed_ms = (time.monotonic() - start) * 1000
        details["total_duration_ms"] = elapsed_ms

        return ComplianceResult(
            compliant=len(failed) == 0,
            articles_passed=passed,
            articles_failed=failed,
            details=details,
        )

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=[
            "REQ-007-FR2-013",
            "REQ-007-US-002-002",
        ],
    )
    async def check_critical(
        self, action: str, context: dict[str, Any]
    ) -> ComplianceResult:
        """Check only critical safety articles (I, III, VII)."""
        passed: list[int] = []
        failed: list[int] = []
        details: dict[str, Any] = {}

        for article_num in CRITICAL_ARTICLES:
            result = await self.check_article(article_num, action, context)
            if result.compliant:
                passed.extend(result.articles_passed)
            else:
                failed.extend(result.articles_failed)
                details[f"article_{article_num}"] = result.details

        return ComplianceResult(
            compliant=len(failed) == 0,
            articles_passed=passed,
            articles_failed=failed,
            details=details,
        )

    def get_articles(self) -> list[Article]:
        """Return all 12 constitutional article definitions."""
        return list(ARTICLES)

    def get_article(self, number: int) -> Article | None:
        """Return a specific article by number."""
        return self._articles.get(number)

    @traces(
        spec="007",
        story="FR-PGOV-003",
        reqs=["REQ-007-FR3-004"],
    )
    async def human_override(self, action: str, actor: str, reason: str) -> None:
        """Record a human override for an action.

        Asserts BOTH override facts as ground facts from here — the
        constitutional.pl rule no longer performs assertz side effects
        (ADR-0005/0025, ratified 2026-07-24): facts come from outside the
        rulebase.
        """
        fact_scope = __import__(
            "orchestrata_ultimate.core.enums", fromlist=["FactScope"]
        ).FactScope.SESSION
        self._prolog_client.bridge.assert_fact(
            "human_override_active",
            [action, actor, reason],
            scope=fact_scope,
        )
        self._prolog_client.bridge.assert_fact(
            "human_consent_given",
            [action, actor],
            scope=fact_scope,
        )
        _logger.info(
            "Human override recorded: action=%s actor=%s reason=%s",
            action,
            actor,
            reason,
        )

    async def requires_human_consent(
        self, action: str, env: Mapping[str, str] | None = None
    ) -> bool:
        """Check if an action requires human consent.

        On a Prolog error, fails CLOSED (requires consent) when governance
        strict mode is on (the default); set
        ``ORCHESTRATA_GOVERNANCE_STRICT=false`` (or pass ``env=``) to restore
        the permissive legacy behavior.
        """
        try:
            return self._prolog_client.bridge.query_bool(
                f"requires_human_consent('{action}')"
            )
        except Exception as exc:
            _logger.error("requires_human_consent Prolog query failed: %s", exc)
            return governance_strict(env)