hyperactor_mesh/

proc_mesh.rs

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 * All rights reserved.
 *
 * This source code is licensed under the BSD-style license found in the
 * LICENSE file in the root directory of this source tree.
 */

use std::any::type_name;
use std::collections::HashMap;
use std::collections::HashSet;
use std::fmt;
use std::hash::Hash;
use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;

use hyperactor::ActorAddr;
use hyperactor::ActorId;
use hyperactor::ActorRef;
use hyperactor::Endpoint as _;
use hyperactor::Handler;
use hyperactor::ProcAddr;
use hyperactor::RemoteMessage;
use hyperactor::RemoteSpawn;
use hyperactor::accum::IdleFlushReducerOpts;
use hyperactor::actor::ActorStatus;
use hyperactor::actor::remote::Remote;
use hyperactor::context;
use hyperactor::id::Label;
use hyperactor::supervision::ActorSupervisionEvent;
use hyperactor_config::CONFIG;
use hyperactor_config::ConfigAttr;
use hyperactor_config::Flattrs;
use hyperactor_config::NonZeroUsize;
use hyperactor_config::attrs::declare_attrs;
use hyperactor_remote::ActorSpawner;
use hyperactor_remote::ActorSpawnerEndpoint;
use hyperactor_remote::KeepaliveLink;
use hyperactor_remote::proc_spawner::actor_spawner_uid;
use hyperactor_telemetry::hash_to_u64;
use ndslice::Extent;
use ndslice::ViewExt as _;
use ndslice::view;
use ndslice::view::CollectMeshExt;
use ndslice::view::Ranked;
use ndslice::view::Region;
use serde::Deserialize;
use serde::Serialize;
use typeuri::Named;

use crate::ActorMesh;
use crate::ActorMeshRef;
use crate::Error;
use crate::HostMeshRef;
use crate::ValueMesh;
use crate::actor_mesh::ACTOR_MESH_ID;
use crate::actor_mesh::ActorMeshStopHandle;
use crate::casting::CAST_POINT;
use crate::host_mesh::GET_PROC_STATE_MAX_IDLE;
use crate::host_mesh::host_agent::GetHostProcStates;
use crate::host_mesh::host_agent::ProcState;
use crate::host_mesh::mesh_to_rankedvalues_with_default;
use crate::mesh_controller::ActorMeshControlPlane;
use crate::mesh_controller::ActorMeshController;
use crate::mesh_id::ActorMeshId;
use crate::mesh_id::ProcMeshId;
use crate::mesh_id::ResourceId;
use crate::proc_agent;
use crate::proc_agent::ActorState;
use crate::proc_agent::ProcAgent;
use crate::resource;
use crate::resource::GetRankStatus;
use crate::resource::Status;
use crate::supervision::MeshFailure;

declare_attrs! {
    /// Whether ordinary actor mesh spawns use the direct spawn path.
    @meta(CONFIG = ConfigAttr::new(
        Some("HYPERACTOR_MESH_USE_DIRECT_SPAWN".to_string()),
        Some("use_direct_spawn".to_string()),
    ))
    pub attr USE_DIRECT_SPAWN: bool = false;

    /// The maximum idle time between updates while spawning actor
    /// meshes.
    @meta(CONFIG = ConfigAttr::new(
        Some("HYPERACTOR_MESH_ACTOR_SPAWN_MAX_IDLE".to_string()),
        Some("actor_spawn_max_idle".to_string()),
    ))
    pub attr ACTOR_SPAWN_MAX_IDLE: Duration = Duration::from_secs(30);

    /// The maximum idle time between updates while waiting for a response to GetState
    /// from ProcAgent.
    @meta(CONFIG = ConfigAttr::new(
        Some("HYPERACTOR_MESH_GET_ACTOR_STATE_MAX_IDLE".to_string()),
        Some("get_actor_state_max_idle".to_string()),
    ))
    pub attr GET_ACTOR_STATE_MAX_IDLE: Duration = Duration::from_secs(30);
}

/// Returns the telemetry `meshes.id` value for an actor mesh.
pub fn telemetry_actor_mesh_id(proc_mesh_id: &ProcMeshId, actor_mesh_id: &ActorMeshId) -> u64 {
    hash_to_u64(&(proc_mesh_id, actor_mesh_id))
}

/// A reference to a single [`hyperactor::Proc`].
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct ProcRef {
    proc_id: ProcAddr,
    /// The rank of this proc at creation.
    create_rank: usize,
    /// The agent managing this proc.
    agent: ActorRef<ProcAgent>,
}

impl ProcRef {
    /// Create a new proc ref from the provided id, create rank and agent.
    pub fn new(proc_id: ProcAddr, create_rank: usize, agent: ActorRef<ProcAgent>) -> Self {
        Self {
            proc_id,
            create_rank,
            agent,
        }
    }

    pub fn proc_addr(&self) -> &ProcAddr {
        &self.proc_id
    }

    pub(crate) fn actor_addr(&self, id: &ActorMeshId) -> ActorAddr {
        self.proc_id.actor_addr_uid(id.uid().clone())
    }

    /// Return the actor spawner hosted by this proc.
    pub(crate) fn spawner(&self) -> ActorRef<ActorSpawner> {
        ActorRef::attest(self.proc_id.actor_addr_uid(actor_spawner_uid()))
    }
}

/// A mesh of processes.
#[derive(Debug)]
pub struct ProcMesh {
    #[allow(dead_code)]
    id: ProcMeshId,
    current_ref: ProcMeshRef,
    controller: Option<ActorRef<crate::mesh_controller::ProcMeshController>>,
}

enum SpawnedActorMesh<A: RemoteSpawn> {
    Managed {
        mesh: ActorMesh<A>,
        statuses: crate::StatusMesh,
    },
    Data(ActorMesh<A>),
}

impl ProcMesh {
    pub(crate) fn create(
        id: ProcMeshId,
        extent: Extent,
        hosts: HostMeshRef,
        ranks: Vec<ProcRef>,
    ) -> crate::Result<Self> {
        let region = extent.into();
        let ranks = Arc::new(ranks);

        // Set the global supervision sink to the first ProcAgent's
        // supervision event handler. Last-mesh-wins semantics: if a
        // previous mesh installed a sink, it is replaced.
        if let Some(first) = ranks.first() {
            crate::global_context::set_global_supervision_sink(
                first.agent.port::<ActorSupervisionEvent>(),
            );
        }

        let current_ref = ProcMeshRef::new(id.clone(), region, ranks, Some(hosts)).unwrap();

        // Notify telemetry that the ProcAgent mesh was created.
        {
            let name_str = id.to_string();
            let mesh_id_hash = hash_to_u64(&id);

            let hm = current_ref
                .host_mesh
                .as_ref()
                .expect("ProcMesh always has a host mesh");
            let parent_mesh_id = hash_to_u64(hm.id());
            let parent_view_json = serde_json::to_string(hm.region())
                .unwrap_or_else(|e| format!("encountered error when serializing region: {}", e));

            hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
                id: mesh_id_hash,
                timestamp: std::time::SystemTime::now(),
                class: "Proc".to_string(),
                given_name: id
                    .display_label()
                    .map(|l| l.as_str())
                    .unwrap_or("unnamed")
                    .to_string(),
                full_name: name_str,
                shape_json: serde_json::to_string(&current_ref.region.extent()).unwrap_or_default(),
                parent_mesh_id: Some(parent_mesh_id),
                parent_view_json: Some(parent_view_json),
            });

            // Notify telemetry of each ProcAgent actor in this mesh.
            // These are skipped in Proc::spawn_inner. mesh_id directly points to proc mesh.
            let now = std::time::SystemTime::now();
            for rank in current_ref.ranks.iter() {
                let actor_addr = rank.agent.actor_addr();

                hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
                    id: hyperactor_telemetry::hash_to_u64(actor_addr.id()),
                    timestamp: now,
                    mesh_id: mesh_id_hash,
                    rank: rank.create_rank as u64,
                    full_name: actor_addr.to_string(),
                    display_name: None,
                });
            }
        }

        Ok(Self {
            id,
            current_ref,
            controller: None,
        })
    }

    /// Set or clear the controller actor managing this mesh.
    pub(crate) fn set_controller(
        &mut self,
        controller: Option<ActorRef<crate::mesh_controller::ProcMeshController>>,
    ) {
        self.controller = controller;
    }

    /// Stop this mesh gracefully.
    ///
    /// If a `ProcMeshController` is present (owned meshes spawned from a host
    /// mesh), the stop is delegated to the controller via `resource::Stop`;
    /// the controller's handler awaits `HostMeshRef::stop_proc_mesh`, which
    /// casts `Stop` + `WaitRankStatus{min_status: Stopped}` to the
    /// HostAgents and waits up to `PROC_STOP_MAX_IDLE` for every proc to
    /// reach `Stopped`. We then serialize behind that handler with a
    /// `GetState` to read the final statuses out of the controller's
    /// `health_state`.
    pub async fn stop(&mut self, cx: &impl context::Actor, reason: String) -> anyhow::Result<()> {
        if let Some(controller) = self.controller.take() {
            let id = self.id.resource_id().clone();
            controller.post(
                cx,
                resource::Stop {
                    id: id.clone(),
                    reason,
                },
            );

            // The controller processes messages serially, so by the time it
            // gets to this `GetState`, its `health_state.statuses` already
            // reflects the outcome of `stop_proc_mesh` (Stopping, Stopped,
            // Failed, or Timeout on `PROC_STOP_MAX_IDLE` exhaustion).
            let (port, mut rx) = cx.mailbox().open_port();
            controller.post(
                cx,
                resource::GetState::<resource::mesh::State<()>> {
                    id: id.clone(),
                    reply: port.bind(),
                },
            );

            let statuses = rx.recv().await?;
            let Some(state) = &statuses.state else {
                anyhow::bail!(
                    "non-existent state in GetState reply from controller: {}",
                    controller.actor_addr()
                );
            };
            // `is_terminating` accepts Stopping, Stopped, Failed, and
            // Timeout. The controller's Stop handler has already awaited
            // (or timed out) the underlying HostAgent wait, so any rank
            // still in Running here means the controller never processed
            // the stop for that rank.
            let all_stopped = state.statuses.values().all(|s| s.is_terminating());
            if !all_stopped {
                anyhow::bail!(
                    "proc mesh {} not all procs reached terminating state after stop: {:?}",
                    id,
                    state.statuses,
                );
            }
            return Ok(());
        }

        let region = self.region.clone();
        let procs = self.current_ref.proc_ids().collect::<Vec<ProcAddr>>();
        // We use the proc mesh region rather than the host mesh region
        // because the host agent stores one entry per proc, not per host.
        self.current_ref
            .host_mesh
            .as_ref()
            .expect("ProcMesh always has a host mesh")
            .stop_proc_mesh(cx, &self.id, procs, region, reason)
            .await
            .map(|_| ())
            .map_err(anyhow::Error::from)
    }
}

impl fmt::Display for ProcMesh {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.current_ref)
    }
}

impl Deref for ProcMesh {
    type Target = ProcMeshRef;

    fn deref(&self) -> &Self::Target {
        &self.current_ref
    }
}

impl Drop for ProcMesh {
    fn drop(&mut self) {
        tracing::info!(
            name = "ProcMeshStatus",
            proc_mesh = %self.id,
            status = "Dropped",
        );
    }
}

