[monarch_hyperactor] propagate proc status in supervision events for proc failures

Currently, we report the agent to be Stopped. This is accurate, but confusing, and could be better attributed.

Here, we synthesize an actor failure by:

1) attributing the fault to the corresponding actor in the monitored actor mesh;
2) elevating the proc_status (which contains mode of failure, exit code, etc) into the actor failure, making it clear it is a process failure

In the future:
1) We will have a more general "Failure" struct, explicitly capturing host, proc, actor, etc., failures.
2) We will attribute the actor failure (it is the most proximate), but move the proc failure to the "cause" (i.e., proc failure caused actor to fail), which is the most correct and clear.

Differential Revision: [D86993889](https://our.internmc.facebook.com/intern/diff/D86993889/)

**NOTE FOR REVIEWERS**: This PR has internal Meta-specific changes or comments, please review them on [Phabricator](https://our.internmc.facebook.com/intern/diff/D86993889/)!

ghstack-source-id: 323095346
Pull Request resolved: #1877

Author: mariusae

hyperactor_mesh/src/v1/host_mesh/mesh_agent.rs

@@ -355,7 +355,7 @@ impl Handler<resource::GetState<ProcState>> for HostMeshAgent {
         cx: &Context<Self>,
         get_state: resource::GetState<ProcState>,
     ) -> anyhow::Result<()> {
-        let manager = self
+        let manager: Option<&BootstrapProcManager> = self
             .host
             .as_mut()
             .expect("host")

monarch_hyperactor/src/v1/actor_mesh.rs

@@ -15,13 +15,15 @@ use hyperactor::Actor;
 use hyperactor::ActorHandle;
 use hyperactor::ActorRef;
 use hyperactor::RemoteMessage;
+use hyperactor::actor::ActorErrorKind;
 use hyperactor::actor::ActorStatus;
 use hyperactor::actor::Referable;
 use hyperactor::actor::RemotableActor;
 use hyperactor::clock::Clock;
 use hyperactor::clock::RealClock;
 use hyperactor::context;
 use hyperactor::supervision::ActorSupervisionEvent;
+use hyperactor_mesh::bootstrap::ProcStatus;
 use hyperactor_mesh::dashmap::DashMap;
 use hyperactor_mesh::proc_mesh::mesh_agent::ActorState;
 use hyperactor_mesh::resource;
@@ -502,7 +504,7 @@ async fn actor_states_monitor<A, F>(
                 ActorSupervisionEvent::new(
                     cx.instance().self_id().clone(),
                     ActorStatus::generic_failure(format!(
-                        "Unable to query for proc states: {:?}",
+                        "unable to query for proc states: {:?}",
                         e
                     )),
                     None,
@@ -520,18 +522,27 @@ async fn actor_states_monitor<A, F>(
         }
         if let Some(proc_states) = proc_states.unwrap() {
             // Check if the proc mesh is still alive.
-            if let Some((rank, state)) = proc_states
+            if let Some((point, state)) = proc_states
                 .iter()
                 .find(|(_rank, state)| state.status.is_terminating())
             {
                 send_state_change(
-                    rank.rank(),
+                    point.rank(),
                     ActorSupervisionEvent::new(
-                        state
-                            .state
-                            .map(|s| s.mesh_agent.actor_id().clone())
-                            .unwrap_or(cx.instance().self_id().clone()),
-                        ActorStatus::Stopped,
+                        // Attribute this to the monitored actor, even if the underlying
+                        // cause is a proc_failure. We propagate the cause explicitly.
+                        mesh.get(point.rank()).unwrap().actor_id().clone(),
+                        // TODO: allow "actor supervision event" to be general, and
+                        // make the proc failure the cause.
+                        ActorStatus::Failed(ActorErrorKind::Generic(format!(
+                            "process failure: {}",
+                            state
+                                .state
+                                .and_then(|state| state.proc_status)
+                                .unwrap_or_else(|| ProcStatus::Failed {
+                                    reason: "unknown".to_string()
+                                })
+                        ))),
                         None,
                         None,
                     ),
@@ -555,7 +566,7 @@ async fn actor_states_monitor<A, F>(
                 ActorSupervisionEvent::new(
                     cx.instance().self_id().clone(),
                     ActorStatus::generic_failure(format!(
-                        "Unable to query for actor states: {:?}",
+                        "unable to query for actor states: {:?}",
                         e
                     )),
                     None,