palladium_runtime/

bounded_cluster.rs

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::engine::EngineHandle;
use crate::fs::{FileSystem, TokioFileSystem};
use crate::reactor::{Reactor, TokioReactor};
use crate::registry::ActorRegistry;
use dashmap::DashMap;
use palladium_actor::{ActorPath, AddrHash, EngineId, RemoteMessage};
use palladium_transport::network::{Network, TokioNetwork};
use palladium_transport::{
    mailbox, QuicTransport, QuicTransportConfig, TcpTransport, TcpTransportConfig, Transport,
    TransportError,
};
use rustls::pki_types::ServerName;
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio_rustls::TlsConnector;

trait SpawnPeerInstaller: Send + Sync {
    fn ensure_peer(&self, engine_id: EngineId, addr: SocketAddr, server_name: Option<String>);
}

impl<R: Reactor + Clone, N: Network + Clone> SpawnPeerInstaller for BoundedTransportHandle<R, N> {
    fn ensure_peer(&self, engine_id: EngineId, addr: SocketAddr, server_name: Option<String>) {
        let source_addr = AddrHash::synthetic(engine_id.as_str().as_bytes());
        match server_name {
            Some(server_name) => self.add_peer(&PeerSpec {
                engine_id: engine_id.clone(),
                addr,
                control_plane_addr: None,
                server_name: Some(server_name),
            }),
            None => self.add_peer(&PeerSpec {
                engine_id: engine_id.clone(),
                addr,
                control_plane_addr: None,
                server_name: None,
            }),
        }
        match self {
            Self::Tcp(transport) => {
                let _ = transport.register_remote_source_addr(engine_id, source_addr);
            }
            Self::Quic(transport) => {
                let _ = transport.register_remote_source_addr(engine_id, source_addr);
            }
        }
    }
}

fn spawn_peer_installers() -> &'static DashMap<String, Arc<dyn SpawnPeerInstaller>> {
    static INSTALLERS: std::sync::OnceLock<DashMap<String, Arc<dyn SpawnPeerInstaller>>> =
        std::sync::OnceLock::new();
    INSTALLERS.get_or_init(DashMap::new)
}

pub(crate) fn register_spawn_peer_installer<R: Reactor + Clone, N: Network + Clone>(
    engine_id: &EngineId,
    transport: &BoundedTransportHandle<R, N>,
) {
    spawn_peer_installers().insert(
        engine_id.as_str().to_string(),
        Arc::new(transport.clone()) as Arc<dyn SpawnPeerInstaller>,
    );
}

pub(crate) fn maybe_register_spawn_caller_peer(
    host_engine_id: &EngineId,
    caller_engine_id: &EngineId,
    caller_transport_addr: SocketAddr,
    caller_server_name: Option<String>,
) {
    if let Some(installer) = spawn_peer_installers().get(host_engine_id.as_str()) {
        installer.ensure_peer(
            caller_engine_id.clone(),
            caller_transport_addr,
            caller_server_name,
        );
    }
}

#[derive(Clone)]
pub struct BoundedClusterConfig {
    pub transport: BoundedTransportConfig,
    pub peers: Vec<PeerSpec>,
    pub declared_actors: Vec<DeclaredRemoteActor>,
    pub roles: Vec<BoundedClusterRole>,
    pub placement_policies: Vec<BoundedPlacementPolicy>,
}

impl BoundedClusterConfig {
    pub fn validate(&self) -> Result<(), ClusterError> {
        validate_static_config(self)
    }
}

#[derive(Clone)]
pub enum BoundedTransportConfig {
    Tcp(TcpTransportConfig),
    Quic(QuicTransportConfig),
}