/// A reference to a ProcMesh, consisting of a set of ranked [`ProcRef`]s,
/// arranged into a region. ProcMeshes are named, uniquely identifying the
/// ProcMesh from which the reference was derived.
///
/// ProcMeshes can be sliced to create new ProcMeshes with a subset of the
/// original ranks.
///
/// `ProcMeshRef::sliced` is intentionally pure. A sliced proc mesh can still
/// expose dense `ProcRef`s, while the backing `ProcAgent` `ActorMeshRef`
/// carries a lazy cast-domain descriptor that installs itself on first cast.
#[derive(Debug, Clone, Named, Serialize, Deserialize)]
pub struct ProcMeshRef {
    id: ProcMeshId,
    region: Region,
    ranks: Arc<Vec<ProcRef>>,
    /// Actor mesh for the `ProcAgent`s backing this proc mesh view.
    ///
    /// `ProcMeshRef::sliced` derives a sliced agent mesh with a lazy cast
    /// descriptor. The first cast through that view installs the descriptor on
    /// the caller's sender stream.
    ///
    /// The `ProcMeshRef` itself keeps dense `ProcRef`s so it can later
    /// materialize this field from the current view without consulting the
    /// parent mesh or using a temporary proc.
    proc_agent_mesh: ActorMeshRef<ProcAgent>,
    // Some if this was spawned from a host mesh, else none.
    host_mesh: Option<HostMeshRef>,
}
wirevalue::register_type!(ProcMeshRef);

// The proc-agent actor mesh is derived from `ranks`, so it is not part of
// `ProcMeshRef` identity.
impl PartialEq for ProcMeshRef {
    fn eq(&self, other: &Self) -> bool {
        self.id == other.id
            && self.region == other.region
            && self.ranks == other.ranks
            && self.host_mesh == other.host_mesh
    }
}

impl Eq for ProcMeshRef {}

impl std::hash::Hash for ProcMeshRef {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.id.hash(state);
        self.region.hash(state);
        self.ranks.hash(state);
        self.host_mesh.hash(state);
    }
}

impl ProcMeshRef {
    /// Create a new ProcMeshRef from the given id, region, ranks, and so on.
    #[allow(clippy::result_large_err)]
    fn new(
        id: ProcMeshId,
        region: Region,
        ranks: Arc<Vec<ProcRef>>,
        host_mesh: Option<HostMeshRef>,
    ) -> crate::Result<Self> {
        if ranks.is_empty() {
            return Err(crate::Error::ConfigurationError(anyhow::anyhow!(
                "empty proc meshes are not supported"
            )));
        }
        if region.num_ranks() != ranks.len() {
            return Err(crate::Error::InvalidRankCardinality {
                expected: region.num_ranks(),
                actual: ranks.len(),
            });
        }
        let proc_agent_mesh = Self::proc_agent_mesh_ref(&id, &region, &ranks)?;
        Ok(Self {
            id,
            region,
            ranks,
            proc_agent_mesh,
            host_mesh,
        })
    }

    /// Create a singleton ProcMeshRef, given the provided ProcRef and id.
    /// This is used to support creating local singleton proc meshes to support `this_proc()`
    /// in python client actors.
    pub fn new_singleton(id: ProcMeshId, proc_ref: ProcRef) -> crate::Result<Self> {
        let region: Region = Extent::unity().into();
        let ranks = Arc::new(vec![proc_ref]);
        let proc_agent_mesh = Self::proc_agent_mesh_ref(&id, &region, &ranks)?;
        Ok(Self {
            id,
            region,
            ranks,
            proc_agent_mesh,
            host_mesh: None,
        })
    }

    pub fn id(&self) -> &ProcMeshId {
        &self.id
    }

    pub fn host_mesh_id(&self) -> Option<&crate::mesh_id::HostMeshId> {
        self.host_mesh.as_ref().map(|h| h.id())
    }

    /// Returns the HostMeshRef that owns this ProcMeshRef, if any.
    pub fn hosts(&self) -> Option<&HostMeshRef> {
        self.host_mesh.as_ref()
    }

    pub(crate) fn agent_mesh(&self) -> &ActorMeshRef<ProcAgent> {
        &self.proc_agent_mesh
    }

    fn proc_agent_mesh_ref(
        proc_mesh_id: &ProcMeshId,
        region: &Region,
        ranks: &[ProcRef],
    ) -> crate::Result<ActorMeshRef<ProcAgent>> {
        let agent_label = ranks
            .first()
            .unwrap()
            .agent
            .actor_addr()
            .label()
            .cloned()
            .unwrap_or_else(|| Label::new(proc_agent::PROC_AGENT_ACTOR_NAME).unwrap());
        let id = ActorMeshId::singleton(agent_label);

        let members = Arc::new(
            ranks
                .iter()
                .map(|rank| rank.agent.actor_addr().clone())
                .collect_mesh::<ValueMesh<_>>(region.clone())
                .map_err(|error| crate::Error::ConfigurationError(error.into()))?,
        );

        Ok(ActorMeshRef::new_managed(
            id,
            Some(proc_mesh_id.clone()),
            region.clone(),
            None,
            members,
        ))
    }

    /// Query the state of all actors in this mesh matching the given id.
    pub async fn actor_states(
        &self,
        cx: &impl context::Actor,
        id: ActorMeshId,
    ) -> crate::Result<ValueMesh<resource::State<ActorState>>> {
        self.actor_states_with_keepalive(cx, id, None).await
    }

    /// Query the state of all actors in this mesh matching the given id.
    /// If keepalive is Some, use a message that indicates to the recipient
    /// that the owner of the mesh is still alive, along with the expiry time
    /// after which the actor should be considered orphaned. Else, use a normal
    /// state query.
    pub(crate) async fn actor_states_with_keepalive(
        &self,
        cx: &impl context::Actor,
        id: ActorMeshId,
        keepalive: Option<std::time::SystemTime>,
    ) -> crate::Result<ValueMesh<resource::State<ActorState>>> {
        let (port, mut rx) = cx.mailbox().open_port::<resource::State<ActorState>>();
        let mut port = port.bind();
        // If this proc dies or some other issue renders the reply undeliverable,
        // the reply does not need to be returned to the sender.
        port.return_undeliverable(false);
        // TODO: Use accumulation to get back a single value (representing whether
        // *any* of the actors failed) instead of a mesh.
        let get_state = resource::GetState::<ActorState> {
            id: id.resource_id().clone(),
            reply: port,
        };
        if let Some(expires_after) = keepalive {
            self.proc_agent_mesh.cast(
                cx,
                resource::KeepaliveGetState {
                    expires_after,
                    get_state,
                },
            )?;
        } else {
            self.proc_agent_mesh.cast(cx, get_state)?;
        }
        let expected = self.ranks.len();
        // Map each managed actor's identity to its rank in *this* (current)
        // view. Locations may differ between the controller and worker, while
        // `create_rank` is relative to the view where the actor was first spawned.
        let rank_of_actor: HashMap<ActorId, usize> = self
            .ranks
            .iter()
            .enumerate()
            .map(|(rank, proc_ref)| (proc_ref.actor_addr(&id).id().clone(), rank))
            .collect();
        let mut states = Vec::with_capacity(expected);
        let timeout = hyperactor_config::global::get(GET_ACTOR_STATE_MAX_IDLE);
        for _ in 0..expected {
            // The agent runs on the same process as the running actor, so if some
            // fatal event caused the process to crash (e.g. OOM, signal, process exit),
            // the agent will be unresponsive.
            // We handle this by setting a timeout on the recv, and if we don't get a
            // message we assume the agent is dead and return a failed state.
            let state = tokio::time::timeout(timeout, rx.recv()).await;
            if let Ok(state) = state {
                // Handle non-timeout receiver error.
                let state = state?;
                match state.state {
                    Some(ref inner) => {
                        // Position by this actor's rank in the current view, not
                        // its historical `create_rank`.
                        let rank = *rank_of_actor.get(inner.actor_id.id()).ok_or_else(|| {
                            anyhow::anyhow!(
                                "state reply from {} is not a member of the queried view",
                                inner.actor_id,
                            )
                        })?;
                        states.push((rank, state));
                    }
                    None => {
                        // A received `None` is a definitive NotExist answer;
                        // return immediately rather than waiting for the timeout
                        // padding (which is only for ranks that never reply).
                        return Err(Error::NotExist(state.id));
                    }
                }
            } else {
                tracing::error!(
                    "timeout waiting for a message after {:?} from proc mesh agent in mesh {}",
                    timeout,
                    self.proc_agent_mesh
                );
                // Timeout error, stop reading from the receiver and send back what we have so far,
                // padding with failed states.
                let all_ranks = (0..self.ranks.len()).collect::<HashSet<_>>();
                let completed_ranks = states.iter().map(|(rank, _)| *rank).collect::<HashSet<_>>();
                let mut leftover_ranks = all_ranks.difference(&completed_ranks).collect::<Vec<_>>();
                assert_eq!(leftover_ranks.len(), expected - states.len());
                while states.len() < expected {
                    let rank = *leftover_ranks
                        .pop()
                        .expect("leftover ranks should not be empty");
                    let agent = self.proc_agent_mesh.get(rank).expect("agent should exist");
                    let agent_id = agent.actor_addr().clone();
                    states.push((
                        // We populate with any ranks leftover at the time of the timeout.
                        rank,
                        resource::State {
                            id: id.resource_id().clone(),
                            status: resource::Status::Timeout(timeout),
                            // We don't know the ActorAddr that used to live on this rank.
                            // But we do know the mesh agent id, so we'll use that.
                            // Use u64::MAX so this synthetic state always wins
                            // last-writer-wins ordering against real streamed updates.
                            generation: u64::MAX,
                            timestamp: std::time::SystemTime::now(),
                            state: Some(ActorState {
                                actor_id: agent_id.clone(),
                                create_rank: rank,
                                supervision_events: vec![ActorSupervisionEvent::new(
                                    agent_id,
                                    None,
                                    ActorStatus::generic_failure(format!(
                                        "timeout waiting for message from proc mesh agent while querying for \"{}\". The process likely crashed",
                                        id,
                                    )),
                                    None,
                                )],
                            }),
                        },
                    ));
                }
                break;
            }
        }
        // Ensure that all ranks have replied. Note that if the mesh is sliced,
        // not all create_ranks may be in the mesh.
        // Sort by rank, so that the resulting mesh is ordered.
        states.sort_by_key(|(rank, _)| *rank);
        let vm = states
            .into_iter()
            .map(|(_, state)| state)
            .collect_mesh::<ValueMesh<_>>(self.region.clone())?;
        Ok(vm)
    }

