orchestrata_ultimate.orchestrator.intent_results

Reference › orchestrata_ultimate › orchestrator › intent_results

The three result records that the later stages of the intent pipeline produce, one per stage. They are imported by intent_types and by the agents-side intent adapter, and traced to spec 003 (stories US-002 and US-003).

ValidationResult holds the ValidationOutcome (see intent_enums), the intent id, a reason, a list of violations, and the Prolog query and result strings when the rule base was consulted; its is_valid property is the convenience check. RoutingResult records which volks can serve the intent (target_volks), which one is primary, and why; is_routed reports whether any volk was chosen. DispatchResult records the outcome of sending the intent to a volk as a CXP packet: the packet id, target volk, success flag, an optional error message and the delivery latency in milliseconds. Each has a to_dict for serialization, and each stamps a UTC timestamp when created. The stages that fill them in are intent_validator and intent_router.

At a glance

Class diagram

Dependency diagram

API

Classes

ValidationResult (dataclass)

Decorators: dataclass, traces(spec='003', story='US-003', reqs=['REQ-003-US-003-008', 'REQ-003-US-003-021']) · Line 21

Result of intent validation.

RoutingResult (dataclass)

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

Result of intent routing.

DispatchResult (dataclass)

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

Result of dispatching an intent to a Volk via CXP.

Source

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

"""Intent Broker result types.

Defines result dataclasses returned by intent validation, routing,
and dispatch operations in the Orchestrata system.

Spec: 003-intent
Author: Mohamed
"""

# Spec: 003 - Intent Broker

from dataclasses import dataclass, field
from datetime import UTC, datetime

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.intent_enums import ValidationOutcome


@dataclass
@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-008", "REQ-003-US-003-021"])
class ValidationResult:
    """Result of intent validation."""

    outcome: ValidationOutcome
    intent_id: str
    reason: str = ""
    violations: list[str] = field(default_factory=list)
    prolog_query: str | None = None
    prolog_result: str | None = None
    validated_at: datetime = field(default_factory=lambda: datetime.now(UTC))

    @property
    @traces(
        spec="003",
        story="US-003",
        reqs=["REQ-003-US-003-008", "REQ-003-US-003-021"],
    )
    def is_valid(self) -> bool:
        return self.outcome == ValidationOutcome.ALLOWED

    @traces(
        spec="003",
        story="US-002",
        reqs=[
            "REQ-003-US-002-002",
            "REQ-003-US-002-007",
            "REQ-003-US-002-015",
            "REQ-003-US-002-019",
            "REQ-003-US-002-028",
        ],
    )
    def to_dict(self) -> dict:
        """Serialize ValidationResult to a dictionary."""
        return {
            "outcome": self.outcome.value,
            "intent_id": self.intent_id,
            "reason": self.reason,
            "violations": self.violations,
            "prolog_query": self.prolog_query,
            "prolog_result": self.prolog_result,
            "validated_at": self.validated_at.isoformat(),
        }


@dataclass
@traces(spec="003", story="US-002", reqs=["REQ-003-US-002-001"])
class RoutingResult:
    """Result of intent routing."""

    intent_id: str
    target_volks: list[str] = field(default_factory=list)
    primary_volk: str | None = None
    routing_reason: str = ""
    routed_at: datetime = field(default_factory=lambda: datetime.now(UTC))

    @property
    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-001"])
    def is_routed(self) -> bool:
        return len(self.target_volks) > 0

    @traces(
        spec="003",
        story="US-002",
        reqs=[
            "REQ-003-US-002-002",
            "REQ-003-US-002-007",
            "REQ-003-US-002-015",
            "REQ-003-US-002-019",
            "REQ-003-US-002-028",
        ],
    )
    def to_dict(self) -> dict:
        """Serialize RoutingResult to a dictionary."""
        return {
            "intent_id": self.intent_id,
            "target_volks": self.target_volks,
            "primary_volk": self.primary_volk,
            "routing_reason": self.routing_reason,
            "routed_at": self.routed_at.isoformat(),
        }


@dataclass
@traces(spec="003", story="US-002", reqs=["REQ-003-US-002-001"])
class DispatchResult:
    """Result of dispatching an intent to a Volk via CXP."""

    intent_id: str
    packet_id: str
    target_volk: str
    success: bool
    error_message: str | None = None
    dispatched_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    delivery_latency_ms: float | None = None

    @traces(
        spec="003",
        story="US-002",
        reqs=[
            "REQ-003-US-002-002",
            "REQ-003-US-002-007",
            "REQ-003-US-002-015",
            "REQ-003-US-002-019",
            "REQ-003-US-002-028",
        ],
    )
    def to_dict(self) -> dict:
        """Serialize DispatchResult to a dictionary."""
        return {
            "intent_id": self.intent_id,
            "packet_id": self.packet_id,
            "target_volk": self.target_volk,
            "success": self.success,
            "error_message": self.error_message,
            "dispatched_at": self.dispatched_at.isoformat(),
            "delivery_latency_ms": self.delivery_latency_ms,
        }