hyperactor/

actor.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.
 */

#![allow(dead_code)] // Allow until this is used outside of tests.

//! This module contains all the core traits required to define and manage actors.

use std::any::TypeId;
use std::borrow::Cow;
use std::fmt;
use std::fmt::Debug;
use std::future::Future;
use std::future::IntoFuture;
use std::ops::Deref;
use std::pin::Pin;
use std::sync::Arc;
use std::time::SystemTime;

use async_trait::async_trait;
use enum_as_inner::EnumAsInner;
use futures::FutureExt;
use futures::future::BoxFuture;
use hyperactor_config::Flattrs;
use serde::Deserialize;
use serde::Serialize;
use tokio::sync::watch;
use tokio::task::JoinHandle;
use typeuri::Named;

use crate as hyperactor; // for macros
use crate::ActorAddr;
use crate::ActorRef;
use crate::Addr;
#[cfg(test)]
use crate::Client;
use crate::Data;
use crate::EndpointLocation;
use crate::Message;
use crate::RemoteMessage;
use crate::context;
use crate::endpoint::Endpoint;
use crate::mailbox::DeliveryFailure;
use crate::mailbox::DeliveryFailureKind;
use crate::mailbox::ExpiredDelivery;
use crate::mailbox::InvalidReference;
use crate::mailbox::MailboxError;
use crate::mailbox::MailboxSenderError;
use crate::mailbox::MessageEnvelope;
use crate::mailbox::PortHandle;
use crate::mailbox::TransportFailureReason;
use crate::mailbox::Undeliverable;
use crate::mailbox::UndeliverableReason;
use crate::proc::Context;
use crate::proc::HandlerPorts;
use crate::proc::Instance;
use crate::proc::InstanceCell;
use crate::proc::Proc;
use crate::supervision::ActorSupervisionEvent;

pub mod remote;

/// The shutdown mode requested for an actor.
#[derive(
    Clone,
    Copy,
    Debug,
    Serialize,
    Deserialize,
    PartialEq,
    Eq,
    typeuri::Named
)]
pub enum StopMode {
    /// Stop without draining ordinary queued work first.
    Stop,
    /// Stop after draining already accepted ordinary queued work.
    DrainAndStop,
}
wirevalue::register_type!(StopMode);

/// An Actor is an independent, asynchronous thread of execution. Each
/// actor instance has a mailbox, whose messages are delivered through
/// the method [`Actor::handle`].
///
/// Actors communicate with each other by way of message passing.
/// Actors are assumed to be _deterministic_: that is, the state of an
/// actor is determined by the set (and order) of messages it receives.
#[async_trait]
pub trait Actor: Sized + Send + 'static {
    /// Initialize the actor, after the runtime has been fully initialized.
    /// Init thus provides a mechanism by which an actor can reliably and always
    /// receive some initial event that can be used to kick off further
    /// (potentially delayed) processing.
    async fn init(&mut self, _this: &Instance<Self>) -> Result<(), anyhow::Error> {
        // Default implementation: no init.
        Ok(())
    }

    /// Handle a stop request from the runtime.
    ///
    /// The default implementation closes handler ingress and then
    /// either exits immediately or queues an exit after already
    /// accepted handler work drains. Actors that need to coordinate
    /// asynchronous shutdown work can override this method and call
    /// `Instance::exit()` / `Instance::exit_after_drain()` later,
    /// once they are ready to terminate.
    async fn handle_stop(
        &mut self,
        this: &Instance<Self>,
        mode: StopMode,
        reason: &str,
    ) -> Result<(), anyhow::Error> {
        handle_stop(this, mode, reason)
    }

    /// Cleanup things used by this actor before shutting down. Notably this function
    /// is async and allows more complex cleanup. Simpler cleanup can be handled
    /// by the impl Drop for this Actor.
    /// If err is not None, it is the error that this actor is failing with. Any
    /// errors returned by this function will be logged and ignored.
    /// If err is None, any errors returned by this function will be propagated
    /// as an ActorError.
    /// This function is not called if there is a panic in the actor, as the
    /// actor may be in an indeterminate state. It is also not called if the
    /// process is killed, there is no atexit handler or signal handler.
    async fn cleanup(
        &mut self,
        _this: &Instance<Self>,
        _err: Option<&ActorError>,
    ) -> Result<(), anyhow::Error> {
        // Default implementation: no cleanup.
        Ok(())
    }

    /// This method is used by the runtime to spawn the actor server. It can be
    /// used by actors that require customized runtime setups
    /// (e.g., dedicated actor threads), or want to use a custom tokio runtime.
    #[hyperactor::instrument_infallible]
    fn spawn_server_task<F>(future: F) -> JoinHandle<F::Output>
    where
        F: Future + Send + 'static,
        F::Output: Send + 'static,
    {
        tokio::spawn(future)
    }

    /// Handle actor supervision event. Return `Ok(true)`` if the event is handled here.
    async fn handle_supervision_event(
        &mut self,
        _this: &Instance<Self>,
        event: &ActorSupervisionEvent,
    ) -> Result<bool, anyhow::Error> {
        // Error events are not handled by default and bubble up to the parent.
        // Normal lifecycle events (e.g. clean stop) are absorbed.
        Ok(!event.is_error())
    }

    /// Default delivery-failure event handling behavior.
    async fn handle_delivery_failure_event(
        &mut self,
        cx: &Instance<Self>,
        undeliverable: Undeliverable<MessageEnvelope>,
    ) -> Result<(), anyhow::Error> {
        handle_delivery_failure_event(self, cx, undeliverable).await
    }

    /// Default undeliverable message handling behavior.
    async fn handle_undeliverable_message(
        &mut self,
        cx: &Instance<Self>,
        reason: UndeliverableReason,
        undeliverable: Undeliverable<MessageEnvelope>,
    ) -> Result<(), anyhow::Error> {
        handle_undeliverable_message(cx, reason, undeliverable)
    }

    /// Default invalid-reference handling behavior.
    async fn handle_invalid_reference(
        &mut self,
        cx: &Instance<Self>,
        invalid: InvalidReference,
        undeliverable: Undeliverable<MessageEnvelope>,
    ) -> Result<(), anyhow::Error> {
        handle_invalid_reference(cx, invalid, undeliverable)
    }

    /// Default expired-delivery handling behavior.
    async fn handle_expired_delivery(
        &mut self,
        cx: &Instance<Self>,
        expired: ExpiredDelivery,
        undeliverable: Undeliverable<MessageEnvelope>,
    ) -> Result<(), anyhow::Error> {
        handle_expired_delivery(cx, expired, undeliverable)
    }

    /// If overridden, we will use this name in place of the
    /// ActorAddr for talking about this actor in supervision error
    /// messages.
    fn display_name(&self) -> Option<String> {
        None
    }
}

/// Default implementation of [`Actor::handle_delivery_failure_event`]. Defined
/// as a free function so that `Actor` implementations that override
/// [`Actor::handle_delivery_failure_event`] can fallback to this default.
pub async fn handle_delivery_failure_event<A: Actor>(
    actor: &mut A,
    cx: &Instance<A>,
    undeliverable: Undeliverable<MessageEnvelope>,
) -> Result<(), anyhow::Error> {
    match undeliverable
        .root_delivery_failure()
        .map(|failure| failure.kind.clone())
    {
        Some(DeliveryFailureKind::InvalidReference(invalid)) => {
            actor
                .handle_invalid_reference(cx, invalid, undeliverable)
                .await
        }
        Some(DeliveryFailureKind::Expired(expired)) => {
            actor
                .handle_expired_delivery(cx, expired, undeliverable)
                .await
        }
        Some(DeliveryFailureKind::Undeliverable(reason)) => {
            actor
                .handle_undeliverable_message(cx, reason, undeliverable)
                .await
        }
        None => anyhow::bail!(undeliverable.into_error()),
    }
}

fn delivery_failure_event_target(undeliverable: &Undeliverable<MessageEnvelope>) -> Addr {
    match undeliverable {
        Undeliverable::Returned(envelope) => envelope.dest().clone().into(),
        Undeliverable::Report(report) => match &report.dest {
            EndpointLocation::Actor(actor) => actor.clone().into(),
            EndpointLocation::Port(port) => port.clone().into(),
            EndpointLocation::Local { actor, .. } => actor.clone().into(),
        },
    }
}

/// Default implementation of [`Actor::handle_undeliverable_message`]. Defined
/// as a free function so that `Actor` implementations that override
/// [`Actor::handle_undeliverable_message`] can fallback to this default.
pub fn handle_undeliverable_message<A: Actor>(
    _cx: &Instance<A>,
    reason: UndeliverableReason,
    undeliverable: Undeliverable<MessageEnvelope>,
) -> Result<(), anyhow::Error> {
    if undeliverable_reason_fails_actor(&reason) {
        anyhow::bail!(undeliverable.into_error());
    }
    Ok(())
}

fn undeliverable_reason_fails_actor(reason: &UndeliverableReason) -> bool {
    matches!(
        reason,
        UndeliverableReason::Transport(transport)
            if matches!(
                &transport.reason,
                TransportFailureReason::OversizedFrame { .. }
            )
    )
}

/// Default implementation of [`Actor::handle_invalid_reference`]. Defined
/// as a free function so that `Actor` implementations that override
/// [`Actor::handle_invalid_reference`] can fallback to this default.
pub fn handle_invalid_reference<A: Actor>(
    _cx: &Instance<A>,
    _invalid: InvalidReference,
    undeliverable: Undeliverable<MessageEnvelope>,
) -> Result<(), anyhow::Error> {
    anyhow::bail!(undeliverable.into_error())
}

/// Default implementation of [`Actor::handle_expired_delivery`]. Defined
/// as a free function so that `Actor` implementations that override
/// [`Actor::handle_expired_delivery`] can fallback to this default.
pub fn handle_expired_delivery<A: Actor>(
    _cx: &Instance<A>,
    _expired: ExpiredDelivery,
    undeliverable: Undeliverable<MessageEnvelope>,
) -> Result<(), anyhow::Error> {
    anyhow::bail!(undeliverable.into_error())
}

/// Default implementation of [`Actor::handle_stop`]. Defined as a free
/// function so that `Actor` implementations that override
/// [`Actor::handle_stop`] can fall back to this default.
pub fn handle_stop<A: Actor>(
    this: &Instance<A>,
    mode: StopMode,
    reason: &str,
) -> Result<(), anyhow::Error> {
    // After `close`, no more messages may be enqueued.
    // exit_after_drain will drain any pending messages before exiting.
    this.close();
    match mode {
        StopMode::Stop => this.exit(reason).map_err(anyhow::Error::from),
        StopMode::DrainAndStop => this.exit_after_drain(reason).map_err(anyhow::Error::from),
    }
}

/// An actor that does nothing. It is used to represent "client only" actors,
/// returned by [`Proc::client`].
#[async_trait]
impl Actor for () {}

impl Referable for () {}

impl Binds<()> for () {
    fn bind(_ports: &HandlerPorts<Self>) {
        // Binds no ports.
    }
}

/// A Handler allows an actor to handle a specific message type.
#[async_trait]
pub trait Handler<M>: Actor {
    /// Handle the next M-typed message.
    async fn handle(&mut self, cx: &Context<Self>, message: M) -> Result<(), anyhow::Error>;
}

/// Blanket Handler impls for bypass-workq message types. Since these messages
/// bypass workq, they will never be sent to actor's handler.
///
/// These exist solely to lock the `Handler<M>` coherence slot for each bypass
/// type, so no specific `impl Handler<BypassType> for SomeActor` can be written.
/// The actual delivery for these types goes through dedicated channels set up
/// in `Instance::new`, not through Handler. See the matching sender-side check
/// in [crate::ordering::Sequencer::assign_seq] and the registry of bypass types
/// in [crate::ordering::is_bypass_workq_type_id].
#[async_trait]
impl<A: Actor> Handler<crate::introspect::IntrospectMessage> for A {
    async fn handle(
        &mut self,
        _cx: &Context<Self>,
        _message: crate::introspect::IntrospectMessage,
    ) -> Result<(), anyhow::Error> {
        unimplemented!("introspect message handler should not be called directly")
    }
}

#[cfg(test)]
#[derive(Debug)]
enum DeliveryFailurePolicy {
    InvalidReference,
    Expired,
    Undeliverable,
}

