orchestrata_ultimate.orchestrator.intent_router

Reference › orchestrata_ultimate › orchestrator › intent_router

The last two stages of the intent pipeline: choosing which volk gets an intent, and sending it there. intent_broker and intent_broker_factory use both classes. The module imports Postman for transport and CXPPacket for the wire format.

IntentRouter.route tries four strategies in order and returns a RoutingResult (see intent_results) with a reason string:

• an explicit target_volk on the intent, if that volk is registered;
• the action and target pair, resolved by the Prolog route_intent/3 rule when a bridge is wired, otherwise by the static INTENT_VOLK_MAPPING from intent_types; a Prolog error or exception falls back to the static table rather than failing;
• a capability lookup named from the action and target;
• the fallback volk (default U:Ox) if registered.

Candidates are filtered to registered volks, and _prefer_healthy_candidates keeps only healthy ones, falling back to all of them, tagged "unhealthy-fallback", when none is healthy. If nothing matches the result has no volks and is_routed is false.

IntentDispatcher.dispatch builds a CXP command packet with _create_packet (priority mapped from the intent, payload carrying the intent id, action, target, path, parameters, constraints, requester, session and hash) and delivers it through Postman.route_sync on a worker thread. Delivery succeeds only when the target's handler returned a response; a None response is reported as "delivery failed: no route". After a successful delivery it records a heartbeat for the target volk, guarded so that a failure there does not fail the dispatch.

At a glance

Class diagram

Dependency diagram

API

Classes

IntentRouter

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

Routes intents to target Volks.

IntentDispatcher

Decorators: traces(spec='003', story='US-004', reqs=['REQ-003-US-004-010']) · Line 254

Dispatches intents to Volks via real CXP transport (Postman).

Source

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

"""Intent routing and CXP dispatch.

Routes validated intents to appropriate Volks based on
action/target mapping and Volk capabilities. Dispatches
via CXP packets through the Postman system.

Spec: 003-intent / US-004
Author: Mohamed
"""

import asyncio
import logging
import time
from datetime import UTC, datetime

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import (
    CXPPacket,
    PacketType,
    Priority,
)
from orchestrata_ultimate.core.protocols import IVolkRegistry, VolkHealthProbe
from orchestrata_ultimate.infra.prolog_bridge import PrologBridge
from orchestrata_ultimate.orchestrator.intent_types import (
    INTENT_VOLK_MAPPING,
    DispatchResult,
    Intent,
    RoutingResult,
)
from orchestrata_ultimate.postman.postman import Postman

logger = logging.getLogger(__name__)

# Prolog Strategy-2 query template (SPEC-003 US-004 REQ-003-US-004-002): asks
# routing.pl's route_intent/3 facts for all candidate Volks, in clause
# (priority) order, for a given action+target pair.
_ROUTE_INTENT_QUERY = "route_intent('{action}', '{target}', Volk)"

# Map IntentPriority to CXP Priority
_PRIORITY_MAP = {
    5: Priority.CRITICAL,  # CRITICAL
    4: Priority.HIGH,  # HIGH
    3: Priority.MEDIUM,  # NORMAL
    2: Priority.LOW,  # LOW
    1: Priority.LOWEST,  # BACKGROUND
}


