Implement remote spawn functionality

Add the ability to spawn processes on remote nodes via distribution:

- Add Spawn/SpawnReply protocol messages for remote spawn RPC
- Create spawn registry for registering spawnable functions by module/function name
- Add pending_spawns tracking in DistributionManager for async request/reply
- Implement node::spawn, spawn_link, and spawn_monitor functions
- Add comprehensive tests for spawn registry and node spawn functions

Remote spawning works by registering functions with
distribution::spawn::register("module", "func", |args| async { ... })
then spawning them with node::spawn(remote_node, "module", "func", args).

Author: scrogson

crates/ambitious/src/distribution/manager.rs

@@ -18,6 +18,9 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 use tokio::sync::{mpsc, oneshot};
 
+/// Type alias for spawn reply (pid, monitor_ref, error).
+type SpawnReply = (Option<Pid>, Option<u64>, Option<String>);
+
 /// The type of a connected node.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum NodeType {
@@ -138,6 +141,9 @@ pub struct DistributionManager {
     /// Pending ping requests awaiting pong responses.
     /// Key is (node_atom, sequence_number), value is the oneshot sender.
     pending_pings: DashMap<(Atom, u64), oneshot::Sender<()>>,
+    /// Pending spawn requests awaiting replies.
+    /// Key is reference, value is the oneshot sender for (pid, monitor_ref, error).
+    pending_spawns: DashMap<u64, oneshot::Sender<SpawnReply>>,
 }
 
 impl DistributionManager {
@@ -157,6 +163,7 @@ impl DistributionManager {
             erlang_nodes: DashMap::new(),
             node_types: DashMap::new(),
             pending_pings: DashMap::new(),
+            pending_spawns: DashMap::new(),
         }
     }
 
@@ -476,6 +483,29 @@ impl DistributionManager {
         }
     }
 
+    /// Register a pending spawn request and return a receiver to await the reply.
+    pub(crate) fn register_spawn(
+        &self,
+        reference: u64,
+    ) -> oneshot::Receiver<(Option<Pid>, Option<u64>, Option<String>)> {
+        let (tx, rx) = oneshot::channel();
+        self.pending_spawns.insert(reference, tx);
+        rx
+    }
+
+    /// Complete a pending spawn request (called when SpawnReply is received).
+    pub(crate) fn complete_spawn(
+        &self,
+        reference: u64,
+        pid: Option<Pid>,
+        monitor_ref: Option<u64>,
+        error: Option<String>,
+    ) {
+        if let Some((_, tx)) = self.pending_spawns.remove(&reference) {
+            let _ = tx.send((pid, monitor_ref, error));
+        }
+    }
+
     /// Send a message to an Erlang/BEAM node.
     ///
     /// The payload should be ETF-encoded bytes. This method decodes them to
@@ -1061,6 +1091,56 @@ async fn handle_incoming_message(from_node: Atom, msg: DistMessage) {
             }
         }
 
+        DistMessage::Spawn {
+            reference,
+            from,
+            module,
+            function,
+            args,
+            link,
+            monitor,
+        } => {
+            // Handle spawn request from remote node
+            let link_to = if link { Some(from) } else { None };
+            let monitor_from = if monitor { Some(from) } else { None };
+
+            let result =
+                super::spawn::execute_spawn(&module, &function, args, link_to, monitor_from);
+
+            // Send reply back to the requesting node
+            if let Some(manager) = DIST_MANAGER.get()
+                && let Some(tx) = manager.get_node_tx(from_node)
+            {
+                let reply = match result {
+                    Ok((pid, monitor_ref)) => DistMessage::SpawnReply {
+                        reference,
+                        pid: Some(pid),
+                        monitor_ref: monitor_ref.map(|r| r.as_raw()),
+                        error: None,
+                    },
+                    Err(e) => DistMessage::SpawnReply {
+                        reference,
+                        pid: None,
+                        monitor_ref: None,
+                        error: Some(e),
+                    },
+                };
+                let _ = tx.try_send(reply);
+            }
+        }
+
+        DistMessage::SpawnReply {
+            reference,
+            pid,
+            monitor_ref,
+            error,
+        } => {
+            // Handle spawn reply - complete the pending spawn request
+            if let Some(manager) = DIST_MANAGER.get() {
+                manager.complete_spawn(reference, pid, monitor_ref, error);
+            }
+        }
+
         _ => {
             tracing::warn!(?msg, "Unexpected message type");
         }

crates/ambitious/src/distribution/mod.rs

@@ -43,6 +43,7 @@ mod node;
 pub mod pg;
 mod process_monitor;
 pub(crate) mod protocol;
+pub mod spawn;
 mod tcp_transport;
 mod traits;
 mod transport;

crates/ambitious/src/distribution/protocol.rs

@@ -142,6 +142,37 @@ pub enum DistMessage {
         /// Serialized PgMessage.
         payload: Vec<u8>,
     },
+
+    // === Remote Spawning ===
+    /// Request to spawn a process on this node.
+    Spawn {
+        /// Unique reference for this spawn request (for matching reply).
+        reference: u64,
+        /// The process requesting the spawn (will receive the reply).
+        from: Pid,
+        /// Module name (used as lookup key).
+        module: String,
+        /// Function name within the module.
+        function: String,
+        /// Serialized arguments for the function.
+        args: Vec<u8>,
+        /// Whether to link the spawned process to the requester.
+        link: bool,
+        /// Whether to monitor the spawned process.
+        monitor: bool,
+    },
+
+    /// Reply to a spawn request.
+    SpawnReply {
+        /// Reference from the spawn request.
+        reference: u64,
+        /// The spawned process PID (if successful).
+        pid: Option<Pid>,
+        /// Monitor reference (if monitor was requested and spawn succeeded).
+        monitor_ref: Option<u64>,
+        /// Error message (if spawn failed).
+        error: Option<String>,
+    },
 }
 
 impl DistMessage {