#[cfg(test)]
fn delivery_failure_policy(message: &Undeliverable<MessageEnvelope>) -> DeliveryFailurePolicy {
    match message.root_delivery_failure().map(|failure| &failure.kind) {
        Some(DeliveryFailureKind::InvalidReference(_)) => DeliveryFailurePolicy::InvalidReference,
        Some(DeliveryFailureKind::Expired(_)) => DeliveryFailurePolicy::Expired,
        Some(DeliveryFailureKind::Undeliverable(_)) | None => DeliveryFailurePolicy::Undeliverable,
    }
}

struct DeliveryFailureLogFields {
    sender: ActorAddr,
    dest: EndpointLocation,
    error: DeliveryFailureLogError,
}

impl DeliveryFailureLogFields {
    fn new(message: &Undeliverable<MessageEnvelope>) -> Self {
        match message {
            Undeliverable::Returned(envelope) => Self {
                sender: envelope.sender().clone(),
                dest: EndpointLocation::Port(envelope.dest().clone()),
                error: DeliveryFailureLogError::DeliveryFailures(
                    envelope.delivery_failures().to_vec(),
                ),
            },
            Undeliverable::Report(report) => Self {
                sender: report.sender.clone(),
                dest: report.dest.clone(),
                error: DeliveryFailureLogError::DeliveryFailures(report.delivery_failures.clone()),
            },
        }
    }
}

enum DeliveryFailureLogError {
    DeliveryFailures(Vec<DeliveryFailure>),
}

impl fmt::Display for DeliveryFailureLogError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::DeliveryFailures(failures) => {
                if failures.is_empty() {
                    return write!(f, "<none>");
                }
                for (index, failure) in failures.iter().enumerate() {
                    if index > 0 {
                        write!(f, "; ")?;
                    }
                    write!(f, "{}", failure)?;
                }
                Ok(())
            }
        }
    }
}

/// This handler provides a default behavior when a message sent by
/// the actor to another is returned due to delivery failure.
#[async_trait]
impl<A: Actor> Handler<Undeliverable<MessageEnvelope>> for A {
    async fn handle(
        &mut self,
        cx: &Context<Self>,
        message: Undeliverable<MessageEnvelope>,
    ) -> Result<(), anyhow::Error> {
        let log_fields = (tracing::enabled!(tracing::Level::DEBUG)
            || tracing::enabled!(tracing::Level::ERROR))
        .then(|| DeliveryFailureLogFields::new(&message));
        let result = self.handle_delivery_failure_event(cx, message).await;
        match result {
            Ok(_) => {
                if let Some(log_fields) = log_fields {
                    tracing::debug!(
                        actor_id = %cx.self_addr(),
                        name = "undeliverable_message_handled",
                        sender = %log_fields.sender,
                        dest = %log_fields.dest,
                        error = %log_fields.error,
                    );
                }
                Ok(())
            }
            Err(e) => {
                if let Some(log_fields) = log_fields {
                    tracing::error!(
                        actor_id = %cx.self_addr(),
                        name = "undeliverable_message",
                        sender = %log_fields.sender,
                        dest = %log_fields.dest,
                        error = %log_fields.error,
                        handler_error = %e,
                    );
                } else {
                    tracing::error!(
                        actor_id = %cx.self_addr(),
                        name = "undeliverable_message",
                        handler_error = %e,
                    );
                }
                Err(e)
            }
        }
    }
}

/// An `Actor` that can be spawned remotely.
///
/// Bounds explained:
/// - `Actor`: only actors may be remotely spawned.
/// - `Referable`: marks the type as eligible for typed remote
///   references (`ActorRef<A>`); required because remote spawn
///   ultimately hands back an `ActorAddr` that higher-level APIs may
///   re-type as `ActorRef<A>`.
/// - `Binds<Self>`: lets the runtime wire this actor's handler ports
///   when it is spawned (the blanket impl calls `handle.bind::<Self>()`).
///
/// `gspawn_root_bind` is a type-erased entry point used by the remote
/// spawn/registry machinery. It takes serialized params and returns
/// the new actor's `ActorAddr`; application code shouldn't call it
/// directly.
#[async_trait]
pub trait RemoteSpawn: Actor + Referable + Binds<Self> {
    /// The type of parameters used to instantiate the actor remotely.
    type Params: RemoteMessage;

    /// Creates a new actor instance given its instantiation parameters.
    /// The `environment` allows whoever is responsible for spawning this actor
    /// to pass in additional context that may be useful.
    async fn new(params: Self::Params, environment: Flattrs) -> anyhow::Result<Self>;

    /// A type-erased entry point to spawn this actor as a root. This is
    /// primarily used by hyperactor's remote actor registration
    /// mechanism.
    // TODO: consider making this 'private' -- by moving it into a non-public trait as in [`cap`].
    fn gspawn_root_bind(
        proc: &Proc,
        uid: crate::id::Uid,
        serialized_params: Data,
        environment: Flattrs,
    ) -> Pin<Box<dyn Future<Output = Result<ActorAddr, anyhow::Error>> + Send>> {
        let proc = proc.clone();
        Box::pin(async move {
            let params =
                bincode::serde::decode_from_slice(&serialized_params, bincode::config::legacy())
                    .map(|(v, _)| v)?;
            let actor = Self::new(params, environment).await?;
            let handle = proc.spawn_with_uid(uid, actor)?;
            // We return only the ActorAddr, not a typed ActorRef.
            // Callers that hold this ID can interact with the actor
            // only via the serialized/opaque messaging path, which
            // makes it safe to export across process boundaries.
            //
            // Note: the actor itself is still `A`-typed here; we
            // merely restrict the *capability* we hand out to an
            // untyped identifier.
            //
            // This will be replaced by a proper export/registry
            // mechanism.
            Ok(handle.bind::<Self>().into_actor_addr())
        })
    }

    /// A type-erased entry point to spawn this actor as a child.
    ///
    /// The returned handle is lifecycle-only; callers that know the concrete
    /// actor type can recover a typed handle with [`AnyActorHandle::downcast`].
    fn gspawn_child(
        proc: &Proc,
        parent: InstanceCell,
        uid: crate::id::Uid,
        serialized_params: Data,
        environment: Flattrs,
    ) -> Pin<Box<dyn Future<Output = Result<AnyActorHandle, anyhow::Error>> + Send>> {
        let proc = proc.clone();
        Box::pin(async move {
            let params =
                bincode::serde::decode_from_slice(&serialized_params, bincode::config::legacy())
                    .map(|(v, _)| v)?;
            let actor = Self::new(params, environment).await?;
            let handle = proc.spawn_child_with_uid(parent, uid, actor)?;
            handle.bind::<Self>();
            Ok(handle.into_any())
        })
    }

    /// The type ID of this actor.
    fn get_type_id() -> TypeId {
        TypeId::of::<Self>()
    }
}

/// If an actor implements Default, we use this as the
/// `RemoteSpawn` implementation, too.
#[async_trait]
impl<A: Actor + Referable + Binds<Self> + Default> RemoteSpawn for A {
    type Params = ();

    async fn new(_params: Self::Params, _environment: Flattrs) -> anyhow::Result<Self> {
        Ok(Default::default())
    }
}

/// Errors that occur while serving actors. Each error is associated
/// with the ID of the actor being served.
#[derive(Debug)]
pub struct ActorError {
    /// The ActorAddr for the actor that generated this error.
    pub actor_id: Box<ActorAddr>,
    /// The kind of error that occurred.
    pub kind: Box<ActorErrorKind>,
}

/// The kinds of actor serving errors.
#[derive(thiserror::Error, Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum ActorErrorKind {
    /// Generic error with a formatted message.
    #[error("{0}")]
    Generic(String),

    /// An error that occurred while trying to handle a supervision event.
    #[error("{0} while handling {1}")]
    ErrorDuringHandlingSupervision(String, Box<ActorSupervisionEvent>),

    /// The actor did not attempt to handle
    #[error("{0}")]
    UnhandledSupervisionEvent(Box<ActorSupervisionEvent>),

    /// The actor was explicitly aborted with the provided reason.
    #[error("actor explicitly aborted due to: {0}")]
    Aborted(String),

    /// The actor's signal channel was closed before the actor loop exited
    /// normally.
    #[error("signal channel closed")]
    SignalChannelClosed,
}

impl ActorErrorKind {
    /// Error while processing actor, i.e., returned by the actor's
    /// processing method.
    pub fn processing(err: anyhow::Error) -> Self {
        // Unbox err from the anyhow err. Check if it is an ActorErrorKind object.
        // If it is directly use it as the new ActorError's ActorErrorKind.
        // This lets us directly pass the ActorErrorKind::UnhandledSupervisionEvent
        // up the handling infrastructure.
        err.downcast::<ActorErrorKind>()
            .unwrap_or_else(|err| Self::Generic(err.to_string()))
    }

    /// Unwound stracktrace of a panic.
    pub fn panic(err: anyhow::Error) -> Self {
        Self::Generic(format!("panic: {}", err))
    }

    /// Error during actor initialization.
    pub fn init(err: anyhow::Error) -> Self {
        Self::Generic(format!("initialization error: {}", err))
    }

    /// Error during actor cleanup.
    pub fn cleanup(err: anyhow::Error) -> Self {
        Self::Generic(format!("cleanup error: {}", err))
    }

    /// An underlying mailbox error.
    pub fn mailbox(err: MailboxError) -> Self {
        Self::Generic(err.to_string())
    }

    /// An underlying mailbox sender error.
    pub fn mailbox_sender(err: MailboxSenderError) -> Self {
        Self::Generic(err.to_string())
    }

    /// The actor's state could not be determined.
    pub fn indeterminate_state() -> Self {
        Self::Generic("actor is in an indeterminate state".to_string())
    }
}

impl ActorError {
    /// Create a new actor server error with the provided id and kind.
    pub(crate) fn new(actor_id: &ActorAddr, kind: ActorErrorKind) -> Self {
        Self {
            actor_id: Box::new(actor_id.clone()),
            kind: Box::new(kind),
        }
    }
}

impl fmt::Display for ActorError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "serving {}: ", self.actor_id)?;
        fmt::Display::fmt(&self.kind, f)
    }
}

impl std::error::Error for ActorError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        self.kind.source()
    }
}

impl From<MailboxError> for ActorError {
    fn from(inner: MailboxError) -> Self {
        Self {
            actor_id: Box::new(inner.actor_addr().clone()),
            kind: Box::new(ActorErrorKind::mailbox(inner)),
        }
    }
}

impl From<MailboxSenderError> for ActorError {
    fn from(inner: MailboxSenderError) -> Self {
        Self {
            actor_id: Box::new(inner.location().actor_addr()),
            kind: Box::new(ActorErrorKind::mailbox_sender(inner)),
        }
    }
}

/// A collection of signals to control the behavior of the actor.
/// Signals are internal runtime control plane messages and should not be
/// sent outside of the runtime.
///
/// These messages are not handled directly by actors; instead, the runtime
/// handles the various signals.
#[derive(Clone, Debug, Serialize, Deserialize, typeuri::Named)]
pub enum Signal {
    /// Stop the actor, after draining messages.
    DrainAndStop(String),

    /// Stop the actor immediately.
    Stop(String),

    /// Exit the actor loop with the provided stop reason.
    ExitRequested(String),

    /// The direct child with the given uid was stopped.
    ChildStopped(crate::id::Uid),

    /// Kill the actor. This will exit the actor loop with an error,
    /// causing a supervision event to propagate up the supervision
    /// hierarchy.
    Kill(String),
}

impl fmt::Display for Signal {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Signal::DrainAndStop(reason) => write!(f, "DrainAndStop({})", reason),
            Signal::Stop(reason) => write!(f, "Stop({})", reason),
            Signal::ExitRequested(reason) => write!(f, "ExitRequested({})", reason),
            Signal::ChildStopped(uid) => write!(f, "ChildStopped({})", uid),
            Signal::Kill(reason) => write!(f, "Kill({})", reason),
        }
    }
}

/// Information about a message handler being processed.
///
/// Uses `Cow<'static, str>` to avoid string copies on the hot path.
/// The typename and arm are typically static strings from `TypeInfo`.
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
pub struct HandlerInfo {
    /// The type name of the message being handled.
    pub typename: Cow<'static, str>,
    /// The enum arm being handled, if the message is an enum.
    pub arm: Option<Cow<'static, str>>,
}

impl HandlerInfo {
    /// Create a new `HandlerInfo` from static strings (zero-copy).
    pub fn from_static(typename: &'static str, arm: Option<&'static str>) -> Self {
        Self {
            typename: Cow::Borrowed(typename),
            arm: arm.map(Cow::Borrowed),
        }
    }

