|
1
|
""" |
|
2
|
Agent-to-agent async messaging via Redis. |
|
3
|
|
|
4
|
Messages are queued in Redis lists, one list per recipient agent. A broadcast |
|
5
|
copies the message to every currently-known agent queue. |
|
6
|
|
|
7
|
All queues live under the ``navegador:msg:`` namespace. |
|
8
|
""" |
|
9
|
|
|
10
|
from __future__ import annotations |
|
11
|
|
|
12
|
import json |
|
13
|
import logging |
|
14
|
import time |
|
15
|
import uuid |
|
16
|
from dataclasses import asdict, dataclass |
|
17
|
from typing import Any |
|
18
|
|
|
19
|
logger = logging.getLogger(__name__) |
|
20
|
|
|
21
|
_QUEUE_PREFIX = "navegador:msg:queue:" |
|
22
|
_ACK_PREFIX = "navegador:msg:ack:" |
|
23
|
_ALL_AGENTS_KEY = "navegador:msg:agents" |
|
24
|
|
|
25
|
|
|
26
|
@dataclass |
|
27
|
class Message: |
|
28
|
"""A single agent-to-agent message.""" |
|
29
|
|
|
30
|
id: str |
|
31
|
from_agent: str |
|
32
|
to_agent: str |
|
33
|
type: str |
|
34
|
payload: dict |
|
35
|
timestamp: float |
|
36
|
acknowledged: bool = False |
|
37
|
|
|
38
|
def to_dict(self) -> dict: |
|
39
|
return asdict(self) |
|
40
|
|
|
41
|
@classmethod |
|
42
|
def from_dict(cls, data: dict) -> "Message": |
|
43
|
return cls( |
|
44
|
id=data["id"], |
|
45
|
from_agent=data["from_agent"], |
|
46
|
to_agent=data["to_agent"], |
|
47
|
type=data["type"], |
|
48
|
payload=data.get("payload", {}), |
|
49
|
timestamp=data["timestamp"], |
|
50
|
acknowledged=data.get("acknowledged", False), |
|
51
|
) |
|
52
|
|
|
53
|
|
|
54
|
class MessageBus: |
|
55
|
""" |
|
56
|
Async message bus for agent-to-agent communication. |
|
57
|
|
|
58
|
Messages are stored in per-agent Redis lists. Acknowledged messages are |
|
59
|
tracked in a separate key set; unacknowledged messages remain in the queue. |
|
60
|
|
|
61
|
Args: |
|
62
|
redis_url: Redis connection URL. |
|
63
|
_redis_client: Optional pre-built Redis client (for testing). |
|
64
|
""" |
|
65
|
|
|
66
|
def __init__(self, redis_url: str, _redis_client: Any = None) -> None: |
|
67
|
self._url = redis_url |
|
68
|
self._redis: Any = _redis_client |
|
69
|
|
|
70
|
# ── Internal ────────────────────────────────────────────────────────────── |
|
71
|
|
|
72
|
def _client(self) -> Any: |
|
73
|
if self._redis is None: |
|
74
|
try: |
|
75
|
import redis # type: ignore[import] |
|
76
|
except ImportError as exc: |
|
77
|
raise ImportError("Install redis: pip install redis") from exc |
|
78
|
self._redis = redis.from_url(self._url) |
|
79
|
return self._redis |
|
80
|
|
|
81
|
def _queue_key(self, agent_id: str) -> str: |
|
82
|
return f"{_QUEUE_PREFIX}{agent_id}" |
|
83
|
|
|
84
|
def _ack_key(self, agent_id: str) -> str: |
|
85
|
return f"{_ACK_PREFIX}{agent_id}" |
|
86
|
|
|
87
|
def _register_agent(self, agent_id: str) -> None: |
|
88
|
"""Track the agent so broadcasts can find it.""" |
|
89
|
self._client().sadd(_ALL_AGENTS_KEY, agent_id) |
|
90
|
|
|
91
|
def _push_message(self, message: Message) -> None: |
|
92
|
"""Push a serialised message onto the recipient's queue.""" |
|
93
|
client = self._client() |
|
94
|
self._register_agent(message.to_agent) |
|
95
|
client.rpush(self._queue_key(message.to_agent), json.dumps(message.to_dict())) |
|
96
|
|
|
97
|
# ── Public API ──────────────────────────────────────────────────────────── |
|
98
|
|
|
99
|
def send( |
|
100
|
self, |
|
101
|
from_agent: str, |
|
102
|
to_agent: str, |
|
103
|
message_type: str, |
|
104
|
payload: dict, |
|
105
|
) -> str: |
|
106
|
""" |
|
107
|
Send a message from one agent to another. |
|
108
|
|
|
109
|
Args: |
|
110
|
from_agent: Sender agent ID. |
|
111
|
to_agent: Recipient agent ID. |
|
112
|
message_type: Semantic type label for the message. |
|
113
|
payload: Arbitrary JSON-serialisable dict. |
|
114
|
|
|
115
|
Returns: |
|
116
|
Unique message ID (UUID4 string). |
|
117
|
""" |
|
118
|
msg = Message( |
|
119
|
id=str(uuid.uuid4()), |
|
120
|
from_agent=from_agent, |
|
121
|
to_agent=to_agent, |
|
122
|
type=message_type, |
|
123
|
payload=payload, |
|
124
|
timestamp=time.time(), |
|
125
|
) |
|
126
|
self._push_message(msg) |
|
127
|
logger.debug("Message sent: %s -> %s [%s]", from_agent, to_agent, msg.id) |
|
128
|
return msg.id |
|
129
|
|
|
130
|
def receive(self, agent_id: str, limit: int = 10) -> list[Message]: |
|
131
|
""" |
|
132
|
Retrieve pending (unacknowledged) messages for an agent. |
|
133
|
|
|
134
|
Messages remain in the queue until acknowledged via :meth:`acknowledge`. |
|
135
|
|
|
136
|
Args: |
|
137
|
agent_id: The receiving agent's ID. |
|
138
|
limit: Maximum number of messages to return (default 10). |
|
139
|
|
|
140
|
Returns: |
|
141
|
List of :class:`Message` objects, oldest first. |
|
142
|
""" |
|
143
|
client = self._client() |
|
144
|
self._register_agent(agent_id) |
|
145
|
raw_items = client.lrange(self._queue_key(agent_id), 0, limit - 1) |
|
146
|
acked_ids: set[str] = set( |
|
147
|
i.decode() if isinstance(i, bytes) else i |
|
148
|
for i in client.smembers(self._ack_key(agent_id)) |
|
149
|
) |
|
150
|
|
|
151
|
messages = [] |
|
152
|
for raw in raw_items: |
|
153
|
if isinstance(raw, bytes): |
|
154
|
raw = raw.decode() |
|
155
|
data = json.loads(raw) |
|
156
|
msg = Message.from_dict(data) |
|
157
|
if msg.id not in acked_ids: |
|
158
|
messages.append(msg) |
|
159
|
return messages |
|
160
|
|
|
161
|
def acknowledge(self, message_id: str, agent_id: str | None = None) -> None: |
|
162
|
""" |
|
163
|
Mark a message as read. |
|
164
|
|
|
165
|
Because messages stay in the queue list (for replay), acknowledgement |
|
166
|
is stored in a separate Redis set. If *agent_id* is not supplied the |
|
167
|
method scans all known agent ack-sets (less efficient). |
|
168
|
|
|
169
|
Args: |
|
170
|
message_id: The message ID to acknowledge. |
|
171
|
agent_id: Optionally scope the acknowledgement to one agent's set. |
|
172
|
""" |
|
173
|
client = self._client() |
|
174
|
if agent_id: |
|
175
|
client.sadd(self._ack_key(agent_id), message_id) |
|
176
|
else: |
|
177
|
# Best-effort: acknowledge in all known agent sets |
|
178
|
all_agents = client.smembers(_ALL_AGENTS_KEY) |
|
179
|
for a in all_agents: |
|
180
|
if isinstance(a, bytes): |
|
181
|
a = a.decode() |
|
182
|
client.sadd(self._ack_key(a), message_id) |
|
183
|
logger.debug("Message acknowledged: %s", message_id) |
|
184
|
|
|
185
|
def broadcast( |
|
186
|
self, |
|
187
|
from_agent: str, |
|
188
|
message_type: str, |
|
189
|
payload: dict, |
|
190
|
) -> list[str]: |
|
191
|
""" |
|
192
|
Send a message to every registered agent. |
|
193
|
|
|
194
|
Args: |
|
195
|
from_agent: Sender agent ID. |
|
196
|
message_type: Semantic type label. |
|
197
|
payload: Arbitrary JSON-serialisable dict. |
|
198
|
|
|
199
|
Returns: |
|
200
|
List of message IDs (one per recipient). |
|
201
|
""" |
|
202
|
client = self._client() |
|
203
|
all_agents = client.smembers(_ALL_AGENTS_KEY) |
|
204
|
message_ids = [] |
|
205
|
for agent_bytes in all_agents: |
|
206
|
to_agent = agent_bytes.decode() if isinstance(agent_bytes, bytes) else agent_bytes |
|
207
|
if to_agent == from_agent: |
|
208
|
continue |
|
209
|
mid = self.send(from_agent, to_agent, message_type, payload) |
|
210
|
message_ids.append(mid) |
|
211
|
logger.debug( |
|
212
|
"Broadcast from %s [%s]: %d recipients", from_agent, message_type, len(message_ids) |
|
213
|
) |
|
214
|
return message_ids |
|
215
|
|