|
1 | | -"""Persist typed inter-agent messages and optionally publish them live.""" |
| 1 | +# pyright: reportWildcardImportFromLibrary=false |
| 2 | +"""Persist typed inter-agent messages and optionally publish them live (implemented by ``agent_core.components.agent_bus.agent_comm``).""" |
2 | 3 |
|
3 | | -from __future__ import annotations |
| 4 | +import sys |
4 | 5 |
|
5 | | -import asyncio |
6 | | -import logging |
7 | | -from enum import StrEnum |
8 | | -from typing import Protocol, runtime_checkable |
| 6 | +import agent_core.components.agent_bus.agent_comm as _implementation |
| 7 | +from agent_core.components.agent_bus.agent_comm import * # noqa: F403 |
9 | 8 |
|
10 | | -from frontier_agent.core.events import EventType |
11 | | -from frontier_agent.core.protocols import EventReader, EventSink |
12 | | -from frontier_agent.core.runtime.events.bus import EventBus |
13 | | -from frontier_agent.core.types import TaskId |
14 | | -from frontier_agent.models.agent_message import AgentMessage |
15 | | -from frontier_agent.models.event import KernelEvent |
16 | | - |
17 | | -logger = logging.getLogger(__name__) |
18 | | - |
19 | | - |
20 | | -@runtime_checkable |
21 | | -class _AgentCommEventStore(EventSink, EventReader, Protocol): |
22 | | - """AgentComm needs both append (EventSink) and cursored reads |
23 | | - (EventReader) — combined here because Python lacks an intersection |
24 | | - type. Public Protocols stay minimal in ``core.protocols``. |
25 | | - """ |
26 | | - |
27 | | - |
28 | | -class DeliveryMode(StrEnum): |
29 | | - """Message delivery mode. |
30 | | -
|
31 | | - TRIGGER: persist + hot queue + EventBus broadcast. |
32 | | - Use when receiver must act immediately (e.g., assertion → critic). |
33 | | - QUEUE: persist + hot queue only (no broadcast). |
34 | | - Use for status updates the receiver pulls when ready. |
35 | | - """ |
36 | | - |
37 | | - TRIGGER = "trigger" |
38 | | - QUEUE = "queue" |
39 | | - |
40 | | - |
41 | | -class AgentComm: |
42 | | - """Sends inter-agent messages and logs them as events. |
43 | | -
|
44 | | - All messages are persisted to EventStore (truth source). |
45 | | - Hot queues are optional in-memory acceleration. |
46 | | - DeliveryMode controls whether EventBus broadcast fires. |
47 | | - """ |
48 | | - |
49 | | - def __init__( |
50 | | - self, |
51 | | - event_store: _AgentCommEventStore, |
52 | | - event_bus: EventBus, |
53 | | - ) -> None: |
54 | | - self._event_store = event_store |
55 | | - self._event_bus = event_bus |
56 | | - self._cursors: dict[tuple[str, str | None], int] = {} |
57 | | - self._hot_queues: dict[str, asyncio.Queue[KernelEvent]] = {} |
58 | | - |
59 | | - async def send( |
60 | | - self, |
61 | | - msg: AgentMessage, |
62 | | - mode: DeliveryMode = DeliveryMode.QUEUE, |
63 | | - ) -> KernelEvent: |
64 | | - """Send an agent message. |
65 | | -
|
66 | | - Always persists to EventStore (truth source). |
67 | | - TRIGGER mode additionally broadcasts via EventBus for immediate wakeup. |
68 | | - """ |
69 | | - event = KernelEvent( |
70 | | - task_id=TaskId(msg.task_id), |
71 | | - event_type=EventType.AGENT_MESSAGE, |
72 | | - from_agent=msg.from_agent, |
73 | | - to_agent=msg.to_agent, |
74 | | - message_type=msg.message_type, |
75 | | - correlation_id=msg.content.get("correlation_id"), |
76 | | - payload={ |
77 | | - "message_id": msg.id, |
78 | | - "from_agent": msg.from_agent, |
79 | | - "to_agent": msg.to_agent, |
80 | | - "message_type": msg.message_type, |
81 | | - "content": msg.content, |
82 | | - "parent_id": msg.parent_id, |
83 | | - "correlation_id": msg.content.get("correlation_id"), |
84 | | - "delivery_mode": mode.value, |
85 | | - }, |
86 | | - ) |
87 | | - |
88 | | - # 1. Persist (always — EventStore is truth source) |
89 | | - persisted = await self._event_store.append(event) |
90 | | - |
91 | | - # 2. Hot queue (always — non-durable acceleration) |
92 | | - queue = self._hot_queues.get(msg.to_agent) |
93 | | - if queue is not None: |
94 | | - try: |
95 | | - queue.put_nowait(persisted) |
96 | | - except asyncio.QueueFull: |
97 | | - logger.debug( |
98 | | - "Hot queue for %s is full; consumer uses EventStore", |
99 | | - msg.to_agent, |
100 | | - ) |
101 | | - |
102 | | - # 3. Broadcast (TRIGGER only — immediate wakeup signal) |
103 | | - if mode == DeliveryMode.TRIGGER: |
104 | | - await self._event_bus.publish( |
105 | | - event.event_type, event.payload, |
106 | | - ) |
107 | | - |
108 | | - logger.debug( |
109 | | - "Agent message %s: %s → %s [%s] mode=%s", |
110 | | - msg.id, msg.from_agent, msg.to_agent, |
111 | | - msg.message_type, mode.value, |
112 | | - ) |
113 | | - return persisted |
114 | | - |
115 | | - async def consume( |
116 | | - self, |
117 | | - agent_id: str, |
118 | | - *, |
119 | | - task_id: str | None = None, |
120 | | - limit: int = 50, |
121 | | - ) -> list[KernelEvent]: |
122 | | - """Consume undelivered messages for an agent using an idempotent cursor. |
123 | | -
|
124 | | - Cursor is per (agent_id, task_id). Restarting from 0 replays all. |
125 | | - """ |
126 | | - cursor_key = (agent_id, task_id) |
127 | | - after_id = self._cursors.get(cursor_key, 0) |
128 | | - events = await self._event_store.get_events_for_agent( |
129 | | - agent_id, |
130 | | - after_id=after_id, |
131 | | - limit=limit, |
132 | | - task_id=task_id, |
133 | | - ) |
134 | | - if events: |
135 | | - self._cursors[cursor_key] = int(events[-1].id) |
136 | | - return events |
137 | | - |
138 | | - def reset_cursor( |
139 | | - self, agent_id: str, task_id: str | None = None, |
140 | | - ) -> None: |
141 | | - """Reset cursor for an agent (e.g., after recovery).""" |
142 | | - self._cursors.pop((agent_id, task_id), None) |
143 | | - |
144 | | - def hot_queue_for( |
145 | | - self, agent_id: str, |
146 | | - ) -> asyncio.Queue[KernelEvent]: |
147 | | - """Return the non-durable hot queue for an agent.""" |
148 | | - return self._hot_queues.setdefault( |
149 | | - agent_id, asyncio.Queue(maxsize=256), |
150 | | - ) |
| 9 | +sys.modules[__name__] = _implementation |
0 commit comments