    /// Create a new `HandlerInfo` from owned strings.
    pub fn from_owned(typename: String, arm: Option<String>) -> Self {
        Self {
            typename: Cow::Owned(typename),
            arm: arm.map(Cow::Owned),
        }
    }
}

impl fmt::Display for HandlerInfo {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match &self.arm {
            Some(arm) => write!(f, "{}.{}", self.typename, arm),
            None => write!(f, "{}", self.typename),
        }
    }
}

/// Why an actor is stopping.
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
pub enum ActorStoppingReason {
    /// The actor is stopping through the normal cooperative shutdown path.
    Requested,
    /// The actor did not respond to hard kill, and teardown stopped waiting on
    /// it normally.
    Zombie(String),
}

impl fmt::Display for ActorStoppingReason {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Requested => write!(f, "requested"),
            Self::Zombie(reason) => write!(f, "zombie: {}", reason),
        }
    }
}

/// The runtime status of an actor.
#[derive(
    Debug,
    Serialize,
    Deserialize,
    PartialEq,
    Eq,
    Clone,
    typeuri::Named,
    EnumAsInner
)]
pub enum ActorStatus {
    /// The actor status is unknown.
    Unknown,
    /// The actor was created, but not yet started.
    Created,
    /// The actor is initializing. It is not yet ready to receive messages.
    Initializing,
    /// The actor is in "client" state: the user is managing the actor's
    /// mailboxes manually.
    Client,
    /// The actor is ready to receive messages, but is currently idle.
    Idle,
    /// The actor has been processing a message, beginning at the specified
    /// instant. The message handler info is included.
    Processing(SystemTime, Option<HandlerInfo>),
    /// The actor is stopping. It is draining messages.
    Stopping(ActorStoppingReason),
    /// The actor is stopped with a provided reason.
    /// It is no longer processing messages.
    Stopped(String),
    /// The actor failed with the provided actor error.
    Failed(ActorErrorKind),
}

impl ActorStatus {
    /// Tells whether the status is a terminal state.
    pub fn is_terminal(&self) -> bool {
        self.is_stopped() || self.is_failed()
    }

    /// Create a normal stopping status.
    pub fn stopping() -> Self {
        Self::Stopping(ActorStoppingReason::Requested)
    }

    /// Create a zombie stopping status.
    pub fn zombie(reason: impl Into<String>) -> Self {
        Self::Stopping(ActorStoppingReason::Zombie(reason.into()))
    }

    /// Tells whether the status is a zombie stopping state.
    pub fn is_zombie(&self) -> bool {
        matches!(self, Self::Stopping(ActorStoppingReason::Zombie(_)))
    }

    /// Create a generic failure status with the provided error message.
    pub fn generic_failure(message: impl Into<String>) -> Self {
        Self::Failed(ActorErrorKind::Generic(message.into()))
    }

    fn span_string(&self) -> &'static str {
        self.arm().unwrap_or_default()
    }
}

impl fmt::Display for ActorStatus {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Unknown => write!(f, "unknown"),
            Self::Created => write!(f, "created"),
            Self::Initializing => write!(f, "initializing"),
            Self::Client => write!(f, "client"),
            Self::Idle => write!(f, "idle"),
            Self::Processing(instant, None) => {
                write!(
                    f,
                    "processing for {}ms",
                    std::time::SystemTime::now()
                        .duration_since(*instant)
                        .unwrap_or_default()
                        .as_millis()
                )
            }
            Self::Processing(instant, Some(handler_info)) => {
                write!(
                    f,
                    "{}: processing for {}ms",
                    handler_info,
                    std::time::SystemTime::now()
                        .duration_since(*instant)
                        .unwrap_or_default()
                        .as_millis()
                )
            }
            Self::Stopping(ActorStoppingReason::Requested) => write!(f, "stopping"),
            Self::Stopping(ActorStoppingReason::Zombie(reason)) => write!(f, "zombie: {}", reason),
            Self::Stopped(reason) => write!(f, "stopped: {}", reason),
            Self::Failed(err) => write!(f, "failed: {}", err),
        }
    }
}

/// ActorHandles represent a (local) serving actor. It is used to access
/// its messaging and signal ports, as well as to synchronize with its
/// lifecycle (e.g., providing joins).  Once dropped, the handle is
/// detached from the underlying actor instance, and there is no longer
/// any way to join it.
///
/// Correspondingly, [`crate::ActorAddr`]s refer to (possibly) remote
/// actors.
pub struct ActorHandle<A: Actor> {
    cell: InstanceCell,
    ports: Arc<HandlerPorts<A>>,
}

/// A handle to a running (local) actor.
impl<A: Actor> ActorHandle<A> {
    pub(crate) fn new(cell: InstanceCell, ports: Arc<HandlerPorts<A>>) -> Self {
        Self { cell, ports }
    }

    /// The actor's cell. Used primarily for testing.
    /// TODO: this should not be a public API.
    pub(crate) fn cell(&self) -> &InstanceCell {
        &self.cell
    }

    /// The [`ActorAddr`] of the actor represented by this handle.
    pub fn actor_addr(&self) -> &ActorAddr {
        self.cell.actor_addr()
    }

    /// Signal the actor to drain its current messages and then stop.
    pub fn drain_and_stop(&self, reason: &str) -> Result<(), ActorError> {
        tracing::info!("ActorHandle::drain_and_stop called: {}", self.actor_addr());
        self.cell.signal(Signal::DrainAndStop(reason.to_string()))
    }

    /// Signal the actor to stop without draining ordinary queued
    /// work first.
    pub fn stop(&self, reason: &str) -> Result<(), ActorError> {
        tracing::info!("actor handle stop called: {}", self.actor_addr());
        self.cell.signal(Signal::Stop(reason.to_string()))
    }

    /// Signal the actor to terminate immediately.
    pub fn kill(&self, reason: &str) -> Result<(), ActorError> {
        tracing::info!("actor handle kill called: {}", self.actor_addr());
        self.cell.signal(Signal::Kill(reason.to_string()))
    }

    /// A watch that observes the lifecycle state of the actor.
    pub fn status(&self) -> watch::Receiver<ActorStatus> {
        self.cell.status().clone()
    }

    /// Return a port for the provided message type handled by the actor.
    pub fn port<M: Message>(&self) -> PortHandle<M>
    where
        A: Handler<M>,
    {
        self.ports.get()
    }

    /// Post `message` to this actor's handler port for `M`, returning an error
    /// if delivery fails (the actor has stopped, its mailbox is closed, or the
    /// underlying channel is disconnected). Unlike [`Endpoint::post`], the
    /// caller observes the failure instead of having it reported through the
    /// actor's lost-message channel.
    pub fn try_post<C, M>(&self, cx: &C, message: M) -> Result<(), MailboxSenderError>
    where
        C: context::Actor,
        M: Message,
        A: Handler<M>,
    {
        self.ports.get::<M>().try_post(cx, message)
    }

    /// TEMPORARY: bind...
    /// TODO: we shoudl also have a default binding(?)
    pub fn bind<R: Binds<A>>(&self) -> ActorRef<R> {
        self.cell.bind(self.ports.as_ref())
    }

    /// Erase this handle's actor type, preserving only lifecycle access.
    pub fn into_any(self) -> AnyActorHandle {
        AnyActorHandle { cell: self.cell }
    }

    /// Convert this handle into a guard that stops the actor when dropped.
    ///
    /// Dropping the returned guard sends a normal stop signal. The guard does
    /// not wait for the actor to stop.
    pub fn into_guard(self) -> ActorGuard<A> {
        ActorGuard { handle: Some(self) }
    }
}

/// A guard that stops an actor when dropped.
pub struct ActorGuard<A: Actor> {
    handle: Option<ActorHandle<A>>,
}

impl<A: Actor> ActorGuard<A> {
    /// Return the actor handle without stopping the actor.
    pub fn into_inner(mut self) -> ActorHandle<A> {
        self.handle
            .take()
            .expect("actor guard must contain a handle")
    }
}

impl<A: Actor> Deref for ActorGuard<A> {
    type Target = ActorHandle<A>;

    fn deref(&self) -> &Self::Target {
        self.handle
            .as_ref()
            .expect("actor guard must contain a handle")
    }
}

impl<A: Actor> Drop for ActorGuard<A> {
    fn drop(&mut self) {
        if let Some(handle) = self.handle.take()
            && let Err(err) = handle.stop("actor guard dropped")
        {
            tracing::debug!(
                actor_id = %handle.actor_addr(),
                "actor guard failed to stop actor: {}",
                err
            );
        }
    }
}

impl<A: Actor> Debug for ActorGuard<A> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
        f.debug_struct("ActorGuard")
            .field(
                "actor_id",
                &self.handle.as_ref().map(|handle| handle.actor_addr()),
            )
            .finish()
    }
}

/// A type-erased handle to a running actor whose concrete type is erased.
///
/// This handle intentionally does not expose typed messaging or binding APIs.
/// Use [`AnyActorHandle::downcast`] to recover a typed [`ActorHandle`] when the
/// concrete actor type is known.
pub struct AnyActorHandle {
    cell: InstanceCell,
}

impl AnyActorHandle {
    /// The [`ActorAddr`] of the actor represented by this handle.
    pub fn actor_id(&self) -> &ActorAddr {
        self.cell.actor_addr()
    }

    /// Signal the actor to drain its current messages and then stop.
    pub fn drain_and_stop(&self, reason: &str) -> Result<(), ActorError> {
        self.cell.signal(Signal::DrainAndStop(reason.to_string()))
    }

    /// Signal the actor to stop without draining ordinary queued work first.
    pub fn stop(&self, reason: &str) -> Result<(), ActorError> {
        self.cell.signal(Signal::Stop(reason.to_string()))
    }

    /// Signal the actor to terminate immediately.
    pub fn kill(&self, reason: &str) -> Result<(), ActorError> {
        self.cell.signal(Signal::Kill(reason.to_string()))
    }

    /// A watch that observes the lifecycle state of the actor.
    pub fn status(&self) -> watch::Receiver<ActorStatus> {
        self.cell.status().clone()
    }

    /// Attempt to recover a typed actor handle.
    pub fn downcast<A: Actor>(&self) -> Option<ActorHandle<A>> {
        self.cell.downcast_handle()
    }

    /// Convert this handle into a guard that stops the actor when dropped.
    ///
    /// Dropping the returned guard sends a normal stop signal. The guard does
    /// not wait for the actor to stop.
    pub fn into_guard(self) -> AnyActorGuard {
        AnyActorGuard { handle: Some(self) }
    }
}

/// A type-erased guard that stops an actor when dropped.
pub struct AnyActorGuard {
    handle: Option<AnyActorHandle>,
}

impl AnyActorGuard {
    /// Return the actor handle without stopping the actor.
    pub fn into_inner(mut self) -> AnyActorHandle {
        self.handle
            .take()
            .expect("actor guard must contain a handle")
    }
}

impl Deref for AnyActorGuard {
    type Target = AnyActorHandle;

    fn deref(&self) -> &Self::Target {
        self.handle
            .as_ref()
            .expect("actor guard must contain a handle")
    }
}

impl Drop for AnyActorGuard {
    fn drop(&mut self) {
        if let Some(handle) = self.handle.take()
            && let Err(err) = handle.stop("actor guard dropped")
        {
            tracing::debug!(
                actor_id = %handle.actor_id(),
                "actor guard failed to stop actor: {}",
                err
            );
        }
    }
}

impl Debug for AnyActorGuard {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
        f.debug_struct("AnyActorGuard")
            .field(
                "actor_id",
                &self.handle.as_ref().map(|handle| handle.actor_id()),
            )
            .finish()
    }
}

/// IntoFuture allows users to await the handle to join it. The future
/// resolves when the actor runtime has fully stopped.
/// The future resolves to the actor's final status.
impl IntoFuture for AnyActorHandle {
    type Output = ActorStatus;
    type IntoFuture = BoxFuture<'static, Self::Output>;

    fn into_future(self) -> Self::IntoFuture {
        let future = async move {
            let mut status_receiver = self.cell.status().clone();
            let result = status_receiver.wait_for(ActorStatus::is_terminal).await;
            match result {
                Err(_) => ActorStatus::Unknown,
                Ok(status) => status.clone(),
            }
        };

        future.boxed()
    }
}

impl Debug for AnyActorHandle {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
        f.debug_struct("AnyActorHandle")
            .field("cell", &"..")
            .finish()
    }
}

