orchestrata_ultimate.agents.cxp_adapter

Reference › orchestrata_ultimate › agents › cxp_adapter

cxp_adapter defines CXPVolkAdapter, a small base class for volks that receive CXP packets. It holds an alias and a role, takes a Postman reference through set_postman, and declares the entry points process_cxp (packet in, optional response packet out), process_intent (the production interface, a dictionary in and a dictionary out) and get_output_display (a human-readable summary of the volk's output). It depends only on the packet type in orchestrata_ultimate.core.packet. See the agents narrative for how volks are fed packets.

At a glance

Class diagram

Dependency diagram

API

Classes

CXPVolkAdapter

Line 16

Base adapter for volks that process CXP packets.

Source

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

"""CXP Volk Adapter — base class for volks that process CXP packets.

Bridges the production async lifecycle interface with the CXP packet
processing pattern from the spike. Volks inherit from this class and
implement process_cxp() to handle incoming CXP packets.
"""

import logging
from typing import Any

from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

_logger = logging.getLogger(__name__)


class CXPVolkAdapter:
    """Base adapter for volks that process CXP packets.

    Provides common state tracking (alias, role, status, summary)
    and the process_cxp() interface. Subclasses implement the
    volk-specific LLM reasoning logic.

    Attributes:
        alias: Volk identifier (e.g., "U:Cx", "U:Ar").
        role: Human-readable role description.
        status: Current work status.
        summary: Current status message.
        current_thread: Active CXP thread ID.
        output_artifacts: Files/artifacts produced by this volk.
    """

    alias: str = ""
    role: str = ""

    def __init__(self) -> None:
        """Initialize volk adapter with default state."""
        self.status: WorkStatus = WorkStatus.PENDING
        self.summary: str = "Idle"
        self.current_thread: str | None = None
        self.output_artifacts: list[str] = []
        self._postman = None

    def set_postman(self, postman: Any) -> None:
        """Set the postman reference for message routing.

        Args:
            postman: The postman/router instance.
        """
        self._postman = postman

    async def process_cxp(
        self,
        packet: CXPPacket,
    ) -> CXPPacket | None:
        """Process an incoming CXP packet and return a response.

        Subclasses must override this method with volk-specific logic.

        Args:
            packet: The incoming CXP packet.

        Returns:
            Response CXP packet, or None if no response.
        """
        raise NotImplementedError(
            f"{self.__class__.__name__}.process_cxp() not implemented"
        )

    async def process_intent(
        self,
        intent: dict[str, Any],
    ) -> dict[str, Any]:
        """Process an intent dictionary (production interface).

        Converts between the dict-based production interface and the
        CXP packet-based interface. Override process_cxp() instead.

        Args:
            intent: Intent dictionary with at least 'action' and 'data'.

        Returns:
            Response dictionary.
        """
        _logger.debug(
            "[%s] Processing intent: %s",
            self.alias,
            intent.get("action", "unknown"),
        )
        return {"status": "processed", "volk": self.alias}

    def get_output_display(self) -> str:
        """Get a human-readable display of volk output.

        Returns:
            Formatted string of output artifacts.
        """
        if not self.output_artifacts:
            return f"[{self.alias}] No output yet"
        lines = [f"[{self.alias}] Output:"]
        for artifact in self.output_artifacts:
            lines.append(f"  - {artifact}")
        return "\n".join(lines)