    /// Get the state of every proc in this proc mesh.
    ///
    /// Casts a single `GetHostProcStates` to the routing host-agent mesh
    /// carrying this mesh's selected global ranks (the mesh may be sliced, so
    /// they need not be a dense `0..n`). The cast may reach hosts that own no
    /// selected procs, but each HostAgent filters locally and only hosts with
    /// matching ranks reply. Replies reduce up the cast tree (fanning in at cast
    /// actor 0) instead of every host dialing this caller. When `keepalive` is
    /// `Some`, each proc's expiry is extended (orphan protection, as with
    /// `KeepaliveGetState`). Returns `None` when this proc mesh is not backed by
    /// a host mesh (local/in-process meshes). On timeout, ranks whose host did
    /// not reply are padded with a `Timeout` state.
    #[allow(clippy::result_large_err)]
    pub async fn states(
        &self,
        cx: &impl context::Actor,
        keepalive: Option<std::time::SystemTime>,
    ) -> crate::Result<Option<ValueMesh<resource::State<ProcState>>>> {
        // Only meaningful when this proc mesh is backed by a host mesh.
        let Some(host_mesh) = self.host_mesh.as_ref() else {
            return Ok(None);
        };
        let region = self.region.clone();
        let timeout = hyperactor_config::global::get(GET_PROC_STATE_MAX_IDLE);

        // Per-rank template seeded with `Timeout` placeholders. Hosts overlay
        // the ranks they own, so any rank never reported keeps its placeholder
        // and the result is a complete mesh with no post-hoc padding.
        let template: ValueMesh<resource::State<ProcState>> = self
            .ranks
            .iter()
            .map(|proc_ref| resource::State {
                id: ResourceId::new(
                    proc_ref.proc_id.uid().clone(),
                    proc_ref.proc_id.label().cloned(),
                ),
                status: resource::Status::Timeout(timeout),
                state: None,
                generation: 0,
                timestamp: std::time::SystemTime::now(),
            })
            .collect_mesh::<ValueMesh<_>>(region.clone())?;
        // Snapshot returned if no host replies before the idle timeout.
        let fallback = template.clone();

        // Accumulator port: receives sparse per-host overlays and emits the
        // merged full mesh (right-wins). The host mesh is a routing
        // over-approximation for sliced proc meshes; HostAgents that own
        // selected ranks post an overlay, others stay silent.
        let (port, rx) = cx.mailbox().open_idle_flush_accum_port(
            template,
            IdleFlushReducerOpts {
                idle_timeout: Duration::from_millis(50),
                abandon_timeout: Duration::from_secs(30),
                expected_updates_per_destination: NonZeroUsize::MIN,
            },
        );

        let reply = port.bind();

        host_mesh.agent_mesh().cast(
            cx,
            GetHostProcStates {
                proc_mesh_id: self.id.clone(),
                region: region.clone(),
                keepalive,
                reply,
            },
        )?;

        // Wait until every rank has reported (moved off its `Timeout`
        // placeholder) or we idle out. Either way the mesh is complete: ranks
        // whose host never replied stay `Timeout`, and a failed proc surfaces as
        // its `Failed` state rather than an error.
        let mesh =
            match resource::wait_mesh(rx, timeout, fallback, |s: &resource::State<ProcState>| {
                !matches!(s.status, resource::Status::Timeout(_))
            })
            .await
            {
                Ok(mesh) | Err(mesh) => mesh,
            };

        Ok(Some(mesh))
    }

    /// Returns an iterator over the proc ids in this mesh.
    pub(crate) fn proc_ids(&self) -> impl Iterator<Item = ProcAddr> {
        self.ranks.iter().map(|proc_ref| proc_ref.proc_id.clone())
    }

    /// Spawn an actor on all of the procs in this mesh, returning a
    /// new ActorMesh.
    ///
    /// Bounds:
    /// - `A: Actor` - the actor actually runs inside each proc.
    /// - `A: Referable` - so we can return typed `ActorRef<A>`s
    ///   inside the `ActorMesh`.
    /// - `A::Params: RemoteMessage` - spawn parameters must be
    ///   serializable and routable.
    pub async fn spawn<A: RemoteSpawn, C: context::Actor>(
        &self,
        cx: &C,
        name: &str,
        params: &A::Params,
    ) -> crate::Result<ActorMesh<A>>
    where
        A::Params: RemoteMessage,
        C::A: Handler<MeshFailure>,
    {
        // Spawning from a string is never a system actor.
        let id = ActorMeshId::instance(Label::strip(name));
        self.spawn_with_name(cx, id, params, None, false).await
    }

    /// Spawn a 'service' actor. Service actors are *singletons*, using
    /// reserved names. The provided name is used verbatim as the actor's
    /// name, and thus it may be persistently looked up by constructing
    /// the appropriate name.
    ///
    /// Note: avoid using service actors if possible; the mechanism will
    /// be replaced by an actor registry.
    pub async fn spawn_service<A: RemoteSpawn, C: context::Actor>(
        &self,
        cx: &C,
        name: &str,
        params: &A::Params,
    ) -> crate::Result<ActorMesh<A>>
    where
        A::Params: RemoteMessage,
        C::A: Handler<MeshFailure>,
    {
        let id = ActorMeshId::singleton(Label::strip(name));
        self.spawn_with_name(cx, id, params, None, false).await
    }

