Skip to main content

monarch_hyperactor/
host_mesh.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9use std::collections::HashMap;
10use std::ops::Deref;
11use std::path::PathBuf;
12use std::sync::OnceLock;
13use std::time::Duration;
14
15use hyperactor::ActorHandle;
16use hyperactor::Endpoint as _;
17use hyperactor::Gateway;
18use hyperactor::Instance;
19use hyperactor::Proc;
20use hyperactor::channel::ChannelAddr;
21use hyperactor::id::Label;
22use hyperactor_mesh::ProcMeshRef;
23use hyperactor_mesh::bootstrap::BootstrapCommand;
24use hyperactor_mesh::bootstrap::ProcBind;
25use hyperactor_mesh::bootstrap::host;
26use hyperactor_mesh::host_mesh;
27use hyperactor_mesh::host_mesh::HostMesh;
28use hyperactor_mesh::host_mesh::HostMeshRef;
29use hyperactor_mesh::host_mesh::PerRankBootstrapFn;
30use hyperactor_mesh::host_mesh::host_agent::GetLocalProcClient;
31use hyperactor_mesh::host_mesh::host_agent::HostAgent;
32use hyperactor_mesh::host_mesh::host_agent::ShutdownHost;
33use hyperactor_mesh::mesh_admin::MeshAdminMessageClient;
34use hyperactor_mesh::mesh_id::ActorMeshId;
35use hyperactor_mesh::mesh_id::HostMeshId;
36use hyperactor_mesh::mesh_id::ProcMeshId;
37use hyperactor_mesh::proc_agent::GetProcClient;
38use hyperactor_mesh::proc_mesh::ProcRef;
39use hyperactor_mesh::proc_mesh::telemetry_actor_mesh_id;
40use hyperactor_mesh::shared_cell::SharedCell;
41use hyperactor_mesh::transport::default_bind_spec;
42use hyperactor_telemetry::hash_to_u64;
43use ndslice::View;
44use ndslice::view::RankedSliceable;
45use pyo3::IntoPyObjectExt;
46use pyo3::exceptions::PyException;
47use pyo3::exceptions::PyRuntimeError;
48use pyo3::exceptions::PyValueError;
49use pyo3::prelude::*;
50use pyo3::types::PyBytes;
51
52use crate::actor::PythonActor;
53use crate::actor::to_py_error;
54use crate::context::PyInstance;
55use crate::proc_mesh::PyProcMesh;
56use crate::pytokio::PyPythonTask;
57use crate::runtime::GilSite;
58use crate::runtime::monarch_with_gil;
59use crate::runtime::monarch_with_gil_blocking;
60use crate::shape::PyExtent;
61use crate::shape::PyPoint;
62use crate::shape::PyRegion;
63
64#[pyclass(
65    name = "BootstrapCommand",
66    module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
67)]
68#[derive(Clone)]
69pub struct PyBootstrapCommand {
70    #[pyo3(get, set)]
71    pub program: String,
72    #[pyo3(get, set)]
73    pub arg0: Option<String>,
74    #[pyo3(get, set)]
75    pub args: Vec<String>,
76    #[pyo3(get, set)]
77    pub env: HashMap<String, String>,
78}
79
80#[pymethods]
81impl PyBootstrapCommand {
82    #[new]
83    fn new(
84        program: String,
85        arg0: Option<String>,
86        args: Vec<String>,
87        env: HashMap<String, String>,
88    ) -> Self {
89        Self {
90            program,
91            arg0,
92            args,
93            env,
94        }
95    }
96
97    fn __repr__(&self) -> String {
98        format!(
99            "BootstrapCommand(program='{}', args={:?}, env={:?})",
100            self.program, self.args, self.env
101        )
102    }
103
104    /// Return a copy of this command with `env` merged on top of its
105    /// environment. Keys in `env` override any conflicting keys in the
106    /// existing environment.
107    fn with_env(&self, env: HashMap<String, String>) -> Self {
108        let mut new_env = self.env.clone();
109        new_env.extend(env);
110        Self {
111            program: self.program.clone(),
112            arg0: self.arg0.clone(),
113            args: self.args.clone(),
114            env: new_env,
115        }
116    }
117}
118
119impl PyBootstrapCommand {
120    pub fn to_rust(&self) -> BootstrapCommand {
121        BootstrapCommand {
122            program: PathBuf::from(&self.program),
123            arg0: self.arg0.clone(),
124            args: self.args.clone(),
125            env: self.env.clone(),
126        }
127    }
128}
129
130#[pyclass(
131    name = "HostMesh",
132    module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
133)]
134#[expect(
135    clippy::large_enum_variant,
136    reason = "PyO3 #[pyclass] enum; Box wrapping interacts with PyO3 codegen and Python interop — separate diff"
137)]
138pub(crate) enum PyHostMesh {
139    Owned(PyHostMeshImpl),
140    Ref(PyHostMeshRefImpl),
141}
142
143impl PyHostMesh {
144    pub(crate) fn new_owned(inner: HostMesh) -> Self {
145        Self::Owned(PyHostMeshImpl(SharedCell::from(inner)))
146    }
147
148    pub(crate) fn new_ref(inner: HostMeshRef) -> Self {
149        Self::Ref(PyHostMeshRefImpl(inner))
150    }
151
152    pub(crate) fn mesh_ref(&self) -> Result<HostMeshRef, anyhow::Error> {
153        match self {
154            PyHostMesh::Owned(inner) => Ok(inner.0.borrow()?.clone()),
155            PyHostMesh::Ref(inner) => Ok(inner.0.clone()),
156        }
157    }
158}
159
160#[pymethods]
161impl PyHostMesh {
162    #[pyo3(signature = (instance, name, per_host, proc_bind = None, per_rank_bootstrap = None))]
163    fn spawn_nonblocking(
164        &self,
165        _py: Python<'_>,
166        instance: &PyInstance,
167        name: String,
168        per_host: &PyExtent,
169        proc_bind: Option<Vec<HashMap<String, String>>>,
170        per_rank_bootstrap: Option<Py<PyAny>>,
171    ) -> PyResult<PyPythonTask> {
172        let host_mesh = self.mesh_ref()?.clone();
173        let per_rank_bootstrap: Option<Box<PerRankBootstrapFn>> = per_rank_bootstrap
174            .map(|callable| -> PyResult<Box<PerRankBootstrapFn>> {
175                Ok(Box::new(move |point| {
176                    monarch_with_gil_blocking(GilSite::Bootstrap, |py| {
177                        let result =
178                            callable
179                                .bind(py)
180                                .call1((PyPoint::from(point),))
181                                .map_err(|e| {
182                                    anyhow::anyhow!("per-rank bootstrap callable raised: {}", e)
183                                })?;
184                        let cmd: PyBootstrapCommand = result.extract().map_err(|e| {
185                            anyhow::anyhow!(
186                                "per-rank bootstrap callable did not return BootstrapCommand: {}",
187                                e
188                            )
189                        })?;
190                        Ok(cmd.to_rust())
191                    })
192                }))
193            })
194            .transpose()?;
195        let instance = instance.clone();
196        let per_host = per_host.clone().into();
197        let proc_bind = proc_bind.map(|v| v.into_iter().map(ProcBind::from).collect());
198        let mesh_impl = async move {
199            let proc_mesh = host_mesh
200                .spawn(
201                    instance.deref(),
202                    &name,
203                    per_host,
204                    proc_bind,
205                    per_rank_bootstrap,
206                )
207                .await
208                .map_err(to_py_error)?;
209            Ok(PyProcMesh::new_owned(proc_mesh))
210        };
211        PyPythonTask::new(mesh_impl)
212    }
213
214    fn with_bootstrap(&self, bootstrap_command: &PyBootstrapCommand) -> PyResult<Self> {
215        match self {
216            PyHostMesh::Owned(inner) => {
217                let cmd = bootstrap_command.to_rust();
218                inner
219                    .0
220                    .try_with_mut(|mesh| mesh.set_bootstrap(cmd))
221                    .map_err(|e| PyException::new_err(e.to_string()))?;
222                Ok(Self::Owned(inner.clone()))
223            }
224            PyHostMesh::Ref(_) => Ok(Self::new_ref(
225                self.mesh_ref()?.with_bootstrap(bootstrap_command.to_rust()),
226            )),
227        }
228    }
229
230    fn sliced(&self, region: &PyRegion) -> PyResult<Self> {
231        Ok(Self::new_ref(
232            self.mesh_ref()?.sliced(region.as_inner().clone()),
233        ))
234    }
235
236    #[getter]
237    fn region(&self) -> PyResult<PyRegion> {
238        Ok(PyRegion::from(self.mesh_ref()?.region()))
239    }
240
241    fn __reduce__<'py>(&self, py: Python<'py>) -> PyResult<(Bound<'py, PyAny>, Bound<'py, PyAny>)> {
242        let mesh_ref = self.mesh_ref()?;
243        if crate::pickle::push_mesh_reference_if_active(crate::actor::MeshRef::Host(Box::new(
244            mesh_ref.clone(),
245        ))) {
246            let pop_fn = PyModule::import(py, "monarch._rust_bindings.monarch_hyperactor.pickle")?
247                .getattr("pop_mesh_reference")?;
248            return Ok((pop_fn, pyo3::types::PyTuple::empty(py).into_any()));
249        }
250        let bytes = bincode::serde::encode_to_vec(&mesh_ref, bincode::config::legacy())
251            .map_err(|e| PyErr::new::<PyValueError, _>(e.to_string()))?;
252        let py_bytes = (PyBytes::new(py, &bytes),).into_bound_py_any(py).unwrap();
253        let from_bytes =
254            PyModule::import(py, "monarch._rust_bindings.monarch_hyperactor.host_mesh")?
255                .getattr("py_host_mesh_from_bytes")?;
256        Ok((from_bytes, py_bytes))
257    }
258
259    fn __eq__(&self, other: &PyHostMesh) -> PyResult<bool> {
260        Ok(self.mesh_ref()? == other.mesh_ref()?)
261    }
262
263    fn shutdown(&self, instance: &PyInstance) -> PyResult<PyPythonTask> {
264        match self {
265            PyHostMesh::Owned(inner) => {
266                let instance = instance.clone();
267                let mesh_borrow = inner.0.clone();
268                let fut = async move {
269                    match mesh_borrow.take().await {
270                        Ok(mut mesh) => {
271                            mesh.shutdown(instance.deref()).await?;
272                            Ok(())
273                        }
274                        Err(_) => {
275                            // Don't return an exception, silently ignore the stop request
276                            // because it was already done.
277                            tracing::info!("shutdown was already called on host mesh");
278                            Ok(())
279                        }
280                    }
281                };
282                PyPythonTask::new(fut)
283            }
284            PyHostMesh::Ref(_) => Err(PyRuntimeError::new_err(
285                "cannot shut down `HostMesh` that is a reference instead of owned",
286            )),
287        }
288    }
289
290    fn stop(&self, instance: &PyInstance) -> PyResult<PyPythonTask> {
291        match self {
292            PyHostMesh::Owned(inner) => {
293                let instance = instance.clone();
294                let mesh_borrow = inner.0.clone();
295                let fut = async move {
296                    match mesh_borrow.take().await {
297                        Ok(mut mesh) => {
298                            mesh.stop(instance.deref()).await?;
299                            Ok(())
300                        }
301                        Err(_) => {
302                            tracing::info!("stop was already called on host mesh");
303                            Ok(())
304                        }
305                    }
306                };
307                PyPythonTask::new(fut)
308            }
309            PyHostMesh::Ref(_) => Err(PyRuntimeError::new_err(
310                "cannot stop `HostMesh` that is a reference instead of owned",
311            )),
312        }
313    }
314}
315
316#[derive(Clone)]
317#[pyclass(
318    name = "HostMeshImpl",
319    module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
320)]
321pub(crate) struct PyHostMeshImpl(SharedCell<HostMesh>);
322
323#[derive(Debug, Clone)]
324#[pyclass(
325    name = "HostMeshRefImpl",
326    module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
327)]
328pub(crate) struct PyHostMeshRefImpl(HostMeshRef);
329
330impl PyHostMeshRefImpl {
331    fn __repr__(&self) -> PyResult<String> {
332        Ok(format!("<HostMeshRefImpl {:?}>", self.0))
333    }
334}
335
336/// Static storage for the root client instance when using host-based bootstrap.
337static ROOT_CLIENT_INSTANCE_FOR_HOST: OnceLock<Instance<PythonActor>> = OnceLock::new();
338
339/// Static storage for the host mesh agent created by bootstrap_host().
340static HOST_MESH_AGENT_FOR_HOST: OnceLock<ActorHandle<HostAgent>> = OnceLock::new();
341
342/// Static storage for the host shutdown handle created by bootstrap_host().
343/// Used during shutdown_context to join the mailbox server and flush
344/// receive-side acks.
345static HOST_SHUTDOWN_HANDLE: OnceLock<
346    tokio::sync::Mutex<Option<hyperactor_mesh::bootstrap::HostShutdownHandle>>,
347> = OnceLock::new();
348
349/// Bootstrap the client host and root client actor.
350///
351/// This creates a proper Host with BootstrapProcManager, spawns the root client
352/// actor on the Host's local_proc.
353///
354/// Returns a tuple of (HostMesh, ProcMesh, PyInstance) where:
355/// - PyHostMesh: the bootstrapped (local) host mesh; and
356/// - PyProcMesh: the local ProcMesh on this HostMesh; and
357/// - PyInstance: the root client actor instance, on the ProcMesh.
358///
359/// The HostMesh is served on the default transport.
360///
361/// If ``via`` is set to a ZMQ-style address of a remote host's duplex
362/// server, the local host's gateway is attached to that remote
363/// gateway: outbound traffic to unknown destinations is forwarded over
364/// the duplex, and inbound traffic from the duplex is delivered to
365/// local procs by the gateway's routing. The local host still owns
366/// its own frontend address; ``via`` only adds a forwarding path. The
367/// returned ``PyHostMesh`` / ``PyProcMesh`` / ``PyInstance`` all live
368/// on the local host's procs.
369///
370/// This should be called only once, at process initialization.
371#[pyfunction]
372#[pyo3(signature = (bootstrap_cmd, via=None))]
373fn bootstrap_host(
374    bootstrap_cmd: Option<PyBootstrapCommand>,
375    via: Option<&str>,
376) -> PyResult<PyPythonTask> {
377    let bootstrap_cmd = match bootstrap_cmd {
378        Some(cmd) => cmd.to_rust(),
379        None => BootstrapCommand::current().map_err(|e| PyException::new_err(e.to_string()))?,
380    };
381    let via_addr = via
382        .map(|s| {
383            ChannelAddr::from_zmq_url(s)
384                .map_err(|e| PyValueError::new_err(format!("via address: {}", e)))
385        })
386        .transpose()?;
387
388    PyPythonTask::new(async move {
389        // Pass the via address into `host` so the gateway's `serve_via`
390        // attach runs *after* the host's own frontend serve but *before*
391        // any proc or actor ref is minted. That ordering makes the via
392        // session the newest active serve, so refs advertise the
393        // routable `Via` location and out-of-cluster return traffic
394        // flows back over the duplex. The host owns the resulting serve
395        // handle and tears it down on drop.
396        let gateway = Gateway::global().clone();
397
398        let (host_mesh_agent, shutdown_handle) = host(
399            default_bind_spec().binding_addr(),
400            Some(bootstrap_cmd),
401            None,
402            false,
403            None,
404            gateway,
405            via_addr,
406        )
407        .await
408        .map_err(|e| PyException::new_err(e.to_string()))?;
409
410        // Store the agent and shutdown handle for later shutdown
411        HOST_MESH_AGENT_FOR_HOST.set(host_mesh_agent.clone()).ok();
412        HOST_SHUTDOWN_HANDLE.get_or_init(|| tokio::sync::Mutex::new(Some(shutdown_handle)));
413
414        let host_mesh_id = HostMeshId::singleton(Label::new("local").unwrap());
415        let host_mesh = HostMeshRef::from_host_agent(host_mesh_id, host_mesh_agent.bind())
416            .map_err(|e| PyException::new_err(e.to_string()))?;
417
418        // Register C so MeshAdminAgent can discover it ("A/C
419        // invariant" - hyperactor_mesh/src/mesh_admin.rs).
420        hyperactor_mesh::global_context::register_client_host(host_mesh.clone());
421
422        // We require a temporary instance to make a call to the host/proc agent.
423        let temp_proc = Proc::isolated();
424        let temp_instance = temp_proc.client("temp");
425
426        let local_proc_agent: hyperactor::ActorHandle<hyperactor_mesh::proc_agent::ProcAgent> =
427            host_mesh_agent
428                .get_local_proc(&temp_instance)
429                .await
430                .map_err(|e| PyException::new_err(e.to_string()))?;
431
432        let proc_mesh = ProcMeshRef::new_singleton(
433            ProcMeshId::singleton(Label::new("local").unwrap()),
434            ProcRef::new(
435                local_proc_agent.actor_addr().proc_addr(),
436                0,
437                local_proc_agent.bind(),
438            ),
439        )
440        .map_err(|e| PyException::new_err(e.to_string()))?;
441
442        let local_proc = local_proc_agent
443            .get_proc(&temp_instance)
444            .await
445            .map_err(|e| PyException::new_err(e.to_string()))?;
446
447        let (instance, _handle) = monarch_with_gil(GilSite::Bootstrap, |py| {
448            PythonActor::bootstrap_client_inner(py, local_proc, &ROOT_CLIENT_INSTANCE_FOR_HOST)
449        })
450        .await;
451
452        // Notify telemetry of the bootstrap host mesh, proc mesh, and client actor.
453        {
454            let now = std::time::SystemTime::now();
455
456            let host_name_str = host_mesh.id().to_string();
457            let host_mesh_id = hash_to_u64(host_mesh.id());
458            hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
459                id: host_mesh_id,
460                timestamp: now,
461                class: "Host".to_string(),
462                given_name: host_mesh
463                    .id()
464                    .display_label()
465                    .map(|l| l.as_str())
466                    .unwrap_or("unnamed")
467                    .to_string(),
468                full_name: host_name_str,
469                shape_json: serde_json::to_string(&host_mesh.region().extent()).unwrap_or_default(),
470                parent_mesh_id: None,
471                parent_view_json: None,
472            });
473
474            let host_agent_addr = host_mesh_agent.actor_addr();
475            hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
476                id: hyperactor_telemetry::hash_to_u64(host_agent_addr.id()),
477                timestamp: now,
478                mesh_id: host_mesh_id,
479                rank: 0,
480                full_name: host_agent_addr.to_string(),
481                display_name: None,
482            });
483
484            let proc_id_str = proc_mesh.id().to_string();
485            let proc_mesh_id = hash_to_u64(proc_mesh.id());
486            hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
487                id: proc_mesh_id,
488                timestamp: now,
489                class: "Proc".to_string(),
490                given_name: proc_mesh
491                    .id()
492                    .display_label()
493                    .map(|l| l.as_str())
494                    .unwrap_or("unnamed")
495                    .to_string(),
496                full_name: proc_id_str,
497                shape_json: serde_json::to_string(&proc_mesh.region().extent()).unwrap_or_default(),
498                parent_mesh_id: Some(host_mesh_id),
499                parent_view_json: None,
500            });
501
502            let proc_agent_addr = local_proc_agent.actor_addr();
503            hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
504                id: hyperactor_telemetry::hash_to_u64(proc_agent_addr.id()),
505                timestamp: now,
506                mesh_id: proc_mesh_id,
507                rank: 0,
508                full_name: proc_agent_addr.to_string(),
509                display_name: None,
510            });
511
512            let client_mesh_actor_id = ActorMeshId::singleton(Label::new("client").unwrap());
513            let client_mesh_name = format!("{}/client", proc_mesh.id());
514            let client_mesh_id = telemetry_actor_mesh_id(proc_mesh.id(), &client_mesh_actor_id);
515            hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
516                id: client_mesh_id,
517                timestamp: now,
518                class: <PythonActor as typeuri::Named>::typename().to_string(),
519                given_name: "client".to_string(),
520                full_name: client_mesh_name,
521                shape_json: serde_json::to_string(&proc_mesh.region().extent()).unwrap_or_default(),
522                parent_mesh_id: Some(proc_mesh_id),
523                parent_view_json: None,
524            });
525
526            hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
527                id: hyperactor_telemetry::hash_to_u64(instance.self_addr().id()),
528                timestamp: now,
529                mesh_id: client_mesh_id,
530                rank: 0,
531                full_name: instance.self_addr().to_string(),
532                display_name: Some("<root>".to_string()),
533            });
534        }
535
536        Ok((
537            PyHostMesh::new_ref(host_mesh),
538            PyProcMesh::new_ref(proc_mesh),
539            PyInstance::from(instance),
540        ))
541    })
542}
543
544#[pyfunction]
545fn py_host_mesh_from_bytes(bytes: &Bound<'_, PyBytes>) -> PyResult<PyHostMesh> {
546    let r: PyResult<HostMeshRef> =
547        bincode::serde::decode_from_slice(bytes.as_bytes(), bincode::config::legacy())
548            .map(|(v, _)| v)
549            .map_err(|e| PyErr::new::<PyValueError, _>(e.to_string()));
550    r.map(PyHostMesh::new_ref)
551}
552
553#[pyfunction]
554fn shutdown_local_host_mesh() -> PyResult<PyPythonTask> {
555    let agent = HOST_MESH_AGENT_FOR_HOST
556        .get()
557        .ok_or_else(|| PyException::new_err("No local host mesh to shutdown"))?
558        .clone();
559
560    PyPythonTask::new(async move {
561        // Create a temporary instance to send the shutdown message
562        let temp_proc = hyperactor::Proc::isolated();
563        let instance = temp_proc.client("shutdown_requester");
564
565        tracing::info!(
566            "sending shutdown_host request to agent {}",
567            agent.actor_addr()
568        );
569        // Use same defaults as HostMesh::shutdown():
570        // - MESH_TERMINATE_TIMEOUT = 10 seconds
571        // - MESH_TERMINATE_CONCURRENCY = 16
572
573        let (port, _) = instance.open_port::<usize>();
574        let mut port = port.bind();
575        // We don't need the ack, and this temporary proc doesn't have a mailbox
576        // receiver set up anyways. Just ignore the message.
577        port.return_undeliverable(false);
578        agent.post(
579            &instance,
580            ShutdownHost {
581                timeout: Duration::from_secs(10),
582                max_in_flight: 16,
583                rank: hyperactor_mesh::resource::Rank::new(0),
584                ack: port,
585            },
586        );
587
588        // Join the host's mailbox server to flush receive-side acks
589        // before the process exits.
590        if let Some(lock) = HOST_SHUTDOWN_HANDLE.get()
591            && let Some(handle) = lock.lock().await.take()
592        {
593            handle.join().await;
594        }
595
596        Ok(())
597    })
598}
599
600/// Opaque capability token for `ActorRef<MeshAdminAgent>` across the
601/// Python boundary. No methods, no getters — Python never inspects
602/// this. It exists solely to transport the typed ref from
603/// `_spawn_admin` to `_start_periodic_snapshots_http`.
604#[pyclass(
605    name = "PyMeshAdminRef",
606    module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
607)]
608#[derive(Clone)]
609pub struct PyMeshAdminRef(hyperactor::ActorRef<hyperactor_mesh::mesh_admin::MeshAdminAgent>);
610
611impl PyMeshAdminRef {
612    pub fn actor_ref(&self) -> hyperactor::ActorRef<hyperactor_mesh::mesh_admin::MeshAdminAgent> {
613        self.0.clone()
614    }
615}
616
617/// `MESH_ADMIN_ADDR` config.
618///
619/// Python-facing wrapper around
620/// [`hyperactor_mesh::host_mesh::spawn_admin`].
621#[pyfunction]
622fn _spawn_admin(
623    host_meshes: Vec<PyRef<'_, PyHostMesh>>,
624    instance: &PyInstance,
625    admin_addr: Option<String>,
626    telemetry_url: Option<String>,
627) -> PyResult<PyPythonTask> {
628    if host_meshes.is_empty() {
629        return Err(PyException::new_err("at least one mesh is required"));
630    }
631
632    let admin_addr = admin_addr
633        .map(|s| {
634            s.parse::<std::net::SocketAddr>()
635                .map_err(|e| PyException::new_err(format!("invalid admin_addr '{}': {}", s, e)))
636        })
637        .transpose()?;
638
639    let mesh_refs = host_meshes
640        .iter()
641        .map(|m| -> PyResult<HostMeshRef> { Ok(m.mesh_ref()?.clone()) })
642        .collect::<PyResult<Vec<HostMeshRef>>>()?;
643
644    let instance = instance.clone();
645    PyPythonTask::new(async move {
646        let admin_ref =
647            host_mesh::spawn_admin(&mesh_refs, instance.deref(), admin_addr, telemetry_url)
648                .await
649                .map_err(|e| PyException::new_err(e.to_string()))?;
650        let admin_url = admin_ref
651            .get_admin_addr(instance.deref())
652            .await
653            .map_err(|e| PyException::new_err(e.to_string()))?
654            .addr
655            .ok_or_else(|| PyException::new_err("mesh admin agent did not report an address"))?;
656        Ok((admin_url, PyMeshAdminRef(admin_ref)))
657    })
658}
659
660pub fn register_python_bindings(hyperactor_mod: &Bound<'_, PyModule>) -> PyResult<()> {
661    let f = wrap_pyfunction!(py_host_mesh_from_bytes, hyperactor_mod)?;
662    f.setattr(
663        "__module__",
664        "monarch._rust_bindings.monarch_hyperactor.host_mesh",
665    )?;
666    hyperactor_mod.add_function(f)?;
667
668    let f2 = wrap_pyfunction!(bootstrap_host, hyperactor_mod)?;
669    f2.setattr(
670        "__module__",
671        "monarch._rust_bindings.monarch_hyperactor.host_mesh",
672    )?;
673    hyperactor_mod.add_function(f2)?;
674
675    let f3 = wrap_pyfunction!(shutdown_local_host_mesh, hyperactor_mod)?;
676    f3.setattr(
677        "__module__",
678        "monarch._rust_bindings.monarch_hyperactor.host_mesh",
679    )?;
680    hyperactor_mod.add_function(f3)?;
681
682    let f4 = wrap_pyfunction!(_spawn_admin, hyperactor_mod)?;
683    f4.setattr(
684        "__module__",
685        "monarch._rust_bindings.monarch_hyperactor.host_mesh",
686    )?;
687    hyperactor_mod.add_function(f4)?;
688
689    hyperactor_mod.add_class::<PyHostMesh>()?;
690    hyperactor_mod.add_class::<PyBootstrapCommand>()?;
691    hyperactor_mod.add_class::<PyMeshAdminRef>()?;
692    Ok(())
693}