impl BoundedTransportConfig {
    pub fn engine_id(&self) -> &EngineId {
        match self {
            Self::Tcp(config) => &config.engine_id,
            Self::Quic(config) => &config.engine_id,
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PeerSpec {
    pub engine_id: EngineId,
    pub addr: SocketAddr,
    pub control_plane_addr: Option<SocketAddr>,
    pub server_name: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct DeclaredRemoteActor {
    pub path: ActorPath,
    pub owner: EngineId,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BoundedClusterRole {
    pub name: String,
    pub owner: EngineId,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BoundedPlacementPolicy {
    pub actor_class: String,
    pub role: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ClusterError {
    DuplicatePeer(EngineId),
    LocalPeerDeclared(EngineId),
    DuplicateDeclaredActor(ActorPath),
    LocalActorDeclaredRemote(ActorPath),
    UnknownPeerOwner { path: ActorPath, owner: EngineId },
    DuplicateRole(String),
    UnknownRoleOwner { role: String, owner: EngineId },
    RoleOwnerMissingControlPlaneEndpoint { role: String, owner: EngineId },
    DuplicatePlacementClass(String),
    UnknownPlacementRole { actor_class: String, role: String },
    Transport(TransportError),
}

#[derive(Clone, Debug, PartialEq)]
pub struct RemoteSpawnSpec {
    pub path: ActorPath,
    pub type_name: String,
    pub config: Option<Value>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RemoteSpawnError {
    UnknownPeer(EngineId),
    UnknownRole(String),
    UnknownPlacementClass(String),
    LocalPathCollision(ActorPath),
    RemoteOwnershipConflict { path: ActorPath, owner: EngineId },
    RemotePathAlreadyRunning(ActorPath),
    MissingControlPlaneEndpoint(EngineId),
    InvalidServerName(String),
    ControlPlane(String),
    SpawnTimedOut(ActorPath),
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RemoteSupervisionState {
    Running { restart_count: u32 },
    Restarting { restart_count: u32 },
    ActorUnavailable,
    SupervisorUnavailable,
    TransportUnavailable,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RemoteSupervisionStatus {
    pub path: ActorPath,
    pub supervisor: EngineId,
    pub state: RemoteSupervisionState,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RemoteSupervisionError {
    UnknownRemoteActor(ActorPath),
    UnknownPeer(EngineId),
    MissingControlPlaneEndpoint(EngineId),
}

#[derive(Clone)]
pub(crate) enum ControlPlaneProtocol {
    Tcp,
    Quic,
}

#[derive(Clone)]
pub(crate) struct ControlPlaneClientConfig {
    pub(crate) protocol: ControlPlaneProtocol,
    pub(crate) tls: palladium_transport::TlsConfig,
    pub(crate) wait_timeout: Duration,
}

#[derive(Clone)]
pub(crate) struct BoundedClusterTopology {
    pub(crate) peers: Vec<PeerSpec>,
    pub(crate) roles: Vec<BoundedClusterRole>,
    pub(crate) placement_policies: Vec<BoundedPlacementPolicy>,
}

const MIN_REMOTE_SPAWN_WAIT_TIMEOUT: Duration = Duration::from_secs(5);
const REMOTE_SUPERVISION_PROBE_TIMEOUT: Duration = Duration::from_millis(150);

impl From<TransportError> for ClusterError {
    fn from(value: TransportError) -> Self {
        Self::Transport(value)
    }
}

impl From<TransportError> for RemoteSpawnError {
    fn from(value: TransportError) -> Self {
        Self::ControlPlane(format!("transport route install failed: {value:?}"))
    }
}

#[derive(Clone)]
pub enum BoundedTransportHandle<R: Reactor = TokioReactor, N: Network = TokioNetwork> {
    Tcp(Arc<TcpTransport<R, N>>),
    Quic(Arc<QuicTransport<R, N>>),
}

impl<R: Reactor + Clone, N: Network + Clone> BoundedTransportHandle<R, N> {
    pub fn as_tcp(&self) -> Option<&Arc<TcpTransport<R, N>>> {
        match self {
            Self::Tcp(transport) => Some(transport),
            Self::Quic(_) => None,
        }
    }

    pub fn as_quic(&self) -> Option<&Arc<QuicTransport<R, N>>> {
        match self {
            Self::Tcp(_) => None,
            Self::Quic(transport) => Some(transport),
        }
    }

    pub fn add_peer(&self, peer: &PeerSpec) {
        match self {
            Self::Tcp(transport) => {
                if let Some(server_name) = &peer.server_name {
                    transport.add_peer_with_server_name(
                        peer.engine_id.clone(),
                        peer.addr,
                        server_name.clone(),
                    );
                } else {
                    transport.add_peer(peer.engine_id.clone(), peer.addr);
                }
            }
            Self::Quic(transport) => {
                if let Some(server_name) = &peer.server_name {
                    transport.add_peer_with_server_name(
                        peer.engine_id.clone(),
                        peer.addr,
                        server_name.clone(),
                    );
                } else {
                    transport.add_peer(peer.engine_id.clone(), peer.addr);
                }
            }
        }
    }

    pub fn declare_remote_path(
        &self,
        owner: EngineId,
        path: &ActorPath,
    ) -> Result<(), TransportError> {
        match self {
            Self::Tcp(transport) => transport.declare_remote_path(owner, path),
            Self::Quic(transport) => transport.declare_remote_path(owner, path),
        }
    }

    pub fn advertise_path(&self, addr: AddrHash, path: &ActorPath) -> Result<(), TransportError> {
        match self {
            Self::Tcp(transport) => transport.advertise_path(addr, path),
            Self::Quic(transport) => transport.advertise_path(addr, path),
        }
    }

    pub fn register_source_addr(&self, addr: AddrHash) -> Result<(), TransportError> {
        match self {
            Self::Tcp(transport) => transport.register(addr, mailbox(1).0),
            Self::Quic(transport) => transport.register(addr, mailbox(1).0),
        }
    }

    pub fn undeclare_remote_path(&self, path: &ActorPath) -> Result<(), TransportError> {
        match self {
            Self::Tcp(transport) => transport.undeclare_remote_path(path),
            Self::Quic(transport) => transport.undeclare_remote_path(path),
        }
    }

    pub fn can_route(&self, destination: AddrHash) -> bool {
        match self {
            Self::Tcp(transport) => transport.can_route(destination),
            Self::Quic(transport) => transport.can_route(destination),
        }
    }

    pub async fn wait_for_peer_connection(
        &self,
        engine_id: &EngineId,
        timeout: Duration,
    ) -> Result<(), TransportError> {
        match self {
            Self::Tcp(transport) => transport.wait_for_peer_connection(engine_id, timeout).await,
            Self::Quic(transport) => transport.wait_for_peer_connection(engine_id, timeout).await,
        }
    }
}

#[derive(Clone)]
pub struct BoundedClusterHandle<
    R: Reactor = TokioReactor,
    N: Network = TokioNetwork,
    F: FileSystem = TokioFileSystem,
> {
    engine: EngineHandle<R, N, F>,
    local_engine_id: EngineId,
    control_plane: ControlPlaneClientConfig,
    transport: BoundedTransportHandle<R, N>,
    peers: Vec<PeerSpec>,
    declared_actors: Vec<DeclaredRemoteActor>,
    roles: Vec<BoundedClusterRole>,
    placement_policies: Vec<BoundedPlacementPolicy>,
    supervised_owners: Arc<DashMap<String, EngineId>>,
    registry: Arc<ActorRegistry<R>>,
}

impl<R: Reactor + Clone, N: Network + Clone, F: FileSystem + Clone> BoundedClusterHandle<R, N, F> {
    pub(crate) fn new(
        engine: EngineHandle<R, N, F>,
        local_engine_id: EngineId,
        control_plane: ControlPlaneClientConfig,
        transport: BoundedTransportHandle<R, N>,
        topology: BoundedClusterTopology,
        registry: Arc<ActorRegistry<R>>,
    ) -> Self {
        Self {
            engine,
            local_engine_id,
            control_plane,
            transport,
            peers: topology.peers,
            declared_actors: Vec::new(),
            roles: topology.roles,
            placement_policies: topology.placement_policies,
            supervised_owners: Arc::new(DashMap::new()),
            registry,
        }
    }

    pub fn transport(&self) -> &BoundedTransportHandle<R, N> {
        &self.transport
    }

    pub fn peers(&self) -> &[PeerSpec] {
        &self.peers
    }

    pub fn declared_actors(&self) -> &[DeclaredRemoteActor] {
        &self.declared_actors
    }

    pub fn roles(&self) -> &[BoundedClusterRole] {
        &self.roles
    }

    pub fn placement_policies(&self) -> &[BoundedPlacementPolicy] {
        &self.placement_policies
    }

    pub(crate) fn advertise_existing_local_routes(&self) -> Result<(), ClusterError> {
        self.transport
            .register_source_addr(self.engine.source_addr)
            .map_err(ClusterError::Transport)?;

        for info in self
            .registry
            .snapshot(&crate::introspection::ActorQuery::default(), 0)
        {
            if info.state != crate::introspection::ActorState::Running {
                continue;
            }
            if let Some(slot) = self.registry.get_by_path(&info.path) {
                self.transport
                    .advertise_path(slot.addr, &info.path)
                    .map_err(ClusterError::Transport)?;
            }
        }
        Ok(())
    }

    pub fn add_peer(&mut self, peer: PeerSpec) -> Result<(), ClusterError> {
        validate_peer_spec(&peer, &self.local_engine_id, &self.peers)?;
        self.transport.add_peer(&peer);
        self.advertise_existing_local_routes()?;
        self.peers.push(peer);
        Ok(())
    }

    pub fn declare_remote_actor(
        &mut self,
        owner: EngineId,
        path: ActorPath,
    ) -> Result<(), ClusterError> {
        validate_declared_actor_topology(
            &path,
            &owner,
            &self.local_engine_id,
            &self.peers,
            &self.declared_actors,
        )?;
        validate_declared_actor_registry(
            &path,
            &owner,
            &self.local_engine_id,
            &self.declared_actors,
            &self.registry,
        )?;
        self.transport.declare_remote_path(owner.clone(), &path)?;
        self.supervised_owners
            .insert(path.as_str().to_string(), owner.clone());
        self.declared_actors
            .push(DeclaredRemoteActor { path, owner });
        Ok(())
    }

    pub fn undeclare_remote_actor(&mut self, path: &ActorPath) -> Result<(), ClusterError> {
        self.transport.undeclare_remote_path(path)?;
        self.supervised_owners.remove(path.as_str());
        self.declared_actors
            .retain(|declared| &declared.path != path);
        Ok(())
    }

    pub fn can_route(&self, path: &ActorPath) -> bool {
        self.transport.can_route(AddrHash::new(path, 0))
    }

    pub async fn remote_supervision_status(
        &self,
        path: &ActorPath,
    ) -> Result<RemoteSupervisionStatus, RemoteSupervisionError> {
        let supervisor = self
            .supervised_owners
            .get(path.as_str())
            .map(|entry| entry.clone())
            .or_else(|| {
                self.declared_actors
                    .iter()
                    .find(|declared| declared.path == *path)
                    .map(|declared| declared.owner.clone())
            })
            .ok_or_else(|| RemoteSupervisionError::UnknownRemoteActor(path.clone()))?;

        let peer = self
            .peers
            .iter()
            .find(|peer| peer.engine_id == supervisor)
            .ok_or_else(|| RemoteSupervisionError::UnknownPeer(supervisor.clone()))?;
        let control_plane_addr = peer.control_plane_addr.ok_or_else(|| {
            RemoteSupervisionError::MissingControlPlaneEndpoint(supervisor.clone())
        })?;
        let server_name = peer
            .server_name
            .clone()
            .unwrap_or_else(|| supervisor.as_str().to_string());

        if call_control_plane(
            &self.control_plane,
            control_plane_addr,
            &server_name,
            "engine.status",
            json!({}),
        )
        .await
        .is_err()
        {
            return Ok(RemoteSupervisionStatus {
                path: path.clone(),
                supervisor,
                state: RemoteSupervisionState::SupervisorUnavailable,
            });
        }

        match call_control_plane(
            &self.control_plane,
            control_plane_addr,
            &server_name,
            "actor.info",
            json!({ "path": path.as_str() }),
        )
        .await
        {
            Ok(info) => {
                let restart_count = info
                    .get("restart_count")
                    .and_then(Value::as_u64)
                    .unwrap_or(0) as u32;
                let state = match info.get("state").and_then(Value::as_str) {
                    Some("Stopped") => RemoteSupervisionState::Restarting { restart_count },
                    _ => match self
                        .transport
                        .wait_for_peer_connection(&supervisor, REMOTE_SUPERVISION_PROBE_TIMEOUT)
                        .await
                    {
                        Ok(()) => RemoteSupervisionState::Running { restart_count },
                        Err(_) => RemoteSupervisionState::TransportUnavailable,
                    },
                };
                Ok(RemoteSupervisionStatus {
                    path: path.clone(),
                    supervisor,
                    state,
                })
            }
            Err(RemoteSpawnError::ControlPlane(message))
                if message.contains("actor not found:") || message.contains("actor not found") =>
            {
                Ok(RemoteSupervisionStatus {
                    path: path.clone(),
                    supervisor,
                    state: RemoteSupervisionState::ActorUnavailable,
                })
            }
            Err(_) => Ok(RemoteSupervisionStatus {
                path: path.clone(),
                supervisor,
                state: RemoteSupervisionState::SupervisorUnavailable,
            }),
        }
    }

    pub async fn spawn_remote_on_role<M>(
        &self,
        role: &str,
        spec: RemoteSpawnSpec,
    ) -> Result<palladium_actor::Addr<M>, RemoteSpawnError>
    where
        M: RemoteMessage,
        M::Response: serde::Serialize + for<'de> serde::Deserialize<'de> + Send + 'static,
        R: Send + Sync,
    {
        let owner = self
            .roles
            .iter()
            .find(|declared| declared.name == role)
            .map(|declared| declared.owner.clone())
            .ok_or_else(|| RemoteSpawnError::UnknownRole(role.to_string()))?;
        self.spawn_remote::<M>(owner, spec).await
    }

    pub async fn spawn_remote_for_class<M>(
        &self,
        actor_class: &str,
        spec: RemoteSpawnSpec,
    ) -> Result<palladium_actor::Addr<M>, RemoteSpawnError>
    where
        M: RemoteMessage,
        M::Response: serde::Serialize + for<'de> serde::Deserialize<'de> + Send + 'static,
        R: Send + Sync,
    {
        let role = self
            .placement_policies
            .iter()
            .find(|policy| policy.actor_class == actor_class)
            .map(|policy| policy.role.as_str())
            .ok_or_else(|| RemoteSpawnError::UnknownPlacementClass(actor_class.to_string()))?;
        self.spawn_remote_on_role::<M>(role, spec).await
    }

    pub async fn spawn_remote<M>(
        &self,
        target: EngineId,
        spec: RemoteSpawnSpec,
    ) -> Result<palladium_actor::Addr<M>, RemoteSpawnError>
    where
        M: RemoteMessage,
        M::Response: serde::Serialize + for<'de> serde::Deserialize<'de> + Send + 'static,
        R: Send + Sync,
    {
        validate_remote_spawn_request(&spec.path, &target, &self.declared_actors, &self.registry)?;

        let peer = self
            .peers
            .iter()
            .find(|peer| peer.engine_id == target)
            .ok_or_else(|| RemoteSpawnError::UnknownPeer(target.clone()))?;
        let control_plane_addr = peer
            .control_plane_addr
            .ok_or_else(|| RemoteSpawnError::MissingControlPlaneEndpoint(target.clone()))?;
        let server_name = peer
            .server_name
            .clone()
            .unwrap_or_else(|| target.as_str().to_string());
        let caller_server_name = self.local_engine_id.as_str().to_string();

        let deadline = Instant::now() + self.control_plane.wait_timeout;
        loop {
            match call_actor_spawn(
                &self.control_plane,
                control_plane_addr,
                &server_name,
                &self.local_engine_id,
                transport_listen_addr(&self.transport),
                &caller_server_name,
                &spec,
            )
            .await
            {
                Ok(()) => break,
                Err(RemoteSpawnError::ControlPlane(message))
                    if is_retryable_control_plane_error(&message) && Instant::now() <= deadline =>
                {
                    tokio::time::sleep(Duration::from_millis(20)).await;
                }
                Err(err) => return Err(err),
            }
        }

        while Instant::now() <= deadline {
            if actor_is_spawned(
                &self.control_plane,
                control_plane_addr,
                &server_name,
                &spec.path,
            )
            .await?
            {
                self.transport
                    .declare_remote_path(target.clone(), &spec.path)
                    .map_err(RemoteSpawnError::from)?;
                self.supervised_owners
                    .insert(spec.path.as_str().to_string(), target.clone());
                self.transport
                    .wait_for_peer_connection(&target, self.control_plane.wait_timeout)
                    .await
                    .map_err(RemoteSpawnError::from)?;
                return Ok(self.engine.remote_addr_for_path::<M>(&spec.path));
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }

        Err(RemoteSpawnError::SpawnTimedOut(spec.path))
    }
}

async fn call_actor_spawn(
    client: &ControlPlaneClientConfig,
    addr: SocketAddr,
    server_name: &str,
    caller_engine_id: &EngineId,
    caller_transport_addr: SocketAddr,
    caller_server_name: &str,
    spec: &RemoteSpawnSpec,
) -> Result<(), RemoteSpawnError> {
    let params = match &spec.config {
        Some(config) => json!({
            "path": spec.path.as_str(),
            "type_name": spec.type_name,
            "config": config,
            "caller_engine_id": caller_engine_id.as_str(),
            "caller_transport_addr": caller_transport_addr,
            "caller_server_name": caller_server_name,
        }),
        None => json!({
            "path": spec.path.as_str(),
            "type_name": spec.type_name,
            "caller_engine_id": caller_engine_id.as_str(),
            "caller_transport_addr": caller_transport_addr,
            "caller_server_name": caller_server_name,
        }),
    };
    match call_control_plane(client, addr, server_name, "actor.spawn", params).await {
        Ok(_) => {}
        Err(RemoteSpawnError::ControlPlane(message))
            if message.contains("actor already running at") =>
        {
            return Err(RemoteSpawnError::RemotePathAlreadyRunning(
                spec.path.clone(),
            ));
        }
        Err(err) => return Err(err),
    }
    Ok(())
}

fn transport_listen_addr<R: Reactor + Clone, N: Network + Clone>(
    transport: &BoundedTransportHandle<R, N>,
) -> SocketAddr {
    match transport {
        BoundedTransportHandle::Tcp(transport) => transport.local_addr(),
        BoundedTransportHandle::Quic(transport) => transport.local_addr(),
    }
}

fn is_retryable_control_plane_error(message: &str) -> bool {
    let msg = message.to_ascii_lowercase();
    msg.contains("connection refused")
        || msg.contains("connection reset")
        || msg.contains("timed out")
        || msg.contains("broken pipe")
        || msg.contains("not connected")
}

fn validate_remote_spawn_request<R: Reactor>(
    path: &ActorPath,
    owner: &EngineId,
    declared_actors: &[DeclaredRemoteActor],
    registry: &Arc<ActorRegistry<R>>,
) -> Result<(), RemoteSpawnError> {
    if registry.get_by_path(path).is_some() {
        return Err(RemoteSpawnError::LocalPathCollision(path.clone()));
    }

    if let Some(declared) = declared_actors
        .iter()
        .find(|declared| declared.path == *path && declared.owner != *owner)
    {
        return Err(RemoteSpawnError::RemoteOwnershipConflict {
            path: path.clone(),
            owner: declared.owner.clone(),
        });
    }

    Ok(())
}

async fn actor_is_spawned(
    client: &ControlPlaneClientConfig,
    addr: SocketAddr,
    server_name: &str,
    path: &ActorPath,
) -> Result<bool, RemoteSpawnError> {
    match call_control_plane(
        client,
        addr,
        server_name,
        "actor.info",
        json!({ "path": path.as_str() }),
    )
    .await
    {
        Ok(_) => Ok(true),
        Err(RemoteSpawnError::ControlPlane(message))
            if message.contains("actor not found:")
                || message.contains("actor not found")
                || message.contains("User supervisor not available") =>
        {
            Ok(false)
        }
        Err(err) => Err(err),
    }
}

async fn call_control_plane(
    client: &ControlPlaneClientConfig,
    addr: SocketAddr,
    server_name: &str,
    method: &str,
    params: Value,
) -> Result<Value, RemoteSpawnError> {
    let request = json!({
        "id": 1,
        "method": method,
        "params": params,
    });

    match client.protocol {
        ControlPlaneProtocol::Tcp => {
            call_control_plane_tcp(addr, server_name, &client.tls, &request).await
        }
        ControlPlaneProtocol::Quic => {
            call_control_plane_quic(addr, server_name, &client.tls, &request).await
        }
    }
}

async fn call_control_plane_tcp(
    addr: SocketAddr,
    server_name: &str,
    tls: &palladium_transport::TlsConfig,
    request: &Value,
) -> Result<Value, RemoteSpawnError> {
    let config = tls
        .client_config(server_name)
        .map_err(|err| RemoteSpawnError::ControlPlane(format!("tcp tls config failed: {err:?}")))?;
    let name = ServerName::try_from(server_name.to_string())
        .map_err(|_| RemoteSpawnError::InvalidServerName(server_name.to_string()))?;
    let stream = tokio::net::TcpStream::connect(addr)
        .await
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;
    let connector = TlsConnector::from(config);
    let tls_stream = connector
        .connect(name, stream)
        .await
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;

    let (reader, mut writer) = tokio::io::split(tls_stream);
    let mut body = request.to_string();
    body.push('\n');
    writer
        .write_all(body.as_bytes())
        .await
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;
    let mut line = String::new();
    let mut reader = BufReader::new(reader);
    reader
        .read_line(&mut line)
        .await
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;
    parse_control_plane_response(&line)
}

async fn call_control_plane_quic(
    addr: SocketAddr,
    server_name: &str,
    tls: &palladium_transport::TlsConfig,
    request: &Value,
) -> Result<Value, RemoteSpawnError> {
    ensure_rustls_provider();
    let mut client_crypto = (*tls.client_config(server_name).map_err(|err| {
        RemoteSpawnError::ControlPlane(format!("quic tls config failed: {err:?}"))
    })?)
    .clone();
    client_crypto.alpn_protocols = vec![b"pd-control".to_vec()];
    let client_tls: s2n_quic::provider::tls::rustls::Client = client_crypto.into();

    let client = s2n_quic::Client::builder()
        .with_tls(client_tls)
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?
        .with_io("0.0.0.0:0")
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?
        .start()
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;

    let connect = s2n_quic::client::Connect::new(addr).with_server_name(server_name.to_string());
    let mut conn = client
        .connect(connect)
        .await
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;
    let mut stream = conn
        .open_bidirectional_stream()
        .await
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;

    let mut body = request.to_string();
    body.push('\n');
    stream
        .write_all(body.as_bytes())
        .await
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;
    stream
        .close()
        .await
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;

    let mut buf = Vec::new();
    stream
        .read_to_end(&mut buf)
        .await
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;
    let line = String::from_utf8_lossy(&buf);
    parse_control_plane_response(&line)
}

fn parse_control_plane_response(line: &str) -> Result<Value, RemoteSpawnError> {
    let response: Value = serde_json::from_str(line.trim())
        .map_err(|err| RemoteSpawnError::ControlPlane(err.to_string()))?;
    if let Some(error) = response.get("error") {
        let message = error
            .get("message")
            .and_then(Value::as_str)
            .unwrap_or("unknown control-plane error");
        return Err(RemoteSpawnError::ControlPlane(message.to_string()));
    }
    Ok(response.get("result").cloned().unwrap_or(Value::Null))
}

fn ensure_rustls_provider() {
    static INIT: std::sync::Once = std::sync::Once::new();
    INIT.call_once(|| {
        #[cfg(feature = "aws-lc-rs")]
        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
        #[cfg(all(not(feature = "aws-lc-rs"), feature = "ring"))]
        let _ = rustls::crypto::ring::default_provider().install_default();
    });
}

pub(crate) fn remote_spawn_wait_timeout(ask_timeout: Duration) -> Duration {
    ask_timeout.max(MIN_REMOTE_SPAWN_WAIT_TIMEOUT)
}

pub(crate) fn validate_config<R: Reactor>(
    config: &BoundedClusterConfig,
    registry: &Arc<ActorRegistry<R>>,
) -> Result<(), ClusterError> {
    validate_static_config(config)?;

    let local_engine_id = config.transport.engine_id();
    let mut seen_declared = Vec::new();
    for declared in &config.declared_actors {
        validate_declared_actor_registry(
            &declared.path,
            &declared.owner,
            local_engine_id,
            &seen_declared,
            registry,
        )?;
        seen_declared.push(declared.clone());
    }

    Ok(())
}

fn validate_static_config(config: &BoundedClusterConfig) -> Result<(), ClusterError> {
    let local_engine_id = config.transport.engine_id();
    let mut seen_peers = Vec::new();
    for peer in &config.peers {
        validate_peer_spec(peer, local_engine_id, &seen_peers)?;
        seen_peers.push(peer.clone());
    }

    let mut seen_declared = Vec::new();
    for declared in &config.declared_actors {
        validate_declared_actor_topology(
            &declared.path,
            &declared.owner,
            local_engine_id,
            &config.peers,
            &seen_declared,
        )?;
        seen_declared.push(declared.clone());
    }

    let mut seen_roles = Vec::new();
    for role in &config.roles {
        validate_role(role, &config.peers, &seen_roles)?;
        seen_roles.push(role.clone());
    }

    let mut seen_placement_policies = Vec::new();
    for policy in &config.placement_policies {
        validate_placement_policy(policy, &config.roles, &seen_placement_policies)?;
        seen_placement_policies.push(policy.clone());
    }

    Ok(())
}

fn validate_peer_spec(
    peer: &PeerSpec,
    local_engine_id: &EngineId,
    existing: &[PeerSpec],
) -> Result<(), ClusterError> {
    if &peer.engine_id == local_engine_id {
        return Err(ClusterError::LocalPeerDeclared(peer.engine_id.clone()));
    }
    if existing
        .iter()
        .any(|existing| existing.engine_id == peer.engine_id)
    {
        return Err(ClusterError::DuplicatePeer(peer.engine_id.clone()));
    }
    Ok(())
}

fn validate_declared_actor_topology(
    path: &ActorPath,
    owner: &EngineId,
    local_engine_id: &EngineId,
    peers: &[PeerSpec],
    existing: &[DeclaredRemoteActor],
) -> Result<(), ClusterError> {
    if owner == local_engine_id {
        return Err(ClusterError::LocalActorDeclaredRemote(path.clone()));
    }
    if existing.iter().any(|existing| existing.path == *path) {
        return Err(ClusterError::DuplicateDeclaredActor(path.clone()));
    }
    if !peers.iter().any(|peer| peer.engine_id == *owner) {
        return Err(ClusterError::UnknownPeerOwner {
            path: path.clone(),
            owner: owner.clone(),
        });
    }
    Ok(())
}

fn validate_declared_actor_registry<R: Reactor>(
    path: &ActorPath,
    owner: &EngineId,
    local_engine_id: &EngineId,
    existing: &[DeclaredRemoteActor],
    registry: &Arc<ActorRegistry<R>>,
) -> Result<(), ClusterError> {
    if owner == local_engine_id || registry.get_by_path(path).is_some() {
        return Err(ClusterError::LocalActorDeclaredRemote(path.clone()));
    }
    if existing.iter().any(|existing| existing.path == *path) {
        return Err(ClusterError::DuplicateDeclaredActor(path.clone()));
    }
    Ok(())
}

fn validate_role(
    role: &BoundedClusterRole,
    peers: &[PeerSpec],
    existing: &[BoundedClusterRole],
) -> Result<(), ClusterError> {
    if existing.iter().any(|existing| existing.name == role.name) {
        return Err(ClusterError::DuplicateRole(role.name.clone()));
    }
    if !peers.iter().any(|peer| peer.engine_id == role.owner) {
        return Err(ClusterError::UnknownRoleOwner {
            role: role.name.clone(),
            owner: role.owner.clone(),
        });
    }
    if !peers
        .iter()
        .any(|peer| peer.engine_id == role.owner && peer.control_plane_addr.is_some())
    {
        return Err(ClusterError::RoleOwnerMissingControlPlaneEndpoint {
            role: role.name.clone(),
            owner: role.owner.clone(),
        });
    }
    Ok(())
}

fn validate_placement_policy(
    policy: &BoundedPlacementPolicy,
    roles: &[BoundedClusterRole],
    existing: &[BoundedPlacementPolicy],
) -> Result<(), ClusterError> {
    if existing
        .iter()
        .any(|existing| existing.actor_class == policy.actor_class)
    {
        return Err(ClusterError::DuplicatePlacementClass(
            policy.actor_class.clone(),
        ));
    }
    if !roles.iter().any(|role| role.name == policy.role) {
        return Err(ClusterError::UnknownPlacementRole {
            actor_class: policy.actor_class.clone(),
            role: policy.role.clone(),
        });
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{
        call_control_plane, validate_config, BoundedClusterConfig, BoundedClusterRole,
        BoundedPlacementPolicy, BoundedTransportConfig, ClusterError, DeclaredRemoteActor,
        PeerSpec, RemoteSpawnError, RemoteSpawnSpec, RemoteSupervisionState,
    };
    use crate::{Engine, EngineConfig};
    use palladium_actor::{
        Actor, ActorContext, ActorError, ActorPath, AddrHash, ChildSpec, EngineId, Envelope,
        Message, MessagePayload, NamespacePolicy, RestartPolicy, ShutdownPolicy,
    };
    use palladium_transport::{QuicTransportConfig, TcpTransportConfig, TlsConfig, Transport};
    use rcgen::{
        BasicConstraints, Certificate, CertificateParams, ExtendedKeyUsagePurpose, IsCa, KeyPair,
        KeyUsagePurpose,
    };
    use serde_json::json;
    use std::net::{SocketAddr, TcpListener, UdpSocket};
    use std::sync::Arc;
    use std::time::{Duration, Instant};

    fn make_ca() -> (Certificate, KeyPair) {
        let mut params = CertificateParams::default();
        params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
        params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign];
        let ca_key = KeyPair::generate().expect("ca keypair");
        let cert = params.self_signed(&ca_key).expect("ca cert");
        (cert, ca_key)
    }

    fn make_tls_config(common_name: &str, ca: &Certificate, ca_key: &KeyPair) -> TlsConfig {
        let keypair = KeyPair::generate().expect("leaf keypair");
        let params = CertificateParams::new(vec![common_name.to_string()]).expect("params");
        let mut params = params;
        params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
        params.extended_key_usages = vec![
            ExtendedKeyUsagePurpose::ServerAuth,
            ExtendedKeyUsagePurpose::ClientAuth,
        ];
        let cert = params.signed_by(&keypair, ca, ca_key).expect("leaf cert");
        let cert_der = cert.der().to_vec();
        let key_der = rustls::pki_types::PrivatePkcs8KeyDer::from(keypair.serialize_der());
        let ca_der = ca.der().to_vec();
        TlsConfig {
            cert_chain: vec![rustls::pki_types::CertificateDer::from(cert_der)],
            private_key: rustls::pki_types::PrivateKeyDer::from(key_der),
            trusted_cas: vec![rustls::pki_types::CertificateDer::from(ca_der)],
        }
    }

    fn tcp_config(engine_id: &str, tls: TlsConfig) -> TcpTransportConfig {
        TcpTransportConfig {
            engine_id: EngineId::new(engine_id),
            listen_addr: "127.0.0.1:0".parse().unwrap(),
            max_connections_per_peer: 1,
            idle_timeout: Duration::from_secs(1),
            send_buffer_size: 8,
            nodelay: true,
            tls,
            send_delay: Duration::from_millis(0),
        }
    }

    fn reserve_tcp_addr() -> SocketAddr {
        let listener = TcpListener::bind("127.0.0.1:0").expect("reserve tcp addr");
        let addr = listener.local_addr().expect("tcp local addr");
        drop(listener);
        addr
    }

    fn reserve_udp_addr() -> SocketAddr {
        let socket = UdpSocket::bind("127.0.0.1:0").expect("reserve udp addr");
        let addr = socket.local_addr().expect("udp local addr");
        drop(socket);
        addr
    }

    #[test]
    fn bounded_cluster_config_rejects_duplicate_peers() {
        let (ca, ca_key) = make_ca();
        let engine = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let config = BoundedClusterConfig {
            transport: BoundedTransportConfig::Tcp(tcp_config(
                "engine-a.example",
                make_tls_config("engine-a.example", &ca, &ca_key),
            )),
            peers: vec![
                PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: "127.0.0.1:4100".parse().unwrap(),
                    control_plane_addr: None,
                    server_name: None,
                },
                PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: "127.0.0.1:4200".parse().unwrap(),
                    control_plane_addr: None,
                    server_name: None,
                },
            ],
            declared_actors: Vec::new(),
            roles: Vec::new(),
            placement_policies: Vec::new(),
        };

        let err = validate_config(&config, &engine.handle().registry).unwrap_err();
        assert_eq!(
            err,
            ClusterError::DuplicatePeer(EngineId::new("engine-b.example"))
        );
    }

    #[test]
    fn bounded_cluster_config_rejects_unknown_declared_actor_owner() {
        let (ca, ca_key) = make_ca();
        let engine = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let config = BoundedClusterConfig {
            transport: BoundedTransportConfig::Tcp(tcp_config(
                "engine-a.example",
                make_tls_config("engine-a.example", &ca, &ca_key),
            )),
            peers: vec![PeerSpec {
                engine_id: EngineId::new("engine-b.example"),
                addr: "127.0.0.1:4100".parse().unwrap(),
                control_plane_addr: Some("127.0.0.1:5100".parse().unwrap()),
                server_name: None,
            }],
            declared_actors: vec![DeclaredRemoteActor {
                path: ActorPath::parse("/user/missing-owner").unwrap(),
                owner: EngineId::new("engine-c.example"),
            }],
            roles: Vec::new(),
            placement_policies: Vec::new(),
        };

        let err = validate_config(&config, &engine.handle().registry).unwrap_err();
        assert_eq!(
            err,
            ClusterError::UnknownPeerOwner {
                path: ActorPath::parse("/user/missing-owner").unwrap(),
                owner: EngineId::new("engine-c.example"),
            }
        );
    }

    #[test]
    fn bounded_cluster_config_rejects_duplicate_roles() {
        let (ca, ca_key) = make_ca();
        let engine = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let config = BoundedClusterConfig {
            transport: BoundedTransportConfig::Tcp(tcp_config(
                "engine-a.example",
                make_tls_config("engine-a.example", &ca, &ca_key),
            )),
            peers: vec![PeerSpec {
                engine_id: EngineId::new("engine-b.example"),
                addr: "127.0.0.1:4100".parse().unwrap(),
                control_plane_addr: Some("127.0.0.1:5100".parse().unwrap()),
                server_name: None,
            }],
            declared_actors: Vec::new(),
            roles: vec![
                BoundedClusterRole {
                    name: "context-service".to_string(),
                    owner: EngineId::new("engine-b.example"),
                },
                BoundedClusterRole {
                    name: "context-service".to_string(),
                    owner: EngineId::new("engine-b.example"),
                },
            ],
            placement_policies: Vec::new(),
        };

        let err = validate_config(&config, &engine.handle().registry).unwrap_err();
        assert_eq!(
            err,
            ClusterError::DuplicateRole("context-service".to_string())
        );
    }

    #[test]
    fn bounded_cluster_config_rejects_unknown_role_owner() {
        let (ca, ca_key) = make_ca();
        let engine = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let config = BoundedClusterConfig {
            transport: BoundedTransportConfig::Tcp(tcp_config(
                "engine-a.example",
                make_tls_config("engine-a.example", &ca, &ca_key),
            )),
            peers: vec![PeerSpec {
                engine_id: EngineId::new("engine-b.example"),
                addr: "127.0.0.1:4100".parse().unwrap(),
                control_plane_addr: None,
                server_name: None,
            }],
            declared_actors: Vec::new(),
            roles: vec![BoundedClusterRole {
                name: "gateway".to_string(),
                owner: EngineId::new("engine-c.example"),
            }],
            placement_policies: Vec::new(),
        };

        let err = validate_config(&config, &engine.handle().registry).unwrap_err();
        assert_eq!(
            err,
            ClusterError::UnknownRoleOwner {
                role: "gateway".to_string(),
                owner: EngineId::new("engine-c.example"),
            }
        );
    }

    #[test]
    fn bounded_cluster_config_rejects_role_owner_without_control_plane() {
        let (ca, ca_key) = make_ca();
        let config = BoundedClusterConfig {
            transport: BoundedTransportConfig::Tcp(tcp_config(
                "engine-a.example",
                make_tls_config("engine-a.example", &ca, &ca_key),
            )),
            peers: vec![PeerSpec {
                engine_id: EngineId::new("engine-b.example"),
                addr: "127.0.0.1:4100".parse().unwrap(),
                control_plane_addr: None,
                server_name: None,
            }],
            declared_actors: Vec::new(),
            roles: vec![BoundedClusterRole {
                name: "session-host".to_string(),
                owner: EngineId::new("engine-b.example"),
            }],
            placement_policies: Vec::new(),
        };

        let err = config.validate().unwrap_err();
        assert_eq!(
            err,
            ClusterError::RoleOwnerMissingControlPlaneEndpoint {
                role: "session-host".to_string(),
                owner: EngineId::new("engine-b.example"),
            }
        );
    }

    #[test]
    fn bounded_cluster_config_rejects_duplicate_placement_classes() {
        let (ca, ca_key) = make_ca();
        let config = BoundedClusterConfig {
            transport: BoundedTransportConfig::Tcp(tcp_config(
                "engine-a.example",
                make_tls_config("engine-a.example", &ca, &ca_key),
            )),
            peers: vec![PeerSpec {
                engine_id: EngineId::new("engine-b.example"),
                addr: "127.0.0.1:4100".parse().unwrap(),
                control_plane_addr: Some("127.0.0.1:5100".parse().unwrap()),
                server_name: None,
            }],
            declared_actors: Vec::new(),
            roles: vec![BoundedClusterRole {
                name: "context-service".to_string(),
                owner: EngineId::new("engine-b.example"),
            }],
            placement_policies: vec![
                BoundedPlacementPolicy {
                    actor_class: "SessionActor".to_string(),
                    role: "context-service".to_string(),
                },
                BoundedPlacementPolicy {
                    actor_class: "SessionActor".to_string(),
                    role: "context-service".to_string(),
                },
            ],
        };

        let err = config.validate().unwrap_err();
        assert_eq!(
            err,
            ClusterError::DuplicatePlacementClass("SessionActor".to_string())
        );
    }

    #[test]
    fn bounded_cluster_config_rejects_unknown_placement_role() {
        let (ca, ca_key) = make_ca();
        let config = BoundedClusterConfig {
            transport: BoundedTransportConfig::Tcp(tcp_config(
                "engine-a.example",
                make_tls_config("engine-a.example", &ca, &ca_key),
            )),
            peers: vec![PeerSpec {
                engine_id: EngineId::new("engine-b.example"),
                addr: "127.0.0.1:4100".parse().unwrap(),
                control_plane_addr: Some("127.0.0.1:5100".parse().unwrap()),
                server_name: None,
            }],
            declared_actors: Vec::new(),
            roles: vec![BoundedClusterRole {
                name: "context-service".to_string(),
                owner: EngineId::new("engine-b.example"),
            }],
            placement_policies: vec![BoundedPlacementPolicy {
                actor_class: "SessionActor".to_string(),
                role: "missing-role".to_string(),
            }],
        };

        let err = config.validate().unwrap_err();
        assert_eq!(
            err,
            ClusterError::UnknownPlacementRole {
                actor_class: "SessionActor".to_string(),
                role: "missing-role".to_string(),
            }
        );
    }

    #[tokio::test]
    async fn attach_bounded_cluster_rejects_role_owner_without_control_plane() {
        let (ca, ca_key) = make_ca();
        let engine = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });

        let err = match engine
            .handle()
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config(
                    "engine-a.example",
                    make_tls_config("engine-a.example", &ca, &ca_key),
                )),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: "127.0.0.1:4100".parse().unwrap(),
                    control_plane_addr: None,
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: vec![BoundedClusterRole {
                    name: "session-host".to_string(),
                    owner: EngineId::new("engine-b.example"),
                }],
                placement_policies: Vec::new(),
            })
            .await
        {
            Ok(_) => panic!("expected bounded cluster attach to reject missing control plane"),
            Err(err) => err,
        };

        assert_eq!(
            err,
            ClusterError::RoleOwnerMissingControlPlaneEndpoint {
                role: "session-host".to_string(),
                owner: EngineId::new("engine-b.example"),
            }
        );
    }

    #[test]
    fn remote_spawn_wait_timeout_is_not_coupled_to_small_ask_timeout() {
        assert_eq!(
            super::remote_spawn_wait_timeout(Duration::from_millis(25)),
            Duration::from_secs(5)
        );
        assert_eq!(
            super::remote_spawn_wait_timeout(Duration::from_secs(8)),
            Duration::from_secs(8)
        );
    }

    fn tcp_tests_available() -> bool {
        if std::env::var("PD_ENABLE_TCP_TESTS").is_err() {
            return false;
        }
        std::net::TcpListener::bind("127.0.0.1:0").is_ok()
    }

    fn quic_tests_available() -> bool {
        super::ensure_rustls_provider();
        rustls::crypto::CryptoProvider::get_default().is_some()
    }

    async fn wait_for_supervision_state<R, N, F>(
        cluster: &super::BoundedClusterHandle<R, N, F>,
        path: &ActorPath,
        expected: RemoteSupervisionState,
    ) where
        R: crate::reactor::Reactor + Clone + Send + Sync,
        N: palladium_transport::network::Network + Clone,
        F: crate::fs::FileSystem + Clone,
    {
        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            let status = cluster
                .remote_supervision_status(path)
                .await
                .expect("supervision status");
            if status.state == expected {
                return;
            }
            if Instant::now() > deadline {
                panic!(
                    "supervision status did not reach expected state: got {:?}, expected {:?}",
                    status.state, expected
                );
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
    }

    #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
    struct EchoRequest {
        value: u64,
    }

    #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
    struct EchoResponse {
        doubled: u64,
    }

    impl Message for EchoRequest {
        type Response = EchoResponse;
        const TYPE_TAG: u64 = palladium_actor::fnv1a_64("palladium_runtime.test.BoundedEcho");
    }

    struct EchoActor;

    impl<R: crate::reactor::Reactor> Actor<R> for EchoActor {
        fn on_message(
            &mut self,
            ctx: &mut ActorContext<R>,
            envelope: &Envelope,
            payload: MessagePayload,
        ) -> Result<(), ActorError> {
            let req = payload
                .extract::<EchoRequest>()
                .map_err(|_| ActorError::Handler)?;
            let resp = EchoResponse {
                doubled: req.value * 2,
            };
            let resp_env = envelope.response(0);
            ctx.send_raw(resp_env, MessagePayload::local(resp))
                .map_err(|_| ActorError::Handler)
        }
    }

    #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
    struct CrashRequest;

    impl Message for CrashRequest {
        type Response = ();
        const TYPE_TAG: u64 = palladium_actor::fnv1a_64("palladium_runtime.test.BoundedCrash");
    }

    struct SpawnedEchoActor;

    impl<R: crate::reactor::Reactor> Actor<R> for SpawnedEchoActor {
        fn on_message(
            &mut self,
            ctx: &mut ActorContext<R>,
            envelope: &Envelope,
            payload: MessagePayload,
        ) -> Result<(), ActorError> {
            match envelope.type_tag {
                EchoRequest::TYPE_TAG => {
                    let req = payload
                        .extract::<EchoRequest>()
                        .map_err(|_| ActorError::Handler)?;
                    let resp = EchoResponse {
                        doubled: req.value * 2,
                    };
                    let resp_env = envelope.response(0);
                    ctx.send_raw(resp_env, MessagePayload::local(resp))
                        .map_err(|_| ActorError::Handler)
                }
                CrashRequest::TYPE_TAG => Err(ActorError::Handler),
                _ => Err(ActorError::Handler),
            }
        }
    }

    fn bounded_spawn_handler() -> Arc<crate::engine::ActorSpawnFn<crate::TokioReactor>> {
        Arc::new(|type_name: &str, _config: &[u8]| match type_name {
            "bounded-echo" => Ok(Box::new(SpawnedEchoActor)),
            _ => Err(format!("unknown remote spawn type: {type_name}")),
        })
    }

    #[tokio::test]
    async fn bounded_cluster_attach_reaches_remote_actor_over_tcp() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/cluster-echo").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);

        let mut engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            ..Default::default()
        });

        let ns = NamespacePolicy::default_for(&actor_path).unwrap();
        engine_a.add_user_actor(ChildSpec::new(
            "cluster-echo",
            RestartPolicy::Permanent,
            ShutdownPolicy::Timeout(Duration::from_secs(1)),
            ns,
            move || Box::new(EchoActor),
        ));

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls_a)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster a");
        let tcp_a = cluster_a
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport a");

        let cluster_b = handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-b.example", tls_b)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-a.example"),
                    addr: tcp_a.local_addr(),
                    control_plane_addr: None,
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster b");
        let tcp_b = cluster_b
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport b");

        tcp_a.add_peer(EngineId::new("engine-b.example"), tcp_b.local_addr());
        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();

        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));