    /// Spawn a controllerless 'service' actor.
    ///
    /// Like [`spawn_service`](Self::spawn_service), this spawns a singleton
    /// service actor (a reserved name, deduped to one). Unlike `spawn_service`,
    /// the resulting mesh has **no** `ActorMeshController`: no per-caller
    /// controller actor is spawned, and the caller `C` is therefore **not**
    /// required to implement `Handler<MeshFailure>`.
    ///
    /// This is for coordinators that own their own supervision — e.g. a
    /// controllerless singleton that itself spawns controller-ful children —
    /// where a per-caller controller would be redundant.
    #[hyperactor::instrument(fields(
        host_mesh=self.host_mesh_id().map(|id| id.to_string()),
        proc_mesh=self.id.to_string(),
        actor_name=name.to_string(),
    ))]
    pub async fn spawn_controllerless_service<A: RemoteSpawn, C: context::Actor>(
        &self,
        cx: &C,
        name: &str,
        params: &A::Params,
    ) -> crate::Result<ActorMesh<A>>
    where
        A::Params: RemoteMessage,
    {
        tracing::info!(
            name = "ProcMeshStatus",
            status = "ActorMesh::Spawn::Attempt",
        );
        tracing::info!(name = "ActorMeshStatus", status = "Spawn::Attempt");
        let id = ActorMeshId::singleton(Label::strip(name));
        let result = self
            .spawn_managed_mesh_inner(cx, id, params, None)
            .await
            .map(|(mesh, _statuses)| mesh);
        match &result {
            Ok(_) => {
                tracing::info!(
                    name = "ProcMeshStatus",
                    status = "ActorMesh::Spawn::Success",
                );
                tracing::info!(name = "ActorMeshStatus", status = "Spawn::Success");
            }
            Err(error) => {
                tracing::error!(name = "ProcMeshStatus", status = "ActorMesh::Spawn::Failed", %error);
                tracing::error!(name = "ActorMeshStatus", status = "Spawn::Failed", %error);
            }
        }
        result
    }

    /// Spawn an actor on all procs in this mesh using the new data-only mesh API.
    /// `actor_mesh_id` identifies the mesh and its members, and must be an
    /// instance id.
    ///
    /// This initiates an actor spawn on each proc and constructs a stoppable
    /// data mesh from the actor refs and local supervisor handles without
    /// waiting for the remote supervision sessions to link. The mesh is usable
    /// immediately: each proc queues messages for its actor until the actor
    /// starts. If a request cannot be initiated, all previously created local
    /// supervisors receive a best-effort stop request.
    ///
    /// Bounds:
    /// - `A: Actor` - the actor actually runs inside each proc.
    /// - `A: Referable` - so we can return typed `ActorRef<A>`s inside the `ActorMesh`.
    /// - `A::Params: RemoteMessage` - spawn parameters must be serializable and routable.
    pub async fn spawn_data<A: RemoteSpawn, C: context::Actor>(
        &self,
        cx: &C,
        actor_mesh_id: ActorMeshId,
        params: &A::Params,
    ) -> crate::Result<ActorMesh<A>>
    where
        A::Params: RemoteMessage,
    {
        if !actor_mesh_id.uid().is_instance() {
            return Err(Error::ConfigurationError(anyhow::anyhow!(
                "data-backed actor meshes require instance ids"
            )));
        }
        self.spawn_data_mesh_inner(cx, params, actor_mesh_id)
    }

    fn spawn_data_mesh_inner<A: RemoteSpawn, C: context::Actor>(
        &self,
        cx: &C,
        params: &A::Params,
        actor_mesh_id: ActorMeshId,
    ) -> crate::Result<ActorMesh<A>>
    where
        A::Params: RemoteMessage,
    {
        let region = self.region().clone();
        let mut members = Vec::with_capacity(self.ranks.len());
        let mut stop_handles = Vec::with_capacity(self.ranks.len());
        let serialized_params = bincode::serde::encode_to_vec(params, bincode::config::legacy())?;

        let stop_spawned = |handles: &[ActorMeshStopHandle], reason: &str| {
            for handle in handles {
                let _ = handle.stop(reason);
            }
        };

        // Each call only schedules a local supervisor that sends a one-way spawn
        // request. No rank waits for another rank's remote actor initialization.
        for (rank, proc_ref) in self.ranks.iter().enumerate() {
            let params = match bincode::serde::decode_from_slice(
                &serialized_params,
                bincode::config::legacy(),
            ) {
                Ok((params, _)) => params,
                Err(error) => {
                    stop_spawned(&stop_handles, "data mesh spawn failed");
                    return Err(error.into());
                }
            };
            let point = region
                .extent()
                .point_of_rank(rank)
                .expect("proc mesh rank must have a point");
            let mut environment = cx.instance().actor_environment().clone();
            if let Err(error) = environment.set(CAST_POINT, point.clone()) {
                stop_spawned(&stop_handles, "data mesh spawn failed");
                return Err(Error::Other(error.into()));
            }
            // The actor's own environment carries its point, but supervision
            // events are raised on the spawner's side, so they need a copy.
            // TODO: unify the actor environment and supervision labels.
            let mut labels = Flattrs::new();
            labels.set(CAST_POINT, point);
            labels.set(ACTOR_MESH_ID, actor_mesh_id.clone());
            let actor_spawner = proc_ref.spawner();
            let (actor_ref, stop_handle) = match actor_spawner.spawn_uid_with_link_and_ready(
                cx,
                actor_mesh_id.uid().clone(),
                params,
                KeepaliveLink::default(),
                None,
                environment,
                labels,
            ) {
                Ok(spawned) => spawned,
                Err(error) => {
                    stop_spawned(&stop_handles, "data mesh spawn failed");
                    return Err(Error::Other(error));
                }
            };
            members.push(actor_ref);
            stop_handles.push(stop_handle.into_any());
        }

        ActorMesh::try_new_data(actor_mesh_id, region, members, stop_handles)
    }

    /// Spawn an actor on all procs in this mesh under the given
    /// [`ActorMeshId`](crate::mesh_id::ActorMeshId), returning a new `ActorMesh`.
    ///
    /// This is the underlying implementation used by [`spawn`]; it
    /// differs only in that the actor mesh id is passed explicitly
    /// rather than as a `&str`.
    ///
    /// Bounds:
    /// - `A: Actor` - the actor actually runs inside each proc.
    /// - `A: Referable` - so we can return typed `ActorRef<A>`s
    ///   inside the `ActorMesh`.
    /// - `A::Params: RemoteMessage` - spawn parameters must be
    ///   serializable and routable.
    /// - `C::A: Handler<MeshFailure>` - in order to spawn actors,
    ///   the actor must accept messages of type `MeshFailure`. This
    ///   is delivered when the actors spawned in the mesh have a failure that
    ///   isn't handled.
    #[hyperactor::instrument(fields(
        host_mesh=self.host_mesh_id().map(|id| id.to_string()),
        proc_mesh=self.id.to_string(),
        actor_name=name.to_string(),
    ))]
    pub async fn spawn_with_name<A: RemoteSpawn, C: context::Actor>(
        &self,
        cx: &C,
        name: ActorMeshId,
        params: &A::Params,
        supervision_display_name: Option<String>,
        is_system_actor: bool,
    ) -> crate::Result<ActorMesh<A>>
    where
        A::Params: RemoteMessage,
        C::A: Handler<MeshFailure>,
    {
        tracing::info!(
            name = "ProcMeshStatus",
            status = "ActorMesh::Spawn::Attempt",
        );
        tracing::info!(name = "ActorMeshStatus", status = "Spawn::Attempt");
        let result = self
            .spawn_with_name_inner(cx, name, params, supervision_display_name, is_system_actor)
            .await;
        match &result {
            Ok(_) => {
                tracing::info!(
                    name = "ProcMeshStatus",
                    status = "ActorMesh::Spawn::Success",
                );
                tracing::info!(name = "ActorMeshStatus", status = "Spawn::Success");
            }
            Err(error) => {
                tracing::error!(name = "ProcMeshStatus", status = "ActorMesh::Spawn::Failed", %error);
                tracing::error!(name = "ActorMeshStatus", status = "Spawn::Failed", %error);
            }
        }
        result
    }

    async fn spawn_with_name_inner<A: RemoteSpawn, C: context::Actor>(
        &self,
        cx: &C,
        actor_mesh_id: ActorMeshId,
        params: &A::Params,
        supervision_display_name: Option<String>,
        is_system_actor: bool,
    ) -> crate::Result<ActorMesh<A>>
    where
        C::A: Handler<MeshFailure>,
    {
        let spawned = self
            .spawn_mesh_inner(
                cx,
                actor_mesh_id,
                params,
                supervision_display_name.clone(),
                is_system_actor,
            )
            .await?;
        match spawned {
            SpawnedActorMesh::Data(mesh) => Ok(mesh),
            SpawnedActorMesh::Managed { mut mesh, statuses } => {
                // System actors are managed by their owning runtime, not an
                // ActorMeshController.
                if !is_system_actor {
                    // Spawn a unique mesh manager for each actor mesh, so the type of the
                    // mesh can be preserved.
                    let controller: ActorMeshController<A> = ActorMeshController::new(
                        ActorMeshControlPlane::new(mesh.deref().clone(), self.clone()),
                        supervision_display_name,
                        Some(cx.instance().port().bind()),
                        statuses,
                    );
                    // hyperactor::proc AI-3: controller name must include mesh
                    // identity for proc-wide ActorAddr uniqueness. A fixed base name alone
                    // collides across parents because pid allocation is
                    // parent-scoped.
                    let controller_name = format!(
                        "{}_{}",
                        crate::mesh_controller::ACTOR_MESH_CONTROLLER_NAME,
                        mesh.id()
                    );
                    let controller = cx.spawn_with_label(&controller_name, controller);
                    // Controller and ActorMesh both depend on references from each other, break
                    // the cycle by setting the controller after the fact.
                    mesh.set_controller(Some(controller.bind()));
                }
                Ok(mesh)
            }
        }
    }

    async fn spawn_mesh_inner<A: RemoteSpawn, C: context::Actor>(
        &self,
        cx: &C,
        actor_mesh_id: ActorMeshId,
        params: &A::Params,
        supervision_display_name: Option<String>,
        is_system_actor: bool,
    ) -> crate::Result<SpawnedActorMesh<A>> {
        if hyperactor_config::global::get(USE_DIRECT_SPAWN)
            && !is_system_actor
            && actor_mesh_id.uid().is_instance()
        {
            return self
                .spawn_data_mesh_inner(cx, params, actor_mesh_id)
                .map(SpawnedActorMesh::Data);
        }

        let (mesh, statuses) = self
            .spawn_managed_mesh_inner(cx, actor_mesh_id, params, supervision_display_name)
            .await?;
        Ok(SpawnedActorMesh::Managed { mesh, statuses })
    }

    /// The managed spawn path used by singleton services and by ordinary
    /// spawns when [`USE_DIRECT_SPAWN`] is disabled: the create/update cast,
    /// the accumulator port, the bounded
    /// `GetRankStatus::wait` with terminal / timeout mapping, mesh
    /// materialization, and telemetry. It carries **no**
    /// `C::A: Handler<MeshFailure>` bound — that is required only by the
    /// controller construction (`cx.instance().port().bind()`), which the
    /// bounded caller layers on afterward. Returns the mesh together with the
    /// final `StatusMesh` (moved into the controller for non-system actors;
    /// discarded otherwise — system actors and the controllerless path).
    async fn spawn_managed_mesh_inner<A: RemoteSpawn, C: context::Actor>(
        &self,
        cx: &C,
        actor_mesh_id: ActorMeshId,
        params: &A::Params,
        supervision_display_name: Option<String>,
    ) -> crate::Result<(ActorMesh<A>, crate::StatusMesh)> {
        let remote = Remote::collect();
        // `RemoteSpawn` + `register_spawnable!(A)` ensure that `A` has a
        // `SpawnableActor` entry in this registry, so
        // `name_of::<A>()` can resolve its global type name.
        let actor_type = remote
            .name_of::<A>()
            .ok_or(Error::ActorTypeNotRegistered(type_name::<A>().to_string()))?
            .to_string();

        let serialized_params = bincode::serde::encode_to_vec(params, bincode::config::legacy())?;
        self.proc_agent_mesh.cast(
            cx,
            resource::CreateOrUpdate::<proc_agent::ActorSpec> {
                id: actor_mesh_id.resource_id().clone(),
                rank: Default::default(),
                spec: proc_agent::ActorSpec {
                    actor_type: actor_type.clone(),
                    params_data: serialized_params.clone(),
                    actor_environment: cx.instance().actor_environment().clone(),
                },
            },
        )?;

        let region = self.region().clone();
        // Open an accum port that *receives overlays* and *emits full
        // meshes*.
        //
        // NOTE: Mailbox initializes the accumulator state via
        // `Default`, which is an *empty* ValueMesh (0 ranks). Our
        // Accumulator<ValueMesh<T>> implementation detects this on
        // the first update and replaces it with the caller-supplied
        // template (the `self` passed into open_idle_flush_accum_port), which we
        // seed here as "full NotExist over the target region".
        let (port, rx) = cx.mailbox().open_idle_flush_accum_port(
            // Initial state for the accumulator: full mesh seeded to
            // NotExist.
            crate::StatusMesh::from_single(region.clone(), Status::NotExist),
            IdleFlushReducerOpts {
                idle_timeout: Duration::from_millis(50),
                abandon_timeout: Duration::from_secs(30),
                expected_updates_per_destination: NonZeroUsize::MIN,
            },
        );

        let mut reply = port.bind();
        // If this proc dies or some other issue renders the reply undeliverable,
        // the reply does not need to be returned to the sender.
        reply.return_undeliverable(false);
        // Send a message to all ranks. They reply with overlays to
        // `port`.
        self.proc_agent_mesh.cast(
            cx,
            resource::GetRankStatus {
                id: actor_mesh_id.resource_id().clone(),
                // Filled by the cast layer with this recipient's dense view rank.
                rank: Default::default(),
                reply,
            },
        )?;

        let start_time = tokio::time::Instant::now();

        // Wait for all ranks to report a terminal or running status.
        // If any proc reports a failure (via supervision) or the mesh
        // times out, `wait()` returns Err with the final snapshot.
        //
        // `rx` is the accumulator output stream: each time reduced
        // overlays are applied, it emits a new StatusMesh snapshot.
        // `wait()` loops on it, deciding when the stream is
        // "complete" (no more NotExist) or times out.
        let statuses = match GetRankStatus::wait(
            rx,
            self.ranks.len(),
            hyperactor_config::global::get(ACTOR_SPAWN_MAX_IDLE),
            region.clone(), // fallback
        )
        .await
        {
            Ok(statuses) => {
                // Spawn succeeds only if no rank has reported a
                // supervision/terminal state. This preserves the old
                // `first_terminating().is_none()` semantics.
                let has_terminating = statuses.values().any(|s| s.is_terminating());
                if !has_terminating {
                    Ok(statuses)
                } else {
                    let legacy = mesh_to_rankedvalues_with_default(
                        &statuses,
                        Status::NotExist,
                        Status::is_not_exist,
                        self.ranks.len(),
                    );
                    Err(Error::ActorSpawnError { statuses: legacy })
                }
            }
            Err(complete) => {
                // Fill remaining ranks with a timeout status, now
                // handled via the legacy shim.
                let elapsed = start_time.elapsed();
                let legacy = mesh_to_rankedvalues_with_default(
                    &complete,
                    Status::Timeout(elapsed),
                    Status::is_not_exist,
                    self.ranks.len(),
                );
                Err(Error::ActorSpawnError { statuses: legacy })
            }
        }?;

        let actor_mesh_members = Arc::new(
            self.ranks
                .iter()
                .map(|rank| rank.actor_addr(&actor_mesh_id))
                .collect_mesh::<ValueMesh<_>>(self.region().clone())
                .map_err(|error| crate::Error::ConfigurationError(error.into()))?,
        );

        let mesh = ActorMesh::new(
            self.clone(),
            actor_mesh_id.clone(),
            None,
            actor_mesh_members,
        );
        // Notify telemetry that an actor mesh was created.
        {
            let id_str = actor_mesh_id.to_string();

            // Hash the proc mesh id for parent_mesh_id.
            let parent_mesh_id_hash = hash_to_u64(self.id());
            let mesh_id_hash = telemetry_actor_mesh_id(self.id(), &actor_mesh_id);

            hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
                id: mesh_id_hash,
                timestamp: std::time::SystemTime::now(),
                class: supervision_display_name
                    .as_deref()
                    .and_then(python_class_from_supervision_name)
                    .unwrap_or(actor_type),
                given_name: actor_mesh_id
                    .display_label()
                    .map(|l| l.as_str())
                    .unwrap_or("unnamed")
                    .to_string(),
                full_name: id_str,
                shape_json: serde_json::to_string(&self.region().extent()).unwrap_or_default(),
                parent_mesh_id: Some(parent_mesh_id_hash),
                parent_view_json: serde_json::to_string(self.region()).ok(),
            });

            // Notify telemetry of each actor in this mesh. The rank is
            // the actor's position within the actor mesh (not the proc's
            // create_rank, which reflects the original unsliced mesh).
            let now = std::time::SystemTime::now();
            for (rank, proc_ref) in self.ranks.iter().enumerate() {
                let display_name = supervision_display_name.as_ref().map(|sdn| {
                    let point = self.region().extent().point_of_rank(rank).unwrap();
                    crate::actor_display_name(sdn, &point)
                });
                let actor_addr = proc_ref.actor_addr(&actor_mesh_id);
                hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
                    id: hyperactor_telemetry::hash_to_u64(actor_addr.id()),
                    timestamp: now,
                    mesh_id: mesh_id_hash,
                    rank: rank as u64,
                    full_name: actor_addr.to_string(),
                    display_name,
                });
            }
        }

        Ok((mesh, statuses))
    }

    /// Send stop actors message to all mesh agents for a specific actor mesh id.
    #[hyperactor::instrument(fields(
        host_mesh = self.host_mesh_id().map(|id| id.to_string()),
        proc_mesh = self.id.to_string(),
        actor_mesh = actor_mesh_id.to_string(),
    ))]
    pub(crate) async fn stop_actor_by_id(
        &self,
        cx: &impl context::Actor,
        actor_mesh_id: ActorMeshId,
        reason: String,
    ) -> crate::Result<ValueMesh<Status>> {
        tracing::info!(name = "ProcMeshStatus", status = "ActorMesh::Stop::Attempt");
        tracing::info!(name = "ActorMeshStatus", status = "Stop::Attempt");
        let result = self.stop_actor_by_id_inner(cx, actor_mesh_id, reason).await;
        match &result {
            Ok(_) => {
                tracing::info!(name = "ProcMeshStatus", status = "ActorMesh::Stop::Success");
                tracing::info!(name = "ActorMeshStatus", status = "Stop::Success");
            }
            Err(error) => {
                tracing::error!(name = "ProcMeshStatus", status = "ActorMesh::Stop::Failed", %error);
                tracing::error!(name = "ActorMeshStatus", status = "Stop::Failed", %error);
            }
        }
        result
    }

    async fn stop_actor_by_id_inner(
        &self,
        cx: &impl context::Actor,
        actor_mesh_id: ActorMeshId,
        reason: String,
    ) -> crate::Result<ValueMesh<Status>> {
        let region = self.region().clone();
        self.proc_agent_mesh.cast(
            cx,
            resource::Stop {
                id: actor_mesh_id.resource_id().clone(),
                reason,
            },
        )?;

        // Open an accum port that *receives overlays* and *emits full
        // meshes*.
        //
        // NOTE: Mailbox initializes the accumulator state via
        // `Default`, which is an *empty* ValueMesh (0 ranks). Our
        // Accumulator<ValueMesh<T>> implementation detects this on
        // the first update and replaces it with the caller-supplied
        // template (the `self` passed into open_idle_flush_accum_port), which we
        // seed here as "full NotExist over the target region".
        let (port, rx) = cx.mailbox().open_idle_flush_accum_port(
            // Initial state for the accumulator: full mesh seeded to
            // NotExist.
            crate::StatusMesh::from_single(region.clone(), Status::NotExist),
            IdleFlushReducerOpts {
                idle_timeout: Duration::from_millis(50),
                abandon_timeout: Duration::from_secs(30),
                expected_updates_per_destination: NonZeroUsize::MIN,
            },
        );
        // Use WaitRankStatus instead of GetRankStatus so agents defer
        // their reply until the actor reaches terminal state, rather
        // than replying immediately with Stopping.
        self.proc_agent_mesh.cast(
            cx,
            resource::WaitRankStatus {
                id: actor_mesh_id.resource_id().clone(),
                // Filled by the cast layer with this recipient's dense view rank.
                rank: Default::default(),
                min_status: Status::Stopped,
                reply: port.bind(),
            },
        )?;
        let start_time = tokio::time::Instant::now();

        // Reuse actor spawn idle time.
        let max_idle_time = hyperactor_config::global::get(ACTOR_SPAWN_MAX_IDLE);
        match GetRankStatus::wait(
            rx,
            self.ranks.len(),
            max_idle_time,
            region.clone(), // fallback mesh if nothing arrives
        )
        .await
        {
            Ok(statuses) => {
                // Check that all actors are in a terminating state (Stopping
                // or beyond). Failed is ok, because one of these actors may
                // have failed earlier and we're trying to stop the others.
                let all_stopped = statuses.values().all(|s| s.is_terminating());
                if all_stopped {
                    Ok(statuses)
                } else {
                    let legacy = mesh_to_rankedvalues_with_default(
                        &statuses,
                        Status::NotExist,
                        Status::is_not_exist,
                        self.ranks.len(),
                    );
                    Err(Error::ActorStopError { statuses: legacy })
                }
            }
            Err(complete) => {
                // Fill remaining ranks with a timeout status via the
                // legacy shim.
                let legacy = mesh_to_rankedvalues_with_default(
                    &complete,
                    Status::Timeout(start_time.elapsed()),
                    Status::is_not_exist,
                    self.ranks.len(),
                );
                Err(Error::ActorStopError { statuses: legacy })
            }
        }
    }
}