@traces(
    spec="003",
    story="US-004",
    reqs=[
        "REQ-003-US-004-001",
        "REQ-003-US-004-002",
        "REQ-003-US-004-003",
        "REQ-003-US-004-011",
    ],
)
class IntentRouter:
    """Routes intents to target Volks.

    Resolution strategy:
    1. Check explicit target_volk on intent (if set)
    2. Query Prolog route_intent/3 (routing.pl, SPEC-003 US-004
       REQ-003-US-004-002) for action+target -> Volk candidates, in clause
       (priority) order; falls back to the static INTENT_VOLK_MAPPING dict
       when no PrologBridge is wired or the query fails
    3. Verify target Volk is registered and active
    4. Fall back to U:Ox for unresolved intents
    """

    def __init__(
        self,
        volk_registry: IVolkRegistry,
        fallback_volk: str = "U:Ox",
        prolog_bridge: PrologBridge | None = None,
    ) -> None:
        self._registry = volk_registry
        self._fallback = fallback_volk
        self._prolog = prolog_bridge

    @traces(
        spec="003", story="US-004", reqs=["REQ-003-US-004-006", "REQ-003-US-004-019"]
    )
    def _is_volk_healthy(self, alias: str) -> bool:
        """Best-effort health check for a registered Volk.

        The registry is typed as ``IVolkRegistry`` (``VolkRegistryProtocol``),
        which guarantees ``get_volk_health``; treats missing health data
        (``None``) as healthy so routing stays backward-compatible.
        """
        try:
            info = self._registry.get_volk_health(alias)
            return info is None or getattr(info, "is_healthy", True)
        except Exception as e:
            logger.warning(f"Health check failed for {alias}: {e}")
            return True

    @traces(
        spec="003", story="US-004", reqs=["REQ-003-US-004-005", "REQ-003-US-004-018"]
    )
    def _prefer_healthy_candidates(
        self, candidates: list[str]
    ) -> tuple[list[str], bool]:
        """Return only healthy candidates, or all candidates if none are healthy.

        Best-effort health-aware routing: we prefer healthy Volks, but falling
        back to an unhealthy Volk is safer than failing the intent outright
        when no healthy candidate exists.

        Returns:
            (filtered_candidates, did_fallback_to_unhealthy)
        """
        healthy = [c for c in candidates if self._is_volk_healthy(c)]
        if healthy:
            return healthy, False
        logger.warning(
            f"No healthy Volks among candidates {candidates}; routing to all candidates"
        )
        return candidates, True

    @traces(
        spec="003",
        story="US-004",
        reqs=[
            "REQ-003-US-004-001",
            "REQ-003-US-004-003",
            "REQ-003-US-004-005",
            "REQ-003-US-004-006",
            "REQ-003-US-004-012",
            "REQ-003-US-004-016",
            "REQ-003-US-004-018",
            "REQ-003-US-004-019",
        ],
    )
    async def route(self, intent: Intent) -> RoutingResult:
        """Resolve target Volk(s) for an intent."""
        start = time.monotonic()

        # Strategy 1: Explicit target
        if intent.target_volk:
            if self._registry.is_registered(intent.target_volk):
                result = RoutingResult(
                    intent_id=intent.id,
                    target_volks=[intent.target_volk],
                    primary_volk=intent.target_volk,
                    routing_reason=f"Explicit target: {intent.target_volk}",
                )
                self._log_routing(intent, result, start)
                return result

        # Strategy 2: Action+Target mapping. Routing rules are defined in
        # Prolog (routing.pl, route_intent/3 — SPEC-003 US-004
        # REQ-003-US-004-002) when a bridge is wired; the static
        # INTENT_VOLK_MAPPING dict is a no-bridge fallback only, kept in
        # exact parity with routing.pl (see
        # tests/orchestrator/test_intent_router_prolog.py::test_prolog_dict_parity).
        # A Prolog engine failure must never crash routing: any query
        # exception or reported error falls back to the static mapping.
        candidates: list[str] | None = None
        if self._prolog is not None:
            try:
                query_result = self._prolog.query(
                    _ROUTE_INTENT_QUERY.format(
                        action=intent.action.value, target=intent.target.value
                    )
                )
                if query_result.error is None:
                    candidates = [sol["Volk"] for sol in query_result.solutions]
                else:
                    logger.warning(
                        f"Prolog routing query failed for intent {intent.id}: "
                        f"{query_result.error}; falling back to static mapping"
                    )
            except Exception as e:
                logger.warning(
                    f"Prolog routing query raised for intent {intent.id}: {e}; "
                    f"falling back to static mapping"
                )

        if candidates is not None:
            routing_reason = f"Prolog: {intent.action.value}+{intent.target.value}"
        else:
            mapping_key = (intent.action.value, intent.target.value)
            candidates = INTENT_VOLK_MAPPING.get(mapping_key, [])
            routing_reason = f"Mapping: {intent.action.value}+{intent.target.value}"

        # Filter to registered, healthy Volks (prefer healthy; fall back if needed)
        registered = [v for v in candidates if self._registry.is_registered(v)]
        active, fallback = self._prefer_healthy_candidates(registered)

        if active:
            health_note = "unhealthy-fallback" if fallback else "healthy"
            result = RoutingResult(
                intent_id=intent.id,
                target_volks=active,
                primary_volk=active[0],
                routing_reason=f"{routing_reason} ({health_note})",
            )
            self._log_routing(intent, result, start)
            return result

        # Strategy 3: Capability-based lookup
        cap_name = f"{intent.action.value}_{intent.target.value}"
        cap_matches = self._registry.list_volks_by_capability(cap_name)
        if cap_matches:
            aliases = [r.definition.alias for r in cap_matches]
            aliases, fallback = self._prefer_healthy_candidates(aliases)
            health_note = "unhealthy-fallback" if fallback else "healthy"
            result = RoutingResult(
                intent_id=intent.id,
                target_volks=aliases,
                primary_volk=aliases[0],
                routing_reason=f"Capability: {cap_name} ({health_note})",
            )
            self._log_routing(intent, result, start)
            return result

        # Strategy 4: Fallback to orchestrator
        if self._registry.is_registered(self._fallback):
            result = RoutingResult(
                intent_id=intent.id,
                target_volks=[self._fallback],
                primary_volk=self._fallback,
                routing_reason=f"Fallback: {self._fallback}",
            )
            self._log_routing(intent, result, start)
            return result

        # No route found
        result = RoutingResult(
            intent_id=intent.id,
            target_volks=[],
            primary_volk=None,
            routing_reason="No registered Volk found for intent",
        )
        self._log_routing(intent, result, start)
        return result

    @traces(
        spec="003", story="US-004", reqs=["REQ-003-US-004-008", "REQ-003-US-004-022"]
    )
    def _log_routing(self, intent: Intent, result: RoutingResult, start: float) -> None:
        elapsed_ms = (time.monotonic() - start) * 1000
        level = logging.INFO if result.is_routed else logging.WARNING
        logger.log(
            level,
            f"Intent {intent.id} routed to {result.target_volks} "
            f"via {result.routing_reason} in {elapsed_ms:.1f}ms",
        )


