orchestrata_ultimate.agents.postman_adapter

Reference › orchestrata_ultimate › agents › postman_adapter

postman_adapter bridges CXP message delivery to task execution. VolkPostmanAdapter holds a Postman adapter and a TaskExecutor (task_executor). process_next dequeues the next packet for a recipient, converts it with _packet_to_task into a TaskDefinition (types), executes it and returns the TaskResult; delivery is acknowledged after execution. process_all drains a recipient's queue and get_pending_count reports how many packets wait.

At a glance

Class diagram

Dependency diagram

API

Classes

VolkPostmanAdapter

Line 22

Bridges CXP Postman message delivery to Volk task execution.

Source

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

"""Volk Postman Adapter — bridges CXP message delivery to task execution.

Dequeues CXP packets from the Postman, converts them to tasks,
executes via TaskExecutor, and acknowledges delivery.

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

# Spec: 005 - Multi-Agent System

import logging

from orchestrata_ultimate.agents.task_executor import TaskExecutor
from orchestrata_ultimate.agents.types import TaskDefinition, TaskResult
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.cxp.postman_adapter import PostmanAdapter

_logger = logging.getLogger(__name__)


class VolkPostmanAdapter:
    """Bridges CXP Postman message delivery to Volk task execution.

    Polls the Postman for queued packets addressed to specific Volks,
    converts them to TaskDefinitions, and executes them via TaskExecutor.

    Parameters
    ----------
    postman : PostmanAdapter
        CXP Postman adapter for message queue operations.
    executor : TaskExecutor
        Task executor for submitting converted tasks.
    """

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

    async def process_next(self, recipient: str) -> TaskResult | None:
        """Dequeue and process the next packet for a recipient.

        Flow:
        1. Dequeue packet from Postman
        2. Convert to TaskDefinition
        3. Execute via TaskExecutor
        4. Acknowledge the packet
        5. Return TaskResult

        Returns None if no packets are queued.
        """
        packet = await self._postman.dequeue(recipient)
        if packet is None:
            return None

        task = self._packet_to_task(packet)

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

            # Acknowledge successful delivery
            await self._postman.acknowledge(packet)

            _logger.info(
                f"Processed packet {packet.message_id} for {recipient}: "
                f"success={result.success}"
            )
            return result

        except Exception as e:
            # Reject the packet on failure
            await self._postman.reject(packet, str(e))
            _logger.error(f"Failed to process packet {packet.message_id}: {e}")
            return TaskResult(
                task_id=task.task_id,
                success=False,
                error_message=str(e),
            )

    async def process_all(self, recipient: str) -> list[TaskResult]:
        """Process all queued packets for a recipient.

        Returns list of TaskResults for each processed packet.
        """
        results = []
        while True:
            result = await self.process_next(recipient)
            if result is None:
                break
            results.append(result)
        return results

    async def get_pending_count(self, recipient: str) -> int:
        """Get number of packets queued for a recipient."""
        return await self._postman.get_queue_size(recipient)

    def _packet_to_task(self, packet: CXPPacket) -> TaskDefinition:
        """Convert a CXP packet to a TaskDefinition.

        ``correlation_id`` and ``thread_id`` are DIFFERENT granularities and
        both are preserved separately (REQ-012-023). ``thread_id`` identifies
        the conversation; ``correlation_id`` pairs one reply to one specific
        request -- CXP's own :meth:`CXPPacket.create_ack`/``create_error`` set
        both, precisely because two concurrent requests in one thread are
        indistinguishable without the finer key.

        There is deliberately NO fallback to ``thread_id`` when the packet
        carries no ``correlation_id``: substituting the coarser id would make
        "uncorrelated" indistinguishable from "correlated to the whole
        thread", which is the conflation REQ-012-023 exists to forbid. A
        ``None`` here means the inbound packet genuinely had none.
        """
        # Extract task parameters from packet
        parameters = dict(packet.metadata) if packet.metadata else {}
        parameters["message_id"] = packet.message_id
        parameters["from_agent"] = packet.from_agent
        parameters["thread_id"] = packet.thread_id

        # Use packet subject as action, recipient as target
        action = packet.subject or "process"
        target = packet.to_agent or ""

        return TaskDefinition(
            action=action,
            target=target,
            parameters=parameters,
            priority=int(packet.priority),
            correlation_id=packet.correlation_id,
            metadata={
                "source": "cxp",
                "packet_id": packet.message_id,
            },
        )