impl fmt::Display for ProcMeshRef {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}{{{}}}", self.id, self.region)
    }
}

impl view::Ranked for ProcMeshRef {
    type Item = ProcRef;

    fn region(&self) -> &Region {
        &self.region
    }

    fn get(&self, rank: usize) -> Option<&Self::Item> {
        self.ranks.get(rank)
    }
}

impl view::RankedSliceable for ProcMeshRef {
    /// Return a pure slice of this proc mesh.
    ///
    /// The returned `ProcMeshRef` contains the selected dense `ProcRef`s, and
    /// its `proc_agent_mesh` carries a lazy actor-mesh slice descriptor.
    fn sliced(&self, region: Region) -> Self {
        debug_assert!(region.is_subset(view::Ranked::region(self)));
        let ranks = self
            .region()
            .remap(&region)
            .unwrap()
            .map(|index| self.get(index).unwrap().clone())
            .collect::<Vec<_>>();

        Self {
            id: self.id.clone(),
            proc_agent_mesh: self.proc_agent_mesh.sliced(region.clone()),
            region,
            ranks: Arc::new(ranks),
            host_mesh: self.host_mesh.clone(),
        }
    }
}

/// Extract a Python class display name from a supervision display name.
///
/// The supervision display name format is `{instance}.<{module}.{ClassName} {mesh_name}>`.
/// Returns `"Python<ClassName>"` if the format matches, `None` otherwise.
///
/// Scope note: this function is used only by telemetry
/// (`MeshEvent.class`), which needs the Python class as a
/// structured string and has no structured carrier today. It is
/// not on the supervision rendering path.
///
/// TODO: retained only because the telemetry path needs a
/// structured Python-class string and this is the only available
/// source. A follow-up should add a structured carrier (e.g. an
/// `actor_class` field on `ActorSupervisionEvent`, or a dedicated
/// telemetry-side field) and delete this function.
fn python_class_from_supervision_name(sdn: &str) -> Option<String> {
    let inner = sdn.rsplit_once('<')?.1.strip_suffix('>')?;
    let qualified = inner.split_whitespace().next()?;
    let class_name = qualified.rsplit_once('.')?.1;
    Some(format!("Python<{class_name}>"))
}

#[cfg(test)]
mod tests {
    #[cfg(fbcode_build)]
    use std::collections::HashMap;
    #[cfg(fbcode_build)]
    use std::collections::HashSet;
    #[cfg(fbcode_build)]
    use std::net::IpAddr;
    #[cfg(fbcode_build)]
    use std::net::Ipv4Addr;
    #[cfg(fbcode_build)]
    use std::net::SocketAddr;
    #[cfg(fbcode_build)]
    use std::ops::Deref;
    #[cfg(fbcode_build)]
    use std::time::Duration;

    #[cfg(fbcode_build)]
    use hyperactor::ActorEnvironment;
    #[cfg(fbcode_build)]
    use hyperactor::ActorRef;
    #[cfg(fbcode_build)]
    use hyperactor::Endpoint as _;
    #[cfg(fbcode_build)]
    use hyperactor::Instance;
    #[cfg(fbcode_build)]
    use hyperactor::ProcAddr;
    #[cfg(fbcode_build)]
    use hyperactor::accum::IdleFlushReducerOpts;
    #[cfg(fbcode_build)]
    use hyperactor::channel::BindSpec;
    #[cfg(fbcode_build)]
    use hyperactor::channel::ChannelAddr;
    #[cfg(fbcode_build)]
    use hyperactor::context::Mailbox;
    #[cfg(fbcode_build)]
    use hyperactor::id::Label;
    #[cfg(fbcode_build)]
    use hyperactor_config::NonZeroUsize;
    #[cfg(fbcode_build)]
    use ndslice::ViewExt as _;
    #[cfg(fbcode_build)]
    use ndslice::extent;
    #[cfg(fbcode_build)]
    use ndslice::view::Ranked as _;
    #[cfg(fbcode_build)]
    use timed_test::assert_no_process_leak;
    #[cfg(fbcode_build)]
    use timed_test::async_timed_test;
    #[cfg(fbcode_build)]
    use uuid::Uuid;

    #[cfg(fbcode_build)]
    use super::ACTOR_SPAWN_MAX_IDLE;
    #[cfg(fbcode_build)]
    use super::ProcMeshRef;
    #[cfg(fbcode_build)]
    use super::ProcRef;
    #[cfg(fbcode_build)]
    use super::USE_DIRECT_SPAWN;
    #[cfg(fbcode_build)]
    use crate::ActorMesh;
    #[cfg(fbcode_build)]
    use crate::casting::CAST_POINT;
    #[cfg(fbcode_build)]
    use crate::host_mesh::HostMesh;
    #[cfg(fbcode_build)]
    use crate::host_mesh::PROC_SPAWN_MAX_IDLE;
    #[cfg(fbcode_build)]
    use crate::mesh_id::ActorMeshId;
    #[cfg(fbcode_build)]
    use crate::mesh_id::ProcMeshId;
    #[cfg(fbcode_build)]
    use crate::proc_agent::PROC_AGENT_ACTOR_NAME;
    #[cfg(fbcode_build)]
    use crate::proc_agent::ProcAgent;
    #[cfg(fbcode_build)]
    use crate::resource::RankedValues;
    #[cfg(fbcode_build)]
    use crate::resource::Status;
    #[cfg(fbcode_build)]
    use crate::testactor;
    #[cfg(fbcode_build)]
    use crate::testing;
    #[cfg(fbcode_build)]
    use crate::transport::DEFAULT_TRANSPORT;

