enki_core/
message.rs

1//! Message types for inter-agent communication.
2//!
3//! This module provides a unified message structure for agent communication
4//! with typed content, flexible routing, and backward compatibility.
5
6use serde::{Deserialize, Serialize};
7use std::collections::HashMap;
8use std::fmt;
9use uuid::Uuid;
10
11/// Message content types for typed payloads.
12///
13/// This enum provides type-safe message content variants for different
14/// communication patterns between agents.
15#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
16#[serde(tag = "type", content = "data")]
17pub enum MessageContent {
18    /// Raw binary data
19    Binary(Vec<u8>),
20
21    /// UTF-8 text content
22    Text(String),
23
24    /// JSON-serialized structured data
25    Json(serde_json::Value),
26
27    /// Agent request (RPC-style)
28    AgentRequest {
29        method: String,
30        params: serde_json::Value,
31    },
32
33    /// Agent response
34    AgentResponse { result: serde_json::Value },
35
36    /// Tool invocation
37    ToolCall {
38        name: String,
39        arguments: serde_json::Value,
40    },
41
42    /// Tool execution result
43    ToolResult {
44        output: String,
45        error: Option<String>,
46    },
47
48    /// Empty message (e.g., heartbeat, acknowledgment)
49    Empty,
50}
51
52impl MessageContent {
53    /// Get a human-readable type name for this content
54    pub fn type_name(&self) -> &'static str {
55        match self {
56            MessageContent::Binary(_) => "binary",
57            MessageContent::Text(_) => "text",
58            MessageContent::Json(_) => "json",
59            MessageContent::AgentRequest { .. } => "agent_request",
60            MessageContent::AgentResponse { .. } => "agent_response",
61            MessageContent::ToolCall { .. } => "tool_call",
62            MessageContent::ToolResult { .. } => "tool_result",
63            MessageContent::Empty => "empty",
64        }
65    }
66
67    /// Check if content is empty
68    pub fn is_empty(&self) -> bool {
69        matches!(self, MessageContent::Empty)
70    }
71}
72
73/// Message routing target.
74///
75/// Defines how a message should be routed within the mesh.
76#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
77#[serde(tag = "type", content = "target")]
78pub enum MessageTarget {
79    /// Direct message to a specific agent
80    Agent(String),
81
82    /// Broadcast to all agents (excluding sender)
83    Broadcast,
84
85    /// Topic-based routing
86    Topic(String),
87
88    /// Route to specific node in distributed mesh
89    Node {
90        node_id: String,
91        agent: Option<String>,
92    },
93}
94
95impl MessageTarget {
96    /// Create a target for a specific agent
97    pub fn agent(name: impl Into<String>) -> Self {
98        MessageTarget::Agent(name.into())
99    }
100
101    /// Create a broadcast target
102    pub fn broadcast() -> Self {
103        MessageTarget::Broadcast
104    }
105
106    /// Create a topic target
107    pub fn topic(name: impl Into<String>) -> Self {
108        MessageTarget::Topic(name.into())
109    }
110}
111
112impl fmt::Display for MessageTarget {
113    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
114        match self {
115            MessageTarget::Agent(name) => write!(f, "agent:{}", name),
116            MessageTarget::Broadcast => write!(f, "broadcast"),
117            MessageTarget::Topic(topic) => write!(f, "topic:{}", topic),
118            MessageTarget::Node { node_id, agent } => {
119                if let Some(agent_name) = agent {
120                    write!(f, "node:{}@{}", agent_name, node_id)
121                } else {
122                    write!(f, "node:{}", node_id)
123                }
124            }
125        }
126    }
127}
128
129/// Legacy message structure (v0) for backward compatibility.
130///
131/// This is the original message format with binary payloads.
132/// New code should use [`Message`] instead.
133#[derive(Debug, Clone, Serialize, Deserialize)]
134pub struct LegacyMessage {
135    /// Unique message identifier (auto-generated UUID).
136    pub id: String,
137    /// Message topic/type for routing or categorization.
138    pub topic: String,
139    /// Binary payload data.
140    pub payload: Vec<u8>,
141    /// Name of the sending agent.
142    pub sender: String,
143    /// Arbitrary metadata key-value pairs.
144    pub metadata: HashMap<String, String>,
145    /// Creation timestamp in microseconds since epoch.
146    pub created_at: i64,
147}
148
149impl LegacyMessage {
150    /// Create a new legacy message.
151    pub fn new(topic: impl Into<String>, payload: Vec<u8>, sender: impl Into<String>) -> Self {
152        Self {
153            id: Uuid::new_v4().to_string(),
154            topic: topic.into(),
155            payload,
156            sender: sender.into(),
157            metadata: HashMap::new(),
158            created_at: chrono::Utc::now().timestamp_micros(),
159        }
160    }
161
162    /// Get the correlation ID for request/response tracking.
163    pub fn correlation_id(&self) -> Option<String> {
164        self.metadata.get("correlation_id").cloned()
165    }
166
167    /// Set a correlation ID for request/response tracking.
168    pub fn set_correlation_id(&mut self, id: &str) {
169        self.metadata
170            .insert("correlation_id".to_string(), id.to_string());
171    }
172}
173
174/// A message for inter-agent communication (v1).
175///
176/// The new message format with typed content, flexible routing, and metadata.
177///
178/// # Example
179///
180/// ```rust
181/// use enki_core::{Message, MessageContent, MessageTarget};
182///
183/// // Create a text message
184/// let msg = Message::text("my-agent", "Hello, world!")
185///     .to(MessageTarget::agent("other-agent"));
186///
187/// // Create a tool call
188/// let tool_msg = Message::builder("my-agent")
189///     .content(MessageContent::ToolCall {
190///         name: "search".to_string(),
191///         arguments: serde_json::json!({"query": "rust"}),
192///     })
193///     .to(MessageTarget::agent("tool-handler"))
194///     .build();
195/// ```
196#[derive(Debug, Clone, Serialize, Deserialize)]
197pub struct Message {
198    /// Unique message identifier (auto-generated UUID).
199    pub id: String,
200
201    /// Protocol version (currently 1).
202    pub version: u8,
203
204    /// Sender agent name.
205    pub from: String,
206
207    /// Intended recipient(s).
208    pub to: MessageTarget,
209
210    /// Message content type and payload.
211    pub content: MessageContent,
212
213    /// Optional correlation ID for request/response tracking.
214    pub correlation_id: Option<String>,
215
216    /// Optional topic for pub/sub routing.
217    pub topic: Option<String>,
218
219    /// Arbitrary metadata key-value pairs.
220    pub metadata: HashMap<String, String>,
221
222    /// Creation timestamp in microseconds since epoch.
223    pub created_at: i64,
224}
225
226impl Message {
227    /// Create a message builder.
228    ///
229    /// # Example
230    /// ```
231    /// use enki_core::{Message, MessageContent, MessageTarget};
232    ///
233    /// let msg = Message::builder("agent1")
234    ///     .content(MessageContent::Text("Hello".to_string()))
235    ///     .to(MessageTarget::agent("agent2"))
236    ///     .build();
237    /// ```
238    pub fn builder(from: impl Into<String>) -> MessageBuilder {
239        MessageBuilder::new(from)
240    }
241
242    /// Create a text message.
243    pub fn text(from: impl Into<String>, content: impl Into<String>) -> MessageBuilder {
244        MessageBuilder::new(from).content(MessageContent::Text(content.into()))
245    }
246
247    /// Create a JSON message.
248    pub fn json(from: impl Into<String>, data: serde_json::Value) -> MessageBuilder {
249        MessageBuilder::new(from).content(MessageContent::Json(data))
250    }
251
252    /// Create a binary message.
253    pub fn binary(from: impl Into<String>, data: Vec<u8>) -> MessageBuilder {
254        MessageBuilder::new(from).content(MessageContent::Binary(data))
255    }
256
257    /// Create a tool call message.
258    pub fn tool_call(
259        from: impl Into<String>,
260        name: impl Into<String>,
261        arguments: serde_json::Value,
262    ) -> MessageBuilder {
263        MessageBuilder::new(from).content(MessageContent::ToolCall {
264            name: name.into(),
265            arguments,
266        })
267    }
268
269    /// Create a tool result message.
270    pub fn tool_result(
271        from: impl Into<String>,
272        output: impl Into<String>,
273        error: Option<String>,
274    ) -> MessageBuilder {
275        MessageBuilder::new(from).content(MessageContent::ToolResult {
276            output: output.into(),
277            error,
278        })
279    }
280
281    /// Create an agent request message.
282    pub fn agent_request(
283        from: impl Into<String>,
284        method: impl Into<String>,
285        params: serde_json::Value,
286    ) -> MessageBuilder {
287        MessageBuilder::new(from).content(MessageContent::AgentRequest {
288            method: method.into(),
289            params,
290        })
291    }
292
293    /// Create an agent response message.
294    pub fn agent_response(from: impl Into<String>, result: serde_json::Value) -> MessageBuilder {
295        MessageBuilder::new(from).content(MessageContent::AgentResponse { result })
296    }
297
298    /// Create an empty message (heartbeat/ack).
299    pub fn empty(from: impl Into<String>) -> MessageBuilder {
300        MessageBuilder::new(from).content(MessageContent::Empty)
301    }
302}
303
304/// Message builder for fluent construction.
305pub struct MessageBuilder {
306    id: String,
307    version: u8,
308    from: String,
309    to: Option<MessageTarget>,
310    content: Option<MessageContent>,
311    correlation_id: Option<String>,
312    topic: Option<String>,
313    metadata: HashMap<String, String>,
314}
315
316impl MessageBuilder {
317    fn new(from: impl Into<String>) -> Self {
318        Self {
319            id: Uuid::new_v4().to_string(),
320            version: 1,
321            from: from.into(),
322            to: None,
323            content: None,
324            correlation_id: None,
325            topic: None,
326            metadata: HashMap::new(),
327        }
328    }
329
330    /// Set the message ID (usually auto-generated).
331    pub fn id(mut self, id: impl Into<String>) -> Self {
332        self.id = id.into();
333        self
334    }
335
336    /// Set the target.
337    pub fn to(mut self, target: MessageTarget) -> Self {
338        self.to = Some(target);
339        self
340    }
341
342    /// Set the content.
343    pub fn content(mut self, content: MessageContent) -> Self {
344        self.content = Some(content);
345        self
346    }
347
348    /// Set the correlation ID.
349    pub fn correlation_id(mut self, id: impl Into<String>) -> Self {
350        self.correlation_id = Some(id.into());
351        self
352    }
353
354    /// Set the topic.
355    pub fn topic(mut self, topic: impl Into<String>) -> Self {
356        self.topic = Some(topic.into());
357        self
358    }
359
360    /// Add metadata.
361    pub fn metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
362        self.metadata.insert(key.into(), value.into());
363        self
364    }
365
366    /// Build the message.
367    ///
368    /// # Panics
369    /// Panics if target or content was not set.
370    pub fn build(self) -> Message {
371        Message {
372            id: self.id,
373            version: self.version,
374            from: self.from,
375            to: self.to.expect("Message target must be set"),
376            content: self.content.expect("Message content must be set"),
377            correlation_id: self.correlation_id,
378            topic: self.topic,
379            metadata: self.metadata,
380            created_at: chrono::Utc::now().timestamp_micros(),
381        }
382    }
383}
384
385// Backward compatibility conversions
386
387impl From<LegacyMessage> for Message {
388    /// Convert from legacy message format.
389    fn from(legacy: LegacyMessage) -> Self {
390        // Try to detect content type from metadata or topic
391        let content = if let Some(content_type) = legacy.metadata.get("content_type") {
392            match content_type.as_str() {
393                "text" => String::from_utf8(legacy.payload.clone())
394                    .map(MessageContent::Text)
395                    .unwrap_or_else(|_| MessageContent::Binary(legacy.payload.clone())),
396                "json" => serde_json::from_slice(&legacy.payload)
397                    .map(MessageContent::Json)
398                    .unwrap_or_else(|_| MessageContent::Binary(legacy.payload.clone())),
399                _ => MessageContent::Binary(legacy.payload.clone()),
400            }
401        } else {
402            // Default: try UTF-8 text, fallback to binary
403            String::from_utf8(legacy.payload.clone())
404                .map(MessageContent::Text)
405                .unwrap_or_else(|_| MessageContent::Binary(legacy.payload.clone()))
406        };
407
408        // Extract target from metadata
409        let to = legacy
410            .metadata
411            .get("target")
412            .map(|t| MessageTarget::Agent(t.clone()))
413            .unwrap_or(MessageTarget::Broadcast);
414
415        Message {
416            id: legacy.id,
417            version: 1,
418            from: legacy.sender,
419            to,
420            content,
421            correlation_id: legacy.metadata.get("correlation_id").cloned(),
422            topic: Some(legacy.topic),
423            metadata: legacy.metadata,
424            created_at: legacy.created_at,
425        }
426    }
427}
428
429impl TryFrom<Message> for LegacyMessage {
430    type Error = crate::error::Error;
431
432    /// Convert to legacy message format (lossy conversion).
433    fn try_from(msg: Message) -> Result<Self, Self::Error> {
434        use crate::error::Error;
435
436        let (topic, payload, content_type) = match msg.content {
437            MessageContent::Text(text) => ("text".to_string(), text.into_bytes(), Some("text")),
438            MessageContent::Json(value) => (
439                "json".to_string(),
440                serde_json::to_vec(&value).map_err(|e| {
441                    Error::SerializationError(format!("Failed to serialize JSON: {}", e))
442                })?,
443                Some("json"),
444            ),
445            MessageContent::Binary(data) => ("binary".to_string(), data, None),
446            _ => {
447                return Err(Error::ConversionError(
448                    "Unsupported content type for legacy format".to_string(),
449                ))
450            }
451        };
452
453        let mut metadata = msg.metadata;
454        if let Some(corr_id) = msg.correlation_id {
455            metadata.insert("correlation_id".to_string(), corr_id);
456        }
457        if let MessageTarget::Agent(agent) = msg.to {
458            metadata.insert("target".to_string(), agent);
459        }
460        if let Some(ct) = content_type {
461            metadata.insert("content_type".to_string(), ct.to_string());
462        }
463
464        Ok(LegacyMessage {
465            id: msg.id,
466            topic: msg.topic.unwrap_or(topic),
467            payload,
468            sender: msg.from,
469            metadata,
470            created_at: msg.created_at,
471        })
472    }
473}
474
475/// A simple string-based message for text communication.
476///
477/// Use this for simpler text-based interactions instead of binary [`Message`].
478#[derive(Debug, Clone, Serialize, Deserialize)]
479pub struct GenericMessage {
480    /// The message content.
481    pub content: String,
482    /// Optional metadata.
483    pub metadata: std::collections::HashMap<String, String>,
484}
485
486impl GenericMessage {
487    /// Create a new generic message.
488    pub fn new(content: impl Into<String>) -> Self {
489        Self {
490            content: content.into(),
491            metadata: std::collections::HashMap::new(),
492        }
493    }
494}
495
496/// Response to a [`GenericMessage`].
497#[derive(Debug, Clone, Serialize, Deserialize)]
498pub struct GenericResponse {
499    /// The response content.
500    pub content: String,
501}
502
503impl GenericResponse {
504    /// Create a new generic response.
505    pub fn new(content: impl Into<String>) -> Self {
506        Self {
507            content: content.into(),
508        }
509    }
510}
511
512/// Envelope wrapping a message with routing information.
513///
514/// Used internally for message delivery across mesh boundaries.
515#[derive(Debug, Clone, Serialize, Deserialize)]
516pub struct Envelope {
517    /// The wrapped message.
518    pub message: Message,
519    /// Target agent name (within a node).
520    pub target_agent: Option<String>,
521    /// Target node ID (for distributed meshes).
522    pub target_node: Option<String>,
523}
enki_core/message.rs

enki_core/
message.rs
