Change ma_send to require message type

Author: bahner

src/entity.rs

@@ -471,14 +471,21 @@ pub struct RuntimeManifest {
 
 /// Outbound message queued by a plugin via the `ma_send` host function.
 ///
-/// `to`       — recipient DID (or DID-URL).
-/// `reply_to` — if set, marks this as a reply to the given message ID; the
-///              runtime will use `MESSAGE_TYPE_RPC_REPLY` for the wire message.
-/// `content`  — raw payload bytes (e.g. CBOR-encoded RPC atom or JSON).
+/// `to`           — recipient DID (or DID-URL).
+/// `content_type` — MIME type of the payload (e.g. `application/cbor`).
+/// `message_type` — envelope routing type (e.g. `application/x-ma-chat`).  If
+///                  `None` the runtime defaults to `MESSAGE_TYPE_RPC`.  The
+///                  protocol used for delivery is derived from this field; see
+///                  `eventloop::protocol_for`.
+/// `reply_to`     — if set, marks this as a reply; overrides `message_type` to
+///                  `MESSAGE_TYPE_RPC_REPLY` and routes via `/ma/rpc/0.0.1`.
+/// `content`      — raw payload bytes.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct SendEnvelope {
     pub to: String,
     pub content_type: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub message_type: Option<String>,
     #[serde(with = "serde_bytes")]
     pub content: Vec<u8>,
     pub reply_to: Option<String>,

src/eventloop.rs

@@ -10,8 +10,9 @@ use std::time::Duration;
 use anyhow::Result;
 use ma_core::config::Config;
 use ma_core::{
-    Did, Inbox, IpfsGatewayResolver, MaEndpoint, Message, SigningKey, IPFS_PROTOCOL_ID,
-    MESSAGE_TYPE_RPC, MESSAGE_TYPE_RPC_REPLY,
+    Did, Inbox, IpfsGatewayResolver, MaEndpoint, Message, SigningKey, INBOX_PROTOCOL_ID,
+    IPFS_PROTOCOL_ID, MESSAGE_TYPE_CRUD, MESSAGE_TYPE_CRUD_REPLY, MESSAGE_TYPE_IPFS_REQUEST,
+    MESSAGE_TYPE_IPFS_STORE, MESSAGE_TYPE_RPC, MESSAGE_TYPE_RPC_REPLY,
 };
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use tokio::sync::RwLock;
@@ -24,12 +25,27 @@ use crate::ipfs::IpfsServiceState;
 use crate::manifest::ManifestWriter;
 use crate::plugin::EntityRegistry;
 use crate::status::SharedStats;
-use crate::{bootstrap, crud, i18n, ipfs, rpc, status};
+use crate::{bootstrap, crud, i18n, inbox, ipfs, rpc, status};
+
+/// Map a `message_type` string to the iroh delivery protocol.
+///
+/// Only RPC and its reply go to `/ma/rpc/0.0.1`; IPFS requests go to
+/// `/ma/ipfs/0.0.1`; CRUD goes to `/ma/crud/0.0.1`.  Everything else
+/// (message, broadcast, chat, emote, unknown) falls back to `/ma/inbox/0.0.1`.
+fn protocol_for(msg_type: &str) -> &'static str {
+    match msg_type {
+        MESSAGE_TYPE_RPC | MESSAGE_TYPE_RPC_REPLY => rpc::RPC_PROTOCOL_ID,
+        MESSAGE_TYPE_IPFS_REQUEST | MESSAGE_TYPE_IPFS_STORE => IPFS_PROTOCOL_ID,
+        MESSAGE_TYPE_CRUD | MESSAGE_TYPE_CRUD_REPLY => crud::CRUD_PROTOCOL_ID,
+        _ => INBOX_PROTOCOL_ID,
+    }
+}
 
 #[allow(clippy::too_many_arguments, clippy::too_many_lines)]
 pub async fn run(
     mut endpoint: Arc<dyn MaEndpoint>,
     rpc_messages: Inbox<Message>,
+    inbox_messages: Inbox<Message>,
     mut crud_messages: Option<Inbox<Message>>,
     mut ipfs_state: Option<IpfsServiceState>,
     envelope_tx: UnboundedSender<(String, SendEnvelope)>,
@@ -202,12 +218,40 @@ pub async fn run(
                     }
                 }
 
