: v1/logging: avoid spinning up LogForwardActor mesh when redundant (#1722)

Summary:

i’m planning to add more tests (`set_mode` contract, `flush()` behavior, `drop` path) later in this stack. this diff gets the basic `spawn`/forwarding case under test and puts the plumbing in place: dedicated test bootstrap binary, resource wiring, `ensure_python()`, and the `AwaitPyExt` helper so we can drive `PyPythonTask` from rust tests. landing this first to help keep reviews tractable.

Differential Revision: D85919326

Author: shayne-fletcher

hyperactor_mesh/src/testresource.rs

@@ -27,15 +27,15 @@ where
     }
 
     assert!(
-        name.starts_with("monarch/hyperactor_mesh/"),
-        "invalid resource {}: must start with \"monarch/hyperactor_mesh/\"",
+        name.starts_with("monarch/monarch_hyperactor/"),
+        "invalid resource {}: must start with \"monarch/monarch_hyperactor/\"",
         name
     );
 
     let path: PathBuf = name
         .replace(
-            "monarch/hyperactor_mesh/",
-            "../target/debug/hyperactor_mesh_test_",
+            "monarch/monarch_hyperactor/",
+            "../target/debug/monarch_hyperactor_test_",
         )
         .into();
 

monarch_hyperactor/Cargo.toml

@@ -1,4 +1,4 @@
-# @generated by autocargo from //monarch/monarch_hyperactor:[monarch_hyperactor,process_allocator-oss,test_monarch_hyperactor]
+# @generated by autocargo from //monarch/monarch_hyperactor:[monarch_hyperactor,monarch_hyperactor_test_bootstrap,process_allocator-oss,test_monarch_hyperactor]
 
 [package]
 name = "monarch_hyperactor"
@@ -7,6 +7,11 @@ authors = ["Meta"]
 edition = "2021"
 license = "BSD-3-Clause"
 
+[[bin]]
+name = "monarch_hyperactor_test_bootstrap"
+path = "test/bootstrap.rs"
+edition = "2024"
+
 [[bin]]
 name = "process_allocator"
 edition = "2024"
@@ -52,6 +57,7 @@ tokio-util = { version = "0.7.15", features = ["full"] }
 tracing = { version = "0.1.41", features = ["attributes", "valuable"] }
 
 [dev-dependencies]
+buck-resources = "1"
 dir-diff = "0.3"
 
 [features]

monarch_hyperactor/src/pytokio.rs

@@ -50,6 +50,8 @@ use hyperactor::config::CONFIG;
 use hyperactor::config::ConfigAttr;
 use monarch_types::SerializablePyErr;
 use pyo3::IntoPyObjectExt;
+#[cfg(test)]
+use pyo3::PyClass;
 use pyo3::exceptions::PyRuntimeError;
 use pyo3::exceptions::PyStopIteration;
 use pyo3::exceptions::PyTimeoutError;
@@ -547,3 +549,52 @@ pub fn register_python_bindings(hyperactor_mod: &Bound<'_, PyModule>) -> PyResul
 
     Ok(())
 }
+
+/// Ensure the embedded Python interpreter is initialized exactly
+/// once.
+///
+/// Safe to call from multiple threads, multiple times.
+#[cfg(test)]
+pub(crate) fn ensure_python() {
+    static INIT: std::sync::OnceLock<()> = std::sync::OnceLock::new();
+    INIT.get_or_init(|| {
+        pyo3::prepare_freethreaded_python();
+    });
+}
+
+#[cfg(test)]
+// Helper: let us "await" a `PyPythonTask` in Rust.
+//
+// Semantics:
+//   - consume the `PyPythonTask`,
+//   - take the inner future,
+//   - `.await` it on tokio to get `Py<PyAny>`,
+//   - turn that into `Py<T>`.
+pub(crate) trait AwaitPyExt {
+    async fn await_py<T: PyClass>(self) -> Result<Py<T>, PyErr>;
+}
+
+#[cfg(test)]
+impl AwaitPyExt for PyPythonTask {
+    async fn await_py<T: PyClass>(mut self) -> Result<Py<T>, PyErr> {
+        // Take ownership of the inner future.
+        let fut = self
+            .take_task()
+            .expect("PyPythonTask already consumed in await_py");
+
+        // Await a Result<Py<PyAny>, PyErr>.
+        let py_any: Py<PyAny> = fut.await?;
+
+        // Convert Py<PyAny> -> Py<T>.
+        Python::with_gil(|py| {
+            let bound_any = py_any.bind(py);
+
+            // Try extract a Py<T>.
+            let obj: Py<T> = bound_any
+                .extract::<Py<T>>()
+                .expect("spawn() did not return expected Python type");
+
+            Ok(obj)
+        })
+    }
+}