orchestrata_ultimate.agents.communication

Reference › orchestrata_ultimate › agents › communication

communication implements hierarchy-bound messaging between agents using CXP packets (specification 005, user story US-002). AgentCommunication is built over an AgentHierarchy from hierarchy and an optional MultiagentPrologBridge from prolog_bridge. It is used by escalation.

Each agent has an in-memory mailbox. can_communicate decides whether a sender may reach a receiver. The send methods build a CXPPacket with _build_packet and deliver it with _deliver: send_to_parent, send_to_child, broadcast_to_children, send_to_sibling (same parent only) and escalate_message (up the hierarchy). They return the packet, or None when the hierarchy does not allow the route. receive drains an agent's mailbox, peek counts waiting messages and get_metrics returns counters. CommunicationError carries the sender and receiver for reporting.

At a glance

Class diagram

Dependency diagram

API

Classes

CommunicationError

Bases: Exception · Decorators: traces(spec='005', story='US-002', reqs=['REQ-005-AC4-001']) · Line 27

Communication error.

AgentCommunication

Line 36

Manages inter-agent messaging via CXP packets.

Source

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

# mypy: ignore-errors
"""Inter-agent communication via CXP protocol.

Manages message routing between agents in the hierarchy,
enforcing communication boundaries and escalation rules.
"""

# Spec: 005 - Multi-Agent System

from __future__ import annotations

import logging
import uuid
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.agents.hierarchy import AgentHierarchy
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, PacketType

if TYPE_CHECKING:
    from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge

logger = logging.getLogger(__name__)


@traces(spec="005", story="US-002", reqs=["REQ-005-AC4-001"])
class CommunicationError(Exception):
    """Communication error."""

    def __init__(self, message: str, sender: str = "", receiver: str = "") -> None:
        self.sender = sender
        self.receiver = receiver
        super().__init__(message)


class AgentCommunication:
    """Manages inter-agent messaging via CXP packets.

    All communication respects hierarchy boundaries:
    - Parent-child communication is always allowed
    - Sibling communication is allowed within the same parent
    - Cross-hierarchy communication requires escalation
    """

    def __init__(
        self,
        hierarchy: AgentHierarchy,
        prolog_bridge: MultiagentPrologBridge | None = None,
    ) -> None:
        self._hierarchy = hierarchy
        self._prolog = prolog_bridge
        self._mailboxes: dict[str, list[CXPPacket]] = {}
        self._sent_count = 0
        self._received_count = 0

    def _ensure_mailbox(self, agent_id: str) -> None:
        if agent_id not in self._mailboxes:
            self._mailboxes[agent_id] = []

    def _build_packet(
        self,
        from_agent: str,
        to_agent: str,
        message: Any,
        subject: str = "",
        priority: int = 5,
    ) -> CXPPacket:
        return CXPPacket(
            from_agent=from_agent,
            to_agent=to_agent,
            subject=subject or f"agent-message:{from_agent}->{to_agent}",
            payload=message,
            priority=priority,
            message_id=str(uuid.uuid4()),
            thread_id=str(uuid.uuid4()),
            packet_type=PacketType.MESSAGE,
        )

    def can_communicate(self, sender: str, receiver: str) -> bool:
        """Check if sender can communicate with receiver."""
        # Same agent
        if sender == receiver:
            return True
        # Parent-child
        parent = self._hierarchy.get_parent(receiver)
        if parent == sender:
            return True
        parent = self._hierarchy.get_parent(sender)
        if parent == receiver:
            return True
        # Siblings
        siblings = self._hierarchy.get_siblings(sender)
        if receiver in siblings:
            return True
        # Prolog check
        if self._prolog is not None:
            try:
                return self._prolog.can_communicate(sender, receiver)
            except Exception as exc:
                logger.warning(
                    "Prolog can_communicate check failed for '%s'->'%s': %s",
                    sender,
                    receiver,
                    exc,
                    exc_info=True,
                )
        return False

    @traces(spec="005", story="US-002", reqs=["REQ-005-FR7-001", "REQ-005-AC4-001"])
    def send_to_parent(self, agent_id: str, message: Any) -> CXPPacket | None:
        """Send a message to the agent's parent."""
        parent_id = self._hierarchy.get_parent(agent_id)
        if parent_id is None:
            raise CommunicationError("Agent has no parent", sender=agent_id)

        packet = self._build_packet(agent_id, parent_id, message, subject="to-parent")
        self._deliver(parent_id, packet)
        return packet

    @traces(spec="005", story="US-002", reqs=["REQ-005-AC4-001"])
    def send_to_child(
        self, agent_id: str, child_id: str, message: Any
    ) -> CXPPacket | None:
        """Send a message to a specific child."""
        children = self._hierarchy.get_children(agent_id)
        if child_id not in children:
            raise CommunicationError(
                f"{child_id} is not a child of {agent_id}",
                sender=agent_id,
                receiver=child_id,
            )

        packet = self._build_packet(agent_id, child_id, message, subject="to-child")
        self._deliver(child_id, packet)
        return packet

    def broadcast_to_children(self, agent_id: str, message: Any) -> list[CXPPacket]:
        """Broadcast a message to all children."""
        children = self._hierarchy.get_children(agent_id)
        packets = []
        for child_id in children:
            packet = self._build_packet(
                agent_id, child_id, message, subject="broadcast"
            )
            self._deliver(child_id, packet)
            packets.append(packet)
        return packets

    def send_to_sibling(
        self, agent_id: str, sibling_id: str, message: Any
    ) -> CXPPacket | None:
        """Send a message to a sibling (same parent)."""
        siblings = self._hierarchy.get_siblings(agent_id)
        if sibling_id not in siblings:
            raise CommunicationError(
                f"{sibling_id} is not a sibling of {agent_id}",
                sender=agent_id,
                receiver=sibling_id,
            )
        packet = self._build_packet(agent_id, sibling_id, message, subject="to-sibling")
        self._deliver(sibling_id, packet)
        return packet

    @traces(
        spec="005",
        story="US-006",
        reqs=["REQ-005-FR7-001", "REQ-005-US-006-001"],
    )
    def escalate_message(self, agent_id: str, message: Any) -> CXPPacket | None:
        """Escalate a message up the hierarchy."""
        parent_id = self._hierarchy.get_parent(agent_id)
        if parent_id is None:
            raise CommunicationError("Cannot escalate: no parent", sender=agent_id)

        packet = self._build_packet(
            agent_id,
            parent_id,
            message,
            subject="escalation",
            priority=8,
        )
        self._deliver(parent_id, packet)
        return packet

    @traces(spec="005", story="US-004", reqs=["REQ-005-AC6-001"])
    def _deliver(self, recipient: str, packet: CXPPacket) -> None:
        """Deliver a packet to a recipient's mailbox."""
        self._ensure_mailbox(recipient)
        self._mailboxes[recipient].append(packet)
        self._sent_count += 1

    def receive(self, agent_id: str) -> list[CXPPacket]:
        """Receive all pending messages for an agent."""
        self._ensure_mailbox(agent_id)
        messages = list(self._mailboxes[agent_id])
        self._mailboxes[agent_id].clear()
        self._received_count += len(messages)
        return messages

    def peek(self, agent_id: str) -> int:
        """Check how many messages are waiting."""
        return len(self._mailboxes.get(agent_id, []))

    @traces(
        spec="005",
        story="US-004",
        reqs=["REQ-005-AC6-001"],
    )
    def get_metrics(self) -> dict[str, int]:
        return {
            "sent": self._sent_count,
            "received": self._received_count,
            "pending": sum(len(mb) for mb in self._mailboxes.values()),
        }