app/core/messaging.py

"""
Protocol-Based Agent Communication Module.

This module defines explicit message protocols for agent-to-agent communication,
replacing implicit state passing with formal message intents.

Following LangGraph best practices for multi-agent systems:
- Agents communicate via structured messages (not just state dicts)
- Message intents are explicit (TASK_REQUEST, TASK_RESPONSE, etc.)
- Human-in-the-loop checkpoints are supported
"""

from datetime import UTC, datetime
from enum import Enum
from typing import Any, TypedDict

from pydantic import BaseModel, Field


class MessageIntent(str, Enum):
    """
    Explicit message intents for agent communication.

    Following protocol-based communication patterns from:
    https://bix-tech.com/agent-to-agent-communication-with-langgraph-protocol-based-workflows-a-practical-guide/
    """

    # Task Execution
    TASK_REQUEST = "task_request"  # "Please complete this subtask"
    TASK_RESPONSE = "task_response"  # "Here's my output"

    # Control Flow
    ESCALATION = "escalation"  # "I need human help"
    CLARIFICATION_REQUEST = "clarification_request"  # "I need more info"
    APPROVAL_REQUEST = "approval_request"  # "Please approve this output"

    # Error Handling
    ERROR = "error"  # "I failed because..."
    RETRY_REQUEST = "retry_request"  # "Please retry this task"

    # Completion
    TASK_COMPLETE = "task_complete"  # "Task finished successfully"
    PIPELINE_COMPLETE = "pipeline_complete"  # "All tasks finished"


class MessagePriority(str, Enum):
    """Message priority levels for routing decisions."""

    LOW = "low"
    NORMAL = "normal"
    HIGH = "high"
    CRITICAL = "critical"


class AgentMessage(BaseModel):
    """
    Structured message between agents.

    This replaces implicit state passing with explicit message protocols.
    """

    # Message identification
    id: str = Field(default_factory=lambda: datetime.now(UTC).isoformat())

    # Protocol fields
    intent: MessageIntent
    priority: MessagePriority = MessagePriority.NORMAL

    # Communication metadata
    sender: str  # TeamRole value
    recipient: str | None = None  # TeamRole value or None for broadcast

    # Content
    content: str
    metadata: dict[str, Any] = Field(default_factory=dict)

    # Timestamps
    created_at: str = Field(default_factory=lambda: datetime.now(UTC).isoformat())
    processed_at: str | None = None

    # Status
    status: str = "pending"  # pending, processing, processed, failed

    def mark_processed(self) -> "AgentMessage":
        """Mark message as processed."""
        self.processed_at = datetime.now(UTC).isoformat()
        self.status = "processed"
        return self

    def mark_failed(self, error: str) -> "AgentMessage":
        """Mark message as failed."""
        self.metadata["error"] = error
        self.status = "failed"
        return self


class MessageQueue(TypedDict):
    """
    Queue of messages in agent state.

    This replaces the implicit history passing with explicit message queues.
    """

    messages: list[AgentMessage]
    pending_count: int
    processed_count: int


def merge_messages(a: list[AgentMessage], b: list[AgentMessage]) -> list[AgentMessage]:
    """
    Reducer function for merging message lists.

    Combines messages from multiple agents while maintaining order.
    """
    return a + b


def create_task_request_message(
    sender: str,
    recipient: str,
    content: str,
    metadata: dict | None = None,
) -> AgentMessage:
    """
    Factory function to create a task request message.

    Use this when an agent needs to delegate work to another agent.
    """
    return AgentMessage(
        intent=MessageIntent.TASK_REQUEST,
        sender=sender,
        recipient=recipient,
        content=content,
        metadata=metadata or {},
        priority=MessagePriority.HIGH,
    )


def create_task_response_message(
    sender: str,
    content: str,
    status: str = "completed",
    metadata: dict | None = None,
) -> AgentMessage:
    """
    Factory function to create a task response message.

    Use this when an agent completes a task and returns results.
    """
    return AgentMessage(
        intent=MessageIntent.TASK_RESPONSE,
        sender=sender,
        content=content,
        metadata=metadata or {"status": status},
        priority=MessagePriority.NORMAL,
    )


def create_escalation_message(
    sender: str,
    content: str,
    reason: str,
    metadata: dict | None = None,
) -> AgentMessage:
    """
    Factory function to create an escalation message.

    Use this when an agent needs human intervention or cannot proceed.
    """
    return AgentMessage(
        intent=MessageIntent.ESCALATION,
        sender=sender,
        content=content,
        metadata=metadata or {"reason": reason},
        priority=MessagePriority.CRITICAL,
    )


def create_error_message(
    sender: str,
    error: str,
    context: str | None = None,
    metadata: dict | None = None,
) -> AgentMessage:
    """
    Factory function to create an error message.

    Use this when an agent encounters an error it cannot recover from.
    """
    return AgentMessage(
        intent=MessageIntent.ERROR,
        sender=sender,
        content=error,
        metadata=metadata or {"context": context},
        priority=MessagePriority.HIGH,
    )


def create_clarification_message(
    sender: str,
    question: str,
    metadata: dict | None = None,
) -> AgentMessage:
    """
    Factory function to create a clarification request message.

    Use this when an agent needs more information to proceed.
    """
    return AgentMessage(
        intent=MessageIntent.CLARIFICATION_REQUEST,
        sender=sender,
        content=question,
        metadata=metadata or {},
        priority=MessagePriority.NORMAL,
    )


def get_pending_messages(messages: list[AgentMessage]) -> list[AgentMessage]:
    """Filter messages that are still pending."""
    return [m for m in messages if m.status == "pending"]


def get_messages_by_intent(
    messages: list[AgentMessage], intent: MessageIntent
) -> list[AgentMessage]:
    """Filter messages by intent type."""
    return [m for m in messages if m.intent == intent]


def get_messages_for_recipient(
    messages: list[AgentMessage], recipient: str
) -> list[AgentMessage]:
    """Filter messages for a specific recipient."""
    return [m for m in messages if m.recipient == recipient or m.recipient is None]