impl Clone for AnyActorHandle {
    fn clone(&self) -> Self {
        Self {
            cell: self.cell.clone(),
        }
    }
}

/// IntoFuture allows users to await the handle to join it. The future
/// resolves when the actor runtime has fully stopped.
/// The future resolves to the actor's final status.
impl<A: Actor> IntoFuture for ActorHandle<A> {
    type Output = ActorStatus;
    type IntoFuture = BoxFuture<'static, Self::Output>;

    fn into_future(self) -> Self::IntoFuture {
        let future = async move {
            let mut status_receiver = self.cell.status().clone();
            let result = status_receiver.wait_for(ActorStatus::is_terminal).await;
            match result {
                Err(_) => ActorStatus::Unknown,
                Ok(status) => status.clone(),
            }
        };

        future.boxed()
    }
}

impl<A: Actor> Debug for ActorHandle<A> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
        f.debug_struct("ActorHandle").field("cell", &"..").finish()
    }
}

impl<A, M> Endpoint<M> for &ActorHandle<A>
where
    A: Actor + Handler<M>,
    M: Message,
{
    fn endpoint_location(&self) -> EndpointLocation {
        EndpointLocation::Actor(self.actor_addr().clone())
    }

    fn post<C>(self, cx: &C, message: M)
    where
        C: context::Actor,
    {
        Endpoint::post(&self.ports.get(), cx, message)
    }
}

impl<A: Actor> Clone for ActorHandle<A> {
    fn clone(&self) -> Self {
        Self {
            cell: self.cell.clone(),
            ports: self.ports.clone(),
        }
    }
}

/// `Referable` is a marker trait for types that can appear as
/// remote references across process boundaries.
///
/// It is not limited to concrete [`Actor`] implementations. For
/// example, façade types generated by [`behavior!`] implement
/// `Referable` so that you can hand out restricted or stable APIs
/// while still using the same remote messaging machinery.
///
/// Implementing this trait means the type can be identified (`Named`)
/// so the runtime knows what it is.
///
///  In contrast, [`RemoteSpawn`] is the trait that marks *actors*
/// that can actually be **spawned remotely**. A behavior may be a
/// `Referable` but is never a `RemoteSpawn`.
pub trait Referable: Named {}

/// Binds determines how an actor's ports are bound to a specific
/// reference type.
pub trait Binds<A: Actor>: Referable {
    /// Bind ports in this actor.
    fn bind(ports: &HandlerPorts<A>);
}

/// Handles is a marker trait specifying that message type [`M`]
/// is handled by a specific actor type.
pub trait RemoteHandles<M: RemoteMessage>: Referable {}

/// Check if the actor behaves-as the a given behavior (defined by [`behavior!`]).
///
/// ```
/// # use serde::Serialize;
/// # use serde::Deserialize;
/// # use typeuri::Named;
/// # use hyperactor::Actor;
///
/// // First, define a behavior, based on handling a single message type `()`.
/// hyperactor::behavior!(UnitBehavior, ());
///
/// #[derive(Debug, Default)]
/// struct MyActor;
///
/// impl Actor for MyActor {}
///
/// #[async_trait::async_trait]
/// impl hyperactor::Handler<()> for MyActor {
///     async fn handle(
///         &mut self,
///         _cx: &hyperactor::Context<Self>,
///         _message: (),
///     ) -> Result<(), anyhow::Error> {
///         // no-op
///         Ok(())
///     }
/// }
///
/// hyperactor::assert_behaves!(MyActor as UnitBehavior);
/// ```
#[macro_export]
macro_rules! assert_behaves {
    ($ty:ty as $behavior:ty) => {
        const _: fn() = || {
            fn check<B: hyperactor::actor::Binds<$ty>>() {}
            check::<$behavior>();
        };
    };
}

#[cfg(test)]
mod tests {
    use std::assert_matches;
    use std::sync::Mutex;
    use std::time::Duration;

    use rand::seq::SliceRandom;
    use timed_test::async_timed_test;
    use tokio::sync::mpsc;
    use tokio::time::timeout;

    use super::*;
    use crate as hyperactor;
    use crate::Actor;
    use crate::ActorRef;
    use crate::Addr;
    use crate::EndpointLocation;
    use crate::OncePortHandle;
    use crate::PortAddr;
    use crate::PortRef;
    use crate::config;
    use crate::context::Mailbox as _;
    use crate::introspect::IntrospectMessage;
    use crate::introspect::IntrospectResult;
    use crate::introspect::IntrospectView;
    use crate::mailbox::BoxableMailboxSender as _;
    use crate::mailbox::DeliveryFailure;
    use crate::mailbox::DeliveryFailureReport;
    use crate::mailbox::ExpiredDelivery;
    use crate::mailbox::InvalidReference;
    use crate::mailbox::InvalidReferenceReason;
    use crate::mailbox::MailboxSender;
    use crate::mailbox::PortGone;
    use crate::mailbox::PortLocation;
    use crate::mailbox::TransportFailure;
    use crate::mailbox::TransportFailureReason;
    use crate::mailbox::UndeliverableReason;
    use crate::mailbox::monitored_return_handle;
    use crate::ordering::SEQ_INFO;
    use crate::ordering::SeqInfo;
    use crate::port::Port;
    use crate::testing::ids::test_proc_id;
    use crate::testing::pingpong::PingPongActor;
    use crate::testing::pingpong::PingPongMessage;
    use crate::testing::proc_supervison::ProcSupervisionCoordinator; // for macros

    #[derive(Debug)]
    struct EchoActor(PortRef<u64>);

    #[async_trait]
    impl Actor for EchoActor {}

    #[async_trait]
    impl Handler<u64> for EchoActor {
        async fn handle(&mut self, cx: &Context<Self>, message: u64) -> Result<(), anyhow::Error> {
            let Self(port) = self;
            port.post(cx, message);
            Ok(())
        }
    }

    #[derive(Debug)]
    struct DeliveryPolicyActor(PortRef<()>);

    #[async_trait]
    impl Actor for DeliveryPolicyActor {}

    #[async_trait]
    impl Handler<()> for DeliveryPolicyActor {
        async fn handle(&mut self, cx: &Context<Self>, _message: ()) -> Result<(), anyhow::Error> {
            self.0.post(cx, ());
            Ok(())
        }
    }

    fn delivery_policy_envelope(
        sender: &ActorAddr,
        dest: PortAddr,
        failure: DeliveryFailure,
    ) -> MessageEnvelope {
        let mut envelope =
            MessageEnvelope::serialize(sender.clone(), dest, &(), Flattrs::new()).unwrap();
        envelope.push_delivery_failure(failure);
        envelope
    }

    fn delivery_policy_report(
        sender: ActorAddr,
        dest: PortAddr,
        failure: DeliveryFailure,
    ) -> DeliveryFailureReport {
        DeliveryFailureReport::new(
            sender,
            EndpointLocation::Port(dest),
            Some("()".to_string()),
            failure,
        )
    }

    async fn assert_delivery_policy_actor_remains_live(
        make_undeliverable: impl FnOnce(&ActorAddr, PortAddr) -> Undeliverable<MessageEnvelope>,
    ) {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (sync_port, mut sync_rx) = client.open_port::<()>();
        let actor = DeliveryPolicyActor(sync_port.bind());
        let handle = proc.spawn_with_label("delivery_policy", actor);
        let dest = handle.actor_addr().port_addr(Port::from(1234));

        handle.post(
            &client,
            make_undeliverable(handle.actor_addr(), dest.clone()),
        );
        handle.post(&client, ());

        tokio::time::timeout(Duration::from_secs(1), sync_rx.recv())
            .await
            .expect("actor should remain live")
            .expect("sync port should receive response");
        handle.drain_and_stop("test").unwrap();
        assert_matches!(handle.await, ActorStatus::Stopped(reason) if reason == "test");
    }

    async fn assert_delivery_policy_actor_fails(
        make_undeliverable: impl FnOnce(&ActorAddr, PortAddr) -> Undeliverable<MessageEnvelope>,
    ) {
        let proc = Proc::isolated();
        let (_reported, _coordinator) = ProcSupervisionCoordinator::set(&proc).await.unwrap();
        let client = proc.client("client");
        let (sync_port, _sync_rx) = client.open_port::<()>();
        let actor = DeliveryPolicyActor(sync_port.bind());
        let handle = proc.spawn_with_label("delivery_policy", actor);
        let dest = handle.actor_addr().port_addr(Port::from(1234));

        handle.post(
            &client,
            make_undeliverable(handle.actor_addr(), dest.clone()),
        );

        assert_matches!(handle.await, ActorStatus::Failed(_));
    }

    #[tokio::test]
    async fn test_default_transport_undeliverable_policy_does_not_fail_actor() {
        let target = Addr::Proc(test_proc_id("target"));
        let failure = DeliveryFailure::new(UndeliverableReason::Transport(TransportFailure::new(
            target,
            TransportFailureReason::NoRoute,
        )));
        assert_delivery_policy_actor_remains_live(|sender, dest| {
            Undeliverable::Returned(delivery_policy_envelope(sender, dest, failure))
        })
        .await;
    }

    #[tokio::test]
    async fn test_default_transport_report_policy_does_not_fail_actor() {
        let target = Addr::Proc(test_proc_id("target"));
        let failure = DeliveryFailure::new(UndeliverableReason::Transport(TransportFailure::new(
            target,
            TransportFailureReason::NoRoute,
        )));
        assert_delivery_policy_actor_remains_live(|sender, dest| {
            Undeliverable::Report(delivery_policy_report(sender.clone(), dest, failure))
        })
        .await;
    }

    #[tokio::test]
    async fn test_default_oversized_frame_transport_policy_fails_actor() {
        let target = Addr::Proc(test_proc_id("target"));
        let failure = DeliveryFailure::new(UndeliverableReason::Transport(TransportFailure::new(
            target,
            TransportFailureReason::OversizedFrame {
                len: 55001392,
                max: 50000000,
            },
        )));
        assert_delivery_policy_actor_fails(|sender, dest| {
            Undeliverable::Returned(delivery_policy_envelope(sender, dest, failure))
        })
        .await;
    }

    #[tokio::test]
    async fn test_default_oversized_frame_report_policy_fails_actor() {
        let target = Addr::Proc(test_proc_id("target"));
        let failure = DeliveryFailure::new(UndeliverableReason::Transport(TransportFailure::new(
            target,
            TransportFailureReason::OversizedFrame {
                len: 55001392,
                max: 50000000,
            },
        )));
        assert_delivery_policy_actor_fails(|sender, dest| {
            Undeliverable::Report(delivery_policy_report(sender.clone(), dest, failure))
        })
        .await;
    }

    #[tokio::test]
    async fn test_default_port_gone_policy_does_not_fail_actor() {
        let port = test_proc_id("target")
            .actor_addr("actor")
            .port_addr(Port::from(1234));
        let failure = DeliveryFailure::new(UndeliverableReason::PortGone(PortGone::new(
            port,
            Some("()".to_string()),
        )));
        assert_delivery_policy_actor_remains_live(|sender, dest| {
            Undeliverable::Returned(delivery_policy_envelope(sender, dest, failure))
        })
        .await;
    }

    #[tokio::test]
    async fn test_default_invalid_reference_policy_fails_actor() {
        let target = test_proc_id("target").actor_addr("actor");
        let failure = DeliveryFailure::new(InvalidReference::new(
            target,
            InvalidReferenceReason::ActorNotExist,
        ));
        assert_delivery_policy_actor_fails(|sender, dest| {
            Undeliverable::Returned(delivery_policy_envelope(sender, dest, failure))
        })
        .await;
    }

    #[tokio::test]
    async fn test_default_invalid_reference_policy_allows_return_handle_sender_mismatch() {
        let target = test_proc_id("target").actor_addr("actor");
        let failure = DeliveryFailure::new(InvalidReference::new(
            target,
            InvalidReferenceReason::ActorNotExist,
        ));
        assert_delivery_policy_actor_fails(|_actor, dest| {
            Undeliverable::Returned(delivery_policy_envelope(
                &test_proc_id("sender").actor_addr("actor"),
                dest,
                failure,
            ))
        })
        .await;
    }

    #[tokio::test]
    async fn test_default_expired_delivery_policy_fails_actor() {
        let port = test_proc_id("target")
            .actor_addr("actor")
            .port_addr(Port::from(1234));
        let failure = DeliveryFailure::new(ExpiredDelivery::new(port));
        assert_delivery_policy_actor_fails(|sender, dest| {
            Undeliverable::Returned(delivery_policy_envelope(sender, dest, failure))
        })
        .await;
    }

