Separate mycelium client more from supervisor client

Signed-off-by: Lee Smet <lee.smet@hotmail.com>

specs/openrpc.json

@@ -438,6 +438,16 @@
           "Processed"
         ]
       },
+      "TransportStatus": {
+        "type": "string",
+        "enum": [
+          "Queued",
+          "Sent",
+          "Delivered",
+          "Read",
+          "Failed"
+        ]
+      },
       "MessageType": {
         "type": "string",
         "enum": [
@@ -779,6 +789,12 @@
           },
           "status": {
             "$ref": "#/components/schemas/MessageStatus"
+          },
+          "transport_id": {
+            "type": "string"
+          },
+          "transport_status": {
+            "$ref": "#/components/schemas/TransportStatus"
           }
         }
       },

src/clients/mod.rs

@@ -1,7 +1,13 @@
 pub mod supervisor_client;
+pub mod mycelium_client;
+pub mod types;
 
+pub use types::Destination;
 pub use supervisor_client::{
-    Destination,
     SupervisorClient,
     SupervisorClientError,
+};
+pub use mycelium_client::{
+    MyceliumClient,
+    MyceliumClientError,
 };

src/clients/mycelium_client.rs

@@ -0,0 +1,208 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use reqwest::Client as HttpClient;
+
+use serde_json::{Value, json};
+use thiserror::Error;
+
+use crate::models::TransportStatus;
+use crate::clients::Destination;
+
+/// Lightweight client for Mycelium JSON-RPC (send + query status)
+#[derive(Clone)]
+pub struct MyceliumClient {
+    base_url: String,       // e.g. http://127.0.0.1:8990
+    http: HttpClient,
+    id_counter: Arc<AtomicU64>,
+}
+
+#[derive(Debug, Error)]
+pub enum MyceliumClientError {
+    #[error("HTTP error: {0}")]
+    Http(#[from] reqwest::Error),
+    #[error("JSON error: {0}")]
+    Json(#[from] serde_json::Error),
+    #[error("Transport timed out waiting for a reply (408)")]
+    TransportTimeout,
+    #[error("JSON-RPC error: {0}")]
+    RpcError(String),
+    #[error("Invalid response: {0}")]
+    InvalidResponse(String),
+}
+
+impl MyceliumClient {
+    pub fn new(base_url: impl Into<String>) -> Result<Self, MyceliumClientError> {
+        let url = base_url.into();
+        let http = HttpClient::builder().build()?;
+        Ok(Self {
+            base_url: url,
+            http,
+            id_counter: Arc::new(AtomicU64::new(1)),
+        })
+    }
+
+    fn next_id(&self) -> u64 {
+        self.id_counter.fetch_add(1, Ordering::Relaxed)
+    }
+
+    async fn jsonrpc(&self, method: &str, params: Value) -> Result<Value, MyceliumClientError> {
+        let req = json!({
+            "jsonrpc": "2.0",
+            "id": self.next_id(),
+            "method": method,
+            "params": [ params ]
+        });
+        let resp = self.http.post(&self.base_url).json(&req).send().await?;
+        let status = resp.status();
+        let body: Value = resp.json().await?;
+        if let Some(err) = body.get("error") {
+            let code = err.get("code").and_then(|v| v.as_i64()).unwrap_or(0);
+            let msg = err.get("message").and_then(|v| v.as_str()).unwrap_or("unknown error");
+            if code == 408 {
+                return Err(MyceliumClientError::TransportTimeout);
+            }
+            return Err(MyceliumClientError::RpcError(format!("code={code} msg={msg}")));
+        }
+        if !status.is_success() {
+            return Err(MyceliumClientError::RpcError(format!("HTTP {status}, body {body}")));
+        }
+        Ok(body)
+    }
+
+    /// Call messageStatus with an outbound message id (hex string)
+    pub async fn message_status(&self, id_hex: &str) -> Result<TransportStatus, MyceliumClientError> {
+        let params = json!({ "id": id_hex });
+        let body = self.jsonrpc("messageStatus", params).await?;
+        let result = body.get("result").ok_or_else(|| {
+            MyceliumClientError::InvalidResponse(format!("missing result in response: {body}"))
+        })?;
+        // Accept both { status: "..."} and bare "..."
+        let status_str = if let Some(s) = result.get("status").and_then(|v| v.as_str()) {
+            s.to_string()
+        } else if let Some(s) = result.as_str() {
+            s.to_string()
+        } else {
+            return Err(MyceliumClientError::InvalidResponse(format!("unexpected result shape: {result}")));
+        };
+        Self::map_status(&status_str).ok_or_else(|| {
+            MyceliumClientError::InvalidResponse(format!("unknown status: {status_str}"))
+        })
+    }
+
+    fn map_status(s: &str) -> Option<TransportStatus> {
+        match s {
+            "queued" => Some(TransportStatus::Queued),
+            "sent" => Some(TransportStatus::Sent),
+            "delivered" => Some(TransportStatus::Delivered),
+            "read" => Some(TransportStatus::Read),
+            "failed" => Some(TransportStatus::Failed),
+            _ => None,
+        }
+    }
+
+    /// Build params object for pushMessage without performing any network call.
+    /// Exposed for serializer-only tests and reuse.
+    pub(crate) fn build_push_params(
+        dst: &Destination,
+        topic: &str,
+        payload_b64: &str,
+        reply_timeout: Option<u64>,
+    ) -> Value {
+        let dst_v = match dst {
+            Destination::Ip(ip) => json!({ "ip": ip.to_string() }),
+            Destination::Pk(pk) => json!({ "pk": pk }),
+        };
+        let message = json!({
+            "dst": dst_v,
+            "topic": topic,
+            "payload": payload_b64,
+        });
+        let mut params = json!({ "message": message });
+        if let Some(rt) = reply_timeout {
+            params["reply_timeout"] = json!(rt);
+        }
+        params
+    }
+
+    /// pushMessage: send a message with dst/topic/payload. Optional reply_timeout for sync replies.
+    pub async fn push_message(
+        &self,
+        dst: &Destination,
+        topic: &str,
+        payload_b64: &str,
+        reply_timeout: Option<u64>,
+    ) -> Result<Value, MyceliumClientError> {
+        let params = Self::build_push_params(dst, topic, payload_b64, reply_timeout);
+        let body = self.jsonrpc("pushMessage", params).await?;
+        let result = body.get("result").ok_or_else(|| {
+            MyceliumClientError::InvalidResponse(format!("missing result in response: {body}"))
+        })?;
+        Ok(result.clone())
+    }
+
+    /// Helper to extract outbound message id from pushMessage result (InboundMessage or PushMessageResponseId)
+    pub fn extract_message_id_from_result(result: &Value) -> Option<String> {
+        result.get("id").and_then(|v| v.as_str()).map(|s| s.to_string())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::clients::Destination;
+
+    #[test]
+    fn build_push_params_shapes_ip_pk_and_timeout() {
+        // IP destination
+        let p1 = MyceliumClient::build_push_params(
+            &Destination::Ip("2001:db8::1".parse().unwrap()),
+            "supervisor.rpc",
+            "Zm9vYmFy", // "foobar"
+            Some(10),
+        );
+        let msg1 = p1.get("message").unwrap();
+        assert_eq!(msg1.get("topic").unwrap().as_str().unwrap(), "supervisor.rpc");
+        assert_eq!(msg1.get("payload").unwrap().as_str().unwrap(), "Zm9vYmFy");
+        assert_eq!(
+            msg1.get("dst").unwrap().get("ip").unwrap().as_str().unwrap(),
+            "2001:db8::1"
+        );
+        assert_eq!(p1.get("reply_timeout").unwrap().as_u64().unwrap(), 10);
+
+        // PK destination without timeout
+        let p2 = MyceliumClient::build_push_params(
+            &Destination::Pk("bb39b4a3a4efd70f3e05e37887677e02efbda14681d0acd3882bc0f754792c32".into()),
+            "supervisor.rpc",
+            "YmF6", // "baz"
+            None,
+        );
+        let msg2 = p2.get("message").unwrap();
+        assert_eq!(
+            msg2.get("dst").unwrap().get("pk").unwrap().as_str().unwrap(),
+            "bb39b4a3a4efd70f3e05e37887677e02efbda14681d0acd3882bc0f754792c32"
+        );
+        assert!(p2.get("reply_timeout").is_none());
+    }
+
+    #[test]
+    fn extract_message_id_variants() {
+        // PushMessageResponseId
+        let r1 = json!({"id":"0123456789abcdef"});
+        assert_eq!(
+            MyceliumClient::extract_message_id_from_result(&r1).unwrap(),
+            "0123456789abcdef"
+        );
+
+        // InboundMessage-like
+        let r2 = json!({
+            "id":"fedcba9876543210",
+            "srcIp":"449:abcd:0123:defa::1",
+            "payload":"hpV+"
+        });
+        assert_eq!(
+            MyceliumClient::extract_message_id_from_result(&r2).unwrap(),
+            "fedcba9876543210"
+        );
+    }
+}

src/clients/supervisor_client.rs

@@ -1,28 +1,20 @@
-use std::net::IpAddr;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 
 use base64::Engine;
 use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
-use reqwest::Client as HttpClient;
 use serde_json::{Value, json};
 use thiserror::Error;
 
-/// Destination for Mycelium messages
-#[derive(Clone, Debug)]
-pub enum Destination {
-    Ip(IpAddr),
-    Pk(String), // 64-hex public key
-}
+use crate::clients::{Destination, MyceliumClient, MyceliumClientError};
 
 #[derive(Clone)]
 pub struct SupervisorClient {
-    base_url: String,         // e.g. "http://127.0.0.1:8990"
-    destination: Destination, // ip or pk
-    topic: String,            // e.g. "supervisor.rpc"
-    secret: Option<String>,   // optional, required by several supervisor methods
-    http: HttpClient,
-    id_counter: Arc<AtomicU64>, // JSON-RPC id generator (for inner + outer requests)
+    mycelium: Arc<MyceliumClient>, // Delegated Mycelium transport
+    destination: Destination,      // ip or pk
+    topic: String,                 // e.g. "supervisor.rpc"
+    secret: Option<String>,        // optional, required by several supervisor methods
+    id_counter: Arc<AtomicU64>,    // JSON-RPC id generator (for inner supervisor requests)
 }
 
 #[derive(Debug, Error)]
@@ -41,8 +33,37 @@ pub enum SupervisorClientError {
     MissingSecret,
 }
 
+impl From<MyceliumClientError> for SupervisorClientError {
+    fn from(e: MyceliumClientError) -> Self {
+        match e {
+            MyceliumClientError::TransportTimeout => SupervisorClientError::TransportTimeout,
+            MyceliumClientError::RpcError(m) => SupervisorClientError::RpcError(m),
+            MyceliumClientError::InvalidResponse(m) => SupervisorClientError::InvalidResponse(m),
+            MyceliumClientError::Http(err) => SupervisorClientError::Http(err),
+            MyceliumClientError::Json(err) => SupervisorClientError::Json(err),
+        }
+    }
+}
+
 impl SupervisorClient {
-    /// Create a new client. base_url defaults to Mycelium spec "http://127.0.0.1:8990" if empty.
+    /// Preferred constructor: provide a shared Mycelium client.
+    pub fn new_with_client(
+        mycelium: Arc<MyceliumClient>,
+        destination: Destination,
+        topic: impl Into<String>,
+        secret: Option<String>,
+    ) -> Self {
+        Self {
+            mycelium,
+            destination,
+            topic: topic.into(),
+            secret,
+            id_counter: Arc::new(AtomicU64::new(1)),
+        }
+    }
+
+    /// Backward-compatible constructor that builds a Mycelium client from base_url.
+    /// base_url defaults to Mycelium spec "http://127.0.0.1:8990" if empty.
     pub fn new(
         base_url: impl Into<String>,
         destination: Destination,
@@ -53,21 +74,15 @@ impl SupervisorClient {
         if url.is_empty() {
             url = "http://127.0.0.1:8990".to_string();
         }
-        let http = HttpClient::builder().build()?;
-        Ok(Self {
-            base_url: url,
-            destination,
-            topic: topic.into(),
-            secret,
-            http,
-            id_counter: Arc::new(AtomicU64::new(1)),
-        })
+        let mycelium = Arc::new(MyceliumClient::new(url)?);
+        Ok(Self::new_with_client(mycelium, destination, topic, secret))
     }
 
     fn next_id(&self) -> u64 {
         self.id_counter.fetch_add(1, Ordering::Relaxed)
     }
 
+    /// Internal helper used by tests to inspect dst JSON shape.
     fn build_dst(&self) -> Value {
         match &self.destination {
             Destination::Ip(ip) => json!({ "ip": ip.to_string() }),
@@ -89,49 +104,6 @@ impl SupervisorClient {
         Ok(BASE64_STANDARD.encode(s.as_bytes()))
     }
 
-    fn build_push_request(&self, payload_b64: String) -> Value {
-        let dst = self.build_dst();
-        let msg = json!({
-            "dst": dst,
-            "topic": self.topic,
-            "payload": payload_b64
-        });
-
-        // Async path: no reply_timeout attached
-        json!({
-            "jsonrpc": "2.0",
-            "id": self.next_id(),
-            "method": "pushMessage",
-            "params": [ { "message": msg } ]
-        })
-    }
-
-    async fn send_push(&self, req: &Value) -> Result<Value, SupervisorClientError> {
-        let resp = self.http.post(&self.base_url).json(req).send().await?;
-        let status = resp.status();
-        let body: Value = resp.json().await?;
-        // JSON-RPC error object handling
-        if let Some(err) = body.get("error") {
-            let code = err.get("code").and_then(|v| v.as_i64()).unwrap_or(0);
-            let msg = err
-                .get("message")
-                .and_then(|v| v.as_str())
-                .unwrap_or("unknown error");
-            if code == 408 {
-                return Err(SupervisorClientError::TransportTimeout);
-            }
-            return Err(SupervisorClientError::RpcError(format!(
-                "code={code} msg={msg}"
-            )));
-        }
-        if !status.is_success() {
-            return Err(SupervisorClientError::RpcError(format!(
-                "HTTP status {status}, body {body}"
-            )));
-        }
-        Ok(body)
-    }
-
     fn extract_message_id_from_result(result: &Value) -> Option<String> {
         // Two possibilities per Mycelium spec oneOf:
         // - PushMessageResponseId: { "id": "0123456789abcdef" }
@@ -142,34 +114,28 @@ impl SupervisorClient {
             .map(|s| s.to_string())
     }
 
-    /// Generic call: build supervisor JSON-RPC message, wrap in Mycelium pushMessage, return outbound message id (hex).
+    /// Generic call: build supervisor JSON-RPC message, send via Mycelium pushMessage, return outbound message id (hex).
     pub async fn call(&self, method: &str, params: Value) -> Result<String, SupervisorClientError> {
         let inner = self.build_supervisor_payload(method, params);
         let payload_b64 = Self::encode_payload(&inner)?;
-        let push_req = self.build_push_request(payload_b64);
-        let resp = self.send_push(&push_req).await?;
+        let result = self
+            .mycelium
+            .push_message(&self.destination, &self.topic, &payload_b64, None)
+            .await?;
 
-        // Expect "result" to be either inbound message or response id
-        match resp.get("result") {
-            Some(res_obj) => {
-                if let Some(id) = Self::extract_message_id_from_result(res_obj) {
-                    return Ok(id);
-                }
-                // Some servers might return the oneOf wrapped, handle len==1 array defensively (not in spec but resilient)
-                if let Some(arr) = res_obj.as_array()
-                    && arr.len() == 1
-                    && let Some(id) = Self::extract_message_id_from_result(&arr[0])
-                {
-                    return Ok(id);
-                }
-                Err(SupervisorClientError::InvalidResponse(format!(
-                    "result did not contain message id: {res_obj}"
-                )))
-            }
-            None => Err(SupervisorClientError::InvalidResponse(format!(
-                "missing result in response: {resp}"
-            ))),
+        if let Some(id) = MyceliumClient::extract_message_id_from_result(&result) {
+            return Ok(id);
         }
+        // Some servers might return the oneOf wrapped, handle len==1 array defensively (not in spec but resilient)
+        if let Some(arr) = result.as_array()
+            && arr.len() == 1
+            && let Some(id) = MyceliumClient::extract_message_id_from_result(&arr[0])
+        {
+            return Ok(id);
+        }
+        Err(SupervisorClientError::InvalidResponse(format!(
+            "result did not contain message id: {result}"
+        )))
     }
 
     fn need_secret(&self) -> Result<&str, SupervisorClientError> {
@@ -334,8 +300,10 @@ impl SupervisorClient {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::net::IpAddr;
 
     fn mk_client() -> SupervisorClient {
+        // Uses the legacy constructor but will not issue real network calls in these tests.
         SupervisorClient::new(
             "http://127.0.0.1:8990",
             Destination::Pk(
@@ -368,20 +336,21 @@ mod tests {
     }
 
     #[test]
-    fn wraps_supervisor_payload_in_push_message() {
+    fn encodes_supervisor_payload_b64() {
         let c = mk_client();
         let payload = c.build_supervisor_payload("list_runners", json!([]));
         let b64 = SupervisorClient::encode_payload(&payload).unwrap();
-        let req = c.build_push_request(b64);
-        assert_eq!(req.get("method").unwrap().as_str().unwrap(), "pushMessage");
-        let params = req.get("params").unwrap().as_array().unwrap();
-        let msg = params[0].get("message").unwrap();
-        assert!(msg.get("payload").is_some());
+
+        // decode and compare round-trip JSON
+        let raw = base64::engine::general_purpose::STANDARD
+            .decode(b64.as_bytes())
+            .unwrap();
+        let decoded: Value = serde_json::from_slice(&raw).unwrap();
         assert_eq!(
-            msg.get("topic").unwrap().as_str().unwrap(),
-            "supervisor.rpc"
+            decoded.get("method").unwrap().as_str().unwrap(),
+            "list_runners"
         );
-        assert!(msg.get("dst").unwrap().get("pk").is_some());
+        assert_eq!(decoded.get("jsonrpc").unwrap().as_str().unwrap(), "2.0");
     }
 
     #[test]

src/clients/types.rs

@@ -0,0 +1,9 @@
+use std::net::IpAddr;
+
+/// Destination for Mycelium messages (shared by clients)
+#[derive(Clone, Debug)]
+pub enum Destination {
+    Ip(IpAddr),
+    /// 64-hex public key of the receiver node
+    Pk(String),
+}

src/main.rs

@@ -97,6 +97,8 @@ async fn main() {
             concurrency: 32,
             base_url,
             topic: "supervisor.rpc".to_string(),
+            transport_poll_interval_secs: 2,
+            transport_poll_timeout_secs: 300,
         };
         let _auto_handle = herocoordinator::router::start_router_auto(service_for_router, cfg);
     }

src/models.rs

@@ -10,6 +10,6 @@ pub use actor::Actor;
 pub use context::Context;
 pub use flow::{Flow, FlowStatus};
 pub use job::{Job, JobStatus};
-pub use message::{Message, MessageFormatType, MessageStatus, MessageType};
+pub use message::{Message, MessageFormatType, MessageStatus, MessageType, TransportStatus};
 pub use runner::Runner;
 pub use script_type::ScriptType;

src/models/message.rs

@@ -22,6 +22,12 @@ pub struct Message {
     pub timeout_ack: u32,
     /// Seconds for the receiver to send us a reply
     pub timeout_result: u32,
+
+    /// Outbound transport id returned by Mycelium on push
+    pub transport_id: Option<String>,
+    /// Latest transport status as reported by Mycelium
+    pub transport_status: Option<TransportStatus>,
+
     pub job: Vec<Job>,
     pub logs: Vec<Log>,
     pub created_at: Timestamp,
@@ -44,6 +50,15 @@ pub enum MessageStatus {
     Processed,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub enum TransportStatus {
+    Queued,
+    Sent,
+    Delivered,
+    Read,
+    Failed,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum MessageFormatType {
     Html,

src/router.rs

@@ -4,8 +4,8 @@ use serde_json::{Value, json};
 use tokio::sync::Semaphore;
 
 use crate::{
-    clients::{Destination, SupervisorClient},
-    models::{Job, Message, MessageStatus, ScriptType},
+    clients::{Destination, SupervisorClient, MyceliumClient},
+    models::{Job, Message, MessageStatus, ScriptType, TransportStatus},
     service::AppService,
 };
 
@@ -15,7 +15,9 @@ pub struct RouterConfig {
     pub concurrency: usize,
     pub base_url: String, // e.g. http://127.0.0.1:8990
     pub topic: String,    // e.g. "supervisor.rpc"
-                          // secret currently unused (None), add here later if needed
+    // Transport status polling configuration
+    pub transport_poll_interval_secs: u64, // e.g. 2
+    pub transport_poll_timeout_secs: u64,  // e.g. 300 (5 minutes)
 }
 
 /// Start background router loops, one per context.
@@ -32,6 +34,18 @@ pub fn start_router(service: AppService, cfg: RouterConfig) -> Vec<tokio::task::
         let cfg_cloned = cfg.clone();
         let handle = tokio::spawn(async move {
             let sem = Arc::new(Semaphore::new(cfg_cloned.concurrency));
+
+            // Create a shared Mycelium client for this context loop (retry until available)
+            let mycelium = loop {
+                match MyceliumClient::new(cfg_cloned.base_url.clone()) {
+                    Ok(c) => break Arc::new(c),
+                    Err(e) => {
+                        eprintln!("[router ctx={}] MyceliumClient init error: {}", ctx_id, e);
+                        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                    }
+                }
+            };
+
             loop {
                 // Pop next message key (blocking with timeout)
                 match service_cloned.brpop_msg_out(ctx_id, 1).await {
@@ -50,16 +64,19 @@ pub fn start_router(service: AppService, cfg: RouterConfig) -> Vec<tokio::task::
                         };
                         let service_task = service_cloned.clone();
                         let cfg_task = cfg_cloned.clone();
-                        tokio::spawn(async move {
-                            // Ensure permit is dropped at end of task
-                            let _permit = permit;
-                            if let Err(e) =
-                                deliver_one(&service_task, &cfg_task, ctx_id, &key).await
-                            {
-                                eprintln!(
-                                    "[router ctx={}] delivery error for {}: {}",
-                                    ctx_id, key, e
-                                );
+                        tokio::spawn({
+                            let mycelium = mycelium.clone();
+                            async move {
+                                // Ensure permit is dropped at end of task
+                                let _permit = permit;
+                                if let Err(e) =
+                                    deliver_one(&service_task, &cfg_task, ctx_id, &key, mycelium).await
+                                {
+                                    eprintln!(
+                                        "[router ctx={}] delivery error for {}: {}",
+                                        ctx_id, key, e
+                                    );
+                                }
                             }
                         });
                     }
@@ -85,6 +102,7 @@ async fn deliver_one(
     cfg: &RouterConfig,
     context_id: u32,
     msg_key: &str,
+    mycelium: Arc<MyceliumClient>,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     // Parse "message:{caller_id}:{id}"
     let (caller_id, id) = parse_message_key(msg_key)
@@ -118,25 +136,115 @@ async fn deliver_one(
     } else {
         Destination::Ip(runner.address)
     };
-    let client = SupervisorClient::new(
-        cfg.base_url.clone(),
+    let client = SupervisorClient::new_with_client(
+        mycelium.clone(),
         dest,
         cfg.topic.clone(),
         None, // secret
-    )?;
+    );
 
     // Build supervisor method and params from Message
     let method = msg.message.clone();
     let params = build_params(&msg)?;
 
     // Send
-    let _out_id = client.call(&method, params).await?;
+    let out_id = client.call(&method, params).await?;
+
+    // Store transport id and initial Sent status
+    let _ = service
+        .update_message_transport(
+            context_id,
+            caller_id,
+            id,
+            Some(out_id.clone()),
+            Some(TransportStatus::Sent),
+        )
+        .await;
 
     // Mark as acknowledged on success
     service
         .update_message_status(context_id, caller_id, id, MessageStatus::Acknowledged)
         .await?;
 
+    // Spawn transport-status poller
+    {
+        let service_poll = service.clone();
+        let poll_interval = std::time::Duration::from_secs(cfg.transport_poll_interval_secs);
+        let poll_timeout = std::time::Duration::from_secs(cfg.transport_poll_timeout_secs);
+        let out_id_cloned = out_id.clone();
+        let mycelium = mycelium.clone();
+
+        tokio::spawn(async move {
+            let start = std::time::Instant::now();
+            let client = mycelium;
+
+            let mut last_status: Option<TransportStatus> = Some(TransportStatus::Sent);
+
+            loop {
+                if start.elapsed() >= poll_timeout {
+                    let _ = service_poll
+                        .append_message_logs(
+                            context_id,
+                            caller_id,
+                            id,
+                            vec!["Transport-status polling timed out".to_string()],
+                        )
+                        .await;
+                    // leave last known status; do not override
+                    break;
+                }
+
+                match client.message_status(&out_id_cloned).await {
+                    Ok(s) => {
+                        if last_status.as_ref() != Some(&s) {
+                            let _ = service_poll
+                                .update_message_transport(
+                                    context_id,
+                                    caller_id,
+                                    id,
+                                    None,
+                                    Some(s.clone()),
+                                )
+                                .await;
+                            last_status = Some(s.clone());
+                        }
+
+                        // Stop on terminal states
+                        if matches!(s, TransportStatus::Delivered | TransportStatus::Read) {
+                            break;
+                        }
+                        if matches!(s, TransportStatus::Failed) {
+                            let _ = service_poll
+                                .append_message_logs(
+                                    context_id,
+                                    caller_id,
+                                    id,
+                                    vec![format!(
+                                        "Transport failed for outbound id {out_id_cloned}"
+                                    )],
+                                )
+                                .await;
+                            break;
+                        }
+                    }
+                    Err(e) => {
+                        // Log and continue polling
+                        let _ = service_poll
+                            .append_message_logs(
+                                context_id,
+                                caller_id,
+                                id,
+                                vec![format!("messageStatus query error: {e}")],
+                            )
+                            .await;
+                    }
+                }
+
+                tokio::time::sleep(poll_interval).await;
+            }
+        });
+    }
+
     Ok(())
 }
 

src/rpc.rs

@@ -299,6 +299,8 @@ impl MessageCreate {
             timeout,
             timeout_ack,
             timeout_result,
+            transport_id: None,
+            transport_status: None,
             job: job.into_iter().map(JobCreate::into_domain).collect(),
             logs: Vec::new(),
             created_at: ts,

src/service.rs

@@ -1,7 +1,7 @@
 use crate::dag::{DagError, DagResult, FlowDag, build_flow_dag};
 use crate::models::{
     Actor, Context, Flow, FlowStatus, Job, JobStatus, Message, MessageFormatType, MessageStatus,
-    Runner,
+    Runner, TransportStatus,
 };
 use crate::storage::RedisDriver;
 
@@ -508,6 +508,8 @@ impl AppService {
                                     timeout: job.timeout,
                                     timeout_ack: 10,
                                     timeout_result: job.timeout,
+                                    transport_id: None,
+                                    transport_status: None,
                                     job: vec![job.clone()],
                                     logs: Vec::new(),
                                     created_at: ts,
@@ -589,6 +591,8 @@ impl AppService {
                 timeout: job.timeout,
                 timeout_ack: 10,
                 timeout_result: job.timeout,
+                transport_id: None,
+                transport_status: None,
                 job: vec![job.clone()],
                 logs: Vec::new(),
                 created_at: ts,
@@ -817,6 +821,21 @@ impl AppService {
             .await
     }
 
+    pub async fn update_message_transport(
+        &self,
+        context_id: u32,
+        caller_id: u32,
+        id: u32,
+        transport_id: Option<String>,
+        transport_status: Option<TransportStatus>,
+    ) -> Result<(), BoxError> {
+        // Ensure message exists (provides clearer error)
+        let _ = self.redis.load_message(context_id, caller_id, id).await?;
+        self.redis
+            .update_message_transport(context_id, caller_id, id, transport_id, transport_status)
+            .await
+    }
+
     pub async fn update_flow_env_vars_merge(
         &self,
         context_id: u32,

src/storage/redis.rs

@@ -7,7 +7,7 @@ use serde_json::{Map as JsonMap, Value};
 use tokio::sync::Mutex;
 
 use crate::models::{
-    Actor, Context, Flow, FlowStatus, Job, JobStatus, Message, MessageStatus, Runner,
+    Actor, Context, Flow, FlowStatus, Job, JobStatus, Message, MessageStatus, Runner, TransportStatus,
 };
 
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
@@ -358,6 +358,40 @@ impl RedisDriver {
         Ok(())
     }
 
+    /// Message: update transport_id / transport_status (optionally) and bump updated_at
+    pub async fn update_message_transport(
+        &self,
+        db: u32,
+        caller_id: u32,
+        id: u32,
+        transport_id: Option<String>,
+        transport_status: Option<TransportStatus>,
+    ) -> Result<()> {
+        let mut cm = self.manager_for_db(db).await?;
+        let key = Self::message_key(caller_id, id);
+
+        let mut pairs: Vec<(String, String)> = Vec::new();
+
+        if let Some(tid) = transport_id {
+            pairs.push(("transport_id".to_string(), tid));
+        }
+
+        if let Some(ts_status) = transport_status {
+            let status_str = match serde_json::to_value(&ts_status)? {
+                Value::String(s) => s,
+                v => v.to_string(),
+            };
+            pairs.push(("transport_status".to_string(), status_str));
+        }
+
+        // Always bump updated_at
+        let ts = crate::time::current_timestamp();
+        pairs.push(("updated_at".to_string(), ts.to_string()));
+
+        let _: usize = cm.hset_multiple(key, &pairs).await?;
+        Ok(())
+    }
+
     /// Flow: merge env_vars map and bump updated_at
     pub async fn update_flow_env_vars_merge(
         &self,