@traces(spec="003", story="US-004", reqs=["REQ-003-US-004-010"])
class IntentDispatcher:
    """Dispatches intents to Volks via real CXP transport (Postman)."""

    SENDER = "U:Intents"

    def __init__(self, volk_registry: VolkHealthProbe, transport: Postman) -> None:
        self._registry = volk_registry
        self._transport = transport

    @traces(spec="001", reqs=["REQ-001-018"])
    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-010"])
    async def dispatch(self, intent: Intent, target_volk: str) -> DispatchResult:
        """Create and dispatch a CXP packet for the intent via real Postman
        transport (SPEC-003 US-004 REQ-003-US-004-010).

        Delivers the packet through ``Postman.route_sync`` and treats
        delivery as successful only when the target Volk's handler actually
        returned a response. A ``None`` response means no route (no handler
        registered for ``target_volk``) and is reported as a failure —
        dispatch no longer fakes success via a local ``mark_delivered()``
        field flip.
        """
        start = time.monotonic()

        try:
            packet = self._create_packet(intent, target_volk)

            # Real CXP transport: route through Postman to the target volk's
            # registered handler. route_sync is synchronous, so run it off
            # the event loop thread to avoid blocking other coroutines.
            response = await asyncio.to_thread(self._transport.route_sync, packet)

            if response is None:
                elapsed_ms = (time.monotonic() - start) * 1000
                logger.warning(
                    f"Intent {intent.id} dispatch to {target_volk} via packet "
                    f"{packet.header.packet_id} found no route in {elapsed_ms:.1f}ms"
                )
                return DispatchResult(
                    intent_id=intent.id,
                    packet_id=packet.header.packet_id,
                    target_volk=target_volk,
                    success=False,
                    error_message="delivery failed: no route",
                    delivery_latency_ms=elapsed_ms,
                )

            # REQ-001-018: record dispatch-time activity as the target volk's
            # liveness signal (activity-based heartbeat), only after a
            # successful delivery. A periodic push-heartbeat from volks
            # themselves is a future extension. The registry is typed as
            # VolkHealthProbe, which guarantees record_heartbeat; still guard
            # the call itself so a runtime failure inside it (e.g. the
            # registry's best-effort Redis persistence) never fails the
            # dispatch.
            try:
                self._registry.record_heartbeat(target_volk)
            except Exception as heartbeat_error:
                logger.debug(
                    f"Heartbeat recording failed for {target_volk}: {heartbeat_error}"
                )

            elapsed_ms = (time.monotonic() - start) * 1000

            logger.info(
                f"Intent {intent.id} dispatched to {target_volk} "
                f"via packet {packet.header.packet_id} in {elapsed_ms:.1f}ms"
            )

            return DispatchResult(
                intent_id=intent.id,
                packet_id=packet.header.packet_id,
                target_volk=target_volk,
                success=True,
                dispatched_at=datetime.now(UTC),
                delivery_latency_ms=elapsed_ms,
            )

        except Exception as e:
            # The except handler must honour dispatch()'s contract — always
            # return a DispatchResult, never raise — even for a degenerate
            # intent (e.g. None), so guard the id access rather than
            # dereferencing intent.id directly.
            intent_id = getattr(intent, "id", "") or ""
            logger.error(f"Failed to dispatch intent {intent_id} to {target_volk}: {e}")
            return DispatchResult(
                intent_id=intent_id,
                packet_id="",
                target_volk=target_volk,
                success=False,
                error_message=str(e),
            )

    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-010"])
    def _create_packet(self, intent: Intent, target_volk: str) -> CXPPacket:
        """Create a CXP packet for intent dispatch."""
        priority = _PRIORITY_MAP.get(intent.priority.value, Priority.MEDIUM)

        return CXPPacket(
            from_agent=self.SENDER,
            to_agent=target_volk,
            packet_type=PacketType.COMMAND,
            priority=priority,
            correlation_id=intent.correlation_id,
            thread_id=intent.session_id or intent.correlation_id or intent.id,
            ttl_seconds=intent.ttl_seconds,
            payload={
                "intent_id": intent.id,
                "action": intent.action.value,
                "target": intent.target.value,
                "resource_path": intent.resource_path,
                "parameters": intent.parameters,
                "constraints": [c.to_dict() for c in intent.constraints],
                "requester": intent.requester,
                "priority": intent.priority.value,
                "session_id": intent.session_id,
                "intent_hash": intent.hash(),
            },
        )