orchestrata_ultimate.governance.ethical

Reference › orchestrata_ultimate › governance › ethical

The ethical-boundary check, SPEC-007 story US-PGOV-005 (REQ-007-FR5-001, REQ-007-US-005-001, -006, -012). EthicalBoundary implements IEthicalBoundary from interfaces and asks the rules in ethics.pl, through the GovernancePrologClient, whether an action crosses an ethical line. policy_engine is the importer and runs it after the constitutional check.

check(action, context) runs five detectors and returns an EthicalResult from types with the list of concerns, the harm level and the number of checks performed. detect_harm queries ethical_harm_check/3 and treats a solution of medium, high or critical severity as harm, returning a HarmAssessment with type and severity. detect_deception, detect_privacy_violation, detect_bias and _detect_autonomy_violation each return a boolean. A concern is recorded as harm:<type>:<severity>, deception_detected, privacy_violation, bias_detected or autonomy_violation, and the action is ethical only when there are none. A failed harm query is logged as a warning and counts as no harm found.

At a glance

Class diagram

Dependency diagram

API

Classes

EthicalBoundary

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

Enforces ethical boundaries on actions via Prolog.

Source

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

"""Ethical boundary enforcement for Orchestrata governance.

Implements ethical checks via Prolog queries to ethics.pl.
"""

import logging
import time
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.governance.interfaces import IEthicalBoundary
from orchestrata_ultimate.governance.prolog_client import GovernancePrologClient
from orchestrata_ultimate.governance.types import EthicalResult, HarmAssessment

_logger = logging.getLogger(__name__)

# Harm severity levels ordered
HARM_SEVERITY_ORDER = ["none", "low", "medium", "high", "critical"]


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

    Checks for harm, deception, privacy violations, bias,
    and autonomy violations.
    """

    def __init__(self, prolog_client: GovernancePrologClient) -> None:
        self._prolog_client = prolog_client

    @traces(
        spec="007",
        story="US-PGOV-005",
        reqs=[
            "REQ-007-US-005-006",
            "REQ-007-US-005-012",
        ],
    )
    async def check(self, action: str, context: dict[str, Any]) -> EthicalResult:
        """Run all ethical checks on an action."""
        start = time.monotonic()
        concerns: list[str] = []
        harm_level = "none"

        # 1. Harm check
        harm = await self.detect_harm(action)
        if harm.has_harm:
            concerns.append(f"harm:{harm.harm_type}:{harm.severity}")
            harm_level = harm.severity

        # 2. Deception check
        if await self.detect_deception(action):
            concerns.append("deception_detected")

        # 3. Privacy check
        if await self.detect_privacy_violation(action):
            concerns.append("privacy_violation")

        # 4. Bias check
        if await self.detect_bias(action):
            concerns.append("bias_detected")

        # 5. Autonomy check
        if await self._detect_autonomy_violation(action):
            concerns.append("autonomy_violation")

        elapsed_ms = (time.monotonic() - start) * 1000
        ethical = len(concerns) == 0
        return EthicalResult(
            ethical=ethical,
            concerns=concerns,
            harm_level=harm_level,
            details={
                "action": action,
                "duration_ms": elapsed_ms,
                "checks_performed": 5,
            },
        )

    @traces(
        spec="007",
        story="US-PGOV-005",
        reqs=[
            "REQ-007-FR5-001",
            "REQ-007-US-005-001",
        ],
    )
    async def detect_harm(self, action: str) -> HarmAssessment:
        """Detect potential harm from an action via Prolog."""
        try:
            # Resolve the bridge INSIDE the try: unlike detect_deception /
            # detect_privacy_violation / detect_bias / _detect_autonomy_
            # violation below (which all wrap `self._prolog_client.bridge.
            # query_bool(...)` as one expression), this method used to
            # resolve `bridge = self._prolog_client.bridge` *before* the
            # try — so a broken/unavailable prolog_client raised
            # AttributeError straight out of detect_harm(), uncaught,
            # unlike every sibling method. NOTE the shared posture is
            # fail-OPEN: on any prolog failure this falls through to
            # HarmAssessment(has_harm=False), i.e. a broken engine passes
            # all five ethical checks. SPEC-014 ERR-011 wants fail-closed
            # (deny on engine-unavailable) — flipping that posture is a
            # deliberate product decision tracked in the backlog, not a
            # side effect of this consistency fix.
            bridge = self._prolog_client.bridge
            result = bridge.query(f"ethical_harm_check('{action}', HarmType, Severity)")
            if result.success and result.solutions:
                for sol in result.solutions:
                    harm_type = str(sol.get("HarmType", "none"))
                    severity = str(sol.get("Severity", "none"))
                    if severity in ("high", "critical"):
                        return HarmAssessment(
                            has_harm=True,
                            harm_type=harm_type,
                            severity=severity,
                        )
                    if severity == "medium":
                        return HarmAssessment(
                            has_harm=True,
                            harm_type=harm_type,
                            severity=severity,
                        )
        except Exception as exc:
            _logger.warning("Harm detection query failed: %s", exc)

        return HarmAssessment(has_harm=False)

    @traces(
        spec="007",
        story="US-PGOV-005",
        reqs=[
            "REQ-007-FR5-002",
            "REQ-007-US-005-002",
            "REQ-007-US-005-008",
        ],
    )
    async def detect_deception(self, action: str) -> bool:
        """Detect if an action involves deception."""
        try:
            return self._prolog_client.bridge.query_bool(
                f"ethical_deception_check('{action}', true)"
            )
        except Exception as exc:
            _logger.warning("ethical_deception_check failed: %s", exc, exc_info=True)
            return False

    @traces(
        spec="007",
        story="US-PGOV-005",
        reqs=[
            "REQ-007-FR5-003",
            "REQ-007-US-005-003",
            "REQ-007-US-005-009",
        ],
    )
    async def detect_privacy_violation(self, action: str) -> bool:
        """Detect if an action violates privacy."""
        try:
            return self._prolog_client.bridge.query_bool(
                f"ethical_privacy_check('{action}', violation)"
            )
        except Exception as exc:
            _logger.warning("ethical_privacy_check failed: %s", exc, exc_info=True)
            return False

    @traces(
        spec="007",
        story="US-PGOV-005",
        reqs=[
            "REQ-007-FR5-004",
            "REQ-007-US-005-004",
            "REQ-007-US-005-010",
        ],
    )
    async def detect_bias(self, action: str) -> bool:
        """Detect if an action involves bias."""
        try:
            return self._prolog_client.bridge.query_bool(
                f"ethical_bias_check('{action}', biased)"
            )
        except Exception as exc:
            _logger.warning("ethical_bias_check failed: %s", exc, exc_info=True)
            return False

    @traces(
        spec="007",
        story="US-PGOV-005",
        reqs=[
            "REQ-007-FR5-005",
            "REQ-007-US-005-005",
            "REQ-007-US-005-011",
        ],
    )
    async def _detect_autonomy_violation(self, action: str) -> bool:
        """Detect if an action violates user autonomy."""
        try:
            return self._prolog_client.bridge.query_bool(
                f"ethical_autonomy_check('{action}', violation)"
            )
        except Exception as exc:
            _logger.warning("ethical_autonomy_check failed: %s", exc, exc_info=True)
            return False