        let canonical = AddrHash::new(&actor_path, 0);
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while !cluster_b.can_route(&actor_path) {
            if std::time::Instant::now() > deadline {
                panic!("bounded cluster route did not propagate");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        assert!(tcp_b.can_route(canonical));

        let remote = handle_b.remote_addr_for_path::<EchoRequest>(&actor_path);
        let response = remote
            .ask(EchoRequest { value: 12 })
            .await
            .expect("bounded cluster ask");
        assert_eq!(response, EchoResponse { doubled: 24 });

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_add_peer_after_start_backfills_existing_tcp_routes() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/cluster-echo-add-peer").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);

        let mut engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            ..Default::default()
        });

        let ns = NamespacePolicy::default_for(&actor_path).unwrap();
        engine_a.add_user_actor(ChildSpec::new(
            "cluster-echo-add-peer",
            RestartPolicy::Permanent,
            ShutdownPolicy::Timeout(Duration::from_secs(1)),
            ns,
            move || Box::new(EchoActor),
        ));

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let mut cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls_a)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster a");
        let tcp_a = cluster_a
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport a");

        let cluster_b = handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-b.example", tls_b)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-a.example"),
                    addr: tcp_a.local_addr(),
                    control_plane_addr: None,
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster b");
        let tcp_b = cluster_b
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport b");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();

        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));

        let local_ready_deadline = std::time::Instant::now() + Duration::from_secs(5);
        while handle_a.registry.get_by_path(&actor_path).is_none() {
            if std::time::Instant::now() > local_ready_deadline {
                panic!("local actor did not start before peer add");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }

        cluster_a
            .add_peer(PeerSpec {
                engine_id: EngineId::new("engine-b.example"),
                addr: tcp_b.local_addr(),
                control_plane_addr: None,
                server_name: None,
            })
            .expect("add peer b to cluster a after start");

        let canonical = AddrHash::new(&actor_path, 0);
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while !cluster_b.can_route(&actor_path) {
            if std::time::Instant::now() > deadline {
                panic!("bounded cluster route did not propagate after add_peer");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        assert!(tcp_b.can_route(canonical));

        let remote = handle_b.remote_addr_for_path::<EchoRequest>(&actor_path);
        let response = remote
            .ask(EchoRequest { value: 15 })
            .await
            .expect("bounded cluster ask after add_peer");
        assert_eq!(response, EchoResponse { doubled: 30 });

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_declared_remote_actor_installs_canonical_route() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let (ca, ca_key) = make_ca();
        let tls = make_tls_config("engine-a.example", &ca, &ca_key);
        let engine = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let handle = engine.handle();

        let mut cluster = handle
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: "127.0.0.1:4100".parse().unwrap(),
                    control_plane_addr: None,
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster");

        let path = ActorPath::parse("/user/declared-remote").unwrap();
        cluster
            .declare_remote_actor(EngineId::new("engine-b.example"), path.clone())
            .expect("declare remote actor");

        assert!(cluster.can_route(&path));
        assert!(cluster
            .transport()
            .as_tcp()
            .expect("tcp transport")
            .can_route(AddrHash::new(&path, 0)));
    }

    #[tokio::test]
    async fn bounded_cluster_spawn_remote_by_role_rejects_local_path_collision() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/local-remote-collision").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);
        let control_plane_b = reserve_tcp_addr();

        let mut engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            control_plane_tcp_addr: Some(control_plane_b.to_string()),
            control_plane_tls: Some(tls_b.clone()),
            actor_spawn: Some(bounded_spawn_handler()),
            ..Default::default()
        });

        let ns = NamespacePolicy::default_for(&actor_path).unwrap();
        engine_a.add_user_actor(ChildSpec::new(
            "local-remote-collision",
            RestartPolicy::Permanent,
            ShutdownPolicy::Timeout(Duration::from_secs(1)),
            ns,
            move || Box::new(EchoActor),
        ));

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();

        let cluster_b = handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-b.example", tls_b)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster b");
        let tcp_b = cluster_b
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport b");

        let cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls_a)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: tcp_b.local_addr(),
                    control_plane_addr: Some(control_plane_b),
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: vec![BoundedClusterRole {
                    name: "context-service".to_string(),
                    owner: EngineId::new("engine-b.example"),
                }],
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster a");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        std::thread::spawn(move || engine_a.run(shutdown_a_rx));

        let deadline = Instant::now() + Duration::from_secs(5);
        while handle_a.registry.get_by_path(&actor_path).is_none() {
            if Instant::now() > deadline {
                panic!("local collision actor did not start");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }

        let err = cluster_a
            .spawn_remote_on_role::<EchoRequest>(
                "context-service",
                RemoteSpawnSpec {
                    path: actor_path.clone(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect_err("local path collision should fail");
        assert_eq!(err, RemoteSpawnError::LocalPathCollision(actor_path));

        shutdown_a_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_spawn_remote_rejects_conflicting_declared_owner() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/conflicting-remote-owner").unwrap();
        let (ca, ca_key) = make_ca();
        let tls = make_tls_config("engine-a.example", &ca, &ca_key);
        let engine = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let handle = engine.handle();

        let cluster = handle
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls)),
                peers: vec![
                    PeerSpec {
                        engine_id: EngineId::new("engine-b.example"),
                        addr: "127.0.0.1:4100".parse().unwrap(),
                        control_plane_addr: Some("127.0.0.1:5100".parse().unwrap()),
                        server_name: None,
                    },
                    PeerSpec {
                        engine_id: EngineId::new("engine-c.example"),
                        addr: "127.0.0.1:4200".parse().unwrap(),
                        control_plane_addr: Some("127.0.0.1:5200".parse().unwrap()),
                        server_name: None,
                    },
                ],
                declared_actors: vec![DeclaredRemoteActor {
                    path: actor_path.clone(),
                    owner: EngineId::new("engine-b.example"),
                }],
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster");

        let err = cluster
            .spawn_remote::<EchoRequest>(
                EngineId::new("engine-c.example"),
                RemoteSpawnSpec {
                    path: actor_path.clone(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect_err("conflicting declared owner should fail");
        assert_eq!(
            err,
            RemoteSpawnError::RemoteOwnershipConflict {
                path: actor_path,
                owner: EngineId::new("engine-b.example"),
            }
        );
    }

    #[tokio::test]
    async fn bounded_cluster_spawn_remote_over_tcp_returns_typed_handle() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/remote-spawn-tcp").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);
        let control_plane_b = reserve_tcp_addr();

        let engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            control_plane_tcp_addr: Some(control_plane_b.to_string()),
            control_plane_tls: Some(tls_b.clone()),
            actor_spawn: Some(bounded_spawn_handler()),
            ..Default::default()
        });

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls_a)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster a");
        let tcp_a = cluster_a
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport a");

        let cluster_b = handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-b.example", tls_b)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-a.example"),
                    addr: tcp_a.local_addr(),
                    control_plane_addr: None,
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster b");
        let tcp_b = cluster_b
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport b");

        tcp_a.add_peer(EngineId::new("engine-b.example"), tcp_b.local_addr());
        let mut cluster_a = cluster_a;
        cluster_a
            .add_peer(PeerSpec {
                engine_id: EngineId::new("engine-b.example"),
                addr: tcp_b.local_addr(),
                control_plane_addr: Some(control_plane_b),
                server_name: None,
            })
            .expect("add peer with control plane");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();

        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));
        tokio::time::sleep(Duration::from_millis(150)).await;

        let remote = cluster_a
            .spawn_remote::<EchoRequest>(
                EngineId::new("engine-b.example"),
                RemoteSpawnSpec {
                    path: actor_path.clone(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect("spawn remote actor");

        let status = cluster_a
            .remote_supervision_status(&actor_path)
            .await
            .expect("running supervision status");
        assert_eq!(status.supervisor, EngineId::new("engine-b.example"));
        assert_eq!(
            status.state,
            RemoteSupervisionState::Running { restart_count: 0 }
        );

        let response = remote
            .ask(EchoRequest { value: 21 })
            .await
            .expect("bounded remote spawn ask");
        assert_eq!(response, EchoResponse { doubled: 42 });

        let duplicate = cluster_a
            .spawn_remote::<EchoRequest>(
                EngineId::new("engine-b.example"),
                RemoteSpawnSpec {
                    path: actor_path.clone(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect_err("duplicate remote spawn should fail");
        assert_eq!(
            duplicate,
            RemoteSpawnError::RemotePathAlreadyRunning(actor_path.clone())
        );

        let old_host_addr = handle_b
            .registry
            .get_by_path(&actor_path)
            .expect("spawned actor registered on host")
            .addr;
        call_control_plane(
            &cluster_a.control_plane,
            control_plane_b,
            "engine-b.example",
            "actor.restart",
            json!({ "path": actor_path.as_str() }),
        )
        .await
        .expect("restart remote actor on host");

        let restarting_deadline = Instant::now() + Duration::from_secs(5);
        loop {
            match cluster_a.remote_supervision_status(&actor_path).await {
                Ok(status)
                    if status.supervisor == EngineId::new("engine-b.example")
                        && matches!(
                            status.state,
                            RemoteSupervisionState::Restarting { restart_count: 1 }
                        ) =>
                {
                    break;
                }
                Ok(_) | Err(_) if Instant::now() <= restarting_deadline => {
                    tokio::time::sleep(Duration::from_millis(25)).await;
                }
                Ok(status) => panic!("remote actor did not enter restarting state: {status:?}"),
                Err(err) => panic!("remote actor restart status query failed: {err:?}"),
            }
        }

        let restart_deadline = Instant::now() + Duration::from_secs(5);
        loop {
            if handle_b
                .registry
                .get_by_path(&actor_path)
                .map(|slot| {
                    slot.running.load(std::sync::atomic::Ordering::Relaxed)
                        && slot.addr != old_host_addr
                        && slot
                            .restart_count
                            .load(std::sync::atomic::Ordering::Relaxed)
                            > 0
                })
                .unwrap_or(false)
            {
                break;
            }
            if Instant::now() > restart_deadline {
                panic!("remote actor did not restart on host");
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }

        let restart_deadline = Instant::now() + Duration::from_secs(5);
        loop {
            if cluster_a.can_route(&actor_path) {
                break;
            }
            if Instant::now() > restart_deadline {
                panic!("remote actor route did not return after restart");
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }

        let restart_deadline = Instant::now() + Duration::from_secs(5);
        loop {
            match remote.ask(EchoRequest { value: 7 }).await {
                Ok(response) => {
                    assert_eq!(response, EchoResponse { doubled: 14 });
                    break;
                }
                Err(_) if Instant::now() <= restart_deadline => {
                    tokio::time::sleep(Duration::from_millis(25)).await;
                }
                Err(err) => panic!("remote actor did not recover after restart: {err:?}"),
            }
        }

        let restart_status = cluster_a
            .remote_supervision_status(&actor_path)
            .await
            .expect("restart supervision status");
        assert_eq!(restart_status.supervisor, EngineId::new("engine-b.example"));
        assert_eq!(
            restart_status.state,
            RemoteSupervisionState::Running { restart_count: 1 }
        );

        let unknown = cluster_a
            .spawn_remote::<EchoRequest>(
                EngineId::new("engine-c.example"),
                RemoteSpawnSpec {
                    path: ActorPath::parse("/user/unknown-target").unwrap(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect_err("unknown target should fail");
        assert_eq!(
            unknown,
            RemoteSpawnError::UnknownPeer(EngineId::new("engine-c.example"))
        );

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_spawn_remote_over_tcp_by_role_returns_typed_handle() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/remote-spawn-role-tcp").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);
        let control_plane_b = reserve_tcp_addr();

        let engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            control_plane_tcp_addr: Some(control_plane_b.to_string()),
            control_plane_tls: Some(tls_b.clone()),
            actor_spawn: Some(bounded_spawn_handler()),
            ..Default::default()
        });

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let cluster_b = handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-b.example", tls_b)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster b");
        let tcp_b = cluster_b
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport b");

        let cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls_a)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: tcp_b.local_addr(),
                    control_plane_addr: Some(control_plane_b),
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: vec![BoundedClusterRole {
                    name: "context-service".to_string(),
                    owner: EngineId::new("engine-b.example"),
                }],
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster a");
        let tcp_a = cluster_a
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport a");

        let mut cluster_b = cluster_b;
        cluster_b
            .add_peer(PeerSpec {
                engine_id: EngineId::new("engine-a.example"),
                addr: tcp_a.local_addr(),
                control_plane_addr: None,
                server_name: None,
            })
            .expect("add peer a to cluster b");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();

        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));
        tokio::time::sleep(Duration::from_millis(150)).await;

        let remote = cluster_a
            .spawn_remote_on_role::<EchoRequest>(
                "context-service",
                RemoteSpawnSpec {
                    path: actor_path.clone(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect("spawn remote actor by role over tcp");

        let response = remote
            .ask(EchoRequest { value: 11 })
            .await
            .expect("bounded remote spawn ask over tcp by role");
        assert_eq!(response, EchoResponse { doubled: 22 });

        let unknown_role = cluster_a
            .spawn_remote_on_role::<EchoRequest>(
                "missing-role",
                RemoteSpawnSpec {
                    path: ActorPath::parse("/user/missing-role-tcp").unwrap(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect_err("unknown role should fail");
        assert_eq!(
            unknown_role,
            RemoteSpawnError::UnknownRole("missing-role".to_string())
        );

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_spawn_remote_for_class_over_tcp_returns_typed_handle() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/remote-spawn-class-tcp").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);
        let control_plane_b = reserve_tcp_addr();

        let engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            control_plane_tcp_addr: Some(control_plane_b.to_string()),
            control_plane_tls: Some(tls_b.clone()),
            actor_spawn: Some(bounded_spawn_handler()),
            ..Default::default()
        });

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let cluster_b = handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-b.example", tls_b)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster b");
        let tcp_b = cluster_b
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport b");

        let cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls_a)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: tcp_b.local_addr(),
                    control_plane_addr: Some(control_plane_b),
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: vec![BoundedClusterRole {
                    name: "context-service".to_string(),
                    owner: EngineId::new("engine-b.example"),
                }],
                placement_policies: vec![BoundedPlacementPolicy {
                    actor_class: "SessionActor".to_string(),
                    role: "context-service".to_string(),
                }],
            })
            .await
            .expect("attach bounded cluster a");
        let tcp_a = cluster_a
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport a");

        let mut cluster_b = cluster_b;
        cluster_b
            .add_peer(PeerSpec {
                engine_id: EngineId::new("engine-a.example"),
                addr: tcp_a.local_addr(),
                control_plane_addr: None,
                server_name: None,
            })
            .expect("add peer a to cluster b");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();

        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));
        tokio::time::sleep(Duration::from_millis(150)).await;

        let remote = cluster_a
            .spawn_remote_for_class::<EchoRequest>(
                "SessionActor",
                RemoteSpawnSpec {
                    path: actor_path.clone(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect("spawn remote actor by placement class over tcp");

        let response = remote
            .ask(EchoRequest { value: 17 })
            .await
            .expect("bounded remote spawn ask over tcp by placement class");
        assert_eq!(response, EchoResponse { doubled: 34 });

        let status = cluster_a
            .remote_supervision_status(&actor_path)
            .await
            .expect("placement supervision status");
        assert_eq!(status.supervisor, EngineId::new("engine-b.example"));
        assert_eq!(
            status.state,
            RemoteSupervisionState::Running { restart_count: 0 }
        );

        call_control_plane(
            &cluster_a.control_plane,
            control_plane_b,
            "engine-b.example",
            "actor.restart",
            json!({ "path": actor_path.as_str() }),
        )
        .await
        .expect("restart placement actor on host");

        let restarting_deadline = Instant::now() + Duration::from_secs(5);
        loop {
            match cluster_a.remote_supervision_status(&actor_path).await {
                Ok(status)
                    if matches!(
                        status.state,
                        RemoteSupervisionState::Restarting { restart_count: 1 }
                    ) =>
                {
                    assert_eq!(status.supervisor, EngineId::new("engine-b.example"));
                    break;
                }
                Ok(_) | Err(_) if Instant::now() <= restarting_deadline => {
                    tokio::time::sleep(Duration::from_millis(25)).await;
                }
                Ok(status) => {
                    panic!("placement-driven actor did not enter restarting state: {status:?}")
                }
                Err(err) => panic!("placement restart status query failed: {err:?}"),
            }
        }

        wait_for_supervision_state(
            &cluster_a,
            &actor_path,
            RemoteSupervisionState::Running { restart_count: 1 },
        )
        .await;

        let unknown_class = cluster_a
            .spawn_remote_for_class::<EchoRequest>(
                "MissingActorClass",
                RemoteSpawnSpec {
                    path: ActorPath::parse("/user/missing-class-tcp").unwrap(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect_err("unknown placement class should fail");
        assert_eq!(
            unknown_class,
            RemoteSpawnError::UnknownPlacementClass("MissingActorClass".to_string())
        );

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_spawn_remote_over_tcp_by_role_without_reverse_declaration() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let engine_a_id = EngineId::new("engine-a-norev.example");
        let engine_b_id = EngineId::new("engine-b-norev.example");
        let actor_path = ActorPath::parse("/user/remote-spawn-role-tcp-slow").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config(engine_a_id.as_str(), &ca, &ca_key);
        let tls_b = make_tls_config(engine_b_id.as_str(), &ca, &ca_key);
        let control_plane_b = reserve_tcp_addr();

        let engine_a = Engine::with_config(EngineConfig {
            engine_id: engine_a_id.clone(),
            ..Default::default()
        });
        let engine_b = Engine::with_config(EngineConfig {
            engine_id: engine_b_id.clone(),
            control_plane_tcp_addr: Some(control_plane_b.to_string()),
            control_plane_tls: Some(tls_b.clone()),
            actor_spawn: Some(bounded_spawn_handler()),
            ..Default::default()
        });

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let cluster_b = handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config(engine_b_id.as_str(), tls_b)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster b");
        let tcp_b = cluster_b
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport b");

        let cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config(engine_a_id.as_str(), tls_a)),
                peers: vec![PeerSpec {
                    engine_id: engine_b_id.clone(),
                    addr: tcp_b.local_addr(),
                    control_plane_addr: Some(control_plane_b),
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: vec![BoundedClusterRole {
                    name: "context-service".to_string(),
                    owner: engine_b_id.clone(),
                }],
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster a");
        let tcp_a = cluster_a
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport a");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();

        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));

        // Caller knows the static owner already. Remote spawn should not require
        // the host engine to predeclare a reverse peer just to install the
        // caller's canonical path route.
        let _ = tcp_a;

        let remote = cluster_a
            .spawn_remote_on_role::<EchoRequest>(
                "context-service",
                RemoteSpawnSpec {
                    path: actor_path.clone(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect("spawn remote actor by role over tcp without reverse declaration");

        let response = remote
            .ask(EchoRequest { value: 13 })
            .await
            .expect("bounded remote spawn ask over tcp without reverse declaration");
        assert_eq!(response, EchoResponse { doubled: 26 });

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_remote_supervision_reports_actor_unavailable() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/supervision-actor-unavailable").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);
        let control_plane_b = reserve_tcp_addr();

        let engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let mut engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            control_plane_tcp_addr: Some(control_plane_b.to_string()),
            control_plane_tls: Some(tls_b.clone()),
            ..Default::default()
        });
        let ns = NamespacePolicy::default_for(&actor_path).unwrap();
        engine_b.add_user_actor(ChildSpec::new(
            "supervision-actor-unavailable",
            RestartPolicy::Temporary,
            ShutdownPolicy::Timeout(Duration::from_secs(1)),
            ns,
            move || Box::new(EchoActor),
        ));

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let cluster_b = handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-b.example", tls_b)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster b");
        let tcp_b = cluster_b
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport b");

        let mut cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls_a)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: tcp_b.local_addr(),
                    control_plane_addr: Some(control_plane_b),
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster a");
        cluster_a
            .declare_remote_actor(EngineId::new("engine-b.example"), actor_path.clone())
            .expect("declare remote actor");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();
        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));
        tokio::time::sleep(Duration::from_millis(150)).await;

        let remote = handle_a.remote_addr_for_path::<EchoRequest>(&actor_path);
        let response = remote
            .ask(EchoRequest { value: 9 })
            .await
            .expect("initial remote ask");
        assert_eq!(response, EchoResponse { doubled: 18 });
        wait_for_supervision_state(
            &cluster_a,
            &actor_path,
            RemoteSupervisionState::Running { restart_count: 0 },
        )
        .await;

        call_control_plane(
            &cluster_a.control_plane,
            control_plane_b,
            "engine-b.example",
            "actor.kill",
            json!({ "path": actor_path.as_str() }),
        )
        .await
        .expect("kill remote actor on host");

        wait_for_supervision_state(
            &cluster_a,
            &actor_path,
            RemoteSupervisionState::ActorUnavailable,
        )
        .await;

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_remote_supervision_reports_supervisor_unavailable() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/supervision-host-unavailable").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);
        let control_plane_b = reserve_tcp_addr();

        let engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let mut engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            control_plane_tcp_addr: Some(control_plane_b.to_string()),
            control_plane_tls: Some(tls_b.clone()),
            ..Default::default()
        });
        let ns = NamespacePolicy::default_for(&actor_path).unwrap();
        engine_b.add_user_actor(ChildSpec::new(
            "supervision-host-unavailable",
            RestartPolicy::Permanent,
            ShutdownPolicy::Timeout(Duration::from_secs(1)),
            ns,
            move || Box::new(EchoActor),
        ));

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let cluster_b = handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-b.example", tls_b)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster b");
        let tcp_b = cluster_b
            .transport()
            .as_tcp()
            .cloned()
            .expect("tcp transport b");

        let mut cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls_a)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: tcp_b.local_addr(),
                    control_plane_addr: Some(control_plane_b),
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster a");
        cluster_a
            .declare_remote_actor(EngineId::new("engine-b.example"), actor_path.clone())
            .expect("declare remote actor");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();
        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));
        tokio::time::sleep(Duration::from_millis(150)).await;

        let remote = handle_a.remote_addr_for_path::<EchoRequest>(&actor_path);
        let response = remote
            .ask(EchoRequest { value: 3 })
            .await
            .expect("initial remote ask");
        assert_eq!(response, EchoResponse { doubled: 6 });
        wait_for_supervision_state(
            &cluster_a,
            &actor_path,
            RemoteSupervisionState::Running { restart_count: 0 },
        )
        .await;

        shutdown_b_tx.send(()).ok();
        wait_for_supervision_state(
            &cluster_a,
            &actor_path,
            RemoteSupervisionState::SupervisorUnavailable,
        )
        .await;

        shutdown_a_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_remote_supervision_reports_transport_unavailable() {
        if !tcp_tests_available() {
            eprintln!("skipping tcp integration tests: network bind not permitted");
            return;
        }

        let actor_path = ActorPath::parse("/user/supervision-transport-unavailable").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);
        let control_plane_b = reserve_tcp_addr();
        let wrong_transport_addr = reserve_tcp_addr();

        let engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let mut engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            control_plane_tcp_addr: Some(control_plane_b.to_string()),
            control_plane_tls: Some(tls_b),
            ..Default::default()
        });
        let ns = NamespacePolicy::default_for(&actor_path).unwrap();
        engine_b.add_user_actor(ChildSpec::new(
            "supervision-transport-unavailable",
            RestartPolicy::Permanent,
            ShutdownPolicy::Timeout(Duration::from_secs(1)),
            ns,
            move || Box::new(EchoActor),
        ));

        let handle_a = engine_a.handle();
        let mut cluster_a = handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Tcp(tcp_config("engine-a.example", tls_a)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: wrong_transport_addr,
                    control_plane_addr: Some(control_plane_b),
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
            .expect("attach bounded cluster a");
        cluster_a
            .declare_remote_actor(EngineId::new("engine-b.example"), actor_path.clone())
            .expect("declare remote actor");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();
        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));
        tokio::time::sleep(Duration::from_millis(150)).await;

        wait_for_supervision_state(
            &cluster_a,
            &actor_path,
            RemoteSupervisionState::TransportUnavailable,
        )
        .await;

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }

    fn quic_config(engine_id: &str, tls: TlsConfig) -> QuicTransportConfig {
        QuicTransportConfig {
            engine_id: EngineId::new(engine_id),
            listen_addr: "127.0.0.1:0".parse().unwrap(),
            send_buffer_size: 1024,
            idle_timeout: Duration::from_secs(2),
            tls,
        }
    }

    #[tokio::test]
    async fn bounded_cluster_attach_reaches_remote_actor_over_quic() {
        if !quic_tests_available() {
            eprintln!("skipping quic bounded-cluster test: no rustls CryptoProvider configured");
            return;
        }

        let actor_path = ActorPath::parse("/user/cluster-echo-quic").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);

        let mut engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            ..Default::default()
        });

        let ns = NamespacePolicy::default_for(&actor_path).unwrap();
        engine_a.add_user_actor(ChildSpec::new(
            "cluster-echo-quic",
            RestartPolicy::Permanent,
            ShutdownPolicy::Timeout(Duration::from_secs(1)),
            ns,
            move || Box::new(EchoActor),
        ));

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let cluster_a = match handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Quic(quic_config(
                    "engine-a.example",
                    tls_a.clone(),
                )),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
        {
            Ok(cluster) => cluster,
            Err(ClusterError::Transport(err)) => {
                eprintln!("skipping quic bounded-cluster test: {err:?}");
                return;
            }
            Err(err) => panic!("attach bounded cluster a: {err:?}"),
        };
        let quic_a = cluster_a
            .transport()
            .as_quic()
            .cloned()
            .expect("quic transport a");

        let cluster_b = match handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Quic(quic_config("engine-b.example", tls_b)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-a.example"),
                    addr: quic_a.local_addr(),
                    control_plane_addr: None,
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
        {
            Ok(cluster) => cluster,
            Err(ClusterError::Transport(err)) => {
                eprintln!("skipping quic bounded-cluster test: {err:?}");
                return;
            }
            Err(err) => panic!("attach bounded cluster b: {err:?}"),
        };
        let quic_b = cluster_b
            .transport()
            .as_quic()
            .cloned()
            .expect("quic transport b");

        quic_a.add_peer(EngineId::new("engine-b.example"), quic_b.local_addr());
        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();

        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));

        let canonical = AddrHash::new(&actor_path, 0);
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while !cluster_b.can_route(&actor_path) {
            if std::time::Instant::now() > deadline {
                panic!("bounded quic cluster route did not propagate");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        assert!(quic_b.can_route(canonical));

        let remote = handle_b.remote_addr_for_path::<EchoRequest>(&actor_path);
        let ask_deadline = Instant::now() + Duration::from_secs(2);
        loop {
            match remote.ask(EchoRequest { value: 15 }).await {
                Ok(response) => {
                    assert_eq!(response, EchoResponse { doubled: 30 });
                    break;
                }
                Err(palladium_actor::AskError::Send(
                    palladium_actor::SendError::ConnectionFailed,
                )) if Instant::now() <= ask_deadline => {
                    tokio::time::sleep(Duration::from_millis(20)).await;
                }
                Err(err) => panic!("bounded cluster ask over quic: {err:?}"),
            }
        }

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }

    #[tokio::test]
    async fn bounded_cluster_declared_remote_actor_installs_canonical_quic_route() {
        if !quic_tests_available() {
            eprintln!("skipping quic bounded-cluster test: no rustls CryptoProvider configured");
            return;
        }

        let (ca, ca_key) = make_ca();
        let tls = make_tls_config("engine-a.example", &ca, &ca_key);
        let engine = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let handle = engine.handle();

        let mut cluster = match handle
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Quic(quic_config("engine-a.example", tls)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: "127.0.0.1:4100".parse().unwrap(),
                    control_plane_addr: None,
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
        {
            Ok(cluster) => cluster,
            Err(ClusterError::Transport(err)) => {
                eprintln!("skipping quic bounded-cluster test: {err:?}");
                return;
            }
            Err(err) => panic!("attach bounded cluster: {err:?}"),
        };

        let path = ActorPath::parse("/user/declared-remote-quic").unwrap();
        cluster
            .declare_remote_actor(EngineId::new("engine-b.example"), path.clone())
            .expect("declare remote actor");

        assert!(cluster.can_route(&path));
        assert!(cluster
            .transport()
            .as_quic()
            .expect("quic transport")
            .can_route(AddrHash::new(&path, 0)));
    }

    #[tokio::test]
    async fn bounded_cluster_spawn_remote_over_quic_returns_typed_handle() {
        if !quic_tests_available() {
            eprintln!("skipping quic bounded-cluster test: no rustls CryptoProvider configured");
            return;
        }

        let actor_path = ActorPath::parse("/user/remote-spawn-quic").unwrap();
        let (ca, ca_key) = make_ca();
        let tls_a = make_tls_config("engine-a.example", &ca, &ca_key);
        let tls_b = make_tls_config("engine-b.example", &ca, &ca_key);
        let control_plane_b = reserve_udp_addr();

        let engine_a = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-a.example"),
            ..Default::default()
        });
        let engine_b = Engine::with_config(EngineConfig {
            engine_id: EngineId::new("engine-b.example"),
            control_plane_quic_addr: Some(control_plane_b.to_string()),
            control_plane_tls: Some(tls_b.clone()),
            actor_spawn: Some(bounded_spawn_handler()),
            ..Default::default()
        });

        let handle_a = engine_a.handle();
        let handle_b = engine_b.handle();
        handle_a
            .type_registry()
            .register_remote_ask::<EchoRequest>();
        handle_b
            .type_registry()
            .register_remote_ask::<EchoRequest>();

        let cluster_b = match handle_b
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Quic(quic_config("engine-b.example", tls_b)),
                peers: Vec::new(),
                declared_actors: Vec::new(),
                roles: Vec::new(),
                placement_policies: Vec::new(),
            })
            .await
        {
            Ok(cluster) => cluster,
            Err(ClusterError::Transport(err)) => {
                eprintln!("skipping quic bounded-cluster test: {err:?}");
                return;
            }
            Err(err) => panic!("attach bounded cluster b: {err:?}"),
        };
        let quic_b = cluster_b
            .transport()
            .as_quic()
            .cloned()
            .expect("quic transport b");

        let cluster_a = match handle_a
            .attach_bounded_cluster(BoundedClusterConfig {
                transport: BoundedTransportConfig::Quic(quic_config("engine-a.example", tls_a)),
                peers: vec![PeerSpec {
                    engine_id: EngineId::new("engine-b.example"),
                    addr: quic_b.local_addr(),
                    control_plane_addr: Some(control_plane_b),
                    server_name: None,
                }],
                declared_actors: Vec::new(),
                roles: vec![BoundedClusterRole {
                    name: "context-service".to_string(),
                    owner: EngineId::new("engine-b.example"),
                }],
                placement_policies: Vec::new(),
            })
            .await
        {
            Ok(cluster) => cluster,
            Err(ClusterError::Transport(err)) => {
                eprintln!("skipping quic bounded-cluster test: {err:?}");
                return;
            }
            Err(err) => panic!("attach bounded cluster a: {err:?}"),
        };
        let quic_a = cluster_a
            .transport()
            .as_quic()
            .cloned()
            .expect("quic transport a");

        let mut cluster_b = cluster_b;
        cluster_b
            .add_peer(PeerSpec {
                engine_id: EngineId::new("engine-a.example"),
                addr: quic_a.local_addr(),
                control_plane_addr: None,
                server_name: None,
            })
            .expect("add quic peer a to cluster b");

        let (shutdown_a_tx, shutdown_a_rx) = tokio::sync::oneshot::channel::<()>();
        let (shutdown_b_tx, shutdown_b_rx) = tokio::sync::oneshot::channel::<()>();

        std::thread::spawn(move || engine_a.run(shutdown_a_rx));
        std::thread::spawn(move || engine_b.run(shutdown_b_rx));
        tokio::time::sleep(Duration::from_millis(150)).await;

        let remote = cluster_a
            .spawn_remote_on_role::<EchoRequest>(
                "context-service",
                RemoteSpawnSpec {
                    path: actor_path.clone(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect("spawn remote actor over quic by role");

        let response = remote
            .ask(EchoRequest { value: 9 })
            .await
            .expect("bounded remote spawn ask over quic by role");
        assert_eq!(response, EchoResponse { doubled: 18 });

        let unknown_role = cluster_a
            .spawn_remote_on_role::<EchoRequest>(
                "missing-role",
                RemoteSpawnSpec {
                    path: ActorPath::parse("/user/missing-role").unwrap(),
                    type_name: "bounded-echo".to_string(),
                    config: None,
                },
            )
            .await
            .expect_err("unknown role should fail");
        assert_eq!(
            unknown_role,
            RemoteSpawnError::UnknownRole("missing-role".to_string())
        );

        shutdown_a_tx.send(()).ok();
        shutdown_b_tx.send(()).ok();
    }
}