initial commit

1 files changed, 882 insertions, 0 deletions

diff --git a/src/mycelium.rs b/src/mycelium.rs
new file mode 100644
index 0000000..94c5a3d
--- /dev/null
+++ b/src/mycelium.rs
@@ -0,0 +1,882 @@
+use anyhow::{anyhow, Context, Result};
+use async_trait::async_trait;
+use base64::engine::{general_purpose::STANDARD as BASE64, Engine};
+use reqwest::StatusCode;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use serde_bytes::{ByteBuf, Bytes};
+use std::borrow::Cow;
+use std::boxed::Box;
+use std::fs::File;
+use std::io::Write;
+use std::net::TcpListener;
+use std::os::unix::io::FromRawFd;
+use std::path::PathBuf;
+use std::process::Stdio;
+use std::time::Duration;
+use thiserror::Error;
+use tokio::process::{Child, Command};
+use url::Url;
+
+use crate::transport::EncodedMessage;
+use crate::transport::Keepalive;
+use crate::transport::{HelloReply, HelloRequest, SyncEnvelope};
+use crate::transport::{MessageHandler, ParsedMessage, Transport};
+
+const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
+
+const TOPIC_PREFIX: &str = "continuum.v1.";
+const TOPIC_HELLO: &str = "continuum.v1.hello";
+const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply";
+const TOPIC_SYNC: &str = "continuum.v1.sync";
+const TOPIC_PING: &str = "continuum.v1.ping";
+const TOPIC_PONG: &str = "continuum.v1.pong";
+
+const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25);
+
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct InboundMessage {
+    id: String,
+    src_pk: String,
+    topic: String,
+    payload: String,
+}
+
+#[derive(Debug, Serialize)]
+struct OutboundMessage {
+    dst: Destination,
+    topic: String,
+    payload: String,
+}
+
+#[derive(Debug, Serialize)]
+struct Destination {
+    pk: String,
+}
+
+/// Represents where a 32-byte cryptographic key originates.
+#[derive(Debug, Clone)]
+pub enum KeySource {
+    Raw([u8; 32]),
+    Path(PathBuf),
+}
+
+/// Atomic structure ensuring private network credentials are always handled together.
+#[derive(Debug, Clone)]
+pub struct PrivateNetworkConfig {
+    pub name: String,
+    pub key: KeySource,
+}
+
+/// Comprehensive parameters required to spawn or target an instance.
+#[derive(Debug, Clone)]
+pub struct MyceliumSpawnConfig {
+    pub rest_url_opt: Option<Url>,
+    pub key_file_opt: Option<PathBuf>,
+    pub network_key_file: Option<PathBuf>,
+    pub network_name: String,
+    pub tcp_port: Option<u16>,
+    pub api_port: Option<u16>,
+    pub rpc_port: Option<u16>,
+    pub discovery_port: Option<u16>,
+    pub peers: Vec<String>,
+}
+
+impl Default for MyceliumSpawnConfig {
+    fn default() -> Self {
+        Self {
+            rest_url_opt: None,
+            key_file_opt: None,
+            network_key_file: None,
+            network_name: "something".to_string(),
+            tcp_port: None,
+            api_port: None,
+            rpc_port: None,
+            discovery_port: None,
+            peers: vec!["tls://127.0.0.1:9651".to_string()],
+        }
+    }
+}
+
+pub enum MyceliumRuntime {
+    Supervised(Box<SupervisedMycelium>),
+    External,
+}
+
+pub struct MyceliumSetup {
+    pub runtime: MyceliumRuntime,
+    pub rest_url: Url,
+}
+
+impl MyceliumSetup {
+    /// Evaluates the configuration payload and initializes the chosen execution strategy.
+    pub async fn bootstrap(config: MyceliumSpawnConfig) -> Result<Self> {
+        // TODO warn for all ignored options if rest_url_opt is set
+        if let Some(url) = config.rest_url_opt {
+            tracing::info!(
+                "Configured to interface with an external standalone Mycelium instance at: {}",
+                url
+            );
+            return Ok(Self {
+                runtime: MyceliumRuntime::External,
+                rest_url: url,
+            });
+        }
+
+        tracing::info!(
+            "Configured to initialize an embedded user-space Mycelium background worker"
+        );
+
+        let key = match config.key_file_opt {
+            Some(key_file) => KeySource::Path(key_file),
+            None => {
+                let mut bytes = [0u8; 32];
+                getrandom::fill(&mut bytes)?;
+                KeySource::Raw(bytes)
+            }
+        };
+
+        let private_net_config = config
+            .network_key_file
+            .map(|net_key_path| PrivateNetworkConfig {
+                name: config.network_name.clone(),
+                key: KeySource::Path(net_key_path),
+            });
+
+        let supervised_node = SupervisedMycelium::spawn(
+            key,
+            private_net_config,
+            config.tcp_port,
+            config.api_port,
+            config.rpc_port,
+            config.discovery_port,
+            &config.peers,
+        )
+        .await
+        .context("Failed to securely bootstrap embedded Mycelium background worker")?;
+
+        let resolved_url = Url::parse(&format!(
+            "http://127.0.0.1:{}",
+            supervised_node.config.api_port
+        ))?;
+
+        Ok(Self {
+            runtime: MyceliumRuntime::Supervised(Box::new(supervised_node)),
+            rest_url: resolved_url,
+        })
+    }
+
+    /// Blocks until the background daemon drops out. If running externally, hangs open indefinitely.
+    pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
+        match &mut self.runtime {
+            MyceliumRuntime::Supervised(node) => node.wait_for_exit().await,
+            MyceliumRuntime::External => {
+                std::future::pending::<std::io::Result<std::process::ExitStatus>>().await
+            }
+        }
+    }
+}
+
+#[derive(Deserialize, Debug, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct MyceliumAdminInfo {
+    pub node_ip: String,
+    pub node_pubkey: String,
+    pub node_start_time: u64,
+    pub node_subnet: String,
+}
+
+pub async fn fetch_admin_info(rest_url: &Url) -> Result<MyceliumAdminInfo> {
+    let url = rest_url.join("/api/v1/admin")?;
+    let info = reqwest::Client::builder()
+        .timeout(Duration::from_secs(10))
+        .build()?
+        .get(url)
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    Ok(info)
+}
+
+#[derive(Debug, Clone)]
+pub struct ClusterNodeConfig {
+    pub tcp_port: u16,
+    pub api_port: u16,
+    pub rpc_port: u16,
+    pub discovery_port: Option<u16>,
+    pub private_network: Option<String>,
+}
+
+pub struct SupervisedMycelium {
+    child: Child,
+    pub config: ClusterNodeConfig,
+    _key_file_handle: Option<File>,
+    _net_key_file_handle: Option<File>,
+    key_file_path: PathBuf,
+    net_key_file_path: Option<PathBuf>,
+    is_ephemeral_key: bool,
+    is_ephemeral_net_key: bool,
+}
+
+impl SupervisedMycelium {
+    async fn spawn(
+        node_key: KeySource,
+        private_network: Option<PrivateNetworkConfig>,
+        explicit_tcp: Option<u16>,
+        explicit_api: Option<u16>,
+        explicit_rpc: Option<u16>,
+        explicit_discovery_port: Option<u16>,
+        peers: &[String],
+    ) -> Result<Self> {
+        let tcp_port = match explicit_tcp {
+            Some(p) => p,
+            None => Self::find_available_port().context("Failed to allocate TCP wire port")?,
+        };
+        let api_port = match explicit_api {
+            Some(p) => p,
+            None => Self::find_available_port().context("Failed to allocate API port")?,
+        };
+        let rpc_port = match explicit_rpc {
+            Some(p) => p,
+            None => Self::find_available_port().context("Failed to allocate JSON-RPC port")?,
+        };
+        let discovery_port = match explicit_discovery_port {
+            Some(0) => Some(
+                Self::find_available_port().context("Failed to allocate peer discovery port")?,
+            ),
+            p => p,
+        };
+
+        let config = ClusterNodeConfig {
+            tcp_port,
+            api_port,
+            rpc_port,
+            discovery_port,
+            private_network: private_network.as_ref().map(|pn| pn.name.clone()),
+        };
+
+        let parent_pid = std::process::id();
+
+        let (key_file_handle, key_file_path, node_fd, is_ephemeral_key) =
+            Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?;
+
+        let (net_key_file_handle, net_key_file_path, net_fd, is_ephemeral_net_key) =
+            match private_network {
+                Some(net_cfg) => {
+                    let (h, p, fd, eph) =
+                        Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?;
+                    (h, Some(p), fd, eph)
+                }
+                None => (None, None, None, false),
+            };
+
+        let api_addr = format!("127.0.0.1:{}", config.api_port);
+        let rpc_addr = format!("127.0.0.1:{}", config.rpc_port);
+
+        let mut cmd = Command::new("mycelium");
+
+        let fds_to_inherit: Vec<i32> = [node_fd, net_fd].iter().filter_map(|&x| x).collect();
+        if !fds_to_inherit.is_empty() {
+            unsafe {
+                cmd.pre_exec(move || {
+                    for &fd in &fds_to_inherit {
+                        libc::fcntl(fd, libc::F_SETFD, 0);
+                    }
+                    Ok(())
+                });
+            }
+        }
+
+        cmd.arg("--key-file")
+            .arg(&key_file_path)
+            .arg("--tcp-listen-port")
+            .arg(config.tcp_port.to_string())
+            .arg("--disable-quic")
+            .arg("--api-addr")
+            .arg(&api_addr)
+            .arg("--jsonrpc-addr")
+            .arg(&rpc_addr)
+            .arg("--no-tun");
+
+        if let Some(ref discovery_port) = config.discovery_port {
+            cmd.arg("--peer-discovery-port")
+                .arg(discovery_port.to_string());
+        } else {
+            cmd.arg("--disable-peer-discovery");
+        }
+
+        if let Some(ref net_name) = config.private_network {
+            cmd.arg("--network-name").arg(net_name);
+        }
+        if let Some(ref net_key_path) = net_key_file_path {
+            cmd.arg("--network-key-file").arg(net_key_path);
+        }
+
+        if !peers.is_empty() {
+            cmd.arg("--peers").arg(peers.join(" "));
+        }
+
+        cmd.env("RUST_LOG", "info");
+
+        cmd.stdout(Stdio::piped());
+        cmd.stderr(Stdio::piped());
+
+        let mut child = cmd
+            .spawn()
+            .context("Failed to execute mycelium process daemon")?;
+
+        // 1. FIXED: Pass-through streaming for child process STDOUT
+        if let Some(mut stdout) = child.stdout.take() {
+            tokio::spawn(async move {
+                let mut host_stdout = tokio::io::stdout();
+                let _ = tokio::io::copy(&mut stdout, &mut host_stdout).await;
+            });
+        }
+
+        // 2. FIXED: Pass-through streaming for child process STDERR
+        if let Some(mut stderr) = child.stderr.take() {
+            tokio::spawn(async move {
+                let mut host_stderr = tokio::io::stderr();
+                let _ = tokio::io::copy(&mut stderr, &mut host_stderr).await;
+            });
+        }
+
+        let node = Self {
+            child,
+            config,
+            _key_file_handle: key_file_handle,
+            _net_key_file_handle: net_key_file_handle,
+            key_file_path,
+            net_key_file_path,
+            is_ephemeral_key,
+            is_ephemeral_net_key,
+        };
+
+        //let admin_info = node.poll_admin_endpoint().await?;
+        //Ok((node, admin_info))
+        Ok(node)
+    }
+
+    fn resolve_key_strategy(
+        desc_tag: &str,
+        source: KeySource,
+        parent_pid: u32,
+    ) -> Result<(Option<File>, PathBuf, Option<i32>, bool)> {
+        match source {
+            KeySource::Raw(raw_bytes) => {
+                let (handle, fd) = Self::create_ephemeral_file(desc_tag, raw_bytes)?;
+                let pseudo_path = PathBuf::from(format!("/proc/{}/fd/{}", parent_pid, fd));
+                Ok((Some(handle), pseudo_path, Some(fd), true))
+            }
+            KeySource::Path(existing_path) => Ok((None, existing_path, None, false)),
+        }
+    }
+
+    fn find_available_port() -> Result<u16> {
+        let listener = TcpListener::bind("127.0.0.1:0")?;
+        let addr = listener.local_addr()?;
+        Ok(addr.port())
+    }
+
+    fn create_ephemeral_file(desc_tag: &str, raw_key: [u8; 32]) -> Result<(File, i32)> {
+        #[cfg(target_os = "linux")]
+        {
+            let memfd_name = std::ffi::CString::new(desc_tag)?;
+            let fd = unsafe { libc::memfd_create(memfd_name.as_ptr(), 0) };
+            if fd < 0 {
+                return Err(anyhow!(
+                    "memfd_create failed natively inside Linux kernel for tag: {}",
+                    desc_tag
+                ));
+            }
+            let mut file = unsafe { File::from_raw_fd(fd) };
+            file.write_all(&raw_key)?;
+            file.flush()?;
+            Ok((file, fd))
+        }
+
+        #[cfg(not(target_os = "linux"))]
+        {
+            let mut path = std::env::tempdir();
+            path.push(format!("{}_{}.key", desc_tag, uuid::Uuid::new_v4()));
+            let mut file = File::create(&path)?;
+            file.write_all(&raw_key)?;
+            file.flush()?;
+            Ok((file, -1))
+        }
+    }
+
+    //async fn poll_admin_endpoint(&mut self) -> Result<MyceliumAdminInfo> {
+    //    let client = reqwest::Client::new();
+    //    let url = format!("http://127.0.0.1:{}/api/v1/admin", self.config.api_port);
+    //
+    //    for _ in 0..30 {
+    //        if let Some(status) = self.child.try_wait()? {
+    //            return Err(anyhow!(
+    //                "myceliumd process closed prematurely with status: {}",
+    //                status
+    //            ));
+    //        }
+    //        if let Ok(resp) = client.get(&url).send().await {
+    //            if resp.status().is_success() {
+    //                let info: MyceliumAdminInfo = resp.json().await?;
+    //                return Ok(info);
+    //            }
+    //        }
+    //        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+    //    }
+    //    Err(anyhow!(
+    //        "Timeout waiting for local mycelium administration socket to respond"
+    //    ))
+    //}
+
+    pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
+        self.child.wait().await
+    }
+}
+
+impl Drop for SupervisedMycelium {
+    fn drop(&mut self) {
+        let _ = self.child.start_kill();
+
+        #[cfg(not(target_os = "linux"))]
+        {
+            if self.is_ephemeral_key {
+                let _ = std::fs::remove_file(&self.key_file_path);
+            }
+            if self.is_ephemeral_net_key {
+                if let Some(ref path) = self.net_key_file_path {
+                    let _ = std::fs::remove_file(path);
+                }
+            }
+        }
+    }
+}
+
+fn is_pubkey(s: &str) -> bool {
+    s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit())
+}
+
+pub struct MyceliumTransport {
+    // Wrap your zero-allocation lifecycle supervisor directly inside the driver plane
+    pub setup: tokio::sync::Mutex<MyceliumSetup>,
+    pub client: reqwest::Client,
+    pub poll_client: reqwest::Client,
+    pub messages_endpoint: Url,
+    pubkey_basepath: Url,
+    local_id: String,
+}
+
+impl MyceliumTransport {
+    pub async fn connect(setup: MyceliumSetup) -> Result<Self> {
+        let rest_url = setup.rest_url.clone();
+
+        let client = reqwest::Client::builder()
+            .timeout(Duration::from_secs(10))
+            .build()?;
+        let poll_client = reqwest::Client::builder()
+            .timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30))
+            .build()?;
+
+        let admin: MyceliumAdminInfo = client
+            .get(rest_url.join("/api/v1/admin")?)
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await
+            .context("Querying local Mycelium identity")?;
+
+        Ok(Self {
+            setup: tokio::sync::Mutex::new(setup),
+            client,
+            poll_client,
+            messages_endpoint: rest_url.join("/api/v1/messages")?,
+            pubkey_basepath: rest_url.join("/api/v1/pubkey/")?,
+            local_id: admin.node_pubkey,
+        })
+    }
+
+    async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result<String> {
+        #[derive(Deserialize)]
+        struct PublicKeyResponse {
+            #[serde(rename = "publicKey")]
+            public_key: String,
+        }
+
+        let response: PublicKeyResponse = self
+            .client
+            .get(self.pubkey_basepath.join(&ip.to_string())?)
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await?;
+        Ok(response.public_key)
+    }
+}
+
+#[async_trait]
+impl Transport for MyceliumTransport {
+    fn local_id(&self) -> &str {
+        &self.local_id
+    }
+
+    async fn resolve_peer(&self, hint: &str) -> Result<String> {
+        if is_pubkey(hint) {
+            return Ok(hint.to_owned());
+        }
+
+        if let Ok(ip) = hint.parse::<std::net::Ipv6Addr>() {
+            return self.lookup_pubkey(&ip).await;
+        }
+
+        let ip = tokio::time::timeout(Duration::from_secs(10), tokio::net::lookup_host((hint, 0)))
+            .await
+            .context("DNS lookup timed out")??
+            .find_map(|addr| match addr.ip() {
+                std::net::IpAddr::V6(ip) => Some(ip),
+                _ => None,
+            })
+            .ok_or_else(|| anyhow!("{hint}: no AAAA record found"))?;
+
+        self.lookup_pubkey(&ip).await
+    }
+
+    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
+        let outbound = format_message(peer, message)?;
+        Ok(EncodedMessage {
+            peer: peer.to_owned(),
+            body: serde_json::to_vec(&outbound)?.into(),
+        })
+    }
+
+    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
+        tracing::debug!(
+            payload_bytes = message.body.len(),
+            "Sending message"
+        );
+        let response = self
+            .client
+            .post(self.messages_endpoint.clone())
+            .header(reqwest::header::CONTENT_TYPE, "application/json")
+            .body(message.body.clone())
+            .send()
+            .await?;
+
+        let status = response.status();
+        if !status.is_success() {
+            tracing::error!(status = %status, "Mycelium rejected outbound message");
+            return Err(TransportError::MyceliumRejected {
+                peer: message.peer.clone(),
+                status,
+            }
+            .into());
+        }
+        Ok(())
+    }
+
+    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
+        tracing::info!(
+            endpoint = self.messages_endpoint.as_str(),
+            "Starting Mycelium message consumer"
+        );
+
+        loop {
+            tracing::trace!("Polling Mycelium inbox");
+
+            let response = match self
+                .poll_client
+                .get(self.messages_endpoint.clone())
+                .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)])
+                .query(&[("topic", encode_topic(TOPIC_PREFIX))])
+                .send()
+                .await
+            {
+                Ok(resp) => resp,
+                Err(error) => {
+                    tracing::warn!(error = %error, "Mycelium inbox poll failed");
+                    tokio::time::sleep(Duration::from_millis(250)).await;
+                    continue;
+                }
+            };
+
+            let status = response.status();
+
+            if status == StatusCode::NO_CONTENT {
+                tracing::trace!("Mycelium inbox is empty");
+                tokio::time::sleep(EMPTY_QUEUE_DELAY).await;
+                continue;
+            }
+
+            if !status.is_success() {
+                tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status");
+                tokio::time::sleep(Duration::from_millis(500)).await;
+                continue;
+            }
+
+            let raw_bytes = match response.bytes().await {
+                Ok(bytes) => bytes,
+                Err(error) => {
+                    tracing::warn!(error = %error, "Failed to read raw Mycelium response stream");
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                    continue;
+                }
+            };
+
+            let inbound_message = match serde_json::from_slice::<InboundMessage>(&raw_bytes) {
+                Ok(msg) => msg,
+                Err(error) => {
+                    tracing::error!(
+                        error = %error,
+                        raw_payload = %String::from_utf8_lossy(&raw_bytes),
+                        "Failed to decode inbound Mycelium message structure"
+                    );
+                    tokio::time::sleep(Duration::from_millis(500)).await;
+                    continue;
+                }
+            };
+
+            tracing::debug!(
+                id = %inbound_message.id,
+                payload_bytes = raw_bytes.len(),
+                "Received inbound Mycelium message"
+            );
+
+            match parse_message(&inbound_message) {
+                Ok((sender, parsed)) => {
+                    if let Err(error) = handler(sender, parsed).await {
+                        tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message");
+                    }
+                }
+                Err(transport_error) => {
+                    tracing::error!(
+                        error = %transport_error,
+                        topic = %inbound_message.topic,
+                        "Dropping poisoned packet: parsing failed gracefully"
+                    );
+                    // No sleep required: we successfully popped a malformed packet and want to process the next one immediately
+                }
+            }
+        }
+    }
+
+    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
+        let mut guard = self.setup.lock().await;
+        guard.wait_for_exit().await
+    }
+}
+
+fn encode_topic(topic: &str) -> String {
+    BASE64.encode(topic.as_bytes())
+}
+
+fn decode_topic(topic: &str) -> Result<String, TransportError> {
+    let bytes = BASE64.decode(topic)?;
+    Ok(String::from_utf8(bytes)?)
+}
+
+fn encode_payload(payload: &Vec<u8>) -> String {
+    BASE64.encode(payload)
+}
+
+fn decode_payload(topic: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
+    let bytes = BASE64.decode(payload)?;
+    match topic {
+        TOPIC_HELLO => {
+            let msg = bytes_to_hello_request(&bytes)?;
+            Ok(ParsedMessage::Hello(msg))
+        }
+        TOPIC_HELLO_REPLY => {
+            let msg = bytes_to_hello_reply(&bytes)?;
+            Ok(ParsedMessage::HelloReply(msg))
+        }
+        TOPIC_SYNC => {
+            let msg = bytes_to_sync_envelope(&bytes)?;
+            Ok(ParsedMessage::Sync(msg))
+        }
+        TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)),
+        TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)),
+        t => {
+            Err(TransportError::UnknownTopic(t.to_string()))
+        }
+    }
+}
+
+fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), TransportError> {
+    let sender = message.src_pk.clone();
+
+    if sender.is_empty() {
+        return Err(TransportError::NoSender);
+    }
+
+    let topic = decode_topic(&message.topic)?;
+    let parsed = decode_payload(&topic, &message.payload)?;
+
+    Ok((sender, parsed))
+}
+
+fn format_message(
+    recipient: &str,
+    message: &ParsedMessage,
+) -> Result<OutboundMessage, TransportError> {
+    let (topic, payload_bytes) = match message {
+        ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?),
+        ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?),
+        ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?),
+        ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?),
+        ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?),
+    };
+    Ok(OutboundMessage {
+        dst: Destination {
+            pk: recipient.to_owned(),
+        },
+        topic: encode_topic(topic),
+        payload: encode_payload(&payload_bytes),
+    })
+}
+
+#[derive(Serialize, Deserialize)]
+struct KeepaliveShadow {
+    heads_digest: ByteBuf,
+}
+
+fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+    rmp_serde::to_vec(&KeepaliveShadow {
+        heads_digest: ByteBuf::from(msg.heads_digest.to_vec()),
+    })
+}
+
+fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> {
+    let shadow: KeepaliveShadow = rmp_serde::from_slice(bytes)?;
+    let heads_digest = shadow
+        .heads_digest
+        .as_slice()
+        .try_into()
+        .map_err(|_| TransportError::InvalidPayload("heads digest must be 32 bytes".into()))?;
+    Ok(Keepalive { heads_digest })
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct HelloRequestShadow {}
+
+#[derive(Serialize, Deserialize)]
+pub struct HelloReplyShadow<'a> {
+    // Cow allows zero-copy references when saving, but creates an owned String when loading
+    pub cluster_id: Cow<'a, str>,
+    #[serde(default)]
+    pub peers: Vec<String>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct SyncEnvelopeShadow<'a> {
+    pub cluster_id: Cow<'a, str>,
+    #[serde(with = "automerge_msg_codec")]
+    pub message: automerge::sync::Message,
+}
+
+// =====================================================================
+// 2. CUSTOM CODEC
+// =====================================================================
+mod automerge_msg_codec {
+    use super::*;
+    use automerge::sync::Message;
+
+    pub fn serialize<S>(msg: &Message, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        let bytes = msg.clone().encode();
+        Bytes::new(&bytes).serialize(serializer)
+    }
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Message, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let byte_buf = ByteBuf::deserialize(deserializer)?;
+        Message::decode(&byte_buf)
+            .map_err(|e| serde::de::Error::custom(format!("Automerge decode failed: {:?}", e)))
+    }
+}
+
+// =====================================================================
+// 3. CLEAN CONVERSION INTERFACE
+// =====================================================================
+
+// --- HelloRequest ---
+pub fn hello_request_to_bytes(_msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+    rmp_serde::to_vec(&HelloRequestShadow {})
+}
+pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, rmp_serde::decode::Error> {
+    let _: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
+    Ok(HelloRequest {})
+}
+
+// --- HelloReply ---
+pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+    let shadow = HelloReplyShadow {
+        cluster_id: Cow::Borrowed(&msg.cluster_id),
+        peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
+    };
+    rmp_serde::to_vec(&shadow)
+}
+pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, rmp_serde::decode::Error> {
+    let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?;
+    Ok(HelloReply {
+        cluster_id: shadow.cluster_id.into_owned(),
+        peers: shadow.peers,
+    })
+}
+
+// --- SyncEnvelope ---
+pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+    let shadow = SyncEnvelopeShadow {
+        cluster_id: Cow::Borrowed(&msg.cluster_id),
+        message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules
+    };
+    rmp_serde::to_vec(&shadow)
+}
+pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, rmp_serde::decode::Error> {
+    let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?;
+    Ok(SyncEnvelope {
+        cluster_id: shadow.cluster_id.into_owned(),
+        message: shadow.message,
+    })
+}
+
+#[derive(Error, Debug)]
+pub enum TransportError {
+    #[error("Mycelium delivered a message without a sender public key")]
+    NoSender,
+
+    #[error("Base64 decoding failed: {0}")]
+    Base64Decode(#[from] base64::DecodeError),
+
+    #[error("UTF-8 decoding failed: {0}")]
+    Utf8Decode(#[from] std::string::FromUtf8Error),
+
+    #[error("MessagePack serialization failed: {0}")]
+    MessagePackEncode(#[from] rmp_serde::encode::Error),
+
+    #[error("MessagePack deserialization failed: {0}")]
+    MessagePackDecode(#[from] rmp_serde::decode::Error),
+
+    #[error("Unknown topic: {0}")]
+    UnknownTopic(String),
+
+    #[error("Invalid payload: {0}")]
+    InvalidPayload(String),
+
+    #[error("Network HTTP request failed: {0}")]
+    Http(#[from] reqwest::Error),
+
+    #[error("Mycelium server rejected outbound message to {peer} (HTTP {status})")]
+    MyceliumRejected {
+        peer: String,
+        status: reqwest::StatusCode,
+    },
+}