    #[test]
    fn test_delivery_failure_policy_ignores_attrs() {
        hyperactor_config::attrs::declare_attrs! {
            attr TEST_DELIVERY_FAILURE_ATTR: String;
        }

        let sender = test_proc_id("sender").actor_addr("actor");
        let dest = sender.port_addr(Port::from(1234));
        let mut attrs = Flattrs::new();
        attrs.set(TEST_DELIVERY_FAILURE_ATTR, "context".to_string());

        let transport = delivery_policy_envelope(
            &sender,
            dest.clone(),
            DeliveryFailure::with_attrs(
                UndeliverableReason::Transport(TransportFailure::new(
                    dest.clone(),
                    TransportFailureReason::NoRoute,
                )),
                attrs.clone(),
            ),
        );
        assert_matches!(
            delivery_failure_policy(&Undeliverable::Returned(transport)),
            DeliveryFailurePolicy::Undeliverable
        );

        let invalid_reference = delivery_policy_envelope(
            &sender,
            dest.clone(),
            DeliveryFailure::with_attrs(
                InvalidReference::new(dest.clone(), InvalidReferenceReason::PortNeverAllocated),
                attrs.clone(),
            ),
        );
        assert_matches!(
            delivery_failure_policy(&Undeliverable::Returned(invalid_reference)),
            DeliveryFailurePolicy::InvalidReference
        );

        let expired = delivery_policy_envelope(
            &sender,
            dest.clone(),
            DeliveryFailure::with_attrs(ExpiredDelivery::new(dest), attrs),
        );
        assert_matches!(
            delivery_failure_policy(&Undeliverable::Returned(expired)),
            DeliveryFailurePolicy::Expired
        );
    }

    #[test]
    fn test_delivery_failure_policy_uses_report_root_failure() {
        let sender = test_proc_id("sender").actor_addr("actor");
        let dest = sender.port_addr(Port::from(1234));
        let report = DeliveryFailureReport::new(
            sender,
            EndpointLocation::Port(dest.clone()),
            Some("()".to_string()),
            DeliveryFailure::new(ExpiredDelivery::new(dest)),
        );

        assert_matches!(
            delivery_failure_policy(&Undeliverable::Report(report)),
            DeliveryFailurePolicy::Expired
        );
    }

