orchestrata_ultimate.verification.integrations.intent_broker

Reference › orchestrata_ultimate › verification › integrations › intent_broker

IntentBrokerAdapter turns an Intent Broker intent into verification constraints (Spec 035 US-009). It is used by the orchestrator's verification_gate and by the verification session manager. intent_to_constraints reads the intent's action and requirements list and builds one Constraint per requirement: the predicate defaults to the action, the expected value to true, and the description to the requirement's own. _infer_type maps the requirement's category word to a ConstraintType: performance, speed and latency to performance; quality, coverage and testing to quality; security, auth and safety to security; anything else to functional. Each processed intent is remembered in memory and counted by processed_count. The adapter makes no network call.

At a glance

Class diagram

Dependency diagram

API

Classes

IntentBrokerAdapter

Decorators: traces(spec='035', story='US-009', reqs=['REQ-035-US-009-001', 'REQ-035-US-009-002', 'REQ-035-US-009-003', 'REQ-035-US-… · Line 26

Adapts Intent Broker intents to verification constraints.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/integrations/intent_broker.py

"""Intent Broker integration adapter."""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.types import Constraint, ConstraintType

logger = logging.getLogger(__name__)


@traces(
    spec="035",
    story="US-009",
    reqs=[
        "REQ-035-US-009-001",
        "REQ-035-US-009-002",
        "REQ-035-US-009-003",
        "REQ-035-US-009-009",
        "REQ-035-US-009-010",
        "REQ-035-US-009-015",
        "REQ-035-US-009-016",
        "REQ-035-US-009-017",
    ],
)
class IntentBrokerAdapter:
    """Adapts Intent Broker intents to verification constraints."""

    def __init__(self) -> None:
        self._processed: list[dict[str, Any]] = []

    @traces(
        spec="035",
        story="US-009",
        reqs=[
            "REQ-035-US-009-001",
            "REQ-035-US-009-002",
            "REQ-035-US-009-010",
            "REQ-035-US-009-015",
            "REQ-035-US-009-016",
        ],
    )
    def intent_to_constraints(self, intent: dict[str, Any]) -> list[Constraint]:
        """Convert an intent to verification constraints."""
        constraints = []
        action = intent.get("action", "")
        requirements = intent.get("requirements", [])

        for req in requirements:
            constraint = Constraint(
                type=self._infer_type(req),
                predicate=req.get("predicate", action),
                expected=req.get("expected", True),
                description=req.get("description", ""),
            )
            constraints.append(constraint)

        self._processed.append(
            {"intent": intent, "constraints_count": len(constraints)}
        )
        return constraints

    @traces(spec="035", story="US-009", reqs=["REQ-035-US-009-003"])
    def _infer_type(self, requirement: dict[str, Any]) -> ConstraintType:
        """Infer constraint type from a requirement."""
        category = requirement.get("category", "").lower()
        if category in ("performance", "speed", "latency"):
            return ConstraintType.PERFORMANCE
        elif category in ("quality", "coverage", "testing"):
            return ConstraintType.QUALITY
        elif category in ("security", "auth", "safety"):
            return ConstraintType.SECURITY
        return ConstraintType.FUNCTIONAL

    @property
    @traces(spec="035", story="US-007", reqs=["REQ-035-US-007-017"])
    def processed_count(self) -> int:
        return len(self._processed)