orchestrata_ultimate.verification.protocols.verification_cxp

Reference › orchestrata_ultimate › verification › protocols › verification_cxp

VerificationCXPAdapter creates and parses the CXP messages of the verification workflow (Spec 035). It is used by the coordinator. Each create_* method returns a plain dictionary with from_agent, to_agent, a subject and a payload, and the default sender and receiver follow the role names: U:Ox (orchestrator), U:Exec (execution), U:Vfy (verification) and U:Sup (supervisor). The messages are constraint delivery (U:Ox to U:Vfy), verification request (U:Exec to U:Vfy), verification result (U:Vfy to U:Ox), escalation (U:Vfy to U:Ox), feedback (U:Vfy to U:Exec), proof request (U:Vfy to U:Sup) and proof submission (U:Sup to U:Vfy). parse_message extracts subject, sender, receiver and payload from a packet, with empty defaults. The adapter only builds and reads dictionaries; it does not send them.

At a glance

Class diagram

Dependency diagram

API

Classes

VerificationCXPAdapter

Decorators: traces(spec='035', story='US-007', reqs=['REQ-035-US-007-009', 'REQ-035-US-007-017', 'REQ-035-US-007-021', 'REQ-035-US-… · Line 30

Creates and parses CXP messages for verification workflows.

Source

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

"""CXP protocol adapters for verification messages."""

# Spec: 035 - Verification Agent

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.types import (
    CheckpointName,
    Constraint,
    VerificationResult,
    VerificationSession,
)

logger = logging.getLogger(__name__)


@traces(
    spec="035",
    story="US-007",
    reqs=[
        "REQ-035-US-007-009",
        "REQ-035-US-007-017",
        "REQ-035-US-007-021",
        "REQ-035-US-009-004",
        "REQ-035-US-009-011",
    ],
)
class VerificationCXPAdapter:
    """Creates and parses CXP messages for verification workflows."""

    @traces(spec="035", story="US-7", reqs=["REQ-035-US-007-009", "REQ-035-US-009-004"])
    def create_constraint_message(
        self,
        constraints: list[Constraint],
        from_agent: str = "U:Ox",
        to_agent: str = "U:Vfy",
    ) -> dict[str, Any]:
        """Create a CXP packet for constraint delivery."""
        return {
            "from_agent": from_agent,
            "to_agent": to_agent,
            "subject": "VERIFICATION_CONSTRAINTS",
            "payload": {
                "constraints": [c.to_dict() for c in constraints],
                "count": len(constraints),
            },
        }

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-012"])
    @traces(
        spec="035",
        story="US-010",
        reqs=[
            "REQ-035-US-004-014",
            "REQ-035-US-004-020",
            "REQ-035-US-005-007",
            "REQ-035-US-005-022",
            "REQ-035-US-007-012",
            "REQ-035-US-007-013",
            "REQ-035-US-007-014",
        ],
    )
    def create_verification_request(
        self,
        checkpoint: CheckpointName,
        artifacts: dict[str, Any],
        from_agent: str = "U:Exec",
        to_agent: str = "U:Vfy",
    ) -> dict[str, Any]:
        """Create a CXP packet for verification request."""
        return {
            "from_agent": from_agent,
            "to_agent": to_agent,
            "subject": "VERIFICATION_REQUEST",
            "payload": {
                "checkpoint": checkpoint.value,
                "artifacts": artifacts,
            },
        }

    @traces(
        spec="035", story="US-009", reqs=["REQ-035-US-009-004", "REQ-035-US-009-011"]
    )
    def create_verification_result(
        self,
        results: list[VerificationResult],
        analysis: dict[str, Any] | None = None,
        from_agent: str = "U:Vfy",
        to_agent: str = "U:Ox",
    ) -> dict[str, Any]:
        """Create a CXP packet for verification results."""
        return {
            "from_agent": from_agent,
            "to_agent": to_agent,
            "subject": "VERIFICATION_RESULT",
            "payload": {
                "results": [r.to_dict() for r in results],
                "analysis": analysis or {},
                "all_satisfied": all(r.is_satisfied for r in results),
            },
        }

    @traces(spec="035", story="US-008", reqs=["REQ-035-US-008-005"])
    def create_escalation_message(
        self,
        session: VerificationSession,
        from_agent: str = "U:Vfy",
        to_agent: str = "U:Ox",
    ) -> dict[str, Any]:
        """Create a CXP packet for escalation."""
        failed = [cid for cid, r in session.results.items() if not r.is_satisfied]
        return {
            "from_agent": from_agent,
            "to_agent": to_agent,
            "subject": "VERIFICATION_ESCALATION",
            "payload": {
                "session_id": session.id,
                "attempt_count": session.attempt_count,
                "failed_constraints": failed,
                "reason": "Max verification attempts exceeded",
            },
        }

    @traces(spec="035", story="US-008", reqs=["REQ-035-US-008-012"])
    def create_feedback_message(
        self,
        feedback: dict[str, Any],
        from_agent: str = "U:Vfy",
        to_agent: str = "U:Exec",
    ) -> dict[str, Any]:
        """Create a CXP packet for feedback."""
        return {
            "from_agent": from_agent,
            "to_agent": to_agent,
            "subject": "VERIFICATION_FEEDBACK",
            "payload": feedback,
        }

    @traces(
        spec="035", story="US-007", reqs=["REQ-035-US-007-009", "REQ-035-US-007-021"]
    )
    def create_proof_request(
        self,
        constraint_id: str,
        from_agent: str = "U:Vfy",
        to_agent: str = "U:Sup",
    ) -> dict[str, Any]:
        """Create a CXP packet requesting a proof."""
        return {
            "from_agent": from_agent,
            "to_agent": to_agent,
            "subject": "PROOF_REQUEST",
            "payload": {"constraint_id": constraint_id},
        }

    @traces(spec="035", story="US-007", reqs=["REQ-035-US-007-017"])
    def create_proof_submission(
        self,
        proof: dict[str, Any],
        from_agent: str = "U:Sup",
        to_agent: str = "U:Vfy",
    ) -> dict[str, Any]:
        """Create a CXP packet submitting a proof."""
        return {
            "from_agent": from_agent,
            "to_agent": to_agent,
            "subject": "PROOF_SUBMISSION",
            "payload": proof,
        }

    def parse_message(self, packet: dict[str, Any]) -> dict[str, Any]:
        """Parse a CXP verification message."""
        return {
            "subject": packet.get("subject", ""),
            "from_agent": packet.get("from_agent", ""),
            "to_agent": packet.get("to_agent", ""),
            "payload": packet.get("payload", {}),
        }