    #[cfg(fbcode_build)]
    async fn execute_spawn_actor() {
        hyperactor_telemetry::initialize_logging(hyperactor_telemetry::DefaultTelemetryClock {});

        let instance = testing::instance();

        let mut hm = testing::host_mesh(2).await;
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 1), None, None)
            .await
            .unwrap();
        let actor_mesh = proc_mesh.spawn(instance, "test", &()).await.unwrap();
        testactor::assert_mesh_shape(actor_mesh).await;

        let _ = hm.shutdown(instance).await;
    }

    #[async_timed_test(timeout_secs = 300)]
    #[cfg(fbcode_build)]
    async fn spawn_data_uses_default_actor_spawner() {
        let instance = testing::instance();
        let mut hm = testing::host_mesh(1).await;
        let proc_mesh = hm
            .spawn(instance, "spawn_data", extent!(gpus = 2), None, None)
            .await
            .expect("spawn proc mesh");

        let mut actor_mesh: ActorMesh<testactor::TestActor> = proc_mesh
            .spawn_data(
                instance,
                ActorMeshId::instance(Label::strip("default_spawner")),
                &(),
            )
            .await
            .expect("spawn data actor mesh through default spawner");
        assert_eq!(actor_mesh.region().num_ranks(), 2);
        let actor_uids: HashSet<_> = actor_mesh
            .values()
            .map(|actor| actor.actor_addr().id().uid().clone())
            .collect();
        assert_eq!(
            actor_uids.len(),
            1,
            "all data mesh actors should share one uid"
        );

        // Message every rank before any of them can have started.
        let (port, mut rx) = instance.mailbox().open_port();
        for actor in actor_mesh.values() {
            actor.post(instance, testactor::GetActorId(port.bind()));
        }
        let mut replied = HashSet::new();
        for _ in 0..actor_mesh.region().num_ranks() {
            let (actor_addr, _seq) = tokio::time::timeout(Duration::from_secs(60), rx.recv())
                .await
                .expect("timed out waiting for a rank to reply")
                .expect("reply port closed");
            replied.insert(actor_addr);
        }
        assert_eq!(
            replied,
            actor_mesh
                .values()
                .map(|actor| actor.actor_addr().clone())
                .collect::<HashSet<_>>(),
            "every rank should receive the message sent before it started"
        );

        let mut actor_mesh_clone = actor_mesh.clone();
        actor_mesh
            .stop(instance, "test complete".to_string())
            .await
            .expect("stop data actor mesh");
        actor_mesh_clone
            .stop(instance, "test complete".to_string())
            .await
            .expect("stopping a data actor mesh clone should be idempotent");

        hm.shutdown(instance)
            .await
            .expect("host mesh shutdown should complete");
    }

    #[async_timed_test(timeout_secs = 300)]
    #[cfg(fbcode_build)]
    async fn spawn_uses_direct_path_when_configured_but_services_remain_managed() {
        let config = hyperactor_config::global::lock();
        let _mode = config.override_key(USE_DIRECT_SPAWN, true);
        let instance = testing::instance();
        let mut hm = testing::host_mesh(1).await;
        let proc_mesh = hm
            .spawn(
                instance,
                "configured_data_spawn",
                extent!(gpus = 2),
                None,
                None,
            )
            .await
            .expect("spawn proc mesh");

        let mut actor_mesh: ActorMesh<testactor::TestActor> = proc_mesh
            .spawn(instance, "configured_data_actor", &())
            .await
            .expect("initiate configured data actor mesh spawn");
        assert!(
            actor_mesh.as_managed().is_none(),
            "configured instance spawn should return a data mesh"
        );
        assert_eq!(
            actor_mesh.region().num_ranks(),
            2,
            "direct spawn should cover all proc mesh ranks"
        );
        assert!(
            actor_mesh.values().all(|actor| {
                actor.actor_addr().label().map(|label| label.as_str())
                    == Some("configured_data_actor")
            }),
            "direct spawn should preserve the requested actor label"
        );

        let service: ActorMesh<testactor::TestActor> = proc_mesh
            .spawn_service(instance, "configured_managed_service", &())
            .await
            .expect("spawn managed service");
        assert!(
            service.as_managed().is_some(),
            "service spawn should ignore the data mesh flag"
        );

        let system_actor: ActorMesh<testactor::TestActor> = proc_mesh
            .spawn_with_name(
                instance,
                ActorMeshId::instance(Label::strip("configured_managed_system_actor")),
                &(),
                None,
                true,
            )
            .await
            .expect("spawn managed system actor");
        assert!(
            system_actor.as_managed().is_some(),
            "system actor spawn should ignore the direct spawn flag"
        );

        actor_mesh
            .stop(instance, "test complete".to_string())
            .await
            .expect("stop configured data actor mesh");
        hm.shutdown(instance)
            .await
            .expect("host mesh shutdown should complete");
    }

    #[async_timed_test(timeout_secs = 300)]
    #[cfg(fbcode_build)]
    async fn actor_environment_survives_two_remote_hops_and_shapes_cast_point() {
        const SENTINEL: u64 = 0xA0FE;
        let operation_timeout = Duration::from_secs(120);

        let config = hyperactor_config::global::lock();
        let _guard = config.override_key(PROC_SPAWN_MAX_IDLE, operation_timeout);
        let _guard2 = config.override_key(
            hyperactor::config::HOST_SPAWN_READY_TIMEOUT,
            operation_timeout,
        );

        let base = testing::instance();
        let persistent_point = extent!(persistent = 3).point_of_rank(2).unwrap();
        let mut environment = ActorEnvironment::default();
        environment
            .set(testactor::ACTOR_ENVIRONMENT_TEST_TAG, SENTINEL)
            .expect("seed persistent tag");
        environment
            .set(CAST_POINT, persistent_point.clone())
            .expect("seed persistent cast-point collision");
        let root = base
            .proc()
            .actor_instance_in_environment::<testing::TestRootClient>(
                &format!("actor_environment_root_{}", Uuid::now_v7()),
                environment,
            )
            .expect("create seeded root instance");

        let mut hm = testing::host_mesh(1).await;
        let proc_mesh = hm
            .spawn(
                &root.instance,
                "actor_environment",
                extent!(gpus = 2),
                None,
                None,
            )
            .await
            .expect("spawn proc mesh");
        let first_hop = proc_mesh
            .range("gpus", 0..1)
            .expect("first proc-mesh slice");
        let second_hop = proc_mesh
            .range("gpus", 1..2)
            .expect("second proc-mesh slice");
        let first_point = first_hop
            .region()
            .extent()
            .point_of_rank(0)
            .expect("first slice has rank zero");
        let second_point = second_hop
            .region()
            .extent()
            .point_of_rank(0)
            .expect("second slice has rank zero");
        let first_proc_addr = first_hop
            .get(0)
            .expect("first proc")
            .proc_addr()
            .to_string();
        let second_proc_addr = second_hop
            .get(0)
            .expect("second proc")
            .proc_addr()
            .to_string();
        assert_ne!(
            first_proc_addr, second_proc_addr,
            "the two hops must target distinct proc addresses",
        );
        assert!(
            persistent_point != first_point && persistent_point != second_point,
            "seeded and assigned cast points must be distinct",
        );

        let (reply, mut replies) = root
            .instance
            .open_port::<testactor::ActorEnvironmentObservation>();
        first_hop
            .spawn_controllerless_service::<testactor::ActorEnvironmentProbe, _>(
                &root.instance,
                "actor_environment_hop_1",
                &testactor::ActorEnvironmentProbeParams {
                    label: "hop_1".to_string(),
                    reply: reply.bind(),
                    nested: Some((
                        second_hop,
                        "actor_environment_hop_2".to_string(),
                        "hop_2".to_string(),
                    )),
                },
            )
            .await
            .expect("spawn first environment probe");

        let expected = HashMap::from([
            ("hop_1".to_string(), (first_point, first_proc_addr)),
            ("hop_2".to_string(), (second_point, second_proc_addr)),
        ]);
        let mut seen = HashSet::new();
        let reply_deadline = tokio::time::Instant::now() + operation_timeout;
        for _ in 0..2 {
            let observed = tokio::time::timeout_at(reply_deadline, replies.recv())
                .await
                .expect("environment probe reply timed out")
                .expect("environment probe reply port closed");
            let (expected_point, expected_proc_addr) = expected
                .get(&observed.label)
                .unwrap_or_else(|| panic!("unexpected probe label {}", observed.label));
            assert!(
                seen.insert(observed.label.clone()),
                "received duplicate observation from {}",
                observed.label,
            );
            assert_eq!(
                observed.persistent_tag,
                Some(SENTINEL),
                "{} must inherit the persistent tag",
                observed.label,
            );
            assert_eq!(
                observed.constructor_tag,
                Some(SENTINEL),
                "{} constructor must see the persistent tag",
                observed.label,
            );
            assert_eq!(
                observed.stored_point,
                Some(expected_point.clone()),
                "{} must store its caller-shaped construction point",
                observed.label,
            );
            assert_eq!(
                observed.constructor_point,
                Some(expected_point.clone()),
                "{} constructor must see its real rank-local cast point",
                observed.label,
            );
            assert_eq!(
                &observed.proc_addr, expected_proc_addr,
                "{} must run on its requested proc",
                observed.label,
            );
        }
        assert_eq!(seen.len(), expected.len(), "both hops must report");

        hm.shutdown(&root.instance)
            .await
            .expect("host mesh shutdown should complete");
    }

    #[async_timed_test(timeout_secs = 300)]
    #[cfg(fbcode_build)]
    async fn direct_spawn_shapes_cast_point() {
        const SENTINEL: u64 = 0xD1EC;
        let config = hyperactor_config::global::lock();
        let _mode = config.override_key(USE_DIRECT_SPAWN, true);

        let base = testing::instance();
        let persistent_point = extent!(persistent = 3).point_of_rank(2).unwrap();
        let mut environment = ActorEnvironment::default();
        environment
            .set(testactor::ACTOR_ENVIRONMENT_TEST_TAG, SENTINEL)
            .expect("seed persistent tag");
        environment
            .set(CAST_POINT, persistent_point.clone())
            .expect("seed persistent cast-point collision");
        let root = base
            .proc()
            .actor_instance_in_environment::<testing::TestRootClient>(
                &format!("direct_spawn_root_{}", Uuid::now_v7()),
                environment,
            )
            .expect("create seeded root instance");

        let mut hm = testing::host_mesh(1).await;
        let proc_mesh = hm
            .spawn(
                &root.instance,
                "direct_spawn_env",
                extent!(gpus = 2),
                None,
                None,
            )
            .await
            .expect("spawn proc mesh");
        let expected: HashMap<_, _> = (0..proc_mesh.region().num_ranks())
            .map(|rank| {
                (
                    proc_mesh.get(rank).expect("proc").proc_addr().to_string(),
                    proc_mesh.region().extent().point_of_rank(rank).unwrap(),
                )
            })
            .collect();
        assert!(
            !expected.values().any(|point| *point == persistent_point),
            "seeded and assigned cast points must be distinct",
        );

        let (reply, mut replies) = root
            .instance
            .open_port::<testactor::ActorEnvironmentObservation>();
        let mut actor_mesh: ActorMesh<testactor::ActorEnvironmentProbe> = proc_mesh
            .spawn_with_name(
                &root.instance,
                ActorMeshId::instance(Label::strip("direct_spawn_probe")),
                &testactor::ActorEnvironmentProbeParams {
                    label: "direct".to_string(),
                    reply: reply.bind(),
                    nested: None,
                },
                None,
                false,
            )
            .await
            .expect("spawn direct environment probes");
        assert!(
            actor_mesh.as_managed().is_none(),
            "configured instance spawn should return a data mesh"
        );

        let mut seen = HashSet::new();
        for _ in 0..expected.len() {
            let observed = tokio::time::timeout(Duration::from_secs(120), replies.recv())
                .await
                .expect("environment probe reply timed out")
                .expect("environment probe reply port closed");
            let expected_point = expected
                .get(&observed.proc_addr)
                .unwrap_or_else(|| panic!("unexpected probe proc {}", observed.proc_addr));
            assert!(
                seen.insert(observed.proc_addr.clone()),
                "received duplicate observation from {}",
                observed.proc_addr,
            );
            assert_eq!(
                observed.constructor_tag,
                Some(SENTINEL),
                "constructor must see the persistent tag"
            );
            assert_eq!(
                observed.constructor_point.as_ref(),
                Some(expected_point),
                "constructor must see its rank-local point, not the caller's"
            );
            assert_eq!(
                observed.stored_point.as_ref(),
                Some(expected_point),
                "instance must store its rank-local point"
            );
        }

        actor_mesh
            .stop(&root.instance, "test complete".to_string())
            .await
            .expect("stop direct environment probes");
        hm.shutdown(&root.instance)
            .await
            .expect("host mesh shutdown should complete");
    }

    /// Reuse a per-proc singleton service across two overlapping proc-mesh views
    /// whose shared proc is renumbered — a whole view and a `gpus` sub-slice, so
    /// the shared proc (gpu 1) is rank 1 in the whole and rank 0 in the slice.
    /// `spawn_service` creates one deduped actor per proc, so the slice must
    /// keep the whole's actor on the shared proc.
    ///
    /// One `host_mesh(1)` fixture drives three uniquely-named scenarios to keep
    /// the harness footprint minimal (one Tokio runtime, one host process, no
    /// process-global config override): reuse whole-then-slice, reuse
    /// slice-then-whole, and a deferred `WaitRankStatus` over the slice. Each
    /// spawn must resolve (a reused spawn positions its readiness overlay at the
    /// consuming-view rank, so the accumulator completes), the slice must
    /// reference the whole's shared actor with no new actors, and a Stopped-waiter
    /// registered while the service runs must flush at slice-local ranks.
    /// The assertions exercise the ProcAgent readiness path; the per-view
    /// controllers remain idle because this test does not trigger supervision.
    /// Snapshot assertions also verify that both reuse orders return actor state
    /// in the queried view's rank order.
    #[async_timed_test(timeout_secs = 120)]
    #[cfg(fbcode_build)]
    async fn test_singleton_service_reuse_over_overlapping_views() {
        let instance = testing::instance();
        let mut hm = testing::host_mesh(1).await;
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 2), None, None)
            .await
            .unwrap();
        let whole = proc_mesh.deref().clone();
        let slice = proc_mesh.range("gpus", 1..2).unwrap();

        // The slice reuses the whole's actor on the shared proc; no new actors.
        fn assert_shared(
            whole_mesh: &ActorMesh<testactor::TestActor>,
            slice_mesh: &ActorMesh<testactor::TestActor>,
        ) {
            let whole_ids: HashSet<_> = whole_mesh
                .values()
                .map(|actor_ref| actor_ref.actor_addr().id().clone())
                .collect();
            let slice_ids: HashSet<_> = slice_mesh
                .values()
                .map(|actor_ref| actor_ref.actor_addr().id().clone())
                .collect();
            assert_eq!(whole_ids.len(), 2, "whole view spans two procs");
            assert_eq!(slice_ids.len(), 1, "slice spans one proc");
            assert!(
                slice_ids.is_subset(&whole_ids),
                "slice reuses the shared proc's actor: {slice_ids:?} vs {whole_ids:?}",
            );
            assert_eq!(
                whole_ids.union(&slice_ids).count(),
                2,
                "no new actors created for the overlapping slice",
            );
        }

        // Scenario 1: spawn over the whole, then reuse over the renumbered slice.
        let w1 = whole
            .spawn_service::<testactor::TestActor, _>(instance, "svc_whole_first", &())
            .await
            .unwrap();
        let s1 = slice
            .spawn_service::<testactor::TestActor, _>(instance, "svc_whole_first", &())
            .await
            .unwrap();
        assert_shared(&w1, &s1);
        let s1_states = slice.actor_states(instance, s1.id().clone()).await.unwrap();
        for rank in 0..slice.region().num_ranks() {
            assert_eq!(
                s1_states
                    .get(rank)
                    .and_then(|state| state.state.as_ref())
                    .map(|state| &state.actor_id),
                s1.get(rank).map(|actor| actor.actor_addr()),
                "whole-first state is positioned at slice rank {rank}",
            );
        }

        // Scenario 2: spawn over the slice first, then reuse over the whole.
        let s2 = slice
            .spawn_service::<testactor::TestActor, _>(instance, "svc_slice_first", &())
            .await
            .unwrap();
        let w2 = whole
            .spawn_service::<testactor::TestActor, _>(instance, "svc_slice_first", &())
            .await
            .unwrap();
        assert_shared(&w2, &s2);
        let w2_states = whole.actor_states(instance, w2.id().clone()).await.unwrap();
        for rank in 0..whole.region().num_ranks() {
            assert_eq!(
                w2_states
                    .get(rank)
                    .and_then(|state| state.state.as_ref())
                    .map(|state| &state.actor_id),
                w2.get(rank).map(|actor| actor.actor_addr()),
                "slice-first state is positioned at whole-view rank {rank}",
            );
        }

        // Scenario 3: a `WaitRankStatus{Stopped}` registered over the renumbered
        // slice while the service runs is deferred and flushed on stop. The
        // flushed overlays land at slice-local ranks, within the slice region the
        // accumulator was opened over, so the wait completes. Registering before
        // the stop makes the deferred branch deterministic.
        whole
            .spawn_service::<testactor::TestActor, _>(instance, "svc_wait", &())
            .await
            .unwrap();
        let sw = slice
            .spawn_service::<testactor::TestActor, _>(instance, "svc_wait", &())
            .await
            .unwrap();
        let id = sw.id().resource_id().clone();
        let region = slice.region().clone();
        let num_ranks = region.num_ranks();
        let (port, rx) = instance.mailbox().open_idle_flush_accum_port(
            crate::StatusMesh::from_single(region.clone(), Status::NotExist),
            IdleFlushReducerOpts {
                idle_timeout: Duration::from_millis(50),
                abandon_timeout: Duration::from_secs(30),
                expected_updates_per_destination: NonZeroUsize::MIN,
            },
        );
        slice
            .agent_mesh()
            .cast(
                instance,
                crate::resource::WaitRankStatus {
                    id: id.clone(),
                    // Filled by the cast layer with this recipient's dense view rank.
                    rank: Default::default(),
                    min_status: Status::Stopped,
                    reply: port.bind(),
                },
            )
            .unwrap();
        slice
            .agent_mesh()
            .cast(
                instance,
                crate::resource::Stop {
                    id: id.clone(),
                    reason: "test".to_string(),
                },
            )
            .unwrap();
        let statuses = crate::resource::GetRankStatus::wait(
            rx,
            num_ranks,
            hyperactor_config::global::get(ACTOR_SPAWN_MAX_IDLE),
            region,
        )
        .await
        .unwrap();
        assert!(
            statuses.values().all(|s| s.is_terminating()),
            "deferred waiters flushed at slice-local ranks: {statuses:?}",
        );

        hm.shutdown(instance)
            .await
            .expect("host mesh shutdown should complete");
    }

    /// Actor-state replies match the queried view when the controller carries
    /// an IPv4 location and the worker reports IPv6 for the same actor identity.
    #[async_timed_test(timeout_secs = 120)]
    #[cfg(fbcode_build)]
    async fn test_actor_states_matches_actor_identity_across_ipv4_and_ipv6() {
        let instance = testing::instance();
        let config = hyperactor_config::global::lock();
        let _guard = config.override_key(
            DEFAULT_TRANSPORT,
            BindSpec::Addr(ChannelAddr::Tcp("[::1]:0".parse().unwrap())),
        );
        let mut host_mesh = HostMesh::local().await.unwrap();
        let mut worker_mesh = host_mesh
            .spawn(
                instance,
                "location_mismatch_worker",
                extent!(gpus = 1),
                None,
                None,
            )
            .await
            .unwrap();

        let worker_proc_addr = worker_mesh.get(0).unwrap().proc_addr().clone();
        let ChannelAddr::Tcp(worker_socket_addr) = worker_proc_addr.addr() else {
            panic!("worker proc must use TCP")
        };
        assert!(worker_socket_addr.is_ipv6(), "worker must use IPv6");
        let proc_agent: ActorRef<ProcAgent> =
            ActorRef::attest(worker_proc_addr.actor_addr(PROC_AGENT_ACTOR_NAME).clone());
        let controller_proc_addr = ProcAddr::new(
            worker_proc_addr.id().clone(),
            ChannelAddr::Tcp(SocketAddr::new(
                IpAddr::V4(Ipv4Addr::LOCALHOST),
                worker_socket_addr.port(),
            ))
            .into(),
        );
        let controller_view = ProcMeshRef::new_singleton(
            ProcMeshId::singleton(Label::strip("location_mismatch_view")),
            ProcRef::new(controller_proc_addr, 0, proc_agent),
        )
        .unwrap();

        let actor_mesh = controller_view
            .spawn_controllerless_service::<testactor::TestActor, _>(
                instance,
                "location_mismatch_actor",
                &(),
            )
            .await
            .unwrap();
        let expected_actor = actor_mesh.get(0).unwrap().actor_addr();
        let states = controller_view
            .actor_states(instance, actor_mesh.id().clone())
            .await
            .unwrap();
        let reported_actor = &states
            .get(0)
            .and_then(|state| state.state.as_ref())
            .unwrap()
            .actor_id;

        assert_ne!(
            reported_actor, expected_actor,
            "IPv6 worker and IPv4 controller addresses must differ",
        );
        assert_eq!(
            reported_actor.id(),
            expected_actor.id(),
            "worker and controller addresses must identify the same actor",
        );
        assert!(
            matches!(reported_actor.addr(), ChannelAddr::Tcp(addr) if addr.is_ipv6()),
            "worker must report an IPv6 location",
        );
        assert!(
            matches!(expected_actor.addr(), ChannelAddr::Tcp(addr) if addr.is_ipv4()),
            "controller must retain an IPv4 location",
        );

        worker_mesh
            .stop(instance, "test complete".to_string())
            .await
            .unwrap();
        host_mesh.shutdown(instance).await.unwrap();
    }

    #[async_timed_test(timeout_secs = 120)]
    #[cfg(fbcode_build)]
    async fn test_spawn_actor_casting() {
        let config = hyperactor_config::global::lock();
        let _guard = config.override_key(PROC_SPAWN_MAX_IDLE, Duration::from_secs(120));
        let _guard2 = config.override_key(
            hyperactor::config::HOST_SPAWN_READY_TIMEOUT,
            Duration::from_secs(120),
        );
        execute_spawn_actor().await;
    }

    /// Spawn an actor mesh, then do a random number of casts to bump the seq
    /// numbers for all actors participating in the cast. This avoids the test
    /// mistakenly passing.
    #[cfg(fbcode_build)]
    async fn spawn_for_seq_test(
        cx: &Instance<testing::TestRootClient>,
        proc_mesh: &super::ProcMeshRef,
    ) -> ActorMesh<testactor::TestActor> {
        let actor_mesh: ActorMesh<testactor::TestActor> =
            proc_mesh.spawn(cx, "test", &()).await.unwrap();

        let instance = cx
            .proc()
            .client(&format!("random_casts_{}", Uuid::now_v7()));
        let n = 1;
        for _ in 0..n {
            actor_mesh.cast(&instance, ()).unwrap();
        }
        println!(
            "did {} casts with sequencer session id {}",
            n,
            instance.sequencer().session_id()
        );
        actor_mesh
    }

    #[async_timed_test(timeout_secs = 60)]
    #[cfg(fbcode_build)]
    async fn test_seq_from_same_sender_to_different_meshes() {
        let config = hyperactor_config::global::lock();
        let _guard = config.override_key(PROC_SPAWN_MAX_IDLE, Duration::from_secs(60));
        let _guard2 = config.override_key(
            hyperactor::config::HOST_SPAWN_READY_TIMEOUT,
            Duration::from_secs(60),
        );

        hyperactor_telemetry::initialize_logging_for_test();
        let instance = testing::instance();
        let session_id = instance.sequencer().session_id();

        let mut hm = testing::host_mesh(2).await;
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 1), None, None)
            .await
            .unwrap();
        let proc_mesh_ref = proc_mesh.deref();

        // Sequence numbers are scoped based on the (client, dest) pair.
        // So casts to different meshes from the same client instance would
        // result in seq 1 for all casts.
        let handles = (0..3)
            .map(|_| {
                let proc_mesh_ref_clone = proc_mesh_ref.clone();
                tokio::spawn(async move {
                    let actor_mesh = spawn_for_seq_test(instance, &proc_mesh_ref_clone).await;
                    let expected_seqs = vec![1; 2];
                    testactor::assert_casting_correctness(
                        &actor_mesh,
                        instance,
                        Some((session_id, expected_seqs)),
                    )
                    .await;
                })
            })
            .collect::<Vec<_>>();
        futures::future::join_all(handles).await;

        let _ = hm.shutdown(instance).await;
    }

    /// Verify that the seq numbers are assigned correctly when we cast to
    /// different views of the same root mesh.
    #[async_timed_test(timeout_secs = 60)]
    #[cfg(fbcode_build)]
    async fn test_seq_from_same_sender_to_different_views() {
        let config = hyperactor_config::global::lock();
        let _guard = config.override_key(PROC_SPAWN_MAX_IDLE, Duration::from_secs(60));
        let _guard2 = config.override_key(
            hyperactor::config::HOST_SPAWN_READY_TIMEOUT,
            Duration::from_secs(60),
        );

        hyperactor_telemetry::initialize_logging_for_test();

        let instance = testing::instance();
        let session_id = instance.sequencer().session_id();

        let mut hm = testing::host_mesh(3).await;
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 1), None, None)
            .await
            .unwrap();

        let actor_mesh = spawn_for_seq_test(instance, &proc_mesh).await;

        // First cast. The seq should be 1 for all actors.
        let expected_seqs = vec![1; 3];
        testactor::assert_casting_correctness(
            &actor_mesh,
            instance,
            Some((session_id, expected_seqs)),
        )
        .await;

        // Verify casting to the sliced actor mesh
        let sliced_actor_mesh = actor_mesh.range("hosts", 1..3).unwrap();
        // Second cast. The seq should be 2 for actors in the sliced mesh.
        let expected_seqs = vec![2; 2];
        testactor::assert_casting_correctness(
            &sliced_actor_mesh,
            instance,
            Some((session_id, expected_seqs)),
        )
        .await;

        // Verify casting to a different sliced actor mesh
        let sliced_actor_mesh = actor_mesh.range("hosts", 0..2).unwrap();
        // For actors in the previous sliced mesh, the seq should be 3 since
        // this is the third cast for them. For other actors, the seq should
        // be 2.
        let expected_seqs = vec![2, 3];
        testactor::assert_casting_correctness(
            &sliced_actor_mesh,
            instance,
            Some((session_id, expected_seqs)),
        )
        .await;

        let _ = hm.shutdown(instance).await;
    }

    #[async_timed_test(timeout_secs = 60)]
    #[cfg(fbcode_build)]
    async fn test_seq_from_different_senders() {
        let config = hyperactor_config::global::lock();
        let _guard = config.override_key(PROC_SPAWN_MAX_IDLE, Duration::from_secs(60));
        let _guard2 = config.override_key(
            hyperactor::config::HOST_SPAWN_READY_TIMEOUT,
            Duration::from_secs(60),
        );

        hyperactor_telemetry::initialize_logging_for_test();

        use hyperactor::Proc;
        use hyperactor::channel::ChannelTransport;

        let proc = Proc::direct(ChannelTransport::Unix.any(), "test_0".to_string()).unwrap();
        let instance = proc
            .actor_instance::<testing::TestRootClient>("test_client")
            .unwrap()
            .instance;
        let first_instance = proc
            .actor_instance::<testing::TestRootClient>("first_client")
            .unwrap()
            .instance;
        let second_instance = proc
            .actor_instance::<testing::TestRootClient>("second_client")
            .unwrap()
            .instance;
        let third_instance = proc
            .actor_instance::<testing::TestRootClient>("third_client")
            .unwrap()
            .instance;

        let mut hm = testing::host_mesh(2).await;
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 1), None, None)
            .await
            .unwrap();

        let actor_mesh = spawn_for_seq_test(&instance, &proc_mesh).await;

        // Sequence numbers are calculated based on the sequencer, i.e. the
        // client name. So three casts would result in seq 1 for all actors.
        for inst in [&first_instance, &second_instance, &third_instance] {
            let expected_seqs = vec![1; 2];
            let session_id = inst.sequencer().session_id();
            testactor::assert_casting_correctness(
                &actor_mesh,
                inst,
                Some((session_id, expected_seqs)),
            )
            .await;
        }

        let _ = hm.shutdown(&instance).await;
    }

    #[cfg(fbcode_build)]
    #[assert_no_process_leak]
    #[tokio::test]
    async fn test_failing_spawn_actor() {
        hyperactor_telemetry::initialize_logging(hyperactor_telemetry::DefaultTelemetryClock {});

        let config = hyperactor_config::global::lock();
        let _mode = config.override_key(USE_DIRECT_SPAWN, false);
        let _guard = config.override_key(PROC_SPAWN_MAX_IDLE, Duration::from_secs(60));
        let _guard2 = config.override_key(
            hyperactor::config::HOST_SPAWN_READY_TIMEOUT,
            Duration::from_secs(60),
        );

        let instance = testing::instance();

        let mut hm = testing::host_mesh(1).await;
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 1), None, None)
            .await
            .unwrap();
        let err = proc_mesh
            .spawn::<testactor::FailingCreateTestActor, Instance<testing::TestRootClient>>(
                instance,
                "testfail",
                &(),
            )
            .await
            .unwrap_err();
        let statuses = err.into_actor_spawn_error().unwrap();
        assert_eq!(
            statuses,
            RankedValues::from((0..1, Status::Failed("test failure".to_string()))),
        );

        let _ = hm.shutdown(instance).await;
    }

    #[cfg(fbcode_build)]
    #[assert_no_process_leak]
    #[tokio::test]
    async fn test_failing_spawn_controllerless() {
        hyperactor_telemetry::initialize_logging(hyperactor_telemetry::DefaultTelemetryClock {});

        let config = hyperactor_config::global::lock();
        let _guard = config.override_key(PROC_SPAWN_MAX_IDLE, Duration::from_secs(60));
        let _guard2 = config.override_key(
            hyperactor::config::HOST_SPAWN_READY_TIMEOUT,
            Duration::from_secs(60),
        );

        let instance = testing::instance();

        let mut hm = testing::host_mesh(1).await;
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 1), None, None)
            .await
            .unwrap();
        // The controllerless path shares spawn_mesh_inner, so it must still run
        // the bounded spawn wait and surface the same ActorSpawnError shape on a
        // terminal init failure — identical to spawn/spawn_service.
        let err = proc_mesh
            .spawn_controllerless_service::<testactor::FailingCreateTestActor, _>(
                instance,
                "testfail_controllerless",
                &(),
            )
            .await
            .unwrap_err();
        let statuses = err.into_actor_spawn_error().unwrap();
        assert_eq!(
            statuses,
            RankedValues::from((0..1, Status::Failed("test failure".to_string()))),
        );

        let _ = hm.shutdown(instance).await;
    }

    #[async_timed_test(timeout_secs = 60)]
    #[cfg(fbcode_build)]
    async fn test_spawn_actor_on_proc_mesh_slice_only_spawns_slice_members() {
        let instance = testing::instance();

        let mut hm = testing::host_mesh(2).await;
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 2), None, None)
            .await
            .unwrap();
        let host1 = proc_mesh.range("hosts", 1..2).unwrap();
        let actor_name = crate::mesh_id::ActorMeshId::instance(
            hyperactor::id::Label::new("slice_only").unwrap(),
        );

        let actor_mesh = host1
            .spawn_with_name::<testactor::TestActor, _>(
                instance,
                actor_name.clone(),
                &(),
                None,
                false,
            )
            .await
            .unwrap();
        testactor::assert_casting_correctness(&actor_mesh, instance, None).await;

        let slice_states = host1
            .actor_states(instance, actor_name.clone())
            .await
            .unwrap();
        assert_eq!(slice_states.extent(), host1.extent());

        let err = proc_mesh
            .actor_states(instance, actor_name.clone())
            .await
            .unwrap_err();
        let expected_name = actor_name.into();
        match err {
            crate::Error::NotExist(name) if name == expected_name => {}
            other => panic!("expected NotExist for {expected_name}, got {other:?}"),
        }

        let _ = hm.shutdown(instance).await;
    }

    /// `proc_states` must resolve exactly the procs of the (possibly sliced)
    /// mesh it is called on — gpu-dim slices, host-dim slices, and
    /// slice-of-slice — rather than re-deriving a dense `0..n` from per-host
    /// counts or from the recipient's stamped (sliced-ordinal) rank.
    #[cfg(fbcode_build)]
    async fn assert_proc_states_match_slice<C: hyperactor::context::Actor>(
        mesh: &super::ProcMeshRef,
        cx: &C,
    ) {
        // The slice's own procs, by their original global rank.
        let mut expected: Vec<usize> = mesh.ranks.iter().map(|r| r.create_rank).collect();
        expected.sort();

        let states = mesh
            .states(cx, None)
            .await
            .unwrap()
            .expect("host-backed mesh yields Some");

        assert_eq!(states.extent(), mesh.extent());

        let mut got: Vec<usize> = states
            .values()
            .map(|s| {
                assert!(
                    !matches!(
                        s.status,
                        Status::NotExist | Status::Failed(_) | Status::Timeout(_)
                    ),
                    "rank should resolve to a live proc, got status {:?}",
                    s.status
                );
                s.state.expect("live proc carries ProcState").create_rank
            })
            .collect();

        got.sort();

        assert_eq!(
            got, expected,
            "proc_states must return exactly this (sub)mesh's procs"
        );
    }

    #[async_timed_test(timeout_secs = 60)]
    #[cfg(fbcode_build)]
    async fn test_proc_states_on_sliced_mesh() {
        let instance = testing::instance();
        let mut hm = testing::host_mesh(2).await;
        // 2 hosts x 4 gpus, host-major: global rank = host * 4 + gpu.
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 4), None, None)
            .await
            .unwrap();

        // Full mesh: ranks 0..8.
        assert_proc_states_match_slice(&proc_mesh, instance).await;
        // gpu-dim slice (gpus 2..4): ranks {2,3,6,7} — the original mis-derivation.
        assert_proc_states_match_slice(&proc_mesh.range("gpus", 2..4).unwrap(), instance).await;
        // host-dim slice (host 1): ranks {4,5,6,7} — the stamped-ordinal trap.
        assert_proc_states_match_slice(&proc_mesh.range("hosts", 1..2).unwrap(), instance).await;
        // slice-of-slice (host 1, gpus 2..4): ranks {6,7} — pins base-rank composition.
        assert_proc_states_match_slice(
            &proc_mesh
                .range("gpus", 2..4)
                .unwrap()
                .range("hosts", 1..2)
                .unwrap(),
            instance,
        )
        .await;

        let _ = hm.shutdown(instance).await;
    }

    #[test]
    fn test_python_class_from_supervision_name() {
        use super::python_class_from_supervision_name;

        assert_eq!(
            python_class_from_supervision_name("instance0.<my_module.MyWorker test_mesh>"),
            Some("Python<MyWorker>".to_string()),
        );
        assert_eq!(
            python_class_from_supervision_name(
                "instance0.<package.submodule.TrainingActor mesh_0>"
            ),
            Some("Python<TrainingActor>".to_string()),
        );
        // No angle brackets — not a Python supervision name.
        assert_eq!(python_class_from_supervision_name("plain_name"), None,);
        // Malformed: missing dot-qualified class name.
        assert_eq!(
            python_class_from_supervision_name("instance0.<NoModule mesh>"),
            None,
        );
    }

    // Compile-only proof that `spawn_controllerless_service` does NOT require
    // `C::A: Handler<MeshFailure>` — the whole point of the controllerless
    // path. This generic wrapper omits that bound and still type-checks; the
    // identical wrapper over `spawn_service` would fail to compile. (Generic
    // fn bodies are type-checked at definition, so this holds without ever
    // being called.)
    #[allow(dead_code)]
    async fn _spawn_controllerless_needs_no_mesh_failure_handler<A, C>(
        pm: &super::ProcMeshRef,
        cx: &C,
        name: &str,
        params: &A::Params,
    ) -> crate::Result<crate::ActorMesh<A>>
    where
        A: hyperactor::RemoteSpawn,
        A::Params: hyperactor::RemoteMessage,
        C: hyperactor::context::Actor,
    {
        pm.spawn_controllerless_service::<A, C>(cx, name, params)
            .await
    }

    #[cfg(fbcode_build)]
    async fn execute_spawn_controllerless() {
        hyperactor_telemetry::initialize_logging(hyperactor_telemetry::DefaultTelemetryClock {});

        let instance = testing::instance();

        let mut hm = testing::host_mesh(2).await;
        let proc_mesh = hm
            .spawn(&instance, "test", extent!(gpus = 1), None, None)
            .await
            .unwrap();

        // A controllerless service mesh has no ActorMeshController...
        let controllerless = proc_mesh
            .spawn_controllerless_service::<testactor::TestActor, _>(
                instance,
                "svc_controllerless",
                &(),
            )
            .await
            .unwrap();
        assert!(
            controllerless
                .as_managed()
                .expect("controllerless service mesh should be managed")
                .controller()
                .is_none(),
            "spawn_controllerless_service must not create a controller",
        );

        // ...whereas an ordinary service mesh does (contrast).
        let with_controller = proc_mesh
            .spawn_service::<testactor::TestActor, _>(instance, "svc_with_controller", &())
            .await
            .unwrap();
        assert!(
            with_controller
                .as_managed()
                .expect("service mesh should be managed")
                .controller()
                .is_some(),
            "spawn_service must create a controller",
        );

        let _ = hm.shutdown(instance).await;
    }

    #[async_timed_test(timeout_secs = 120)]
    #[cfg(fbcode_build)]
    async fn test_spawn_controllerless_service() {
        let config = hyperactor_config::global::lock();
        let _guard = config.override_key(PROC_SPAWN_MAX_IDLE, Duration::from_secs(120));
        let _guard2 = config.override_key(
            hyperactor::config::HOST_SPAWN_READY_TIMEOUT,
            Duration::from_secs(120),
        );
        execute_spawn_controllerless().await;
    }
}