    #[tokio::test]
    async fn test_server_basic() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, mut rx) = client.open_port();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);
        handle.post(&client, 123u64);
        handle.drain_and_stop("test").unwrap();
        handle.await;

        assert_eq!(rx.drain(), vec![123u64]);
    }

    #[tokio::test]
    async fn test_actor_handle_guard_stops_actor_on_drop() {
        let proc = Proc::isolated();
        let handle = proc.spawn(());
        let mut status = handle.status();

        {
            let _guard = handle.into_guard();
        }

        let stopped = timeout(
            Duration::from_secs(5),
            status.wait_for(|status| {
                matches!(status, ActorStatus::Stopped(reason) if reason == "actor guard dropped")
            }),
        )
        .await
        .unwrap()
        .unwrap()
        .clone();

        match stopped {
            ActorStatus::Stopped(reason) => assert_eq!(reason, "actor guard dropped"),
            status => panic!("actor guard should stop actor, got {status}"),
        }
    }

    #[tokio::test]
    async fn test_actor_handle_guard_into_inner_disarms_stop() {
        let proc = Proc::isolated();
        let handle = proc.spawn(());
        let mut status = handle.status();

        let guard = handle.into_guard();
        let _ = guard.status();
        let guarded_handle = guard.into_inner();
        let result = timeout(
            Duration::from_millis(100),
            status.wait_for(ActorStatus::is_terminal),
        )
        .await;
        assert!(result.is_err());

        guarded_handle.drain_and_stop("test").unwrap();
        guarded_handle.await;
    }

    #[tokio::test]
    async fn test_ping_pong() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (undeliverable_msg_tx, _) = client.open_port();

        let ping_actor = PingPongActor::new(Some(undeliverable_msg_tx.bind()), None, None);
        let pong_actor = PingPongActor::new(Some(undeliverable_msg_tx.bind()), None, None);
        let ping_handle = proc.spawn_with_label::<PingPongActor>("ping", ping_actor);
        let pong_handle = proc.spawn_with_label::<PingPongActor>("pong", pong_actor);

        let (local_port, local_receiver) = client.open_once_port();

        ping_handle.post(
            &client,
            PingPongMessage(10, pong_handle.bind(), local_port.bind()),
        );

        assert!(local_receiver.recv().await.unwrap());
    }

    #[tokio::test]
    async fn test_ping_pong_on_handler_error() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (undeliverable_msg_tx, _) = client.open_port();

        // Need to set a supervison coordinator for this Proc because there will
        // be actor failure(s) in this test which trigger supervision.
        let (_reported, _coordinator) = ProcSupervisionCoordinator::set(&proc).await.unwrap();

        let error_ttl = 66;

        let ping_actor =
            PingPongActor::new(Some(undeliverable_msg_tx.bind()), Some(error_ttl), None);
        let pong_actor =
            PingPongActor::new(Some(undeliverable_msg_tx.bind()), Some(error_ttl), None);
        let ping_handle = proc.spawn_with_label::<PingPongActor>("ping", ping_actor);
        let pong_handle = proc.spawn_with_label::<PingPongActor>("pong", pong_actor);

        let (local_port, local_receiver) = client.open_once_port();

        ping_handle.post(
            &client,
            PingPongMessage(
                error_ttl + 1, // will encounter an error at TTL=66
                pong_handle.bind(),
                local_port.bind(),
            ),
        );

        // TODO: Fix this receiver hanging issue in T200423722.
        let res: Result<Result<bool, MailboxError>, tokio::time::error::Elapsed> =
            timeout(Duration::from_secs(5), local_receiver.recv()).await;
        assert!(res.is_err());
    }

    #[derive(Debug)]
    struct InitActor(bool);

    #[async_trait]
    impl Actor for InitActor {
        async fn init(&mut self, _this: &Instance<Self>) -> Result<(), anyhow::Error> {
            self.0 = true;
            Ok(())
        }
    }

    #[async_trait]
    impl Handler<OncePortHandle<bool>> for InitActor {
        async fn handle(
            &mut self,
            cx: &Context<Self>,
            port: OncePortHandle<bool>,
        ) -> Result<(), anyhow::Error> {
            port.post(cx, self.0);
            Ok(())
        }
    }

    #[tokio::test]
    async fn test_init() {
        let proc = Proc::isolated();
        let actor = InitActor(false);
        let handle = proc.spawn(actor);
        let client = proc.client("client");

        let (port, receiver) = client.open_once_port();
        handle.post(&client, port);
        assert!(receiver.recv().await.unwrap());

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    type MultiValues = Arc<Mutex<(u64, String)>>;

    struct MultiValuesTest {
        proc: Proc,
        values: MultiValues,
        handle: ActorHandle<MultiActor>,
        client: Client,
    }

    impl MultiValuesTest {
        async fn new() -> Self {
            let proc = Proc::isolated();
            let values: MultiValues = Arc::new(Mutex::new((0, "".to_string())));
            let actor = MultiActor(values.clone());
            let handle = proc.spawn(actor);
            let client = proc.client("client");
            Self {
                proc,
                values,
                handle,
                client,
            }
        }

        fn send<M>(&self, message: M)
        where
            M: RemoteMessage,
            MultiActor: Handler<M>,
        {
            self.handle.post(&self.client, message)
        }

        async fn sync(&self) {
            let (port, done) = self.client.open_once_port::<bool>();
            self.handle.post(&self.client, port);
            assert!(done.recv().await.unwrap());
        }

        fn get_values(&self) -> (u64, String) {
            self.values.lock().unwrap().clone()
        }
    }

    #[derive(Debug)]
    #[hyperactor::export(handlers = [u64, String])]
    struct MultiActor(MultiValues);

    #[async_trait]
    impl Actor for MultiActor {}

    #[async_trait]
    impl Handler<u64> for MultiActor {
        async fn handle(&mut self, _cx: &Context<Self>, message: u64) -> Result<(), anyhow::Error> {
            let mut vals = self.0.lock().unwrap();
            vals.0 = message;
            Ok(())
        }
    }

    #[async_trait]
    impl Handler<String> for MultiActor {
        async fn handle(
            &mut self,
            _cx: &Context<Self>,
            message: String,
        ) -> Result<(), anyhow::Error> {
            let mut vals = self.0.lock().unwrap();
            vals.1 = message;
            Ok(())
        }
    }

    #[async_trait]
    impl Handler<OncePortHandle<bool>> for MultiActor {
        async fn handle(
            &mut self,
            cx: &Context<Self>,
            message: OncePortHandle<bool>,
        ) -> Result<(), anyhow::Error> {
            message.post(cx, true);
            Ok(())
        }
    }

    #[tokio::test]
    async fn test_multi_handler_refs() {
        let test = MultiValuesTest::new().await;

        test.send(123u64);
        test.send("foo".to_string());
        test.sync().await;
        assert_eq!(test.get_values(), (123u64, "foo".to_string()));

        let myref: ActorRef<MultiActor> = test.handle.bind();

        myref.port().post(&test.client, 321u64);
        test.sync().await;
        assert_eq!(test.get_values(), (321u64, "foo".to_string()));

        myref.port().post(&test.client, "bar".to_string());
        test.sync().await;
        assert_eq!(test.get_values(), (321u64, "bar".to_string()));
    }

    #[tokio::test]
    async fn test_ref_behavior() {
        let test = MultiValuesTest::new().await;

        test.send(123u64);
        test.send("foo".to_string());

        hyperactor::behavior!(MyActorBehavior, u64, String);

        let myref: ActorRef<MyActorBehavior> = test.handle.bind();
        myref.port().post(&test.client, "biz".to_string());
        myref.port().post(&test.client, 999u64);

        test.sync().await;
        assert_eq!(test.get_values(), (999u64, "biz".to_string()));
    }

    #[tokio::test]
    async fn test_actor_handle_downcast() {
        #[derive(Debug, Default)]
        struct NothingActor;

        impl Actor for NothingActor {}

        // Just test that we can round-trip the handle through a downcast.

        let proc = Proc::isolated();
        let handle = proc.spawn(NothingActor);
        let cell = handle.cell();

        // Invalid actor doesn't succeed.
        assert!(cell.downcast_handle::<EchoActor>().is_none());

        let handle = cell.downcast_handle::<NothingActor>().unwrap();
        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    // Returning the sequence number assigned to the message.
    #[derive(Debug)]
    #[hyperactor::export(handlers = [String, Callback])]
    struct GetSeqActor(PortRef<(String, SeqInfo)>);

    #[async_trait]
    impl Actor for GetSeqActor {}

    #[async_trait]
    impl Handler<String> for GetSeqActor {
        async fn handle(
            &mut self,
            cx: &Context<Self>,
            message: String,
        ) -> Result<(), anyhow::Error> {
            let Self(port) = self;
            let seq_info = cx.headers().get(SEQ_INFO).unwrap();
            port.post(cx, (message, seq_info.clone()));
            Ok(())
        }
    }

    // Unlike Handler<String>, where the sender provides the string message
    // directly, in Handler<Callback>, sender needs to provide a port, and
    // handler will reply that port with its own callback port. Then sender can
    // send the string message through this callback port.
    #[derive(Clone, Debug, Serialize, Deserialize, Named)]
    struct Callback(PortRef<PortRef<String>>);

    #[async_trait]
    impl Handler<Callback> for GetSeqActor {
        async fn handle(
            &mut self,
            cx: &Context<Self>,
            message: Callback,
        ) -> Result<(), anyhow::Error> {
            let (handle, mut receiver) = cx.open_port::<String>();
            let callback_ref = handle.bind();
            message.0.post(cx, callback_ref);
            let msg = receiver.recv().await.unwrap();
            self.handle(cx, msg).await
        }
    }

    #[async_timed_test(timeout_secs = 30)]
    async fn test_sequencing_actor_handle_basic() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, mut rx) = client.open_port();

        let actor_handle = proc.spawn(GetSeqActor(tx.bind()));

        // Verify that unbound handle can send message.
        actor_handle.post(&client, "unbound".to_string());
        assert_eq!(
            rx.recv().await.unwrap(),
            ("unbound".to_string(), SeqInfo::Direct)
        );

        let actor_ref: ActorRef<GetSeqActor> = actor_handle.bind();

        let session_id = client.sequencer().session_id();
        let mut expected_seq = 0;
        // Interleave messages sent through the handle and the reference.
        for m in 0..10 {
            actor_handle.post(&client, format!("{m}"));
            expected_seq += 1;
            assert_eq!(
                rx.recv().await.unwrap(),
                (
                    format!("{m}"),
                    SeqInfo::Session {
                        session_id,
                        seq: expected_seq,
                    }
                )
            );

            for n in 0..2 {
                actor_ref.port().post(&client, format!("{m}-{n}"));
                expected_seq += 1;
                assert_eq!(
                    rx.recv().await.unwrap(),
                    (
                        format!("{m}-{n}"),
                        SeqInfo::Session {
                            session_id,
                            seq: expected_seq,
                        }
                    )
                );
            }
        }
    }

    // Test that handler ports share a sequence while non-handler ports get their own.
    #[async_timed_test(timeout_secs = 30)]
    async fn test_sequencing_mixed_handler_and_non_handler_ports() {
        let proc = Proc::isolated();
        let client = proc.client("client");

        // Port for receiving seq info from actor handler
        let (actor_tx, mut actor_rx) = client.open_port();

        // Channel for receiving seq info from non-handler port
        let (non_handler_tx, mut non_handler_rx) = mpsc::unbounded_channel::<Option<SeqInfo>>();

        let actor_handle = proc.spawn(GetSeqActor(actor_tx.bind()));
        let actor_ref: ActorRef<GetSeqActor> = actor_handle.bind();

        // Create a non-handler port using open_enqueue_port
        let non_handler_tx_clone = non_handler_tx.clone();
        let non_handler_port_handle =
            client
                .mailbox()
                .open_enqueue_port(move |headers: Flattrs, _m: ()| {
                    let seq_info = headers.get(SEQ_INFO);
                    non_handler_tx_clone.send(seq_info).unwrap();
                    Ok(())
                });

        // Bind the port to get a port ID
        non_handler_port_handle.bind();
        let non_handler_port_id = match non_handler_port_handle.location() {
            PortLocation::Bound(port_id) => port_id,
            _ => panic!("port_handle should be bound"),
        };
        assert!(!non_handler_port_id.is_handler_port());

        let session_id = client.sequencer().session_id();

        // Send to handler ports via ActorHandle - seq 1
        actor_handle.post(&client, "msg1".to_string());
        assert_eq!(
            actor_rx.recv().await.unwrap().1,
            SeqInfo::Session { session_id, seq: 1 }
        );

        // Send to handler ports via ActorRef - seq 2 (shared with ActorHandle)
        actor_ref.port().post(&client, "msg2".to_string());
        assert_eq!(
            actor_rx.recv().await.unwrap().1,
            SeqInfo::Session { session_id, seq: 2 }
        );

        // Send to non-handler port - has its own sequence starting at 1
        non_handler_port_handle.post(&client, ());
        assert_eq!(
            non_handler_rx.recv().await.unwrap(),
            Some(SeqInfo::Session { session_id, seq: 1 })
        );

        // Send more to handler ports via ActorHandle - seq continues at 3
        actor_handle.post(&client, "msg3".to_string());
        assert_eq!(
            actor_rx.recv().await.unwrap().1,
            SeqInfo::Session { session_id, seq: 3 }
        );

        // Send more to non-handler port - its sequence continues at 2
        non_handler_port_handle.post(&client, ());
        assert_eq!(
            non_handler_rx.recv().await.unwrap(),
            Some(SeqInfo::Session { session_id, seq: 2 })
        );

        // Send via ActorRef again - seq 4
        actor_ref.port().post(&client, "msg4".to_string());
        assert_eq!(
            actor_rx.recv().await.unwrap().1,
            SeqInfo::Session { session_id, seq: 4 }
        );

        actor_handle.drain_and_stop("test cleanup").unwrap();
        actor_handle.await;
    }

    // Test that messages from different clients get independent sequence schemes.
    #[async_timed_test(timeout_secs = 30)]
    async fn test_sequencing_multiple_clients() {
        let proc = Proc::isolated();
        let client1 = proc.client("client1");
        let client2 = proc.client("client2");

        // Port for receiving seq info from actor handler
        let (tx, mut rx) = client1.open_port();

        let actor_handle = proc.spawn(GetSeqActor(tx.bind()));
        let actor_ref: ActorRef<GetSeqActor> = actor_handle.bind();

        // Each client should have a different session_id
        let session_id_1 = client1.sequencer().session_id();
        let session_id_2 = client2.sequencer().session_id();
        assert_ne!(session_id_1, session_id_2);

        // Send from client1 via ActorHandle - seq 1 for session_id_1
        actor_handle.post(&client1, "c1_msg1".to_string());
        assert_eq!(
            rx.recv().await.unwrap().1,
            SeqInfo::Session {
                session_id: session_id_1,
                seq: 1
            }
        );

        // Send from client2 via ActorHandle - seq 1 for session_id_2 (independent)
        actor_handle.post(&client2, "c2_msg1".to_string());
        assert_eq!(
            rx.recv().await.unwrap().1,
            SeqInfo::Session {
                session_id: session_id_2,
                seq: 1
            }
        );

        // Send from client1 via ActorRef - seq 2 for session_id_1
        actor_ref.port().post(&client1, "c1_msg2".to_string());
        assert_eq!(
            rx.recv().await.unwrap().1,
            SeqInfo::Session {
                session_id: session_id_1,
                seq: 2
            }
        );

        // Send from client2 via ActorRef - seq 2 for session_id_2
        actor_ref.port().post(&client2, "c2_msg2".to_string());
        assert_eq!(
            rx.recv().await.unwrap().1,
            SeqInfo::Session {
                session_id: session_id_2,
                seq: 2
            }
        );

        // Interleave more messages to further verify independence
        actor_handle.post(&client1, "c1_msg3".to_string());
        assert_eq!(
            rx.recv().await.unwrap().1,
            SeqInfo::Session {
                session_id: session_id_1,
                seq: 3
            }
        );

        actor_ref.port().post(&client2, "c2_msg3".to_string());
        assert_eq!(
            rx.recv().await.unwrap().1,
            SeqInfo::Session {
                session_id: session_id_2,
                seq: 3
            }
        );

        actor_handle.drain_and_stop("test cleanup").unwrap();
        actor_handle.await;
    }

    // Verify that ordering is guarranteed based on
    //   * (sender actor , client actor, port stream)
    // not
    //   * (sender actor, client actor)
    //
    // For "port stream",
    //   * handler ports of the same actor belongs to the same stream;
    //   * non-handler port has its independent stream.
    //
    // Specifically, in this test,
    //   * client sends a Callback message to dest actor's handler;
    //   * while dest actor is still processing that message, client sends
    //     another non-handler message to dest actor.
    //
    // If the ordering is based on (sender actor, client actor), this test would
    // hang, since dest actor is deadlock on waiting for the 2nd message while
    // still processing the 2nd message.
    //
    // But since port stream is also part of the ordering guarrantee, such
    // deadlock should not happen.
    #[async_timed_test(timeout_secs = 30)]
    async fn test_sequencing_actor_handle_callback() {
        let config = hyperactor_config::global::lock();
        let _guard = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, true);

        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, mut rx) = client.open_port();

        let actor_handle = proc.spawn(GetSeqActor(tx.bind()));
        let actor_ref: ActorRef<GetSeqActor> = actor_handle.bind();

        let (callback_tx, mut callback_rx) = client.open_port();
        // Client sends the 1st message
        actor_ref.post(&client, Callback(callback_tx.bind()));
        let msg_port_ref = callback_rx.recv().await.unwrap();
        // client sends the 2nd message. At this time, GetSeqActor is still
        // processing the 1st message, and waiting for the 2nd message.
        msg_port_ref.post(&client, "finally".to_string());

        let session_id = client.sequencer().session_id();
        // passing this assert means GetSeqActor processed the 2nd message.
        assert_eq!(
            rx.recv().await.unwrap(),
            (
                "finally".to_string(),
                SeqInfo::Session { session_id, seq: 1 }
            )
        );
    }

    // Adding a delay before sending the destination proc. Useful for tests
    // requiring latency injection.
    #[derive(Clone, Debug)]
    struct DelayedMailboxSender {
        relay_tx: mpsc::UnboundedSender<MessageEnvelope>,
    }

    impl DelayedMailboxSender {
        // Use a random latency between 0 and 1 second if the plan is empty.
        fn new(
            // The proc that hosts the dest actor. By posting envelope to this
            // proc, this proc will route that evenlope to the dest actor.
            dest_proc: Proc,
            // Vec index is the message seq - 1, value is the order this message
            // would be relayed to the dest actor. Endpoint actor is responsible to
            // ensure itself processes these messages in order.
            relay_orders: Vec<usize>,
        ) -> Self {
            let (relay_tx, mut relay_rx) = mpsc::unbounded_channel::<MessageEnvelope>();

            tokio::spawn(async move {
                let mut buffer = Vec::new();

                for _ in 0..relay_orders.len() {
                    let envelope = relay_rx.recv().await.unwrap();
                    buffer.push(envelope);
                }

                for m in buffer.clone() {
                    let seq = match m.headers().get(SEQ_INFO) {
                        Some(SeqInfo::Session { seq, .. }) => seq as usize,
                        Some(SeqInfo::Direct) => panic!("expected Session variant"),
                        None => panic!("expected seq info"),
                    };
                    // seq no is one-based.
                    let order = relay_orders[seq - 1];
                    buffer[order] = m;
                }

                let dest_proc_clone = dest_proc.clone();
                for msg in buffer {
                    dest_proc_clone.post(msg, monitored_return_handle());
                }
            });

            Self { relay_tx }
        }
    }

    #[async_trait]
    impl MailboxSender for DelayedMailboxSender {
        fn post_unchecked(
            &self,
            envelope: MessageEnvelope,
            _return_handle: PortHandle<Undeliverable<MessageEnvelope>>,
        ) {
            self.relay_tx.send(envelope).unwrap();
        }
    }

    async fn assert_out_of_order_delivery(expected: Vec<(String, u64)>, relay_orders: Vec<usize>) {
        let local_proc: Proc = Proc::isolated();
        let client = local_proc.client("local");
        let (tx, mut rx) = client.open_port();

        let handle = local_proc.spawn(GetSeqActor(tx.bind()));
        let actor_ref: ActorRef<GetSeqActor> = handle.bind();

        let remote_proc = Proc::configured(
            test_proc_id("remote_0"),
            DelayedMailboxSender::new(local_proc.clone(), relay_orders).boxed(),
        );
        let remote_client = remote_proc.client("remote");
        // Send the messages out in the order of their expected sequence numbers.
        let mut messages = expected.clone();
        messages.sort_by_key(|v| v.1);
        for (message, _seq) in messages {
            actor_ref.post(&remote_client, message);
        }
        let session_id = remote_client.sequencer().session_id();
        for expect in expected {
            let expected = (
                expect.0,
                SeqInfo::Session {
                    session_id,
                    seq: expect.1,
                },
            );
            assert_eq!(rx.recv().await.unwrap(), expected);
        }

        handle.drain_and_stop("test cleanup").unwrap();
        handle.await;
    }

    // Send several messages, use DelayedMailboxSender and the relay orders to
    // ensure these messages will arrive at handler's workq out-of-order.
    // Then verify the actor handler will still process these messages based on
    // their sending order if reordering buffer is enabled.
    #[async_timed_test(timeout_secs = 30)]
    async fn test_sequencing_actor_ref_known_delivery_order() {
        let config = hyperactor_config::global::lock();

        // relay order is second, third, first
        let relay_orders = vec![2, 0, 1];

        // By disabling the actor side re-ordering buffer, the mssages will
        // be processed in the same order as they sent out.
        let _guard = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, false);
        assert_out_of_order_delivery(
            vec![
                ("second".to_string(), 2),
                ("third".to_string(), 3),
                ("first".to_string(), 1),
            ],
            relay_orders.clone(),
        )
        .await;

        // By enabling the actor side re-ordering buffer, the mssages will
        // be re-ordered before being processed.
        let _guard = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, true);
        assert_out_of_order_delivery(
            vec![
                ("first".to_string(), 1),
                ("second".to_string(), 2),
                ("third".to_string(), 3),
            ],
            relay_orders.clone(),
        )
        .await;
    }

    // Send a large nubmer of messages, use DelayedMailboxSender to ensure these
    // messages will arrive at handler's workq in a random order. Then verify the
    // actor handler will still process these messages based on their sending
    // order with reordering buffer enabled.
    #[async_timed_test(timeout_secs = 30)]
    async fn test_sequencing_actor_ref_random_delivery_order() {
        let config = hyperactor_config::global::lock();

        // By enabling the actor side re-ordering buffer, the mssages will
        // be re-ordered before being processed.
        let _guard = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, true);
        let expected = (0..10000)
            .map(|i| (format!("msg{i}"), i + 1))
            .collect::<Vec<_>>();

        let mut relay_orders: Vec<usize> = (0..10000).collect();
        relay_orders.shuffle(&mut rand::rng());
        assert_out_of_order_delivery(expected, relay_orders).await;
    }

    /// Verifies the default blanket introspection handler for a plain
    /// actor.
    ///
    /// This test spawns a simple `EchoActor`, sends it
    /// `IntrospectMessage::Query`, and checks that the returned
    /// `IntrospectResult` matches the framework’s structural default:
    ///
    /// - `identity` matches the actor id
    /// - `attrs` contains actor-runtime keys (status, actor_type, etc.)
    /// - no supervision children are reported
    /// - `supervisor` is None because this actor is spawned as a
    ///   root/top-level actor in the proc (only supervised child actors
    ///   report a supervisor id).
    ///
    /// This exercises the end-to-end introspect task path rather than
    /// calling `live_actor_payload` directly, ensuring the runtime
    /// wiring behaves as expected.
    #[tokio::test]
    async fn test_introspect_query_default_payload() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);

        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );
        let payload = reply_rx.recv().await.unwrap();

        assert_eq!(
            payload.identity,
            crate::introspect::IntrospectRef::Actor(handle.actor_addr().clone())
        );
        assert_valid_attrs(&payload);
        assert_has_attr(&payload, "status");
        assert_has_attr(&payload, "actor_type");
        assert_has_attr(&payload, "created_at");
        assert!(payload.children.is_empty());
        assert!(payload.parent.is_none());

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// Helper: look up an attr in the attrs JSON by short name.
    fn attrs_get(attrs_json: &str, short_name: &str) -> Option<serde_json::Value> {
        use hyperactor_config::INTROSPECT;
        use hyperactor_config::attrs::AttrKeyInfo;
        let fq_name = inventory::iter::<AttrKeyInfo>()
            .find(|info| {
                info.meta
                    .get(INTROSPECT)
                    .is_some_and(|ia| ia.name == short_name)
            })
            .map(|info| info.name)?;
        let obj: serde_json::Value = serde_json::from_str(attrs_json).ok()?;
        obj.get(fq_name).cloned()
    }

    /// Assert that an IntrospectResult has valid JSON attrs (IA-1).
    fn assert_valid_attrs(result: &IntrospectResult) {
        let parsed: serde_json::Value =
            serde_json::from_str(&result.attrs).expect("attrs must be valid JSON");
        assert!(parsed.is_object(), "IA-1: attrs must be a JSON object");
    }

    /// Assert the actor status attr matches expected value.
    fn assert_status(result: &IntrospectResult, expected: &str) {
        let status = attrs_get(&result.attrs, "status")
            .and_then(|v| v.as_str().map(String::from))
            .expect("attrs must contain status");
        assert_eq!(status, expected, "unexpected actor status");
    }

    /// Assert the actor has a specific handler (or None).
    fn assert_handler(result: &IntrospectResult, expected: Option<&str>) {
        let handler =
            attrs_get(&result.attrs, "last_handler").and_then(|v| v.as_str().map(String::from));
        assert_eq!(handler.as_deref(), expected);
    }

    /// Assert the error code attr matches expected value.
    fn assert_error_code(result: &IntrospectResult, expected: &str) {
        let code = attrs_get(&result.attrs, "error_code")
            .and_then(|v| v.as_str().map(String::from))
            .expect("attrs must contain error_code");
        assert_eq!(code, expected);
    }

    /// Assert handler does NOT contain a substring.
    fn assert_handler_not_contains(result: &IntrospectResult, forbidden: &str) {
        if let Some(handler) =
            attrs_get(&result.attrs, "last_handler").and_then(|v| v.as_str().map(String::from))
        {
            assert!(
                !handler.contains(forbidden),
                "handler should not contain '{}'; got: {}",
                forbidden,
                handler
            );
        }
    }

    /// Assert an attr is present by short name.
    fn assert_has_attr(result: &IntrospectResult, short_name: &str) {
        assert!(
            attrs_get(&result.attrs, short_name).is_some(),
            "attrs must contain '{}'",
            short_name
        );
    }

    /// Assert status contains a substring (for non-exact checks
    /// like "processing" on wedged actors).
    fn assert_status_contains(result: &IntrospectResult, substring: &str) {
        let status = attrs_get(&result.attrs, "status")
            .and_then(|v| v.as_str().map(String::from))
            .expect("attrs must contain status");
        assert!(
            status.contains(substring),
            "status should contain '{}'; got: {}",
            substring,
            status
        );
    }

    /// Assert no status_reason attr (IA-3: non-terminal status).
    fn assert_no_status_reason(result: &IntrospectResult) {
        assert!(
            attrs_get(&result.attrs, "status_reason").is_none(),
            "IA-3: must not have status_reason"
        );
    }

    /// Assert a handler is present (any value).
    fn assert_has_handler(result: &IntrospectResult) {
        assert!(
            attrs_get(&result.attrs, "last_handler").is_some(),
            "must have a handler"
        );
    }

    /// Assert no failure attrs are present (IA-4).
    fn assert_no_failure_attrs(result: &IntrospectResult) {
        assert!(
            attrs_get(&result.attrs, "failure_error_message").is_none(),
            "IA-4: must not have failure attrs"
        );
    }

    /// Establishes IA-1 (attrs-json), IA-3 (status-shape), and
    /// IA-4 (failure-shape) for the running-actor path only.
    /// Stopped/failed paths need separate tests (see proc.rs
    /// terminated snapshot tests).
    #[tokio::test]
    async fn test_ia1_ia4_running_actor_attrs() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);

        let payload = crate::introspect::live_actor_payload(handle.cell());

        // IA-1: valid JSON.
        assert_valid_attrs(&payload);

        // IA-3: non-terminal status, no status_reason.
        assert_has_attr(&payload, "status");
        assert_no_status_reason(&payload);

        // IA-4: no failure attrs.
        assert_no_failure_attrs(&payload);

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// AS-1 (snapshot-opacity): an INTROSPECT-tagged actor-supplied attr
    /// is transported into the Actor view verbatim. `LAST_HANDLER` is
    /// INTROSPECT-tagged and left unset by the core builder for a
    /// message-free actor, so it shows additive transport cleanly (and
    /// satisfies the accessor's INTROSPECT-tagging guard).
    #[tokio::test]
    async fn test_actor_attrs_snapshot_appears_verbatim() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let handle = proc.spawn(EchoActor(tx.bind()));

        handle.cell().set_attrs_snapshot(|| {
            let mut attrs = hyperactor_config::Attrs::new();
            attrs.set(crate::introspect::LAST_HANDLER, "seam-probe".to_string());
            attrs
        });

        let payload = crate::introspect::live_actor_payload(handle.cell());
        assert_valid_attrs(&payload);
        assert_eq!(
            attrs_get(&payload.attrs, "last_handler").and_then(|v| v.as_str().map(String::from)),
            Some("seam-probe".to_string()),
            "AS-1: actor-supplied attr must appear verbatim"
        );

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// AS-2 (core-precedence): on a key collision, the core/runtime
    /// value wins over the actor-supplied snapshot.
    #[tokio::test]
    async fn test_actor_attrs_snapshot_core_wins_on_collision() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let handle = proc.spawn(EchoActor(tx.bind()));

        // The snapshot tries to clobber a core key with a bogus value;
        // `build_actor_attrs` sets `messages_processed` unconditionally,
        // so core must win.
        handle.cell().set_attrs_snapshot(|| {
            let mut attrs = hyperactor_config::Attrs::new();
            attrs.set(crate::introspect::MESSAGES_PROCESSED, 999u64);
            attrs
        });

        let payload = crate::introspect::live_actor_payload(handle.cell());
        let messages = attrs_get(&payload.attrs, "messages_processed")
            .and_then(|v| v.as_u64())
            .expect("attrs must contain messages_processed");
        // Fresh actor processed no messages: the exact core value is 0,
        // and it wins over the snapshot's bogus 999.
        assert_eq!(messages, 0, "AS-2: core value must win over the snapshot");

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// AS-3 (snapshot-non-fatal): a panicking snapshot degrades to the
    /// core-only Actor view rather than emptying or invalidating attrs.
    #[tokio::test]
    async fn test_actor_attrs_snapshot_panic_is_non_fatal() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let handle = proc.spawn(EchoActor(tx.bind()));

        handle
            .cell()
            .set_attrs_snapshot(|| -> hyperactor_config::Attrs {
                panic!("snapshot callback boom")
            });

        let payload = crate::introspect::live_actor_payload(handle.cell());
        // Core view survives a panicking snapshot intact (IA-1, IA-5, AS-3).
        assert_valid_attrs(&payload);
        assert_has_attr(&payload, "status");
        assert_has_attr(&payload, "actor_type");

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    // Verifies that QueryChild returns an error for actors without
    // a registered query_child_handler callback. The runtime
    // introspect task responds with the error sentinel payload
    // (`identity == ""`, error attrs with code "not_found",
    // .. }`).
    #[tokio::test]
    async fn test_introspect_query_child_not_found() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);

        let child_ref = crate::Addr::Actor(test_proc_id("nonexistent").actor_addr("child"));
        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::QueryChild {
                child_ref,
                reply: reply_port.bind(),
            },
        );
        let payload = reply_rx.recv().await.unwrap();

        assert_eq!(
            payload.identity,
            crate::introspect::IntrospectRef::Actor(
                test_proc_id("nonexistent").actor_addr("child")
            )
        );
        assert_error_code(&payload, "not_found");

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    // Verifies that with the runtime introspect task, custom
    // `handle_introspect` overrides are not called. The runtime
    // task intercepts IntrospectMessage before it reaches the
    // actor's work queue. An actor with an override still gets
    // standard Actor properties from the runtime task.
    #[tokio::test]
    async fn test_introspect_override() {
        #[derive(Debug, Default)]
        #[hyperactor::export(handlers = [])]
        struct CustomIntrospectActor;

        #[async_trait]
        impl Actor for CustomIntrospectActor {}

        let proc = Proc::isolated();
        let client = proc.client("client");
        let handle = proc.spawn(CustomIntrospectActor);

        handle
            .status()
            .wait_for(|s| matches!(s, ActorStatus::Idle))
            .await
            .unwrap();

        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );
        let payload = reply_rx.recv().await.unwrap();

        // The runtime task returns actor attrs (with status), NOT
        // the override's Host properties.
        assert_has_attr(&payload, "status");

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// Verifies that a child actor spawned via `spawn_child` reports
    /// its parent as `supervisor` in the introspection payload, and
    /// that the parent's payload lists the child in `children`.
    #[tokio::test]
    async fn test_introspect_query_supervision_child() {
        let proc = Proc::isolated();
        let client = proc.client("client");

        // Spawn parent.
        let (tx_parent, _rx_parent) = client.open_port::<u64>();
        let parent_handle =
            proc.spawn_with_label::<EchoActor>("parent", EchoActor(tx_parent.bind()));

        // Spawn child under parent.
        let (tx_child, _rx_child) = client.open_port::<u64>();
        let child_handle =
            proc.spawn_child::<EchoActor>(parent_handle.cell().clone(), EchoActor(tx_child.bind()));

        // Query the child — supervisor should be the parent.
        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        child_handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );
        let child_payload = reply_rx.recv().await.unwrap();

        assert_eq!(
            child_payload.identity,
            crate::introspect::IntrospectRef::Actor(child_handle.actor_addr().clone()),
        );
        // Verify it has actor attrs (status present).
        assert!(
            attrs_get(&child_payload.attrs, "status").is_some(),
            "child should have actor attrs"
        );
        assert_eq!(
            child_payload.parent,
            Some(crate::introspect::IntrospectRef::Actor(
                parent_handle.actor_addr().clone()
            )),
        );

        // Query the parent — children should include the child.
        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        parent_handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );
        let parent_payload = reply_rx.recv().await.unwrap();

        assert!(parent_payload.parent.is_none());
        assert!(
            parent_payload
                .children
                .contains(&crate::introspect::IntrospectRef::Actor(
                    child_handle.actor_addr().clone()
                )),
        );

        child_handle.drain_and_stop("test").unwrap();
        child_handle.await;
        parent_handle.drain_and_stop("test").unwrap();
        parent_handle.await;
    }

    /// A freshly spawned actor that has received no user messages
    /// reports `last_message_handler == None` — the introspect
    /// handler does not leak through. Status is `"idle"` once
    /// initialization completes.
    #[tokio::test]
    async fn test_introspect_fresh_actor_status() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);

        // Wait for the actor to finish initialization.
        handle
            .status()
            .wait_for(|s| matches!(s, ActorStatus::Idle))
            .await
            .unwrap();

        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );
        let payload = reply_rx.recv().await.unwrap();

        assert_status(&payload, "idle");
        assert_handler(&payload, None);

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// After processing a user message, the introspect payload reports
    /// the user message's handler and post-completion status — not
    /// the introspect handler itself (one-behind invariant,
    /// after-user-traffic case).
    #[tokio::test]
    async fn test_introspect_after_user_message() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, mut rx) = client.open_port::<u64>();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);

        // Send a user message and wait for it to be processed.
        handle.post(&client, 42u64);
        let _ = rx.recv().await.unwrap();

        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );
        let payload = reply_rx.recv().await.unwrap();

        assert_status(&payload, "idle");
        assert_has_handler(&payload);
        assert_handler_not_contains(&payload, "IntrospectMessage");

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// Two consecutive introspect queries: with the runtime
    /// introspect task, neither perturbs the actor's state (S2).
    /// Both report the same `last_message_handler` for a fresh
    /// actor — `None`, not `IntrospectMessage`.
    #[tokio::test]
    async fn test_introspect_consecutive_queries() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);

        handle
            .status()
            .wait_for(|s| matches!(s, ActorStatus::Idle))
            .await
            .unwrap();

        // First introspect query.
        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );
        let payload1 = reply_rx.recv().await.unwrap();

        // Second introspect query.
        let (reply_port2, reply_rx2) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port2.bind(),
            },
        );
        let payload2 = reply_rx2.recv().await.unwrap();

        // Neither should show IntrospectMessage as the handler.
        assert_handler(&payload1, None);
        assert_handler(&payload2, None);

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    // test_published_properties_round_trip removed — replaced by
    // test_publish_attrs_round_trip which tests the Attrs-based API.

    /// Verify InstanceCell Attrs storage: `set_published_attrs`
    /// replaces the whole bag, `merge_published_attr` merges a single
    /// key incrementally. (Instance methods are thin wrappers over
    /// these.)
    #[tokio::test]
    async fn test_publish_attrs_round_trip() {
        use hyperactor_config::Attrs;
        use hyperactor_config::declare_attrs;

        declare_attrs! {
            attr TEST_KEY_A: String;
            attr TEST_KEY_B: u64;
        }

        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);

        // Before publishing, attrs are None.
        assert!(handle.cell().published_attrs().is_none());

        // publish_attrs: replace entire bag.
        let mut attrs = Attrs::new();
        attrs.set(TEST_KEY_A, "hello".to_string());
        handle.cell().set_published_attrs(attrs);
        let published = handle.cell().published_attrs().unwrap();
        assert_eq!(published.get(TEST_KEY_A), Some(&"hello".to_string()));

        // publish_attr: merge single key into existing bag.
        handle.cell().merge_published_attr(TEST_KEY_B, 42u64);
        let published = handle.cell().published_attrs().unwrap();
        assert_eq!(published.get(TEST_KEY_A), Some(&"hello".to_string()));
        assert_eq!(published.get(TEST_KEY_B), Some(&42u64));

        // publish_attr: overwrite existing key.
        handle
            .cell()
            .merge_published_attr(TEST_KEY_A, "world".to_string());
        let published = handle.cell().published_attrs().unwrap();
        assert_eq!(published.get(TEST_KEY_A), Some(&"world".to_string()));

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// Verify the query_child_handler callback: register a callback,
    /// invoke it via `query_child()`, and confirm the response.
    #[tokio::test]
    async fn test_query_child_handler_round_trip() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, _rx) = client.open_port::<u64>();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);

        // Before registering, query_child returns None.
        let test_ref = Addr::Actor(test_proc_id("test").actor_addr("child"));
        assert!(handle.cell().query_child(&test_ref).is_none());

        // Register a callback.
        handle.cell().set_query_child_handler(|child_ref| {
            use crate::introspect::IntrospectRef;
            let identity = match child_ref {
                Addr::Proc(p) => IntrospectRef::Proc(p.clone()),
                Addr::Actor(a) => IntrospectRef::Actor(a.clone()),
                Addr::Port(p) => IntrospectRef::Actor(p.actor_addr()),
            };
            IntrospectResult {
                identity,
                attrs: serde_json::json!({
                    "proc_name": "test_proc",
                    "num_actors": 42,
                })
                .to_string(),
                children: Vec::new(),
                parent: None,
                as_of: std::time::SystemTime::now(),
            }
        });

        // Now query_child returns the callback's response.
        let payload = handle
            .cell()
            .query_child(&test_ref)
            .expect("query_child must return payload");
        assert_eq!(
            payload.identity,
            crate::introspect::IntrospectRef::Actor(test_proc_id("test").actor_addr("child"))
        );
        let attrs: serde_json::Value =
            serde_json::from_str(&payload.attrs).expect("attrs must be valid JSON");
        assert_eq!(
            attrs.get("proc_name").and_then(|v| v.as_str()),
            Some("test_proc")
        );
        assert_eq!(attrs.get("num_actors").and_then(|v| v.as_u64()), Some(42));

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// Exercises S1 (see `introspect` module doc).
    ///
    /// Sends a wedging message, then queries introspect while the
    /// actor is blocked. The response must arrive and report live
    /// processing status.
    #[tokio::test]
    async fn test_introspect_wedged() {
        #[derive(Debug, Default)]
        #[hyperactor::export(handlers = [u64])]
        struct WedgedActor;

        #[async_trait]
        impl Actor for WedgedActor {}

        #[async_trait]
        impl Handler<u64> for WedgedActor {
            async fn handle(
                &mut self,
                _cx: &Context<Self>,
                _message: u64,
            ) -> Result<(), anyhow::Error> {
                // Block forever.
                std::future::pending::<()>().await;
                Ok(())
            }
        }

        let proc = Proc::isolated();
        let client = proc.client("client");
        let handle = proc.spawn(WedgedActor);

        // Wait for idle before sending the wedging message.
        handle
            .status()
            .wait_for(|s| matches!(s, ActorStatus::Idle))
            .await
            .unwrap();

        // Send a u64 to wedge the actor in its handler.
        handle.post(&client, 1u64);

        // Wait for the handler to start blocking.
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Send introspect query via the dedicated introspect port.
        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );

        // Must not hang — the introspect task runs independently.
        let payload = tokio::time::timeout(Duration::from_secs(5), reply_rx.recv())
            .await
            .expect("introspect should not hang on a wedged actor")
            .unwrap();

        assert_status_contains(&payload, "processing");
        assert_handler_not_contains(&payload, "IntrospectMessage");
    }

    /// Exercises S2 (see `introspect` module doc).
    ///
    /// After a user message, two consecutive introspect queries both
    /// report the user message handler.
    #[tokio::test]
    async fn test_introspect_no_perturbation() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let (tx, mut rx) = client.open_port::<u64>();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn(actor);

        // Wait for idle before sending the user message.
        handle
            .status()
            .wait_for(|s| matches!(s, ActorStatus::Idle))
            .await
            .unwrap();

        // Send a user message and wait for it to be processed.
        handle.post(&client, 42u64);
        let _ = rx.recv().await.unwrap();

        // First introspect query.
        let (reply_port1, reply_rx1) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port1.bind(),
            },
        );
        let payload1 = reply_rx1.recv().await.unwrap();

        // Second introspect query.
        let (reply_port2, reply_rx2) = client.open_once_port::<IntrospectResult>();
        handle.actor_addr().introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port2.bind(),
            },
        );
        let payload2 = reply_rx2.recv().await.unwrap();

        // Both should report the user message handler, not IntrospectMessage.
        assert_handler_not_contains(&payload1, "IntrospectMessage");
        assert_handler_not_contains(&payload2, "IntrospectMessage");
        // Consecutive queries must agree (compare parsed, not raw
        // strings — HashMap key ordering is non-deterministic).
        let attrs1: serde_json::Value = serde_json::from_str(&payload1.attrs).unwrap();
        let attrs2: serde_json::Value = serde_json::from_str(&payload2.attrs).unwrap();
        assert_eq!(attrs1, attrs2, "consecutive queries should be identical");

        handle.drain_and_stop("test").unwrap();
        handle.await;
    }

    /// Exercises CI-1 (see `proc` module doc).
    ///
    /// Unlike a plain `client()`, which drops the introspect
    /// receiver so queries are silently discarded, an
    /// `introspectable_instance` has a live `serve_introspect` task
    /// and is fully navigable in admin tooling.
    #[tokio::test]
    async fn test_introspectable_instance_responds_to_query() {
        let proc = Proc::isolated();
        let (bridge, handle) = proc.introspectable_instance("bridge").unwrap();
        let actor_id: crate::ActorAddr = handle.actor_addr().clone();

        let (reply_port, reply_rx) = bridge.open_once_port::<IntrospectResult>();
        actor_id.introspect_port().post(
            &bridge,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );
        let payload = reply_rx.recv().await.unwrap();

        // CI-1: introspectable_instance reports status "client"
        // and actor_type "()" (the unit type).
        assert_eq!(
            payload.identity,
            crate::introspect::IntrospectRef::Actor(actor_id.clone())
        );
        assert_status(&payload, "client");
        let actor_type =
            attrs_get(&payload.attrs, "actor_type").and_then(|v| v.as_str().map(String::from));
        assert_eq!(
            actor_type.as_deref(),
            Some("()"),
            "CI-1: actor_type must be \"()\""
        );
    }

    /// Contrast with CI-1: a plain `client()` does NOT respond to
    /// `IntrospectMessage::Query`. Its introspect receiver is dropped
    /// in `Proc::client()`, so the message is silently discarded
    /// and the reply port never receives a value.
    ///
    /// Callers that need TUI visibility must use
    /// `introspectable_instance` instead.
    #[tokio::test]
    async fn test_instance_does_not_respond_to_query() {
        let proc = Proc::isolated();
        let client = proc.client("client");
        let mailbox = proc.client("mailbox");
        let mailbox_id: crate::ActorAddr = mailbox.self_addr().clone();

        let (reply_port, reply_rx) = client.open_once_port::<IntrospectResult>();
        mailbox_id.introspect_port().post(
            &client,
            IntrospectMessage::Query {
                view: IntrospectView::Actor,
                reply: reply_port.bind(),
            },
        );

        // The introspect receiver was dropped in `client()`, so the
        // message is silently discarded and the reply never arrives.
        let result = tokio::time::timeout(Duration::from_millis(100), reply_rx.recv()).await;
        assert!(
            result.is_err(),
            "client() must not respond to IntrospectMessage (introspect receiver dropped)"
        );
    }

    /// Exercises CI-2 (see `proc` module doc).
    ///
    /// Dropping the instance shuts down and joins `serve_introspect`
    /// before terminal status is published.
    #[tokio::test]
    async fn test_introspectable_instance_snapshot_on_drop() {
        let proc = Proc::isolated();
        let (instance, handle) = proc.introspectable_instance("bridge").unwrap();
        let actor_id = handle.actor_addr().clone();

        assert!(
            proc.all_actor_ids().contains(&actor_id),
            "should appear in all_actor_ids while live"
        );

        drop(instance);
        handle.await;

        let snapshot = proc.terminated_snapshot(&actor_id).unwrap();
        let actor_status = attrs_get(&snapshot.attrs, "status")
            .and_then(|v| v.as_str().map(String::from))
            .expect("snapshot attrs must contain status");
        assert!(
            actor_status.starts_with("stopped"),
            "CI-2: snapshot actor_status should be stopped, got: {}",
            actor_status
        );
    }

    #[test]
    fn zombie_status_is_not_terminal() {
        let status = ActorStatus::zombie("hard kill did not finish");

        assert!(!status.is_terminal(), "zombie status is not terminal");
        assert!(status.is_stopping(), "zombie status is stopping");
        assert!(status.is_zombie(), "zombie predicate should match");
    }
}