orchestrata_ultimate.agents.intent_adapter

Reference › orchestrata_ultimate › agents › intent_adapter

intent_adapter is the door from the Intent Broker into the runtime (specification 003, user story US-004). IntentTaskAdapter takes a TaskExecutor (task_executor). dispatch converts an Intent into a TaskDefinition with _intent_to_task (using _resolve_target to pick the volk alias), executes it through the executor and returns a DispatchResult for the broker. The task types come from types.

At a glance

Class diagram

Dependency diagram

API

Classes

IntentTaskAdapter

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

Converts Intents to TaskDefinitions and dispatches them.

Source

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

"""Intent Task Adapter — converts Intents to TaskDefinitions.

Bridges the IntentBroker pipeline to the Volk task execution system.
Maps IntentAction/IntentTarget to task action/target and returns
DispatchResult for IntentBroker compatibility.

Author: Mohamed
Spec: 005-multi-agent (amended)
"""

# Spec: 005 - Multi-Agent System

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

from orchestrata_ultimate.agents.task_executor import TaskExecutor
from orchestrata_ultimate.agents.types import TaskDefinition
from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.orchestrator.intent_results import DispatchResult
    from orchestrata_ultimate.orchestrator.intent_types import Intent

_logger = logging.getLogger(__name__)


# Map IntentPriority values to CXP 0-10 scale
_PRIORITY_MAP = {
    1: 2,  # BACKGROUND -> 2
    2: 4,  # LOW -> 4
    3: 5,  # NORMAL -> 5
    4: 7,  # HIGH -> 7
    5: 10,  # CRITICAL -> 10
}


@traces(spec="003", story="US-004", reqs=["REQ-003-US-004-001"])
class IntentTaskAdapter:
    """Converts Intents to TaskDefinitions and dispatches them.

    Provides an alternative dispatch target in the intent pipeline,
    routing intents to Volk task execution instead of CXP packets.

    Parameters
    ----------
    executor : TaskExecutor
        Task executor for submitting converted tasks.
    """

    def __init__(self, executor: TaskExecutor) -> None:
        self._executor = executor

    async def dispatch(self, intent: Intent) -> DispatchResult:
        """Convert an Intent to a TaskDefinition and execute it.

        Maps:
        - IntentAction -> task.action
        - IntentTarget -> task.target (via INTENT_VOLK_MAPPING)
        - IntentPriority -> task.priority (CXP 0-10 scale)
        - intent.parameters -> task.parameters
        - intent.resource_path -> task.parameters["resource_path"]

        Returns a DispatchResult for IntentBroker compatibility.
        """
        from orchestrata_ultimate.orchestrator.intent_results import (  # noqa: PLC0415
            DispatchResult,
        )

        try:
            task = self._intent_to_task(intent)
            result = await self._executor.submit(task)

            return DispatchResult(
                intent_id=intent.id,
                packet_id=result.task_id,
                target_volk=task.target,
                success=result.success,
                error_message=result.error_message,
                delivery_latency_ms=result.execution_time_ms,
            )
        except Exception as e:
            _logger.error(f"Intent dispatch failed: {e}")
            return DispatchResult(
                intent_id=intent.id,
                packet_id="",
                target_volk="",
                success=False,
                error_message=str(e),
            )

    def _intent_to_task(self, intent: Intent) -> TaskDefinition:
        """Convert an Intent to a TaskDefinition."""
        # Resolve target alias from INTENT_VOLK_MAPPING
        target_alias = self._resolve_target(intent)

        # Map priority
        priority = _PRIORITY_MAP.get(intent.priority.value, 5)

        # Build parameters
        parameters = dict(intent.parameters)
        if intent.resource_path:
            parameters["resource_path"] = intent.resource_path
        parameters["intent_id"] = intent.id
        parameters["intent_action"] = intent.action.value
        parameters["intent_target"] = intent.target.value

        return TaskDefinition(
            action=intent.action.value,
            target=target_alias,
            parameters=parameters,
            priority=priority,
            requester=intent.requester,
            correlation_id=intent.correlation_id,
        )

    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-012"])
    def _resolve_target(self, intent: Intent) -> str:
        """Resolve the target Volk alias for an intent.

        Uses INTENT_VOLK_MAPPING first, falls back to intent.target_volk.
        """
        from orchestrata_ultimate.orchestrator.intent_types import (  # noqa: PLC0415
            INTENT_VOLK_MAPPING,
        )

        # Check explicit target
        if intent.target_volk:
            return intent.target_volk

        # Look up in mapping
        key = (intent.action.value, intent.target.value)
        volks = INTENT_VOLK_MAPPING.get(key, [])
        if volks:
            return volks[0]  # Primary target

        # Default to orchestrator
        return "U:Ox"