use crate::inproc::InprocRegistry;
use crate::{InboxItem, MessageKind, PubKey, TrustedPeers};
use meerkat_core::PlainEventSource;
use meerkat_core::types::{ContentBlock, HandlingMode, RenderMetadata};
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
use std::borrow::Cow;
use uuid::Uuid;
#[derive(Debug, Clone, PartialEq)]
pub struct PlainMessage {
pub body: String,
pub source: PlainEventSource,
pub interaction_id: Option<uuid::Uuid>,
pub handling_mode: HandlingMode,
pub blocks: Option<Vec<ContentBlock>>,
pub render_metadata: Option<RenderMetadata>,
}
impl PlainMessage {
pub fn to_user_message_text(&self) -> String {
meerkat_core::interaction::format_external_event_projection(
&self.source.to_string(),
Some(&self.body),
)
}
}
#[derive(Debug, Clone)]
pub enum DrainedMessage {
Authenticated(CommsMessage),
Plain(PlainMessage),
}
pub fn drain_inbox_item(
item: &InboxItem,
trusted_peers: &TrustedPeers,
require_peer_auth: bool,
) -> Option<DrainedMessage> {
match item {
InboxItem::External { .. } => {
CommsMessage::from_inbox_item(item, trusted_peers, require_peer_auth)
.map(DrainedMessage::Authenticated)
}
InboxItem::PlainEvent {
body,
source,
handling_mode,
interaction_id,
blocks,
render_metadata,
} => Some(DrainedMessage::Plain(PlainMessage {
body: body.clone(),
source: *source,
handling_mode: *handling_mode,
interaction_id: *interaction_id,
blocks: blocks.clone(),
render_metadata: render_metadata.clone(),
})),
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MessageIntent {
Delegate,
Status,
Cancel,
Ack,
Review,
Calculate,
Query,
#[serde(rename = "mob.peer_added")]
PeerAdded,
#[serde(rename = "mob.peer_retired")]
PeerRetired,
#[serde(untagged)]
Custom(String),
}
impl MessageIntent {
pub fn custom(s: impl Into<String>) -> Self {
Self::Custom(s.into())
}
pub fn as_str(&self) -> &str {
match self {
Self::Delegate => "delegate",
Self::Status => "status",
Self::Cancel => "cancel",
Self::Ack => "ack",
Self::Review => "review",
Self::Calculate => "calculate",
Self::Query => "query",
Self::PeerAdded => "mob.peer_added",
Self::PeerRetired => "mob.peer_retired",
Self::Custom(s) => s.as_str(),
}
}
}
impl From<String> for MessageIntent {
fn from(s: String) -> Self {
match s.as_str() {
"delegate" => Self::Delegate,
"status" => Self::Status,
"cancel" => Self::Cancel,
"ack" => Self::Ack,
"review" => Self::Review,
"calculate" => Self::Calculate,
"query" => Self::Query,
"mob.peer_added" => Self::PeerAdded,
"mob.peer_retired" => Self::PeerRetired,
_ => Self::Custom(s),
}
}
}
impl From<&str> for MessageIntent {
fn from(s: &str) -> Self {
Self::from(s.to_string())
}
}
impl std::fmt::Display for MessageIntent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.as_str())
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum CommsContent {
Message {
body: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
blocks: Option<Vec<ContentBlock>>,
},
Request {
request_id: Uuid,
intent: MessageIntent,
params: JsonValue,
#[serde(default, skip_serializing_if = "Option::is_none")]
blocks: Option<Vec<ContentBlock>>,
},
Response {
in_reply_to: Uuid,
status: CommsStatus,
result: JsonValue,
#[serde(default, skip_serializing_if = "Option::is_none")]
blocks: Option<Vec<ContentBlock>>,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum CommsStatus {
Accepted,
Completed,
Failed,
}
impl From<crate::Status> for CommsStatus {
fn from(s: crate::Status) -> Self {
match s {
crate::Status::Accepted => CommsStatus::Accepted,
crate::Status::Completed => CommsStatus::Completed,
crate::Status::Failed => CommsStatus::Failed,
}
}
}
impl From<CommsStatus> for crate::Status {
fn from(s: CommsStatus) -> Self {
match s {
CommsStatus::Accepted => crate::Status::Accepted,
CommsStatus::Completed => crate::Status::Completed,
CommsStatus::Failed => crate::Status::Failed,
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CommsMessage {
pub envelope_id: Uuid,
pub from_peer: String,
pub from_pubkey: PubKey,
pub content: CommsContent,
}
impl CommsMessage {
pub(crate) fn from_external_with_resolved_peer(
envelope: &crate::Envelope,
from_peer: String,
) -> Option<Self> {
let content = match &envelope.kind {
MessageKind::Message { body, blocks, .. } => CommsContent::Message {
body: body.clone(),
blocks: blocks.clone(),
},
MessageKind::Request {
intent,
params,
blocks,
..
} => CommsContent::Request {
request_id: envelope.id,
intent: MessageIntent::from(intent.as_str()),
params: params.clone(),
blocks: blocks.clone(),
},
MessageKind::Lifecycle { .. } => return None,
MessageKind::Response {
in_reply_to,
status,
result,
blocks,
..
} => CommsContent::Response {
in_reply_to: *in_reply_to,
status: (*status).into(),
result: result.clone(),
blocks: blocks.clone(),
},
MessageKind::Ack { .. } => return None,
};
Some(CommsMessage {
envelope_id: envelope.id,
from_peer,
from_pubkey: envelope.from,
content,
})
}
pub fn from_inbox_item(
item: &InboxItem,
trusted_peers: &TrustedPeers,
require_peer_auth: bool,
) -> Option<Self> {
let envelope = match item {
InboxItem::External { envelope } => envelope,
InboxItem::PlainEvent { .. } => return None,
};
let from_peer = match trusted_peers
.get_peer(&envelope.from)
.map(|peer| peer.name.clone())
{
Some(name) => name,
None if require_peer_auth => return None,
None => InprocRegistry::global()
.get_name_by_pubkey(&envelope.from)
.unwrap_or_else(|| envelope.from.to_pubkey_string()),
};
Self::from_external_with_resolved_peer(envelope, from_peer)
}
pub(crate) fn from_classified_entry(
entry: &crate::inbox::ClassifiedInboxEntry,
) -> Option<Self> {
let envelope = match &entry.item {
InboxItem::External { envelope } => envelope,
InboxItem::PlainEvent { .. } => return None,
};
let from_peer = entry.from_peer.clone().unwrap_or_else(|| {
InprocRegistry::global()
.get_name_by_pubkey(&envelope.from)
.unwrap_or_else(|| envelope.from.to_pubkey_string())
});
Self::from_external_with_resolved_peer(envelope, from_peer)
}
pub fn to_user_message_text(&self) -> String {
match &self.content {
CommsContent::Message { body, blocks } => {
let text = match blocks {
Some(b) if !b.is_empty() => meerkat_core::types::text_content(b),
_ => body.clone(),
};
meerkat_core::format_peer_message_projection(&self.from_peer, &text)
}
CommsContent::Request {
request_id,
intent,
params,
..
} => meerkat_core::format_peer_request_projection(
self.from_pubkey.to_peer_id(),
Some(&self.from_peer),
request_id,
intent.as_str(),
params,
),
CommsContent::Response {
in_reply_to,
status,
result,
blocks,
} => {
meerkat_core::format_peer_response_projection(
&self.from_peer,
in_reply_to,
match status {
CommsStatus::Accepted => meerkat_core::ResponseStatus::Accepted,
CommsStatus::Completed => meerkat_core::ResponseStatus::Completed,
CommsStatus::Failed => meerkat_core::ResponseStatus::Failed,
},
result,
) + &blocks
.as_ref()
.filter(|blocks| !blocks.is_empty())
.map(|blocks| format!("\n{}", meerkat_core::types::text_content(blocks)))
.unwrap_or_default()
}
}
}
}
impl meerkat_core::TurnBoundaryMessage for CommsMessage {
fn render_for_prompt(&self) -> Cow<'_, str> {
Cow::Owned(self.to_user_message_text())
}
}
#[cfg(test)]
impl CommsMessage {
pub fn from_envelope(
envelope: &crate::Envelope,
trusted_peers: &TrustedPeers,
require_peer_auth: bool,
) -> Option<Self> {
Self::from_inbox_item(
&InboxItem::External {
envelope: envelope.clone(),
},
trusted_peers,
require_peer_auth,
)
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
use crate::{Envelope, Keypair, Signature, TrustedPeer};
fn make_keypair() -> Keypair {
Keypair::generate()
}
fn make_trusted_peers(name: &str, pubkey: &PubKey) -> TrustedPeers {
TrustedPeers {
peers: vec![TrustedPeer {
name: name.to_string(),
pubkey: *pubkey,
addr: "tcp://127.0.0.1:4200".to_string(),
meta: crate::PeerMeta::default(),
}],
}
}
fn make_envelope(from: &Keypair, to: PubKey, kind: MessageKind) -> Envelope {
let mut envelope = Envelope {
id: Uuid::new_v4(),
from: from.public_key(),
to,
kind,
sig: Signature::new([0u8; 64]),
};
envelope.sign(from);
envelope
}
#[test]
fn test_comms_message_struct() {
let msg = CommsMessage {
envelope_id: Uuid::new_v4(),
from_peer: "test-peer".to_string(),
from_pubkey: PubKey::new([1u8; 32]),
content: CommsContent::Message {
body: "hello".to_string(),
blocks: None,
},
};
assert_eq!(msg.from_peer, "test-peer");
}
#[test]
fn test_comms_content_variants() {
let _ = CommsContent::Message {
body: "hello".to_string(),
blocks: None,
};
let _ = CommsContent::Request {
request_id: Uuid::new_v4(),
intent: MessageIntent::Review,
params: serde_json::json!({"pr": 42}),
blocks: None,
};
let _ = CommsContent::Response {
in_reply_to: Uuid::new_v4(),
status: CommsStatus::Completed,
result: serde_json::json!({"approved": true}),
blocks: None,
};
}
#[test]
fn test_comms_message_from_inbox_item() {
let sender = make_keypair();
let receiver = make_keypair();
let trusted = make_trusted_peers("sender-agent", &sender.public_key());
let envelope = make_envelope(
&sender,
receiver.public_key(),
MessageKind::Message {
body: "hello world".to_string(),
blocks: None,
handling_mode: None,
},
);
let item = InboxItem::External { envelope };
let msg = CommsMessage::from_inbox_item(&item, &trusted, true);
assert!(msg.is_some());
let msg = msg.unwrap();
assert_eq!(msg.from_peer, "sender-agent");
match msg.content {
CommsContent::Message { body, blocks } => {
assert_eq!(body, "hello world");
assert_eq!(blocks, None);
}
_ => unreachable!("expected Message"),
}
}
#[test]
fn test_comms_message_from_inbox_item_request() {
let sender = make_keypair();
let receiver = make_keypair();
let trusted = make_trusted_peers("requester", &sender.public_key());
let envelope = make_envelope(
&sender,
receiver.public_key(),
MessageKind::Request {
intent: "review".to_string(),
params: serde_json::json!({"pr": 123}),
blocks: None,
handling_mode: None,
},
);
let request_id = envelope.id;
let item = InboxItem::External { envelope };
let msg = CommsMessage::from_inbox_item(&item, &trusted, true).unwrap();
match msg.content {
CommsContent::Request {
request_id: rid,
intent,
params,
blocks,
} => {
assert_eq!(rid, request_id);
assert_eq!(intent, MessageIntent::Review);
assert_eq!(params["pr"], 123);
assert_eq!(blocks, None);
}
_ => unreachable!("expected Request"),
}
}
#[test]
fn test_comms_message_from_inbox_item_request_custom_intent() {
let sender = make_keypair();
let receiver = make_keypair();
let trusted = make_trusted_peers("requester", &sender.public_key());
let envelope = make_envelope(
&sender,
receiver.public_key(),
MessageKind::Request {
intent: "review-pr".to_string(),
params: serde_json::json!({"pr": 456}),
blocks: None,
handling_mode: None,
},
);
let item = InboxItem::External { envelope };
let msg = CommsMessage::from_inbox_item(&item, &trusted, true).unwrap();
match msg.content {
CommsContent::Request { intent, params, .. } => {
assert_eq!(intent, MessageIntent::Custom("review-pr".to_string()));
assert_eq!(params["pr"], 456);
}
_ => unreachable!("expected Request"),
}
}
#[test]
fn test_comms_message_from_inbox_item_response() {
let sender = make_keypair();
let receiver = make_keypair();
let trusted = make_trusted_peers("responder", &sender.public_key());
let orig_request_id = Uuid::new_v4();
let envelope = make_envelope(
&sender,
receiver.public_key(),
MessageKind::Response {
in_reply_to: orig_request_id,
status: crate::Status::Completed,
result: serde_json::json!({"approved": true}),
blocks: None,
handling_mode: None,
},
);
let item = InboxItem::External { envelope };
let msg = CommsMessage::from_inbox_item(&item, &trusted, true).unwrap();
match msg.content {
CommsContent::Response {
in_reply_to,
status,
result,
blocks: _,
} => {
assert_eq!(in_reply_to, orig_request_id);
assert_eq!(status, CommsStatus::Completed);
assert_eq!(result["approved"], true);
}
_ => unreachable!("expected Response"),
}
}
#[test]
fn test_comms_message_from_inbox_item_ignores_ack() {
let sender = make_keypair();
let receiver = make_keypair();
let trusted = make_trusted_peers("sender", &sender.public_key());
let envelope = make_envelope(
&sender,
receiver.public_key(),
MessageKind::Ack {
in_reply_to: Uuid::new_v4(),
},
);
let item = InboxItem::External { envelope };
let msg = CommsMessage::from_inbox_item(&item, &trusted, true);
assert!(
msg.is_none(),
"Acks should not be converted to CommsMessage"
);
}
#[test]
fn test_comms_message_from_inbox_item_ignores_unknown_peer() {
let sender = make_keypair();
let other = make_keypair();
let receiver = make_keypair();
let trusted = make_trusted_peers("other", &other.public_key());
let envelope = make_envelope(
&sender,
receiver.public_key(),
MessageKind::Message {
body: "hello".to_string(),
blocks: None,
handling_mode: None,
},
);
let item = InboxItem::External { envelope };
let msg = CommsMessage::from_inbox_item(&item, &trusted, true);
assert!(msg.is_none(), "Unknown peers should return None");
}
#[test]
fn test_comms_message_from_inbox_item_uses_inproc_name_without_auth() {
let sender = make_keypair();
let receiver = make_keypair();
let trusted = TrustedPeers::new();
let sender_name = format!("sender-no-auth-{name}", name = Uuid::new_v4().simple());
let (_, sender_inbox_sender) = crate::Inbox::new();
InprocRegistry::global().register(
sender_name.clone(),
sender.public_key(),
sender_inbox_sender,
);
let envelope = make_envelope(
&sender,
receiver.public_key(),
MessageKind::Message {
body: "hello".to_string(),
blocks: None,
handling_mode: None,
},
);
let item = InboxItem::External { envelope };
let msg = CommsMessage::from_inbox_item(&item, &trusted, false).unwrap();
assert_eq!(msg.from_peer, sender_name);
match msg.content {
CommsContent::Message { body, .. } => assert_eq!(body, "hello"),
_ => unreachable!("expected Message"),
}
assert!(InprocRegistry::global().unregister(&sender.public_key()));
}
#[test]
fn test_comms_message_formatting() {
let msg = CommsMessage {
envelope_id: Uuid::new_v4(),
from_peer: "review-agent".to_string(),
from_pubkey: PubKey::new([1u8; 32]),
content: CommsContent::Message {
body: "Please review PR #42".to_string(),
blocks: None,
},
};
let text = msg.to_user_message_text();
assert!(text.contains("COMMS MESSAGE"));
assert!(text.contains("review-agent"));
assert!(text.contains("Please review PR #42"));
}
#[test]
fn test_comms_message_formatting_request() {
let request_id = Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap();
let msg = CommsMessage {
envelope_id: request_id,
from_peer: "coding-agent".to_string(),
from_pubkey: PubKey::new([1u8; 32]),
content: CommsContent::Request {
request_id,
intent: MessageIntent::Review,
params: serde_json::json!({"pr": 42}),
blocks: None,
},
};
let text = msg.to_user_message_text();
assert!(text.contains("COMMS REQUEST"));
assert!(text.contains("coding-agent"));
assert!(text.contains("review"));
assert!(text.contains("550e8400-e29b-41d4-a716-446655440000"));
assert!(text.contains("send_response"));
assert!(text.contains(&format!(
"\"peer_id\":\"{}\"",
PubKey::new([1u8; 32]).to_peer_id()
)));
assert!(text.contains("\"display_name\":\"coding-agent\""));
assert!(text.contains("\"in_reply_to\":\"550e8400-e29b-41d4-a716-446655440000\""));
assert!(text.contains("\"status\":\"completed\""));
assert!(text.contains("Do not answer this request with send_message"));
}
#[test]
fn test_comms_message_formatting_response() {
let in_reply_to = Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap();
let msg = CommsMessage {
envelope_id: Uuid::new_v4(),
from_peer: "review-agent".to_string(),
from_pubkey: PubKey::new([1u8; 32]),
content: CommsContent::Response {
in_reply_to,
status: CommsStatus::Completed,
result: serde_json::json!({"approved": true}),
blocks: None,
},
};
let text = msg.to_user_message_text();
assert!(text.contains("COMMS RESPONSE"));
assert!(text.contains("review-agent"));
assert!(text.contains("completed"));
assert!(text.contains("550e8400-e29b-41d4-a716-446655440000"));
}
#[test]
fn test_message_intent_standard_variants() {
assert_eq!(MessageIntent::Delegate.as_str(), "delegate");
assert_eq!(MessageIntent::Status.as_str(), "status");
assert_eq!(MessageIntent::Cancel.as_str(), "cancel");
assert_eq!(MessageIntent::Ack.as_str(), "ack");
assert_eq!(MessageIntent::Review.as_str(), "review");
assert_eq!(MessageIntent::Calculate.as_str(), "calculate");
assert_eq!(MessageIntent::Query.as_str(), "query");
}
#[test]
fn test_message_intent_custom_variant() {
let custom = MessageIntent::custom("my-custom-intent");
assert_eq!(custom.as_str(), "my-custom-intent");
assert_eq!(
custom,
MessageIntent::Custom("my-custom-intent".to_string())
);
}
#[test]
fn test_message_intent_from_string() {
assert_eq!(MessageIntent::from("delegate"), MessageIntent::Delegate);
assert_eq!(MessageIntent::from("status"), MessageIntent::Status);
assert_eq!(MessageIntent::from("cancel"), MessageIntent::Cancel);
assert_eq!(MessageIntent::from("ack"), MessageIntent::Ack);
assert_eq!(MessageIntent::from("review"), MessageIntent::Review);
assert_eq!(MessageIntent::from("calculate"), MessageIntent::Calculate);
assert_eq!(MessageIntent::from("query"), MessageIntent::Query);
assert_eq!(
MessageIntent::from("unknown-intent"),
MessageIntent::Custom("unknown-intent".to_string())
);
}
#[test]
fn test_message_intent_display() {
assert_eq!(format!("{}", MessageIntent::Review), "review");
assert_eq!(
format!("{}", MessageIntent::Custom("foo".to_string())),
"foo"
);
}
#[test]
fn test_message_intent_serialization() {
let json = serde_json::to_string(&MessageIntent::Delegate).unwrap();
assert_eq!(json, "\"delegate\"");
let json = serde_json::to_string(&MessageIntent::Review).unwrap();
assert_eq!(json, "\"review\"");
let json = serde_json::to_string(&MessageIntent::Custom("custom-op".to_string())).unwrap();
assert_eq!(json, "\"custom-op\"");
}
#[test]
fn test_message_intent_deserialization() {
let intent: MessageIntent = serde_json::from_str("\"delegate\"").unwrap();
assert_eq!(intent, MessageIntent::Delegate);
let intent: MessageIntent = serde_json::from_str("\"review\"").unwrap();
assert_eq!(intent, MessageIntent::Review);
let intent: MessageIntent = serde_json::from_str("\"my-custom\"").unwrap();
assert_eq!(intent, MessageIntent::Custom("my-custom".to_string()));
}
#[test]
fn test_drained_message_from_external() {
let sender = make_keypair();
let receiver = make_keypair();
let trusted = make_trusted_peers("sender-agent", &sender.public_key());
let envelope = make_envelope(
&sender,
receiver.public_key(),
MessageKind::Message {
body: "hello".to_string(),
blocks: None,
handling_mode: None,
},
);
let item = InboxItem::External { envelope };
let drained = drain_inbox_item(&item, &trusted, true);
assert!(drained.is_some());
match drained.unwrap() {
DrainedMessage::Authenticated(msg) => {
assert_eq!(msg.from_peer, "sender-agent");
}
DrainedMessage::Plain(_) => panic!("Expected Authenticated"),
}
}
#[test]
fn test_drained_message_from_plain_event() {
use meerkat_core::PlainEventSource;
let trusted = TrustedPeers::new();
let item = InboxItem::PlainEvent {
body: "New email arrived".to_string(),
source: PlainEventSource::Tcp,
handling_mode: HandlingMode::Queue,
interaction_id: None,
blocks: None,
render_metadata: None,
};
let drained = drain_inbox_item(&item, &trusted, true);
assert!(drained.is_some());
match drained.unwrap() {
DrainedMessage::Plain(msg) => {
assert_eq!(msg.body, "New email arrived");
assert_eq!(msg.source, PlainEventSource::Tcp);
assert_eq!(msg.interaction_id, None);
}
DrainedMessage::Authenticated(_) => panic!("Expected Plain"),
}
}
#[test]
fn test_plain_message_to_user_message_text() {
use meerkat_core::PlainEventSource;
let msg = PlainMessage {
body: "CPU > 95% on prod-3".to_string(),
source: PlainEventSource::Webhook,
handling_mode: HandlingMode::Queue,
interaction_id: None,
blocks: None,
render_metadata: None,
};
let text = msg.to_user_message_text();
assert_eq!(text, "[EVENT via webhook] CPU > 95% on prod-3");
}
#[test]
fn test_plain_message_formatting_all_sources() {
use meerkat_core::PlainEventSource;
for (source, label) in [
(PlainEventSource::Tcp, "tcp"),
(PlainEventSource::Uds, "uds"),
(PlainEventSource::Stdin, "stdin"),
(PlainEventSource::Webhook, "webhook"),
(PlainEventSource::Rpc, "rpc"),
] {
let msg = PlainMessage {
body: "test".to_string(),
source,
handling_mode: HandlingMode::Queue,
interaction_id: None,
blocks: None,
render_metadata: None,
};
assert!(
msg.to_user_message_text().contains(label),
"PlainMessage should contain source label '{label}'"
);
}
}
}