orchestrata_ultimate.agents.constraints

Reference › orchestrata_ultimate › agents › constraints

constraints validates agent operations against the constitutional articles (specification 005, user story US-005). ConstitutionalConstraints is the validator, Constraint is a dataclass describing one rule (id, name, article number, scope and severity), and ConstraintResult carries the outcome: a valid flag, a list of violation messages and a severity. It is used by AgentLifecycle in lifecycle and depends on the facade in prolog_bridge.

validate_at_spawn checks an agent definition before it is spawned. Its helpers _check_alias_and_resources (alias present, CPU, memory and depth limits) and _check_prolog_registration (the registration check in the Prolog rules, which appends a violation on denial) fill the violation list. validate_at_runtime checks a named action by an existing agent against each constraint through _check_constraint. check_fact_visibility answers whether an agent may see facts of a given scope. The validator accepts a custom list of constraints and an optional MultiagentPrologBridge; the curated agents page describes the governance model.

At a glance

Class diagram

Dependency diagram

API

Classes

ConstraintResult (dataclass)

Decorators: dataclass, traces(spec='005', story='US-005', reqs=['REQ-005-AC3-001', 'REQ-005-FR4-001']) · Line 33

Result of a constraint validation.

Constraint (dataclass)

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

A single constitutional constraint.

ConstitutionalConstraints

Decorators: traces(spec='005', story='US-005', reqs=['REQ-005-AC3-001', 'REQ-005-FR4-001', 'REQ-005-FR6-001', 'REQ-005-US-005-001']) · Line 121

Validates agent operations against constitutional articles.

Source

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

# mypy: ignore-errors
"""Constitutional constraint validation for agents.

Validates agent operations against constitutional articles,
enforces scope boundaries, and checks resource limits.
"""

# Spec: 005 - Multi-Agent System

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge

logger = logging.getLogger(__name__)


@dataclass
@traces(
    spec="005",
    story="US-005",
    reqs=[
        "REQ-005-AC3-001",
        "REQ-005-FR4-001",
    ],
)
class ConstraintResult:
    """Result of a constraint validation."""

    valid: bool
    violations: list[str] = field(default_factory=list)
    severity: str = "none"  # "none", "warning", "error", "critical"

    @property
    @traces(
        spec="005",
        story="US-005",
        reqs=["REQ-005-AC3-001", "REQ-005-FR4-001"],
    )
    def has_violations(self) -> bool:
        return len(self.violations) > 0


@dataclass
@traces(
    spec="005",
    story="US-005",
    reqs=["REQ-005-US-005-001"],
)
class Constraint:
    """A single constitutional constraint."""

    id: str
    name: str
    article: int
    description: str = ""
    scope: str = "all"  # "spawn", "runtime", "terminate"
    severity: str = "error"

    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "name": self.name,
            "article": self.article,
            "description": self.description,
            "scope": self.scope,
            "severity": self.severity,
        }


# Default constitutional constraints
DEFAULT_CONSTRAINTS = [
    Constraint(
        "c-art1-scope", "Scope Limitation", 1, "Agent operates within defined scope"
    ),
    Constraint(
        "c-art3-context",
        "Context Preservation",
        3,
        "Context must be preserved during operations",
    ),
    Constraint(
        "c-art4-lifespan", "Lifespan Control", 4, "Agent cannot extend own lifespan"
    ),
    Constraint("c-art5-di", "Dependency Injection", 5, "Dependencies must be injected"),
    Constraint(
        "c-art7-failure", "Failure Context", 7, "Failure context must be preserved"
    ),
    Constraint(
        "c-art9-escalation",
        "Escalation Required",
        9,
        "Cross-scope requires escalation",
        scope="runtime",
    ),
    Constraint(
        "c-art11-immutable",
        "Immutable After Spawn",
        11,
        "Constraints immutable after spawn",
    ),
]


