src/protocol/envelope.py

"""OFP Envelope handling"""

from typing import Any, Dict, List, Optional
from dataclasses import dataclass, asdict
from datetime import datetime

from ..utils.helpers import generate_message_id, get_timestamp
from ..utils.config import settings


@dataclass
class Envelope:
    """
    OpenFloor Protocol Envelope structure
    Based on: https://openfloor.dev/protocol/specifications/inter-agent-message.md
    """
    version: str
    sender: str
    recipients: List[str]
    message: Dict[str, Any]
    timestamp: str
    message_id: str
    correlation_id: Optional[str] = None
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert envelope to dictionary"""
        data = asdict(self)
        # Remove None values
        return {k: v for k, v in data.items() if v is not None}
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Envelope":
        """Create envelope from dictionary"""
        return cls(
            version=data.get("version", settings.OFP_VERSION),
            sender=data["sender"],
            recipients=data["recipients"],
            message=data["message"],
            timestamp=data.get("timestamp", get_timestamp()),
            message_id=data.get("message_id", generate_message_id()),
            correlation_id=data.get("correlation_id")
        )


def create_envelope(
    sender: str,
    recipients: List[str],
    message: Dict[str, Any],
    correlation_id: Optional[str] = None
) -> Envelope:
    """
    Create a new OFP envelope
    
    Args:
        sender: Agent ID of sender
        recipients: List of recipient agent IDs or ["broadcast"]
        message: Message content (InterAgentMessage)
        correlation_id: Optional correlation ID for request-response
    
    Returns:
        Envelope object
    """
    return Envelope(
        version=settings.OFP_VERSION,
        sender=sender,
        recipients=recipients,
        message=message,
        timestamp=get_timestamp(),
        message_id=generate_message_id(),
        correlation_id=correlation_id
    )


def create_inter_agent_message(
    message_type: str,
    content: Any,
    dialog_event: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    Create InterAgentMessage structure
    Based on: https://openfloor.dev/protocol/specifications/inter-agent-message.md
    
    Args:
        message_type: Type of message (e.g., "agent_response", "floor_request")
        content: Message content (can be text, JSON, etc.)
        dialog_event: Optional DialogEventObject
        meta Optional metadata
    
    Returns:
        InterAgentMessage dictionary
    """
    message = {
        "message_type": message_type,
        "content": content,
        "timestamp": get_timestamp()
    }
    
    if dialog_event:
        message["dialog_event"] = dialog_event
    
    if metadata:
        message["metadata"] = metadata
    
    return message