+                // Drain /ma/inbox/0.0.1
+                while let Some(mut message) = inbox_messages.pop(now) {
+                    debug!(
+                        from = %message.from,
+                        to = %message.to,
+                        message_type = %message.message_type,
+                        "{}", i18n::t("inbox-message-received")
+                    );
+                    let ctx = inbox::InboxHandlerCtx {
+                        our_did: Arc::from(our_did.as_str()),
+                        entity_registry: entity_registry.clone(),
+                        kubo_rpc_url: Arc::from(kubo_url.as_str()),
+                    };
+                    tokio::spawn(async move {
+                        if let Err(err) = tokio::time::timeout(
+                            Duration::from_secs(30),
+                            inbox::handle_inbox_message(&message, &ctx),
+                        )
+                        .await
+                        .unwrap_or_else(|_| Err(anyhow::anyhow!("inbox handler timed out")))
+                        {
+                            warn!(error = %err, from = %message.from, "inbox message rejected");
+                        }
+                        message.content.zeroize();
+                        message.signature.zeroize();
+                    });
+                }
+
                 // Drain plugin outbox — envelopes sent fire-and-forget by ma_send/ma_reply.
                 while let Ok((fragment, env)) = envelope_rx.try_recv() {
                     let msg_type = if env.reply_to.is_some() {
                         MESSAGE_TYPE_RPC_REPLY
                     } else {
-                        MESSAGE_TYPE_RPC
+                        env.message_type.as_deref().unwrap_or(MESSAGE_TYPE_RPC)
                     };
                     let sender_did_url = format!("{our_did}#{fragment}");
                     let recipient = match Did::try_from(env.to.as_str()) {
@@ -232,6 +276,7 @@ pub async fn run(
                         }
                     };
                     msg.reply_to = env.reply_to;
+                    let protocol = protocol_for(msg_type);
                     // Spawn each delivery independently so one unreachable peer
                     // cannot block others. Cap the outbox-open at 5 seconds.
                     let ep   = Arc::clone(&endpoint);
@@ -240,7 +285,7 @@ pub async fn run(
                     tokio::spawn(async move {
                         match tokio::time::timeout(
                             Duration::from_secs(5),
-                            ep.outbox(res.as_ref(), &base, rpc::RPC_PROTOCOL_ID),
+                            ep.outbox(res.as_ref(), &base, protocol),
                         )
                         .await
                         {

src/inbox.rs

@@ -0,0 +1,93 @@
+//! Handler for the `/ma/inbox/0.0.1` service.
+//!
+//! Inbox messages are fire-and-forget: fragment-addressed messages are routed
+//! to the target entity's `handle_cast` (stateless) or `handle_call` (stateful,
+//! to allow state persistence), and the result is discarded.  No reply is sent.
+//! Unfragmented (broadcast / runtime-level) messages are logged and dropped.
+
+use std::sync::Arc;
+
+use anyhow::Result;
+use tracing::{info, warn};
+
+use crate::entity::{CastInput, LocalMessage, PluginKind, PluginMsg};
+use crate::plugin::EntityRegistry;
+
+pub struct InboxHandlerCtx {
+    pub our_did: Arc<str>,
+    pub entity_registry: EntityRegistry,
+    pub kubo_rpc_url: Arc<str>,
+}
+
+pub async fn handle_inbox_message(message: &ma_core::Message, ctx: &InboxHandlerCtx) -> Result<()> {
+    let fragment = extract_fragment(&message.to, &ctx.our_did);
+
+    let Some(fragment) = fragment else {
+        info!(
+            from = %message.from,
+            to = %message.to,
+            message_type = %message.message_type,
+            "inbox: unfragmented message dropped"
+        );
+        return Ok(());
+    };
+
+    let entity = ctx.entity_registry.read().await.get(fragment).cloned();
+
+    let Some(entity) = entity else {
+        warn!(fragment = %fragment, "inbox: unknown entity fragment; dropped");
+        return Ok(());
+    };
+
+    info!(
+        fragment = %fragment,
+        from = %message.from,
+        message_type = %message.message_type,
+        "inbox: dispatching to entity"
+    );
+
+    let local_msg = LocalMessage {
+        id: message.id.clone(),
+        from: message.from.clone(),
+        to: message.to.clone(),
+        created_at: message.created_at,
+        expires: message.exp,
+        reply_to: message.reply_to.clone(),
+        content_type: message.content_type.clone(),
+        content: message.content.clone(),
+    };
+    let cast_input = CastInput {
+        msg: PluginMsg::from(&local_msg),
+    };
+
+    let result = match entity.kind {
+        PluginKind::Stateless => entity.handle_cast(&cast_input).await?,
+        PluginKind::Stateful => entity.handle_call(&cast_input).await?,
+    };
+
+    // Persist state if the entity called ma_set_state during this dispatch.
+    if let Some(state_bytes) = result.pending_state {
+        let kubo_url = Arc::clone(&ctx.kubo_rpc_url);
+        let fragment_str = entity.fragment.clone();
+        let entity_arc = Arc::clone(&entity);
+        tokio::spawn(async move {
+            match crate::kubo::dag_put(&kubo_url, &state_bytes).await {
+                Ok(cid) => {
+                    entity_arc.mark_saved(state_bytes);
+                    info!(fragment = %fragment_str, cid = %cid, "inbox: entity state persisted");
+                }
+                Err(e) => {
+                    warn!(fragment = %fragment_str, error = %e, "inbox: failed to persist entity state");
+                }
+            }
+        });
+    }
+
+    Ok(())
+}
+
+/// Strip `<our_did>#` from `to` and return the bare fragment, if present.
+fn extract_fragment<'a>(to: &'a str, our_did: &str) -> Option<&'a str> {
+    let prefix = format!("{our_did}#");
+    to.strip_prefix(prefix.as_str())
+}

src/main.rs

@@ -4,6 +4,7 @@ mod crud;
 mod entity;
 mod eventloop;
 mod i18n;
+mod inbox;
 mod ipfs;
 mod kubo;
 mod manifest;
@@ -23,7 +24,7 @@ use cid::Cid;
 use clap::Parser;
 use ma_core::config::{Config, MaArgs};
 use ma_core::ipfs::IpfsDidPublisher;
-use ma_core::{ipns_from_secret, Ipld, IPFS_PROTOCOL_ID};
+use ma_core::{ipns_from_secret, Ipld, INBOX_PROTOCOL_ID, IPFS_PROTOCOL_ID};
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -224,6 +225,7 @@ async fn main() -> Result<()> {
     }
     let mut endpoint = ma_core::new_ma_endpoint(secrets.iroh_secret_key, ipv6_enabled).await?;
     let rpc_messages = endpoint.service(rpc::RPC_PROTOCOL_ID);
+    let inbox_messages = endpoint.service(INBOX_PROTOCOL_ID);
     let ipfs_messages = if ipfs_publisher_enabled {
         Some(endpoint.service(IPFS_PROTOCOL_ID))
     } else {
@@ -563,6 +565,7 @@ async fn main() -> Result<()> {
     eventloop::run(
         endpoint,
         rpc_messages,
+        inbox_messages,
         crud_messages,
         ipfs_state,
         envelope_tx,

src/plugin/backend.rs

@@ -91,6 +91,7 @@ host_fn!(ma_reply_fn(user_data: OutboxCtx; input: Vec<u8>) -> Vec<u8> {
     let envelope = SendEnvelope {
         to: req.msg.from,
         content_type: req.content_type,
+        message_type: None,
         content: req.content,
         reply_to: Some(req.msg.id),
     };
@@ -211,8 +212,8 @@ host_fn!(ma_avatar_id_fn(user_data: AvatarIdCtx; input: Vec<u8>) -> Vec<u8> {
 
 // ── ma_call host function ─────────────────────────────────────────────────────────────────────────
 
-/// Hard cap on any single Wasm export invocation (init / handle_cast /
-/// handle_call).  Enforced by extism via wasmtime epoch interruption — a
+/// Hard cap on any single Wasm export invocation (init / `handle_cast` /
+/// `handle_call`).  Enforced by extism via wasmtime epoch interruption — a
 /// plugin stuck in an infinite loop gets aborted and the worker thread
 /// survives, instead of being wedged forever.
 pub(super) const WASM_CALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

src/plugin/mod.rs

@@ -381,7 +381,7 @@ impl EntityPlugin {
 mod wasm_repro {
     //! Local reproduction harness for plugin WASM crashes.
     //! Requires a local Kubo node with the plugin CID pinned.
-    //! Run: cargo test wasm_repro -- --ignored --nocapture
+    //! Run: cargo test `wasm_repro` -- --ignored --nocapture
 
     use std::collections::BTreeMap;
 

src/rpc.rs

@@ -332,6 +332,7 @@ async fn handle_entity_plugin_message(
                     crate::entity::SendEnvelope {
                         to: format!("{}#{}", ctx.our_did, req.parent),
                         content_type: ma_core::CONTENT_TYPE_TERM.to_string(),
+                        message_type: None,
                         content: err_content,
                         reply_to: None,
                     },