@traces(
    spec="005",
    story="US-005",
    reqs=[
        "REQ-005-AC3-001",
        "REQ-005-FR4-001",
        "REQ-005-FR6-001",
        "REQ-005-US-005-001",
    ],
)
class ConstitutionalConstraints:
    """Validates agent operations against constitutional articles.

    Uses Prolog for rule evaluation when available, with
    deny-by-default fallback on error.
    """

    def __init__(
        self,
        prolog_bridge: MultiagentPrologBridge | None = None,
        constraints: list[Constraint] | None = None,
    ) -> None:
        self._prolog = prolog_bridge
        self._constraints = constraints or list(DEFAULT_CONSTRAINTS)
        self._agent_constraints: dict[str, list[Constraint]] = {}

    @traces(
        spec="005",
        story="US-005",
        reqs=[
            "REQ-005-AC3-001",
            "REQ-005-FR4-001",
        ],
    )
    @traces(
        spec="001",
        story="US-007",
        reqs=[
            "REQ-001-US-007-011",
            "REQ-001-US-007-012",
        ],
    )
    def _check_alias_and_resources(
        self,
        definition: dict[str, Any],
        violations: list[str],
        severity: str,
    ) -> tuple:
        """Validate alias presence, CPU, memory, and depth limits."""
        alias = definition.get("alias", "")
        if not alias:
            violations.append("Agent alias is required")
            severity = "error"

        cpu = definition.get("cpu_cores", 1)
        memory = definition.get("memory_mb", 512)
        if cpu > 8:
            violations.append(f"CPU cores {cpu} exceeds maximum (8)")
            severity = "error"
        if memory > 4096:
            violations.append(f"Memory {memory}MB exceeds maximum (4096MB)")
            severity = "error"

        depth = definition.get("depth", 0)
        if depth > 5:
            violations.append(f"Hierarchy depth {depth} exceeds maximum (5)")
            severity = "error"

        return alias, cpu, memory, severity

    @traces(spec="005", story="US-005", reqs=["REQ-005-FR4-001", "REQ-005-AC3-001"])
    def _check_prolog_registration(
        self,
        alias: str,
        cpu: int,
        memory: int,
        volk_type: str,
        violations: list[str],
        severity: str,
    ) -> str:
        """Run Prolog registration check; append violation on denial."""
        if self._prolog is None:
            return severity
        try:
            if not self._prolog.can_register_volk(alias, volk_type, cpu, memory):
                violations.append(f"Prolog denied registration for {alias}")
                severity = "error"
        except Exception as exc:
            violations.append(f"Prolog validation error: {exc}")
            severity = "error"
        return severity

    @traces(
        spec="005",
        story="US-005",
        reqs=["REQ-005-FR4-001", "REQ-005-AC3-001"],
    )
    def validate_at_spawn(self, definition: dict[str, Any]) -> ConstraintResult:
        """Validate agent definition before spawning."""
        violations: list[str] = []
        severity = "none"

        alias, cpu, memory, severity = self._check_alias_and_resources(
            definition, violations, severity
        )
        severity = self._check_prolog_registration(
            alias, cpu, memory, definition.get("type", "worker"), violations, severity
        )

        return ConstraintResult(
            valid=len(violations) == 0,
            violations=violations,
            severity=severity,
        )

    @traces(
        spec="005",
        story="US-005",
        reqs=[
            "REQ-005-AC3-001",
            "REQ-005-FR4-001",
        ],
    )
    def validate_at_runtime(self, agent_id: str, action: str) -> ConstraintResult:
        """Validate an action at runtime."""
        violations = []
        severity = "none"

        # Check scope constraints
        agent_constraints = self._agent_constraints.get(agent_id, self._constraints)
        for constraint in agent_constraints:
            if constraint.scope not in ("all", "runtime"):
                continue
            if not self._check_constraint(agent_id, action, constraint):
                violations.append(
                    f"Violation of {constraint.name}: {constraint.description}"
                )
                if constraint.severity == "critical":
                    severity = "critical"
                elif constraint.severity == "error" and severity != "critical":
                    severity = "error"
                elif constraint.severity == "warning" and severity not in (
                    "critical",
                    "error",
                ):
                    severity = "warning"

        return ConstraintResult(
            valid=len(violations) == 0,
            violations=violations,
            severity=severity,
        )

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=["REQ-007-US-002-008"],
    )
    def _check_constraint(
        self, agent_id: str, action: str, constraint: Constraint
    ) -> bool:
        """Check a single constraint. Returns True if satisfied."""
        # Use Prolog if available
        if self._prolog is not None:
            try:
                return self._prolog.can_start_volk(agent_id)
            except Exception as exc:
                logger.warning(
                    "Prolog can_start_volk check failed for agent '%s': %s",
                    agent_id,
                    exc,
                    exc_info=True,
                )
                return False  # Deny by default

        # Default: allow (constraints checked at spawn)
        return True

    @traces(
        spec="005",
        story="US-005",
        reqs=["REQ-005-FR6-001"],
    )
    def check_fact_visibility(self, agent_id: str, fact_scope: str) -> bool:
        """Check if an agent can see facts in the given scope."""
        # Own scope is always visible
        if fact_scope == agent_id:
            return True
        # Session scope is visible to all agents in the session
        if fact_scope == "session":
            return True
        # Parent scope is visible
        # Without hierarchy context, default allow for parent/session scopes
        return fact_scope in ("session", "global", agent_id)

    @traces(
        spec="005",
        story="US-005",
        reqs=["REQ-005-AC3-001", "REQ-005-FR4-001"],
    )
    def check_scope(self, agent_id: str, operation: str) -> bool:
        """Check if an operation is within the agent's scope."""
        result = self.validate_at_runtime(agent_id, operation)
        return result.valid

    @traces(
        spec="001",
        story="US-007",
        reqs=["REQ-001-US-007-011", "REQ-001-US-007-012"],
    )
    def check_resource_limits(
        self, agent_id: str, cpu: int = 1, memory: int = 512
    ) -> bool:
        """Check if resource usage is within limits."""
        if cpu > 8:
            return False
        if memory > 4096:
            return False
        if self._prolog is not None:
            try:
                return self._prolog.can_register_volk(agent_id, "worker", cpu, memory)
            except Exception as exc:
                logger.warning(
                    "Prolog can_register_volk check failed for agent '%s': %s",
                    agent_id,
                    exc,
                    exc_info=True,
                )
                return False
        return True

    def get_constraints(self, agent_id: str) -> list[Constraint]:
        """Get constraints for an agent."""
        return self._agent_constraints.get(agent_id, list(self._constraints))

    @traces(
        spec="005",
        story="US-005",
        reqs=["REQ-005-US-005-001"],
    )
    def set_constraints(self, agent_id: str, constraints: list[Constraint]) -> None:
        """Set constraints for an agent (at spawn time only)."""
        self._agent_constraints[agent_id] = list(constraints)

    def get_default_constraints(self) -> list[Constraint]:
        """Get the default constraint set."""
        return list(self._constraints)