56 lines
1.6 KiB
Python
56 lines
1.6 KiB
Python
"""Agent OS message bus runtime."""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Optional
|
|
from enum import Enum
|
|
|
|
|
|
class DeliveryStatus(Enum):
|
|
PENDING = "pending"
|
|
DELIVERED = "delivered"
|
|
FAILED = "failed"
|
|
RETRYING = "retrying"
|
|
EXPIRED = "expired"
|
|
|
|
|
|
@dataclass
|
|
class AgentMessage:
|
|
sender: str
|
|
receiver: str
|
|
event: str
|
|
payload: dict
|
|
# V1R1C5: New fields for message tracking and delivery
|
|
task_id: str = ""
|
|
correlation_id: str = ""
|
|
retry_state: int = 0
|
|
delivery_status: DeliveryStatus = DeliveryStatus.PENDING
|
|
|
|
|
|
class MessageBus:
|
|
def __init__(self):
|
|
self.queue: List[AgentMessage] = []
|
|
|
|
def send(self, message: AgentMessage):
|
|
self.queue.append(message)
|
|
|
|
def broadcast(self, sender: str, event: str, payload: dict, receivers: List[str]):
|
|
for receiver in receivers:
|
|
self.send(AgentMessage(
|
|
sender=sender,
|
|
receiver=receiver,
|
|
event=event,
|
|
payload=payload,
|
|
task_id=payload.get("task_id", "") if isinstance(payload, dict) else "",
|
|
correlation_id=payload.get("correlation_id", "") if isinstance(payload, dict) else "",
|
|
))
|
|
|
|
def consume(self):
|
|
messages = list(self.queue)
|
|
self.queue.clear()
|
|
return messages
|
|
|
|
# R4.2 P0-B compatibility adapter.
|
|
def to_handoff_envelope(message, *, target=None, sequence=1, idempotency_key=None, **scope):
|
|
from message_handoff_adapters import legacy_message_to_envelope
|
|
return legacy_message_to_envelope(message, target=target, sequence=sequence, idempotency_key=idempotency_key, **scope)
|