1use std::collections::HashMap;
338use std::io;
339use std::sync::Arc;
340use std::time::Duration;
341
342use async_trait::async_trait;
343use axum::Json;
344use axum::Router;
345use axum::extract::Path as AxumPath;
346use axum::extract::State;
347use axum::http::StatusCode;
348use axum::response::IntoResponse;
349use axum::routing::get;
350use axum::routing::post;
351use hyperactor::Actor;
352use hyperactor::ActorHandle;
353use hyperactor::ActorRef;
354use hyperactor::Context;
355use hyperactor::Endpoint as _;
356use hyperactor::HandleClient;
357use hyperactor::Handler;
358use hyperactor::Instance;
359use hyperactor::OncePortRef;
360use hyperactor::ProcAddr;
361use hyperactor::RefClient;
362use hyperactor::channel::try_tls_acceptor;
363use hyperactor::introspect::IntrospectMessage;
364use hyperactor::introspect::IntrospectResult;
365use hyperactor::introspect::IntrospectView;
366use hyperactor::mailbox::open_once_port;
367use serde::Deserialize;
368use serde::Serialize;
369use serde_json::Value;
370use tokio::net::TcpListener;
371use tokio_rustls::TlsAcceptor;
372use typeuri::Named;
373
374use crate::config_dump::ConfigDump;
375use crate::config_dump::ConfigDumpResult;
376use crate::host::SERVICE_PROC_NAME;
377use crate::host_mesh::host_agent::HOST_MESH_AGENT_ACTOR_NAME;
378use crate::host_mesh::host_agent::HostAgent;
379use crate::introspect::NodePayload;
380use crate::introspect::NodeProperties;
381use crate::introspect::dto::NodePayloadDto;
382use crate::introspect::to_node_payload;
383use crate::proc_agent::PROC_AGENT_ACTOR_NAME;
384use crate::proc_agent::ProcAgent;
385use crate::pyspy::PySpyDump;
386use crate::pyspy::PySpyOpts;
387use crate::pyspy::PySpyProfile;
388use crate::pyspy::PySpyProfileOpts;
389use crate::pyspy::PySpyProfileResult;
390use crate::pyspy::PySpyResult;
391use crate::pyspy::ValidatedProfileRequest;
392
393async fn query_introspect(
396 cx: &hyperactor::Context<'_, MeshAdminAgent>,
397 actor_id: &hyperactor::ActorAddr,
398 view: hyperactor::introspect::IntrospectView,
399 timeout: Duration,
400 err_ctx: &str,
401) -> Result<IntrospectResult, anyhow::Error> {
402 let introspect_port = actor_id.introspect_port();
403 let (reply_handle, reply_rx) = open_once_port::<IntrospectResult>(cx);
404 let mut reply_ref = reply_handle.bind();
405 reply_ref.return_undeliverable(false);
406 introspect_port.post(
407 cx,
408 IntrospectMessage::Query {
409 view,
410 reply: reply_ref,
411 },
412 );
413 tokio::time::timeout(timeout, reply_rx.recv())
414 .await
415 .map_err(|_| anyhow::anyhow!("timed out {}", err_ctx))?
416 .map_err(|e| anyhow::anyhow!("failed to receive {}: {}", err_ctx, e))
417}
418
419async fn query_child_introspect(
421 cx: &hyperactor::Context<'_, MeshAdminAgent>,
422 actor_id: &hyperactor::ActorAddr,
423 child_ref: hyperactor::Addr,
424 timeout: Duration,
425 err_ctx: &str,
426) -> Result<IntrospectResult, anyhow::Error> {
427 let introspect_port = actor_id.introspect_port();
428 let (reply_handle, reply_rx) = open_once_port::<IntrospectResult>(cx);
429 let mut reply_ref = reply_handle.bind();
430 reply_ref.return_undeliverable(false);
431 introspect_port.post(
432 cx,
433 IntrospectMessage::QueryChild {
434 child_ref,
435 reply: reply_ref,
436 },
437 );
438 tokio::time::timeout(timeout, reply_rx.recv())
439 .await
440 .map_err(|_| anyhow::anyhow!("timed out {}", err_ctx))?
441 .map_err(|e| anyhow::anyhow!("failed to receive {}: {}", err_ctx, e))
442}
443
444pub const MESH_ADMIN_ACTOR_NAME: &str = "mesh_admin";
446
447pub const MESH_ADMIN_BRIDGE_NAME: &str = "mesh_admin_bridge";
464
465#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
468pub struct ApiError {
469 pub code: String,
471 pub message: String,
473 #[serde(skip_serializing_if = "Option::is_none")]
477 pub details: Option<Value>,
478}
479
480#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
482pub struct ApiErrorEnvelope {
483 pub error: ApiError,
484}
485
486impl ApiError {
487 pub fn not_found(message: impl Into<String>, details: Option<Value>) -> Self {
489 Self {
490 code: "not_found".to_string(),
491 message: message.into(),
492 details,
493 }
494 }
495
496 pub fn bad_request(message: impl Into<String>, details: Option<Value>) -> Self {
498 Self {
499 code: "bad_request".to_string(),
500 message: message.into(),
501 details,
502 }
503 }
504}
505
506impl IntoResponse for ApiError {
507 fn into_response(self) -> axum::response::Response {
508 let status = match self.code.as_str() {
509 "not_found" => StatusCode::NOT_FOUND,
510 "bad_request" => StatusCode::BAD_REQUEST,
511 "gateway_timeout" => StatusCode::GATEWAY_TIMEOUT,
512 "service_unavailable" => StatusCode::SERVICE_UNAVAILABLE,
513 _ => StatusCode::INTERNAL_SERVER_ERROR,
514 };
515 let envelope = ApiErrorEnvelope { error: self };
516 (status, Json(envelope)).into_response()
517 }
518}
519
520#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Named)]
525pub struct MeshAdminAddrResponse {
526 pub addr: Option<String>,
527}
528wirevalue::register_type!(MeshAdminAddrResponse);
529
530#[derive(
536 Debug,
537 Clone,
538 PartialEq,
539 Serialize,
540 Deserialize,
541 Handler,
542 HandleClient,
543 RefClient,
544 Named
545)]
546pub enum MeshAdminMessage {
547 GetAdminAddr {
552 #[reply]
553 reply: OncePortRef<MeshAdminAddrResponse>,
554 },
555}
556wirevalue::register_type!(MeshAdminMessage);
557
558#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Named)]
561pub struct ResolveReferenceResponse(pub Result<NodePayload, String>);
562wirevalue::register_type!(ResolveReferenceResponse);
563
564#[derive(
581 Debug,
582 Clone,
583 PartialEq,
584 Serialize,
585 Deserialize,
586 Handler,
587 HandleClient,
588 RefClient,
589 Named
590)]
591pub enum ResolveReferenceMessage {
592 Resolve {
597 reference_string: String,
600 #[reply]
602 reply: OncePortRef<ResolveReferenceResponse>,
603 },
604}
605wirevalue::register_type!(ResolveReferenceMessage);
606
607#[hyperactor::export(handlers = [MeshAdminMessage, ResolveReferenceMessage])]
620pub struct MeshAdminAgent {
621 hosts: HashMap<String, ActorRef<HostAgent>>,
624
625 host_agents_by_actor_id: HashMap<hyperactor::ActorAddr, String>,
635
636 root_client_actor_id: Option<hyperactor::ActorAddr>,
641
642 self_actor_id: Option<hyperactor::ActorAddr>,
647
648 admin_addr_override: Option<std::net::SocketAddr>,
665
666 admin_addr: Option<std::net::SocketAddr>,
669
670 admin_host: Option<String>,
673
674 telemetry_url: Option<String>,
678
679 started_at: String,
681
682 started_by: String,
684}
685
686impl MeshAdminAgent {
687 pub fn new(
705 hosts: Vec<(String, ActorRef<HostAgent>)>,
706 root_client_actor_id: Option<hyperactor::ActorAddr>,
707 admin_addr: Option<std::net::SocketAddr>,
708 telemetry_url: Option<String>,
709 ) -> Self {
710 let host_agents_by_actor_id: HashMap<hyperactor::ActorAddr, String> = hosts
711 .iter()
712 .map(|(addr, agent_ref)| (agent_ref.actor_addr().clone(), addr.clone()))
713 .collect();
714
715 let started_at = chrono::Utc::now().to_rfc3339();
717 let started_by = std::env::var("USER")
718 .or_else(|_| std::env::var("USERNAME"))
719 .unwrap_or_else(|_| "unknown".to_string());
720
721 Self {
722 hosts: hosts.into_iter().collect(),
723 host_agents_by_actor_id,
724 root_client_actor_id,
725 self_actor_id: None,
726 admin_addr_override: admin_addr,
727 admin_addr: None,
728 admin_host: None,
729 telemetry_url,
730 started_at,
731 started_by,
732 }
733 }
734}
735
736impl std::fmt::Debug for MeshAdminAgent {
737 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
738 f.debug_struct("MeshAdminAgent")
739 .field("hosts", &self.hosts.keys().collect::<Vec<_>>())
740 .field("host_agents", &self.host_agents_by_actor_id.len())
741 .field("root_client_actor_id", &self.root_client_actor_id)
742 .field("self_actor_id", &self.self_actor_id)
743 .field("admin_addr", &self.admin_addr)
744 .field("admin_host", &self.admin_host)
745 .field("started_at", &self.started_at)
746 .field("started_by", &self.started_by)
747 .finish()
748 }
749}
750
751#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
758pub struct AdminInfo {
759 pub actor_id: String,
761 pub proc_id: String,
763 pub host: String,
765 pub url: String,
767}
768
769impl AdminInfo {
770 pub fn new(actor_id: String, proc_id: String, url: String) -> anyhow::Result<Self> {
777 let parsed = url::Url::parse(&url)
778 .map_err(|e| anyhow::anyhow!("invalid admin URL '{}': {}", url, e))?;
779 let host = parsed
780 .host_str()
781 .ok_or_else(|| anyhow::anyhow!("admin URL '{}' has no host", url))?
782 .to_string();
783 Ok(Self {
784 actor_id,
785 proc_id,
786 host,
787 url,
788 })
789 }
790}
791
792struct BridgeState {
801 admin_ref: ActorRef<MeshAdminAgent>,
804 bridge_cx: Instance<()>,
812 resolve_semaphore: tokio::sync::Semaphore,
816 _bridge_handle: ActorHandle<()>,
818 telemetry_url: Option<String>,
823 http_client: reqwest::Client,
826 admin_info: AdminInfo,
828}
829
830fn build_http_client() -> reqwest::Client {
837 use std::io::Read;
838
839 if let Some(bundle) = hyperactor::channel::try_tls_pem_bundle() {
840 let mut ca_bytes = Vec::new();
841 if let Ok(mut reader) = bundle.ca.reader()
842 && reader.read_to_end(&mut ca_bytes).is_ok()
843 {
844 let (builder, ca_installed) = crate::mesh_admin_client::add_tls(
845 reqwest::Client::builder(),
846 &ca_bytes,
847 None,
848 None,
849 );
850 if ca_installed {
851 if let Ok(client) = builder.build() {
852 return client;
853 }
854 tracing::warn!(
855 "mesh admin: failed to build reqwest client with root CA; \
856 falling back to default trust store"
857 );
858 }
859 }
860 }
861 reqwest::Client::new()
862}
863
864struct TlsListener {
871 tcp: TcpListener,
872 acceptor: TlsAcceptor,
873}
874
875impl axum::serve::Listener for TlsListener {
876 type Io = tokio_rustls::server::TlsStream<tokio::net::TcpStream>;
877 type Addr = std::net::SocketAddr;
878
879 async fn accept(&mut self) -> (Self::Io, Self::Addr) {
880 loop {
881 let (stream, addr) = match self.tcp.accept().await {
882 Ok(conn) => conn,
883 Err(e) => {
884 tracing::warn!("TCP accept error: {}", e);
885 continue;
886 }
887 };
888
889 match self.acceptor.accept(stream).await {
890 Ok(tls_stream) => return (tls_stream, addr),
891 Err(e) => {
892 tracing::warn!("TLS handshake failed from {}: {}", addr, e);
893 continue;
894 }
895 }
896 }
897 }
898
899 fn local_addr(&self) -> io::Result<Self::Addr> {
900 self.tcp.local_addr()
901 }
902}
903
904#[async_trait]
905impl Actor for MeshAdminAgent {
906 async fn init(&mut self, this: &Instance<Self>) -> Result<(), anyhow::Error> {
927 this.bind::<Self>();
931 this.set_system();
932 self.self_actor_id = Some(this.self_addr().clone());
933
934 let bind_addr = match self.admin_addr_override {
935 Some(addr) => addr,
936 None => hyperactor_config::global::get_cloned(crate::config::MESH_ADMIN_ADDR)
937 .parse_socket_addr()
938 .map_err(|e| anyhow::anyhow!("invalid MESH_ADMIN_ADDR config: {}", e))?,
939 };
940 let listener = TcpListener::bind(bind_addr).await?;
941 let bound_addr = listener.local_addr()?;
942 self.admin_addr = Some(bound_addr);
943
944 let enforce_mtls = cfg!(fbcode_build);
948 let tls_acceptor = try_tls_acceptor(enforce_mtls);
949
950 if enforce_mtls && tls_acceptor.is_none() {
951 return Err(anyhow::anyhow!(
952 "mesh admin requires mTLS but no TLS certificates found; \
953 set HYPERACTOR_TLS_CERT/KEY/CA or ensure Meta cert paths exist \
954 (/var/facebook/x509_identities/server.pem, /var/facebook/rootcanal/ca.pem)"
955 ));
956 }
957
958 let scheme = if tls_acceptor.is_some() {
959 "https"
960 } else {
961 "http"
962 };
963
964 let host = if !bound_addr.ip().is_unspecified() {
974 let ip = bound_addr.ip();
975 if ip.is_loopback() {
976 "localhost".to_string()
977 } else if let std::net::IpAddr::V6(v6) = ip {
978 format!("[{}]", v6)
979 } else {
980 ip.to_string()
981 }
982 } else {
983 advertised_host::from_cert_sans()
984 };
985 self.admin_host = Some(format!("{}://{}:{}", scheme, host, bound_addr.port()));
986
987 let (bridge_cx, bridge_handle) = this
991 .proc()
992 .introspectable_instance(MESH_ADMIN_BRIDGE_NAME)?;
993 bridge_cx.set_system();
994 let admin_url = self
995 .admin_host
996 .clone()
997 .unwrap_or_else(|| "unknown".to_string());
998 let bridge_state = Arc::new(BridgeState {
999 admin_ref: ActorRef::attest(this.self_addr().clone()),
1000 bridge_cx,
1001 resolve_semaphore: tokio::sync::Semaphore::new(hyperactor_config::global::get(
1002 crate::config::MESH_ADMIN_MAX_CONCURRENT_RESOLVES,
1003 )),
1004 _bridge_handle: bridge_handle,
1005 telemetry_url: self.telemetry_url.clone(),
1006 http_client: build_http_client(),
1007 admin_info: AdminInfo::new(
1008 this.self_addr().to_string(),
1009 this.self_addr().proc_addr().to_string(),
1010 admin_url,
1011 )?,
1012 });
1013 let router = create_mesh_admin_router(bridge_state);
1014
1015 if let Some(acceptor) = tls_acceptor {
1016 let tls_listener = TlsListener {
1017 tcp: listener,
1018 acceptor,
1019 };
1020 tokio::spawn(async move {
1021 if let Err(e) = axum::serve(tls_listener, router).await {
1022 tracing::error!("mesh admin server (mTLS) error: {}", e);
1023 }
1024 });
1025 } else {
1026 tokio::spawn(async move {
1028 if let Err(e) = axum::serve(listener, router).await {
1029 tracing::error!("mesh admin server error: {}", e);
1030 }
1031 });
1032 }
1033
1034 tracing::info!(
1035 "mesh admin server listening on {}",
1036 self.admin_host.as_deref().unwrap_or("unknown")
1037 );
1038 Ok(())
1039 }
1040
1041 async fn handle_undeliverable_message(
1055 &mut self,
1056 _cx: &Instance<Self>,
1057 _reason: hyperactor::mailbox::UndeliverableReason,
1058 undeliverable: hyperactor::mailbox::Undeliverable<hyperactor::mailbox::MessageEnvelope>,
1059 ) -> Result<(), anyhow::Error> {
1060 match undeliverable {
1061 hyperactor::mailbox::Undeliverable::Returned(envelope) => {
1062 tracing::debug!(
1063 "admin agent: undeliverable message to {} (port not bound?), ignoring",
1064 envelope.dest(),
1065 );
1066 }
1067 hyperactor::mailbox::Undeliverable::Report(report) => {
1068 tracing::debug!(
1069 "admin agent: undeliverable message report to {} ({}), ignoring",
1070 report.dest,
1071 report.error_msg().unwrap_or_default(),
1072 );
1073 }
1074 }
1075 Ok(())
1076 }
1077
1078 async fn handle_invalid_reference(
1079 &mut self,
1080 _cx: &Instance<Self>,
1081 invalid: hyperactor::mailbox::InvalidReference,
1082 undeliverable: hyperactor::mailbox::Undeliverable<hyperactor::mailbox::MessageEnvelope>,
1083 ) -> Result<(), anyhow::Error> {
1084 tracing::debug!(
1085 %invalid,
1086 "admin agent: invalid reference from introspection probe, ignoring",
1087 );
1088 match undeliverable {
1089 hyperactor::mailbox::Undeliverable::Returned(envelope) => {
1090 tracing::debug!(
1091 "admin agent: undeliverable message to {} (invalid reference), ignoring",
1092 envelope.dest(),
1093 );
1094 }
1095 hyperactor::mailbox::Undeliverable::Report(report) => {
1096 tracing::debug!(
1097 "admin agent: undeliverable message report to {} ({}), ignoring",
1098 report.dest,
1099 report.error_msg().unwrap_or_default(),
1100 );
1101 }
1102 }
1103 Ok(())
1104 }
1105}
1106
1107#[async_trait]
1110impl Handler<MeshAdminMessage> for MeshAdminAgent {
1111 async fn handle(
1118 &mut self,
1119 cx: &Context<Self>,
1120 msg: MeshAdminMessage,
1121 ) -> Result<(), anyhow::Error> {
1122 match msg {
1123 MeshAdminMessage::GetAdminAddr { reply } => {
1124 let resp = MeshAdminAddrResponse {
1125 addr: self.admin_host.clone(),
1126 };
1127 reply.post(cx, resp);
1128 }
1129 }
1130 Ok(())
1131 }
1132}
1133
1134#[async_trait]
1137impl Handler<ResolveReferenceMessage> for MeshAdminAgent {
1138 async fn handle(
1146 &mut self,
1147 cx: &Context<Self>,
1148 msg: ResolveReferenceMessage,
1149 ) -> Result<(), anyhow::Error> {
1150 match msg {
1151 ResolveReferenceMessage::Resolve {
1152 reference_string,
1153 reply,
1154 } => {
1155 let response = ResolveReferenceResponse(
1156 self.resolve_reference(cx, &reference_string)
1157 .await
1158 .map_err(|e| format!("{:#}", e)),
1159 );
1160 reply.post(cx, response);
1161 }
1162 }
1163 Ok(())
1164 }
1165}
1166
1167impl MeshAdminAgent {
1168 async fn resolve_reference(
1187 &self,
1188 cx: &Context<'_, Self>,
1189 reference_string: &str,
1190 ) -> Result<NodePayload, anyhow::Error> {
1191 let node_ref: crate::introspect::NodeRef = reference_string
1192 .parse()
1193 .map_err(|e| anyhow::anyhow!("invalid reference '{}': {}", reference_string, e))?;
1194
1195 match &node_ref {
1196 crate::introspect::NodeRef::Root => Ok(self.build_root_payload()),
1197 crate::introspect::NodeRef::Host(actor_id) => {
1198 self.resolve_host_node(cx, actor_id).await
1199 }
1200 crate::introspect::NodeRef::Proc(proc_id) => {
1201 match self.resolve_proc_node(cx, proc_id).await {
1202 Ok(payload) => Ok(payload),
1203 Err(_) if self.standalone_proc_anchor(proc_id).is_some() => {
1204 self.resolve_standalone_proc_node(cx, proc_id).await
1205 }
1206 Err(e) => Err(e),
1207 }
1208 }
1209 crate::introspect::NodeRef::Actor(actor_id) => {
1210 self.resolve_actor_node(cx, actor_id).await
1211 }
1212 }
1213 }
1214
1215 fn standalone_proc_actors(&self) -> impl Iterator<Item = &hyperactor::ActorAddr> {
1223 std::iter::empty()
1224 }
1225
1226 fn standalone_proc_anchor(&self, proc_id: &ProcAddr) -> Option<&hyperactor::ActorAddr> {
1229 self.standalone_proc_actors()
1230 .find(|actor_id| actor_id.proc_addr() == *proc_id)
1231 }
1232
1233 fn is_standalone_proc_actor(&self, actor_id: &hyperactor::ActorAddr) -> bool {
1235 self.standalone_proc_actors()
1236 .any(|a| a.proc_addr() == actor_id.proc_addr())
1237 }
1238
1239 fn build_root_payload(&self) -> NodePayload {
1245 use crate::introspect::NodeRef;
1246
1247 let children: Vec<NodeRef> = self
1248 .hosts
1249 .values()
1250 .map(|agent| NodeRef::Host(agent.actor_addr().clone()))
1251 .collect();
1252 let system_children: Vec<NodeRef> = Vec::new(); let mut attrs = hyperactor_config::Attrs::new();
1254 attrs.set(crate::introspect::NODE_TYPE, "root".to_string());
1255 attrs.set(crate::introspect::NUM_HOSTS, self.hosts.len());
1256 if let Ok(t) = humantime::parse_rfc3339(&self.started_at) {
1257 attrs.set(crate::introspect::STARTED_AT, t);
1258 }
1259 attrs.set(crate::introspect::STARTED_BY, self.started_by.clone());
1260 attrs.set(crate::introspect::SYSTEM_CHILDREN, system_children.clone());
1261 let attrs_json = serde_json::to_string(&attrs).unwrap_or_else(|_| "{}".to_string());
1262 NodePayload {
1263 identity: NodeRef::Root,
1264 properties: crate::introspect::derive_properties(&attrs_json),
1265 children,
1266 parent: None,
1267 as_of: std::time::SystemTime::now(),
1268 }
1269 }
1270
1271 async fn resolve_host_node(
1280 &self,
1281 cx: &Context<'_, Self>,
1282 actor_id: &hyperactor::ActorAddr,
1283 ) -> Result<NodePayload, anyhow::Error> {
1284 let result = query_introspect(
1285 cx,
1286 actor_id,
1287 hyperactor::introspect::IntrospectView::Entity,
1288 hyperactor_config::global::get(crate::config::MESH_ADMIN_SINGLE_HOST_TIMEOUT),
1289 "querying host agent",
1290 )
1291 .await?;
1292 Ok(crate::introspect::to_node_payload_with(
1293 result,
1294 crate::introspect::NodeRef::Host(actor_id.clone()),
1295 Some(crate::introspect::NodeRef::Root),
1296 ))
1297 }
1298
1299 async fn resolve_proc_node(
1310 &self,
1311 cx: &Context<'_, Self>,
1312 proc_id: &ProcAddr,
1313 ) -> Result<NodePayload, anyhow::Error> {
1314 let host_addr = proc_id.addr().to_string();
1315
1316 let agent = self
1317 .hosts
1318 .get(&host_addr)
1319 .ok_or_else(|| anyhow::anyhow!("host not found: {}", host_addr))?;
1320
1321 let result = query_child_introspect(
1323 cx,
1324 agent.actor_addr(),
1325 hyperactor::Addr::Proc(proc_id.clone()),
1326 hyperactor_config::global::get(crate::config::MESH_ADMIN_QUERY_CHILD_TIMEOUT),
1327 "querying proc details",
1328 )
1329 .await?;
1330
1331 let payload = crate::introspect::to_node_payload_with(
1335 result,
1336 crate::introspect::NodeRef::Proc(proc_id.clone()),
1337 Some(crate::introspect::NodeRef::Host(agent.actor_addr().clone())),
1338 );
1339 if !matches!(payload.properties, NodeProperties::Error { .. }) {
1340 return Ok(payload);
1341 }
1342
1343 let mesh_agent_id = proc_id.actor_addr(PROC_AGENT_ACTOR_NAME);
1345 let result = query_child_introspect(
1346 cx,
1347 &mesh_agent_id,
1348 hyperactor::Addr::Proc(proc_id.clone()),
1349 hyperactor_config::global::get(crate::config::MESH_ADMIN_RESOLVE_ACTOR_TIMEOUT),
1350 "querying proc mesh agent",
1351 )
1352 .await?;
1353
1354 Ok(crate::introspect::to_node_payload_with(
1355 result,
1356 crate::introspect::NodeRef::Proc(proc_id.clone()),
1357 Some(crate::introspect::NodeRef::Host(agent.actor_addr().clone())),
1358 ))
1359 }
1360
1361 async fn resolve_standalone_proc_node(
1375 &self,
1376 cx: &Context<'_, Self>,
1377 proc_id: &ProcAddr,
1378 ) -> Result<NodePayload, anyhow::Error> {
1379 let actor_id = self
1380 .standalone_proc_anchor(proc_id)
1381 .ok_or_else(|| anyhow::anyhow!("no anchor actor for standalone proc {}", proc_id))?;
1382
1383 use crate::introspect::NodeRef;
1384
1385 let (children, system_children) = if self.self_actor_id.as_ref() == Some(actor_id) {
1386 let self_ref = NodeRef::Actor(actor_id.clone());
1387 (vec![self_ref.clone()], vec![self_ref])
1388 } else {
1389 let actor_result = query_introspect(
1390 cx,
1391 actor_id,
1392 hyperactor::introspect::IntrospectView::Actor,
1393 hyperactor_config::global::get(crate::config::MESH_ADMIN_SINGLE_HOST_TIMEOUT),
1394 &format!("querying anchor actor on {}", proc_id),
1395 )
1396 .await?;
1397 let actor_payload = to_node_payload(actor_result);
1398 let anchor_ref = NodeRef::Actor(actor_id.clone());
1399 let anchor_is_system = matches!(
1400 &actor_payload.properties,
1401 NodeProperties::Actor {
1402 is_system: true,
1403 ..
1404 }
1405 );
1406
1407 let mut children = vec![anchor_ref.clone()];
1408 let mut system_children = Vec::new();
1409 if anchor_is_system {
1410 system_children.push(anchor_ref);
1411 }
1412
1413 for child_ref in actor_payload.children {
1414 let child_actor_id = match &child_ref {
1415 NodeRef::Actor(id) => Some(id),
1416 _ => None,
1417 };
1418 if let Some(child_actor_id) = child_actor_id {
1419 let child_is_system = if let Ok(r) = query_introspect(
1420 cx,
1421 child_actor_id,
1422 hyperactor::introspect::IntrospectView::Actor,
1423 hyperactor_config::global::get(
1424 crate::config::MESH_ADMIN_RESOLVE_ACTOR_TIMEOUT,
1425 ),
1426 "querying child actor is_system",
1427 )
1428 .await
1429 {
1430 let p = to_node_payload(r);
1431 matches!(
1432 &p.properties,
1433 NodeProperties::Actor {
1434 is_system: true,
1435 ..
1436 }
1437 )
1438 } else {
1439 false
1440 };
1441 if child_is_system {
1442 system_children.push(child_ref.clone());
1443 }
1444 }
1445 children.push(child_ref);
1446 }
1447 (children, system_children)
1448 };
1449
1450 let proc_name = proc_id
1451 .label()
1452 .map(|l| l.as_str().to_string())
1453 .unwrap_or_else(|| proc_id.id().to_string());
1454
1455 let mut attrs = hyperactor_config::Attrs::new();
1456 attrs.set(crate::introspect::NODE_TYPE, "proc".to_string());
1457 attrs.set(crate::introspect::PROC_NAME, proc_name.clone());
1458 attrs.set(crate::introspect::NUM_ACTORS, children.len());
1459 attrs.set(crate::introspect::SYSTEM_CHILDREN, system_children.clone());
1460 let attrs_json = serde_json::to_string(&attrs).unwrap_or_else(|_| "{}".to_string());
1461
1462 Ok(NodePayload {
1463 identity: NodeRef::Proc(proc_id.clone()),
1464 properties: crate::introspect::derive_properties(&attrs_json),
1465 children,
1466 as_of: std::time::SystemTime::now(),
1467 parent: Some(NodeRef::Root),
1468 })
1469 }
1470
1471 async fn resolve_actor_node(
1485 &self,
1486 cx: &Context<'_, Self>,
1487 actor_id: &hyperactor::ActorAddr,
1488 ) -> Result<NodePayload, anyhow::Error> {
1489 let result = if self.self_actor_id.as_ref() == Some(actor_id) {
1494 cx.introspect_payload()
1495 } else if self.is_standalone_proc_actor(actor_id) {
1496 query_introspect(
1498 cx,
1499 actor_id,
1500 hyperactor::introspect::IntrospectView::Actor,
1501 hyperactor_config::global::get(crate::config::MESH_ADMIN_SINGLE_HOST_TIMEOUT),
1502 &format!("querying actor {}", actor_id),
1503 )
1504 .await?
1505 } else {
1506 let proc_id = actor_id.proc_addr();
1508 let mesh_agent_id = proc_id.actor_addr(PROC_AGENT_ACTOR_NAME);
1509 let terminated = query_child_introspect(
1510 cx,
1511 &mesh_agent_id,
1512 hyperactor::Addr::Actor(actor_id.clone()),
1513 hyperactor_config::global::get(crate::config::MESH_ADMIN_QUERY_CHILD_TIMEOUT),
1514 "querying terminated snapshot",
1515 )
1516 .await
1517 .ok()
1518 .filter(|r| {
1519 let p = crate::introspect::derive_properties(&r.attrs);
1520 !matches!(p, NodeProperties::Error { .. })
1521 });
1522
1523 match terminated {
1524 Some(snapshot) => snapshot,
1525 None => {
1526 query_introspect(
1528 cx,
1529 actor_id,
1530 hyperactor::introspect::IntrospectView::Actor,
1531 hyperactor_config::global::get(
1532 crate::config::MESH_ADMIN_RESOLVE_ACTOR_TIMEOUT,
1533 ),
1534 &format!("querying actor {}", actor_id),
1535 )
1536 .await?
1537 }
1538 }
1539 };
1540 let mut payload = to_node_payload(result);
1541
1542 if self.is_standalone_proc_actor(actor_id) {
1543 payload.parent = Some(crate::introspect::NodeRef::Proc(actor_id.proc_addr()));
1544 return Ok(payload);
1545 }
1546
1547 let proc_id = actor_id.proc_addr();
1548 match &payload.properties {
1549 NodeProperties::Proc { .. } => {
1550 let host_addr = proc_id.addr().to_string();
1551 if let Some(agent) = self.hosts.get(&host_addr) {
1552 payload.parent =
1553 Some(crate::introspect::NodeRef::Host(agent.actor_addr().clone()));
1554 }
1555 }
1556 _ => {
1557 payload.parent = Some(crate::introspect::NodeRef::Proc(proc_id.clone()));
1558 }
1559 }
1560
1561 Ok(payload)
1562 }
1563}
1564
1565fn create_mesh_admin_router(bridge_state: Arc<BridgeState>) -> Router {
1581 Router::new()
1582 .route("/SKILL.md", get(serve_skill_md))
1583 .route("/v1/admin", get(serve_admin_info))
1585 .route("/v1/schema", get(serve_schema))
1586 .route("/v1/schema/admin", get(serve_admin_schema))
1587 .route("/v1/schema/error", get(serve_error_schema))
1588 .route("/v1/openapi.json", get(serve_openapi))
1589 .route("/v1/tree", get(tree_dump))
1590 .route("/v1/query", post(query_proxy))
1591 .route("/v1/pyspy/{*proc_reference}", get(pyspy_bridge))
1592 .route(
1593 "/v1/pyspy_dump/{*proc_reference}",
1594 post(pyspy_dump_and_store),
1595 )
1596 .route(
1597 "/v1/pyspy_profile_svg/{*proc_reference}",
1598 post(pyspy_profile_svg),
1599 )
1600 .route("/v1/config/{*proc_reference}", get(config_bridge))
1601 .route("/v1/{*reference}", get(resolve_reference_bridge))
1602 .with_state(bridge_state)
1603}
1604
1605const SKILL_MD_TEMPLATE: &str = include_str!("mesh_admin_skill.md");
1607
1608fn extract_base_url(headers: &axum::http::HeaderMap) -> String {
1614 let host = headers
1615 .get(axum::http::header::HOST)
1616 .and_then(|v| v.to_str().ok())
1617 .unwrap_or("localhost");
1618 let scheme = headers
1619 .get("x-forwarded-proto")
1620 .and_then(|v| v.to_str().ok())
1621 .unwrap_or("https");
1622 format!("{scheme}://{host}")
1623}
1624
1625async fn serve_admin_info(
1628 State(state): State<Arc<BridgeState>>,
1629) -> axum::response::Json<AdminInfo> {
1630 axum::response::Json(state.admin_info.clone())
1631}
1632
1633async fn serve_admin_schema() -> Result<axum::response::Json<serde_json::Value>, ApiError> {
1635 Ok(axum::response::Json(schema_with_id::<AdminInfo>(
1636 "https://monarch.meta.com/schemas/v1/admin_info",
1637 )?))
1638}
1639
1640async fn serve_skill_md(headers: axum::http::HeaderMap) -> impl axum::response::IntoResponse {
1643 let base = extract_base_url(&headers);
1644 let body = SKILL_MD_TEMPLATE.replace("{base}", &base);
1645 (
1646 [(
1647 axum::http::header::CONTENT_TYPE,
1648 "text/markdown; charset=utf-8",
1649 )],
1650 body,
1651 )
1652}
1653
1654fn schema_with_id<T: schemars::JsonSchema>(id: &str) -> Result<serde_json::Value, ApiError> {
1656 let schema = schemars::schema_for!(T);
1657 let mut value = serde_json::to_value(schema).map_err(|e| ApiError {
1658 code: "internal_error".to_string(),
1659 message: format!("failed to serialize schema: {e}"),
1660 details: None,
1661 })?;
1662 if let Some(obj) = value.as_object_mut() {
1663 obj.insert("$id".into(), serde_json::Value::String(id.into()));
1664 }
1665 Ok(value)
1666}
1667
1668async fn serve_schema() -> Result<axum::response::Json<serde_json::Value>, ApiError> {
1670 Ok(axum::response::Json(schema_with_id::<NodePayloadDto>(
1671 "https://monarch.meta.com/schemas/v1/node_payload",
1672 )?))
1673}
1674
1675async fn serve_error_schema() -> Result<axum::response::Json<serde_json::Value>, ApiError> {
1677 Ok(axum::response::Json(schema_with_id::<ApiErrorEnvelope>(
1678 "https://monarch.meta.com/schemas/v1/error",
1679 )?))
1680}
1681
1682fn hoist_defs(
1686 schema: &mut serde_json::Value,
1687 shared: &mut serde_json::Map<String, serde_json::Value>,
1688) {
1689 if let Some(obj) = schema.as_object_mut() {
1690 if let Some(defs) = obj.remove("$defs")
1691 && let Some(defs_map) = defs.as_object()
1692 {
1693 for (k, v) in defs_map {
1694 shared.insert(k.clone(), v.clone());
1695 }
1696 }
1697 obj.remove("$schema");
1701 }
1702 rewrite_refs(schema);
1703}
1704
1705fn rewrite_refs(value: &mut serde_json::Value) {
1708 match value {
1709 serde_json::Value::Object(map) => {
1710 if let Some(serde_json::Value::String(r)) = map.get_mut("$ref")
1711 && r.starts_with("#/$defs/")
1712 {
1713 *r = r.replace("#/$defs/", "#/components/schemas/");
1714 }
1715 for v in map.values_mut() {
1716 rewrite_refs(v);
1717 }
1718 }
1719 serde_json::Value::Array(arr) => {
1720 for v in arr {
1721 rewrite_refs(v);
1722 }
1723 }
1724 _ => {}
1725 }
1726}
1727
1728pub fn build_openapi_spec() -> serde_json::Value {
1731 let mut node_schema = serde_json::to_value(schemars::schema_for!(NodePayloadDto))
1732 .expect("NodePayload schema must be serializable");
1733 let mut error_schema = serde_json::to_value(schemars::schema_for!(ApiErrorEnvelope))
1734 .expect("ApiErrorEnvelope schema must be serializable");
1735 let mut pyspy_schema = serde_json::to_value(schemars::schema_for!(PySpyResult))
1736 .expect("PySpyResult schema must be serializable");
1737 let mut query_request_schema = serde_json::to_value(schemars::schema_for!(QueryRequest))
1738 .expect("QueryRequest schema must be serializable");
1739 let mut query_response_schema = serde_json::to_value(schemars::schema_for!(QueryResponse))
1740 .expect("QueryResponse schema must be serializable");
1741 let mut pyspy_dump_response_schema =
1742 serde_json::to_value(schemars::schema_for!(PyspyDumpAndStoreResponse))
1743 .expect("PyspyDumpAndStoreResponse schema must be serializable");
1744 let mut admin_info_schema = serde_json::to_value(schemars::schema_for!(AdminInfo))
1745 .expect("AdminInfo schema must be serializable");
1746 let mut profile_opts_schema = serde_json::to_value(schemars::schema_for!(PySpyProfileOpts))
1747 .expect("PySpyProfileOpts schema must be serializable");
1748
1749 let mut shared_schemas = serde_json::Map::new();
1752 hoist_defs(&mut node_schema, &mut shared_schemas);
1753 hoist_defs(&mut error_schema, &mut shared_schemas);
1754 hoist_defs(&mut pyspy_schema, &mut shared_schemas);
1755 hoist_defs(&mut query_request_schema, &mut shared_schemas);
1756 hoist_defs(&mut query_response_schema, &mut shared_schemas);
1757 hoist_defs(&mut pyspy_dump_response_schema, &mut shared_schemas);
1758 hoist_defs(&mut admin_info_schema, &mut shared_schemas);
1759 hoist_defs(&mut profile_opts_schema, &mut shared_schemas);
1760 shared_schemas.insert("NodePayload".into(), node_schema);
1761 shared_schemas.insert("ApiErrorEnvelope".into(), error_schema);
1762 shared_schemas.insert("PySpyResult".into(), pyspy_schema);
1763 shared_schemas.insert("QueryRequest".into(), query_request_schema);
1764 shared_schemas.insert("QueryResponse".into(), query_response_schema);
1765 shared_schemas.insert(
1766 "PyspyDumpAndStoreResponse".into(),
1767 pyspy_dump_response_schema,
1768 );
1769 shared_schemas.insert("AdminInfo".into(), admin_info_schema);
1770 shared_schemas.insert("PySpyProfileOpts".into(), profile_opts_schema);
1771
1772 for value in shared_schemas.values_mut() {
1774 rewrite_refs(value);
1775 }
1776
1777 let error_response = |desc: &str| -> serde_json::Value {
1778 serde_json::json!({
1779 "description": desc,
1780 "content": {
1781 "application/json": {
1782 "schema": { "$ref": "#/components/schemas/ApiErrorEnvelope" }
1783 }
1784 }
1785 })
1786 };
1787
1788 let success_payload = serde_json::json!({
1789 "description": "Resolved NodePayload",
1790 "content": {
1791 "application/json": {
1792 "schema": { "$ref": "#/components/schemas/NodePayload" }
1793 }
1794 }
1795 });
1796
1797 let mut spec = serde_json::json!({
1798 "openapi": "3.1.0",
1799 "info": {
1800 "title": "Monarch Mesh Admin API",
1801 "version": "1.0.0",
1802 "description": "Address-walking introspection API for a Monarch actor mesh. See the Admin Gateway Pattern RFC."
1803 },
1804 "paths": {
1805 "/v1/root": {
1806 "get": {
1807 "summary": "Fetch root node",
1808 "operationId": "getRoot",
1809 "responses": {
1810 "200": success_payload,
1811 "500": error_response("Internal error"),
1812 "503": error_response("Service unavailable (at capacity, retry with backoff)"),
1813 "504": error_response("Gateway timeout (downstream host unresponsive)")
1814 }
1815 }
1816 },
1817 "/v1/{reference}": {
1818 "get": {
1819 "summary": "Resolve a reference to a NodePayload",
1820 "operationId": "resolveReference",
1821 "parameters": [{
1822 "name": "reference",
1823 "in": "path",
1824 "required": true,
1825 "description": "URL-encoded opaque reference string",
1826 "schema": { "type": "string" }
1827 }],
1828 "responses": {
1829 "200": success_payload,
1830 "400": error_response("Bad request (malformed reference)"),
1831 "404": error_response("Address not found"),
1832 "500": error_response("Internal error"),
1833 "503": error_response("Service unavailable (at capacity, retry with backoff)"),
1834 "504": error_response("Gateway timeout (downstream host unresponsive)")
1835 }
1836 }
1837 },
1838 "/v1/schema": {
1839 "get": {
1840 "summary": "JSON Schema for NodePayload (Draft 2020-12)",
1841 "operationId": "getSchema",
1842 "responses": {
1843 "200": {
1844 "description": "JSON Schema document",
1845 "content": { "application/json": {} }
1846 }
1847 }
1848 }
1849 },
1850 "/v1/schema/error": {
1851 "get": {
1852 "summary": "JSON Schema for ApiErrorEnvelope (Draft 2020-12)",
1853 "operationId": "getErrorSchema",
1854 "responses": {
1855 "200": {
1856 "description": "JSON Schema document",
1857 "content": { "application/json": {} }
1858 }
1859 }
1860 }
1861 },
1862 "/v1/admin": {
1863 "get": {
1864 "summary": "Admin self-identification (placement, identity, URL)",
1865 "operationId": "getAdminInfo",
1866 "description": "Returns the admin actor's identity, proc placement, hostname, and URL. Used for placement verification and operational discovery.",
1867 "responses": {
1868 "200": {
1869 "description": "AdminInfo — admin actor placement metadata",
1870 "content": {
1871 "application/json": {
1872 "schema": { "$ref": "#/components/schemas/AdminInfo" }
1873 }
1874 }
1875 }
1876 }
1877 }
1878 },
1879 "/v1/tree": {
1880 "get": {
1881 "summary": "ASCII topology dump (debug)",
1882 "operationId": "getTree",
1883 "responses": {
1884 "200": {
1885 "description": "Human-readable topology tree",
1886 "content": { "text/plain": {} }
1887 }
1888 }
1889 }
1890 },
1891 "/v1/config/{proc_reference}": {
1892 "get": {
1893 "summary": "Config snapshot for a proc",
1894 "operationId": "getConfig",
1895 "description": "Returns the effective CONFIG-marked configuration entries from the target process. Routes to ProcAgent (worker procs) or HostAgent (service proc).",
1896 "parameters": [{
1897 "name": "proc_reference",
1898 "in": "path",
1899 "required": true,
1900 "description": "URL-encoded proc reference (ProcAddr)",
1901 "schema": { "type": "string" }
1902 }],
1903 "responses": {
1904 "200": {
1905 "description": "ConfigDumpResult — sorted list of config entries",
1906 "content": {
1907 "application/json": {
1908 "schema": {
1909 "type": "object",
1910 "properties": {
1911 "entries": {
1912 "type": "array",
1913 "items": {
1914 "type": "object",
1915 "properties": {
1916 "name": { "type": "string" },
1917 "value": { "type": "string" },
1918 "default_value": { "type": ["string", "null"] },
1919 "source": { "type": "string" },
1920 "changed_from_default": { "type": "boolean" },
1921 "env_var": { "type": ["string", "null"] }
1922 }
1923 }
1924 }
1925 }
1926 }
1927 }
1928 }
1929 },
1930 "404": error_response("Proc not found or handler not reachable"),
1931 "500": error_response("Internal error"),
1932 "504": error_response("Gateway timeout")
1933 }
1934 }
1935 },
1936 "/v1/pyspy/{proc_reference}": {
1937 "get": {
1938 "summary": "Py-spy stack dump for a proc",
1939 "operationId": "getPyspy",
1940 "description": "Runs py-spy against the target process and returns structured stack traces. Routes to ProcAgent (worker procs) or HostAgent (service proc).",
1941 "parameters": [{
1942 "name": "proc_reference",
1943 "in": "path",
1944 "required": true,
1945 "description": "URL-encoded proc reference (ProcAddr)",
1946 "schema": { "type": "string" }
1947 }],
1948 "responses": {
1949 "200": {
1950 "description": "PySpyResult — one of Ok, BinaryNotFound, or Failed",
1951 "content": {
1952 "application/json": {
1953 "schema": { "$ref": "#/components/schemas/PySpyResult" }
1954 }
1955 }
1956 },
1957 "400": error_response("Bad request (malformed proc reference)"),
1958 "404": error_response("Proc not found or handler not reachable"),
1959 "500": error_response("Internal error"),
1960 "504": error_response("Gateway timeout")
1961 }
1962 }
1963 },
1964 "/v1/query": {
1965 "post": {
1966 "summary": "Proxy SQL query to the telemetry dashboard",
1967 "operationId": "queryProxy",
1968 "description": "Forwards a SQL query to the Monarch dashboard's DataFusion engine. Requires telemetry_url to be configured.",
1969 "requestBody": {
1970 "required": true,
1971 "content": {
1972 "application/json": {
1973 "schema": { "$ref": "#/components/schemas/QueryRequest" }
1974 }
1975 }
1976 },
1977 "responses": {
1978 "200": {
1979 "description": "Query results",
1980 "content": {
1981 "application/json": {
1982 "schema": { "$ref": "#/components/schemas/QueryResponse" }
1983 }
1984 }
1985 },
1986 "400": error_response("Bad request (invalid SQL or missing sql field)"),
1987 "404": error_response("Dashboard not configured"),
1988 "500": error_response("Internal error"),
1989 "504": error_response("Gateway timeout")
1990 }
1991 }
1992 },
1993 "/v1/pyspy_dump/{proc_reference}": {
1994 "post": {
1995 "summary": "Trigger py-spy dump and store in telemetry",
1996 "operationId": "pyspyDumpAndStore",
1997 "description": "Runs py-spy against the target process, stores the result in the dashboard's DataFusion pyspy tables, and returns the dump_id.",
1998 "parameters": [{
1999 "name": "proc_reference",
2000 "in": "path",
2001 "required": true,
2002 "description": "URL-encoded proc reference (ProcAddr)",
2003 "schema": { "type": "string" }
2004 }],
2005 "responses": {
2006 "200": {
2007 "description": "Dump stored successfully",
2008 "content": {
2009 "application/json": {
2010 "schema": { "$ref": "#/components/schemas/PyspyDumpAndStoreResponse" }
2011 }
2012 }
2013 },
2014 "400": error_response("Bad request (malformed proc reference)"),
2015 "404": error_response("Proc or dashboard not found"),
2016 "500": error_response("Internal error"),
2017 "504": error_response("Gateway timeout")
2018 }
2019 }
2020 }
2021 },
2022 "components": {
2023 "schemas": serde_json::Value::Object(shared_schemas)
2024 }
2025 });
2026
2027 if let Some(paths) = spec.pointer_mut("/paths").and_then(|v| v.as_object_mut()) {
2030 paths.insert(
2031 "/v1/schema/admin".into(),
2032 serde_json::json!({
2033 "get": {
2034 "summary": "JSON Schema for AdminInfo (Draft 2020-12)",
2035 "operationId": "getAdminSchema",
2036 "responses": {
2037 "200": {
2038 "description": "JSON Schema document",
2039 "content": { "application/json": {} }
2040 }
2041 }
2042 }
2043 }),
2044 );
2045 paths.insert(
2046 "/v1/pyspy_profile_svg/{proc_reference}".into(),
2047 serde_json::json!({
2048 "post": {
2049 "summary": "Profile a proc and return SVG flamegraph",
2050 "operationId": "pyspyProfileSvg",
2051 "description": "Runs py-spy record against the target process for the requested duration and returns an SVG flamegraph. Timeout scales with duration_s.",
2052 "parameters": [{
2053 "name": "proc_reference",
2054 "in": "path",
2055 "required": true,
2056 "description": "URL-encoded proc reference (ProcAddr)",
2057 "schema": { "type": "string" }
2058 }],
2059 "requestBody": {
2060 "required": true,
2061 "content": {
2062 "application/json": {
2063 "schema": { "$ref": "#/components/schemas/PySpyProfileOpts" }
2064 }
2065 }
2066 },
2067 "responses": {
2068 "200": {
2069 "description": "SVG flamegraph",
2070 "content": { "image/svg+xml": {} }
2071 },
2072 "400": error_response("Bad request (invalid duration/rate or malformed proc reference)"),
2073 "404": error_response("Proc not found or handler not reachable"),
2074 "500": error_response("Internal error (profile failed or SVG generation failed)"),
2075 "503": error_response("Service unavailable (py-spy not available on target host)"),
2076 "504": error_response("Gateway timeout (subprocess timed out)")
2077 }
2078 }
2079 }),
2080 );
2081 }
2082
2083 spec
2084}
2085
2086async fn serve_openapi() -> Result<axum::response::Json<serde_json::Value>, ApiError> {
2088 Ok(axum::response::Json(build_openapi_spec()))
2089}
2090
2091fn parse_proc_reference(raw: &str) -> Result<(String, ProcAddr), ApiError> {
2094 let trimmed = raw.trim_start_matches('/');
2095 if trimmed.is_empty() {
2096 return Err(ApiError::bad_request("empty proc reference", None));
2097 }
2098 let decoded = urlencoding::decode(trimmed)
2099 .map(|cow| cow.into_owned())
2100 .map_err(|_| {
2101 ApiError::bad_request(
2102 "malformed percent-encoding: decoded bytes are not valid UTF-8",
2103 None,
2104 )
2105 })?;
2106 let proc_id: ProcAddr = decoded
2107 .parse()
2108 .map_err(|e| ApiError::bad_request(format!("invalid proc reference: {}", e), None))?;
2109 Ok((decoded, proc_id))
2110}
2111
2112async fn probe_actor(
2118 cx: &Instance<()>,
2119 agent_id: &hyperactor::ActorAddr,
2120) -> Result<bool, ApiError> {
2121 let port = agent_id.introspect_port();
2122 let (handle, rx) = open_once_port::<IntrospectResult>(cx);
2123 port.post(
2124 cx,
2125 IntrospectMessage::Query {
2126 view: IntrospectView::Entity,
2127 reply: handle.bind(),
2128 },
2129 );
2130
2131 let timeout = hyperactor_config::global::get(crate::config::MESH_ADMIN_QUERY_CHILD_TIMEOUT);
2132 match tokio::time::timeout(timeout, rx.recv()).await {
2133 Ok(Ok(_)) => Ok(true),
2134 Ok(Err(e)) => {
2135 tracing::debug!(
2136 name = "pyspy_probe_recv_failed",
2137 %agent_id,
2138 error = %e,
2139 );
2140 Ok(false)
2141 }
2142 Err(_elapsed) => {
2143 tracing::debug!(
2144 name = "pyspy_probe_timeout",
2145 %agent_id,
2146 );
2147 Ok(false)
2148 }
2149 }
2150}
2151
2152enum ResolvedProcHandler {
2159 Host(ActorRef<HostAgent>),
2160 Proc(ActorRef<ProcAgent>),
2161}
2162
2163impl ResolvedProcHandler {
2164 fn agent_id(&self) -> hyperactor::ActorAddr {
2165 match self {
2166 Self::Host(r) => r.actor_addr().clone(),
2167 Self::Proc(r) => r.actor_addr().clone(),
2168 }
2169 }
2170
2171 async fn pyspy_dump(
2172 &self,
2173 cx: &impl hyperactor::context::Actor,
2174 opts: PySpyOpts,
2175 timeout: std::time::Duration,
2176 ) -> Result<PySpyResult, ApiError> {
2177 let (reply_handle, reply_rx) = open_once_port::<PySpyResult>(cx);
2178 let mut reply_ref = reply_handle.bind();
2179 reply_ref.return_undeliverable(false);
2180 let msg = PySpyDump {
2181 opts,
2182 result: reply_ref,
2183 };
2184 match self {
2185 Self::Host(r) => r.post(cx, msg),
2186 Self::Proc(r) => r.post(cx, msg),
2187 };
2188 tokio::time::timeout(timeout, reply_rx.recv())
2189 .await
2190 .map_err(|_| ApiError {
2191 code: "gateway_timeout".to_string(),
2192 message: "timed out waiting for py-spy dump".to_string(),
2193 details: None,
2194 })?
2195 .map_err(|e| ApiError {
2196 code: "internal_error".to_string(),
2197 message: format!("failed to receive PySpyResult: {}", e),
2198 details: None,
2199 })
2200 }
2201
2202 async fn pyspy_profile(
2203 &self,
2204 cx: &impl hyperactor::context::Actor,
2205 request: ValidatedProfileRequest,
2206 timeout: std::time::Duration,
2207 ) -> Result<PySpyProfileResult, ApiError> {
2208 let (reply_handle, reply_rx) = open_once_port::<PySpyProfileResult>(cx);
2209 let mut reply_ref = reply_handle.bind();
2210 reply_ref.return_undeliverable(false);
2211 let msg = PySpyProfile {
2212 request,
2213 result: reply_ref,
2214 };
2215 match self {
2216 Self::Host(r) => r.post(cx, msg),
2217 Self::Proc(r) => r.post(cx, msg),
2218 };
2219 tokio::time::timeout(timeout, reply_rx.recv())
2220 .await
2221 .map_err(|_| ApiError {
2222 code: "gateway_timeout".to_string(),
2223 message: "timed out waiting for py-spy profile".to_string(),
2224 details: None,
2225 })?
2226 .map_err(|e| ApiError {
2227 code: "internal_error".to_string(),
2228 message: format!("failed to receive PySpyProfileResult: {}", e),
2229 details: None,
2230 })
2231 }
2232
2233 async fn config_dump(
2234 &self,
2235 cx: &impl hyperactor::context::Actor,
2236 timeout: std::time::Duration,
2237 ) -> Result<ConfigDumpResult, ApiError> {
2238 let (reply_handle, reply_rx) = open_once_port::<ConfigDumpResult>(cx);
2239 let mut reply_ref = reply_handle.bind();
2240 reply_ref.return_undeliverable(false);
2241 let msg = ConfigDump { result: reply_ref };
2242 match self {
2243 Self::Host(r) => r.post(cx, msg),
2244 Self::Proc(r) => r.post(cx, msg),
2245 };
2246 tokio::time::timeout(timeout, reply_rx.recv())
2247 .await
2248 .map_err(|_| ApiError {
2249 code: "gateway_timeout".to_string(),
2250 message: "timed out waiting for config dump".to_string(),
2251 details: None,
2252 })?
2253 .map_err(|e| ApiError {
2254 code: "internal_error".to_string(),
2255 message: format!("failed to receive ConfigDumpResult: {}", e),
2256 details: None,
2257 })
2258 }
2259}
2260
2261fn route_proc_handler(raw_proc_reference: &str) -> Result<ResolvedProcHandler, ApiError> {
2265 let (_proc_reference, proc_id) = parse_proc_reference(raw_proc_reference)?;
2266 let is_service = proc_id
2267 .uid()
2268 .as_singleton()
2269 .is_some_and(|label| label.as_str() == SERVICE_PROC_NAME);
2270 if is_service {
2271 let agent_id = proc_id.actor_addr(HOST_MESH_AGENT_ACTOR_NAME);
2272 Ok(ResolvedProcHandler::Host(ActorRef::attest(agent_id)))
2273 } else {
2274 let agent_id = proc_id.actor_addr(PROC_AGENT_ACTOR_NAME);
2275 Ok(ResolvedProcHandler::Proc(ActorRef::attest(agent_id)))
2276 }
2277}
2278
2279async fn resolve_proc_handler(
2281 state: &BridgeState,
2282 raw_proc_reference: &str,
2283) -> Result<ResolvedProcHandler, ApiError> {
2284 let handler = route_proc_handler(raw_proc_reference)?;
2285 let cx = &state.bridge_cx;
2286 if !probe_actor(cx, &handler.agent_id()).await? {
2287 return Err(ApiError::not_found(
2288 format!(
2289 "proc does not have a reachable handler ({})",
2290 raw_proc_reference,
2291 ),
2292 None,
2293 ));
2294 }
2295 Ok(handler)
2296}
2297
2298async fn do_pyspy_dump(
2299 state: &BridgeState,
2300 raw_proc_reference: &str,
2301) -> Result<PySpyResult, ApiError> {
2302 let handler = resolve_proc_handler(state, raw_proc_reference).await?;
2303 let timeout = hyperactor_config::global::get(crate::config::MESH_ADMIN_PYSPY_BRIDGE_TIMEOUT);
2304 handler
2305 .pyspy_dump(
2306 &state.bridge_cx,
2307 PySpyOpts {
2308 threads: false,
2309 native: true,
2310 native_all: true,
2311 nonblocking: false,
2312 },
2313 timeout,
2314 )
2315 .await
2316}
2317
2318async fn pyspy_bridge(
2325 State(state): State<Arc<BridgeState>>,
2326 AxumPath(proc_reference): AxumPath<String>,
2327) -> Result<Json<PySpyResult>, ApiError> {
2328 Ok(Json(do_pyspy_dump(&state, &proc_reference).await?))
2329}
2330
2331async fn do_pyspy_profile(
2332 state: &BridgeState,
2333 raw_proc_reference: &str,
2334 opts: PySpyProfileOpts,
2335) -> Result<PySpyProfileResult, ApiError> {
2336 let max_duration =
2337 hyperactor_config::global::get(crate::config::MESH_ADMIN_PYSPY_MAX_PROFILE_DURATION);
2338 let request = ValidatedProfileRequest::try_new(&opts, max_duration)
2339 .map_err(|msg| ApiError::bad_request(msg, None))?;
2340 let bridge_timeout = request.bridge_timeout();
2341 let handler = resolve_proc_handler(state, raw_proc_reference).await?;
2342 handler
2343 .pyspy_profile(&state.bridge_cx, request, bridge_timeout)
2344 .await
2345}
2346
2347async fn pyspy_profile_svg(
2352 State(state): State<Arc<BridgeState>>,
2353 AxumPath(proc_reference): AxumPath<String>,
2354 Json(opts): Json<PySpyProfileOpts>,
2355) -> Result<axum::response::Response, ApiError> {
2356 let result = do_pyspy_profile(&state, &proc_reference, opts).await?;
2357 match result {
2358 PySpyProfileResult::Ok { svg, .. } => Ok(axum::response::Response::builder()
2359 .header("content-type", "image/svg+xml")
2360 .body(axum::body::Body::from(svg))
2361 .unwrap()),
2362 PySpyProfileResult::BinaryNotFound { searched } => Err(ApiError {
2363 code: "service_unavailable".to_string(),
2364 message: format!(
2365 "py-spy not available on target host; searched: {}",
2366 searched.join(", ")
2367 ),
2368 details: None,
2369 }),
2370 PySpyProfileResult::TimedOut {
2371 timeout_s, stderr, ..
2372 } => Err(ApiError {
2373 code: "gateway_timeout".to_string(),
2374 message: format!(
2375 "py-spy record subprocess timed out after {}s: {}",
2376 timeout_s,
2377 stderr.trim()
2378 ),
2379 details: None,
2380 }),
2381 PySpyProfileResult::ExitFailure { stderr, .. } => Err(ApiError {
2382 code: "profile_failed".to_string(),
2383 message: stderr,
2384 details: None,
2385 }),
2386 PySpyProfileResult::OutputMissing { pid, binary } => Err(ApiError {
2387 code: "profile_output_unusable".to_string(),
2388 message: format!("py-spy exited 0 but SVG file is missing (pid {pid}, {binary})"),
2389 details: None,
2390 }),
2391 PySpyProfileResult::OutputEmpty { pid, binary } => Err(ApiError {
2392 code: "profile_output_unusable".to_string(),
2393 message: format!("py-spy exited 0 but SVG output is empty (pid {pid}, {binary})"),
2394 details: None,
2395 }),
2396 PySpyProfileResult::OutputReadFailure { error, .. } => Err(ApiError {
2397 code: "internal_error".to_string(),
2398 message: format!("failed to read SVG output: {error}"),
2399 details: None,
2400 }),
2401 PySpyProfileResult::WorkerSpawnFailure { error } => Err(ApiError {
2402 code: "internal_error".to_string(),
2403 message: format!("failed to spawn profile worker actor: {error}"),
2404 details: None,
2405 }),
2406 PySpyProfileResult::SubprocessSpawnFailure { error, .. } => Err(ApiError {
2407 code: "internal_error".to_string(),
2408 message: format!("failed to execute py-spy: {error}"),
2409 details: None,
2410 }),
2411 PySpyProfileResult::WaitFailure { error, .. } => Err(ApiError {
2412 code: "internal_error".to_string(),
2413 message: format!("failed to wait for child: {error}"),
2414 details: None,
2415 }),
2416 PySpyProfileResult::TempDirFailure { error, .. } => Err(ApiError {
2417 code: "internal_error".to_string(),
2418 message: format!("failed to create temp dir: {error}"),
2419 details: None,
2420 }),
2421 }
2422}
2423
2424#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
2426pub struct QueryRequest {
2427 pub sql: String,
2429}
2430
2431#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
2433pub struct QueryResponse {
2434 pub rows: serde_json::Value,
2436}
2437
2438#[derive(Debug, Serialize)]
2440struct StorePyspyDumpRequest {
2441 dump_id: String,
2442 proc_ref: String,
2443 pyspy_result_json: String,
2444}
2445
2446#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
2448pub struct PyspyDumpAndStoreResponse {
2449 pub dump_id: String,
2451}
2452
2453fn require_telemetry_url(state: &BridgeState) -> Result<&str, ApiError> {
2456 state.telemetry_url.as_deref().ok_or_else(|| {
2457 ApiError::not_found("dashboard not configured (no telemetry_url provided)", None)
2458 })
2459}
2460
2461async fn query_proxy(
2468 State(state): State<Arc<BridgeState>>,
2469 axum::Json(body): axum::Json<QueryRequest>,
2470) -> Result<axum::Json<QueryResponse>, ApiError> {
2471 let telemetry_url = require_telemetry_url(&state)?;
2472
2473 let resp = state
2474 .http_client
2475 .post(format!("{}/api/query", telemetry_url))
2476 .json(&body)
2477 .send()
2478 .await
2479 .map_err(|e| ApiError {
2480 code: "proxy_error".to_string(),
2481 message: format!("failed to proxy query to dashboard: {}", e),
2482 details: None,
2483 })?;
2484
2485 let status = resp.status();
2486 let resp_body = resp.bytes().await.map_err(|e| ApiError {
2487 code: "proxy_error".to_string(),
2488 message: format!("failed to read dashboard response: {}", e),
2489 details: None,
2490 })?;
2491
2492 if !status.is_success() {
2493 let msg = serde_json::from_slice::<serde_json::Value>(&resp_body)
2495 .ok()
2496 .and_then(|v| v.get("error")?.as_str().map(String::from))
2497 .unwrap_or_else(|| format!("dashboard returned HTTP {status}"));
2498 let code = if status.is_client_error() {
2499 "bad_request"
2500 } else {
2501 "proxy_error"
2502 };
2503 return Err(ApiError {
2504 code: code.to_string(),
2505 message: msg,
2506 details: None,
2507 });
2508 }
2509
2510 let result: QueryResponse = serde_json::from_slice(&resp_body).map_err(|e| ApiError {
2511 code: "proxy_error".to_string(),
2512 message: format!("failed to parse dashboard response: {}", e),
2513 details: None,
2514 })?;
2515
2516 Ok(axum::Json(result))
2517}
2518
2519async fn pyspy_dump_and_store(
2528 State(state): State<Arc<BridgeState>>,
2529 AxumPath(proc_reference): AxumPath<String>,
2530) -> Result<axum::Json<PyspyDumpAndStoreResponse>, ApiError> {
2531 let telemetry_url = require_telemetry_url(&state)?;
2532 let pyspy_result = do_pyspy_dump(&state, &proc_reference).await?;
2533
2534 let dump_id = uuid::Uuid::new_v4().to_string();
2535 let pyspy_json = serde_json::to_string(&pyspy_result).map_err(|e| ApiError {
2536 code: "internal_error".to_string(),
2537 message: format!("failed to serialize PySpyResult: {}", e),
2538 details: None,
2539 })?;
2540
2541 let store_body = StorePyspyDumpRequest {
2542 dump_id: dump_id.clone(),
2543 proc_ref: proc_reference,
2544 pyspy_result_json: pyspy_json,
2545 };
2546
2547 let store_resp = state
2548 .http_client
2549 .post(format!("{}/api/pyspy_dump", telemetry_url))
2550 .json(&store_body)
2551 .send()
2552 .await
2553 .map_err(|e| ApiError {
2554 code: "proxy_error".to_string(),
2555 message: format!("failed to store pyspy dump in dashboard: {}", e),
2556 details: None,
2557 })?;
2558
2559 if !store_resp.status().is_success() {
2560 return Err(ApiError {
2561 code: "proxy_error".to_string(),
2562 message: format!(
2563 "dashboard rejected pyspy dump store: HTTP {}",
2564 store_resp.status()
2565 ),
2566 details: None,
2567 });
2568 }
2569
2570 Ok(axum::Json(PyspyDumpAndStoreResponse { dump_id }))
2571}
2572
2573async fn config_bridge(
2578 State(state): State<Arc<BridgeState>>,
2579 AxumPath(proc_reference): AxumPath<String>,
2580) -> Result<Json<ConfigDumpResult>, ApiError> {
2581 let handler = route_proc_handler(&proc_reference)?;
2582 let timeout =
2583 hyperactor_config::global::get(crate::config::MESH_ADMIN_CONFIG_DUMP_BRIDGE_TIMEOUT);
2584 let result = handler.config_dump(&state.bridge_cx, timeout).await?;
2585 Ok(Json(result))
2586}
2587
2588async fn resolve_reference_bridge(
2601 State(state): State<Arc<BridgeState>>,
2602 AxumPath(reference): AxumPath<String>,
2603) -> Result<Json<NodePayloadDto>, ApiError> {
2604 let reference = reference.trim_start_matches('/');
2606 if reference.is_empty() {
2607 return Err(ApiError::bad_request("empty reference", None));
2608 }
2609 let reference = urlencoding::decode(reference)
2610 .map(|cow| cow.into_owned())
2611 .map_err(|_| {
2612 ApiError::bad_request(
2613 "malformed percent-encoding: decoded bytes are not valid UTF-8",
2614 None,
2615 )
2616 })?;
2617
2618 let _permit = state.resolve_semaphore.try_acquire().map_err(|_| {
2621 tracing::warn!("mesh admin: rejecting resolve request (503): too many concurrent requests");
2622 ApiError {
2623 code: "service_unavailable".to_string(),
2624 message: "too many concurrent introspection requests".to_string(),
2625 details: None,
2626 }
2627 })?;
2628
2629 let cx = &state.bridge_cx;
2630 let resolve_start = std::time::Instant::now();
2631 let response = tokio::time::timeout(
2632 hyperactor_config::global::get(crate::config::MESH_ADMIN_SINGLE_HOST_TIMEOUT),
2633 state.admin_ref.resolve(cx, reference.clone()),
2634 )
2635 .await
2636 .map_err(|_| {
2637 tracing::warn!(
2638 reference = %reference,
2639 elapsed_ms = resolve_start.elapsed().as_millis() as u64,
2640 "mesh admin: resolve timed out (gateway_timeout)",
2641 );
2642 ApiError {
2643 code: "gateway_timeout".to_string(),
2644 message: "timed out resolving reference".to_string(),
2645 details: None,
2646 }
2647 })?
2648 .map_err(|e| ApiError {
2649 code: "internal_error".to_string(),
2650 message: format!("failed to resolve reference: {}", e),
2651 details: None,
2652 })?;
2653
2654 match response.0 {
2655 Ok(payload) => Ok(Json(NodePayloadDto::from(payload))),
2656 Err(error) => Err(ApiError::not_found(error, None)),
2657 }
2658}
2659
2660async fn tree_dump(
2685 State(state): State<Arc<BridgeState>>,
2686 headers: axum::http::header::HeaderMap,
2687) -> Result<String, ApiError> {
2688 let _permit = state.resolve_semaphore.try_acquire().map_err(|_| {
2690 tracing::warn!(
2691 "mesh admin: rejecting tree_dump request (503): too many concurrent requests"
2692 );
2693 ApiError {
2694 code: "service_unavailable".to_string(),
2695 message: "too many concurrent introspection requests".to_string(),
2696 details: None,
2697 }
2698 })?;
2699
2700 let cx = &state.bridge_cx;
2701
2702 let host = headers
2704 .get("host")
2705 .and_then(|v| v.to_str().ok())
2706 .unwrap_or("localhost");
2707 let scheme = headers
2708 .get("x-forwarded-proto")
2709 .and_then(|v| v.to_str().ok())
2710 .unwrap_or("http");
2711 let base_url = format!("{}://{}", scheme, host);
2712
2713 let root_resp = tokio::time::timeout(
2715 hyperactor_config::global::get(crate::config::MESH_ADMIN_TREE_TIMEOUT),
2716 state.admin_ref.resolve(cx, "root".to_string()),
2717 )
2718 .await
2719 .map_err(|_| ApiError {
2720 code: "gateway_timeout".to_string(),
2721 message: "timed out resolving root".to_string(),
2722 details: None,
2723 })?
2724 .map_err(|e| ApiError {
2725 code: "internal_error".to_string(),
2726 message: format!("failed to resolve root: {}", e),
2727 details: None,
2728 })?;
2729
2730 let root = root_resp.0.map_err(|e| ApiError {
2731 code: "internal_error".to_string(),
2732 message: e,
2733 details: None,
2734 })?;
2735
2736 let mut output = String::new();
2737
2738 for child_ref in &root.children {
2742 let child_ref_str = child_ref.to_string();
2743 let resp = tokio::time::timeout(
2744 hyperactor_config::global::get(crate::config::MESH_ADMIN_TREE_TIMEOUT),
2745 state.admin_ref.resolve(cx, child_ref_str.clone()),
2746 )
2747 .await;
2748
2749 let payload = match resp {
2750 Ok(Ok(r)) => r.0.ok(),
2751 _ => None,
2752 };
2753
2754 match payload {
2755 Some(node) if matches!(node.properties, NodeProperties::Host { .. }) => {
2756 let header = match &node.properties {
2757 NodeProperties::Host { addr, .. } => addr.clone(),
2758 _ => child_ref_str.clone(),
2759 };
2760 let host_url = format!("{}/v1/{}", base_url, urlencoding::encode(&child_ref_str));
2761 output.push_str(&format!("{} -> {}\n", header, host_url));
2762
2763 let num_procs = node.children.len();
2764 for (i, proc_ref) in node.children.iter().enumerate() {
2765 let proc_ref_str = proc_ref.to_string();
2766 let is_last_proc = i == num_procs - 1;
2767 let proc_connector = if is_last_proc {
2768 "└── "
2769 } else {
2770 "├── "
2771 };
2772 let proc_name = derive_tree_label(proc_ref);
2773 let proc_url =
2774 format!("{}/v1/{}", base_url, urlencoding::encode(&proc_ref_str));
2775 output.push_str(&format!(
2776 "{}{} -> {}\n",
2777 proc_connector, proc_name, proc_url
2778 ));
2779
2780 let proc_resp = tokio::time::timeout(
2781 hyperactor_config::global::get(crate::config::MESH_ADMIN_TREE_TIMEOUT),
2782 state.admin_ref.resolve(cx, proc_ref_str),
2783 )
2784 .await;
2785 let proc_payload = match proc_resp {
2786 Ok(Ok(r)) => r.0.ok(),
2787 _ => None,
2788 };
2789 if let Some(proc_node) = proc_payload {
2790 let num_actors = proc_node.children.len();
2791 let child_prefix = if is_last_proc { " " } else { "│ " };
2792 for (j, actor_ref) in proc_node.children.iter().enumerate() {
2793 let actor_ref_str = actor_ref.to_string();
2794 let actor_connector = if j == num_actors - 1 {
2795 "└── "
2796 } else {
2797 "├── "
2798 };
2799 let actor_label = derive_actor_label(actor_ref);
2800 let actor_url =
2801 format!("{}/v1/{}", base_url, urlencoding::encode(&actor_ref_str));
2802 output.push_str(&format!(
2803 "{}{}{} -> {}\n",
2804 child_prefix, actor_connector, actor_label, actor_url
2805 ));
2806 }
2807 }
2808 }
2809 output.push('\n');
2810 }
2811 Some(node) if matches!(node.properties, NodeProperties::Proc { .. }) => {
2812 let proc_name = match &node.properties {
2813 NodeProperties::Proc { proc_name, .. } => proc_name.clone(),
2814 _ => child_ref_str.clone(),
2815 };
2816 let proc_url = format!("{}/v1/{}", base_url, urlencoding::encode(&child_ref_str));
2817 output.push_str(&format!("{} -> {}\n", proc_name, proc_url));
2818
2819 let num_actors = node.children.len();
2820 for (j, actor_ref) in node.children.iter().enumerate() {
2821 let actor_ref_str = actor_ref.to_string();
2822 let actor_connector = if j == num_actors - 1 {
2823 "└── "
2824 } else {
2825 "├── "
2826 };
2827 let actor_label = derive_actor_label(actor_ref);
2828 let actor_url =
2829 format!("{}/v1/{}", base_url, urlencoding::encode(&actor_ref_str));
2830 output.push_str(&format!(
2831 "{}{} -> {}\n",
2832 actor_connector, actor_label, actor_url
2833 ));
2834 }
2835 output.push('\n');
2836 }
2837 Some(_node) => {
2838 let label = derive_actor_label(child_ref);
2839 let url = format!("{}/v1/{}", base_url, urlencoding::encode(&child_ref_str));
2840 output.push_str(&format!("{} -> {}\n\n", label, url));
2841 }
2842 _ => {
2843 output.push_str(&format!("{} (unreachable)\n\n", child_ref));
2844 }
2845 }
2846 }
2847 Ok(output)
2848}
2849
2850fn derive_tree_label(node_ref: &crate::introspect::NodeRef) -> String {
2865 match node_ref {
2866 crate::introspect::NodeRef::Root => "root".to_string(),
2867 crate::introspect::NodeRef::Host(id) => id.proc_addr().id().to_string(),
2868 crate::introspect::NodeRef::Proc(id) => id.id().to_string(),
2869 crate::introspect::NodeRef::Actor(id) => {
2870 format!("{}[{}]", id.log_name(), id.uid())
2871 }
2872 }
2873}
2874
2875fn derive_actor_label(node_ref: &crate::introspect::NodeRef) -> String {
2876 match node_ref {
2877 crate::introspect::NodeRef::Root => "root".to_string(),
2878 crate::introspect::NodeRef::Host(id) => id.log_name().to_string(),
2879 crate::introspect::NodeRef::Proc(id) => id.id().to_string(),
2880 crate::introspect::NodeRef::Actor(id) => {
2881 format!("{}[{}]", id.log_name(), id.uid())
2882 }
2883 }
2884}
2885
2886#[non_exhaustive]
2895pub enum PublishedHandle {
2896 Mast(String),
2898}
2899
2900impl PublishedHandle {
2901 pub async fn resolve(self, _port_override: Option<u16>) -> anyhow::Result<String> {
2906 anyhow::bail!(
2907 "publication-based admin handle resolution is not yet implemented: \
2908 mesh admin placement has moved to the caller's local proc. \
2909 Discover the admin URL from startup output or another \
2910 launch-time publication instead."
2911 )
2912 }
2913}
2914
2915#[non_exhaustive]
2920pub enum AdminHandle {
2921 Url(String),
2923 Published(PublishedHandle),
2925 Unsupported(String),
2927}
2928
2929impl AdminHandle {
2930 pub fn parse(addr: &str) -> Self {
2939 if addr.starts_with("mast_conda:///") {
2941 return AdminHandle::Published(PublishedHandle::Mast(addr.to_string()));
2942 }
2943 if let Ok(parsed) = url::Url::parse(addr)
2945 && matches!(parsed.scheme(), "http" | "https")
2946 {
2947 return AdminHandle::Url(addr.to_string());
2948 }
2949 let with_scheme = format!("https://{}", addr);
2952 if let Ok(parsed) = url::Url::parse(&with_scheme)
2953 && parsed.host_str().is_some()
2954 && parsed.port().is_some()
2955 {
2956 return AdminHandle::Url(with_scheme);
2957 }
2958 AdminHandle::Unsupported(addr.to_string())
2959 }
2960
2961 pub async fn resolve(self, port_override: Option<u16>) -> anyhow::Result<String> {
2967 match self {
2968 AdminHandle::Url(url) => Ok(url),
2969 AdminHandle::Published(h) => h.resolve(port_override).await,
2970 AdminHandle::Unsupported(s) => anyhow::bail!(
2971 "unrecognized admin handle '{}': expected https://host:port or mast_conda:///job",
2972 s
2973 ),
2974 }
2975 }
2976}
2977
2978pub async fn resolve_mast_handle(
2984 handle: &str,
2985 port_override: Option<u16>,
2986) -> anyhow::Result<String> {
2987 AdminHandle::Published(PublishedHandle::Mast(handle.to_string()))
2988 .resolve(port_override)
2989 .await
2990}
2991
2992mod advertised_host {
3000 use std::net::IpAddr;
3001
3002 #[derive(Debug, PartialEq, Eq)]
3004 pub(super) enum SanIdentity {
3005 Ip(IpAddr),
3006 Dns(String),
3007 }
3008
3009 pub(super) fn from_cert_sans() -> String {
3021 let hostname = hostname::get()
3022 .unwrap_or_else(|_| "localhost".into())
3023 .into_string()
3024 .unwrap_or_else(|_| "localhost".to_string());
3025
3026 let mut candidates: Vec<(String, SanIdentity)> = Vec::new();
3030
3031 candidates.push((hostname.clone(), SanIdentity::Dns(hostname.clone())));
3033
3034 #[cfg(fbcode_build)]
3036 if let Ok(ip_str) = hyperactor::meta::host_ip::host_ipv6_address()
3037 && let Ok(ip) = ip_str.parse::<IpAddr>()
3038 {
3039 candidates.push((format!("[{}]", ip), SanIdentity::Ip(ip)));
3040 }
3041
3042 let cert_sans = load_cert_sans();
3043 let chosen = pick_candidate(&candidates, &cert_sans, &hostname);
3044
3045 if chosen != hostname && !cert_sans.is_empty() {
3046 tracing::info!("admin URL host '{}' matches cert SAN", chosen);
3047 } else if !cert_sans.is_empty() && !candidates.iter().any(|(_, id)| cert_sans.contains(id))
3048 {
3049 tracing::warn!(
3050 "no admin URL candidate matched cert SANs; falling back to hostname '{}'",
3051 hostname,
3052 );
3053 }
3054
3055 chosen
3056 }
3057
3058 fn load_cert_sans() -> Vec<SanIdentity> {
3065 use std::io::BufReader;
3066
3067 use x509_parser::prelude::*;
3068
3069 let bundle = match hyperactor::channel::try_tls_pem_bundle() {
3070 Some(b) => b,
3071 None => return Vec::new(),
3072 };
3073
3074 let cert_pem = match bundle.cert.reader() {
3075 Ok(r) => {
3076 let mut buf = Vec::new();
3077 if std::io::Read::read_to_end(&mut BufReader::new(r), &mut buf).is_err() {
3078 return Vec::new();
3079 }
3080 buf
3081 }
3082 Err(_) => return Vec::new(),
3083 };
3084
3085 let mut cursor = &cert_pem[..];
3086 let certs: Vec<_> = rustls_pemfile::certs(&mut cursor)
3087 .filter_map(|r| r.ok())
3088 .collect();
3089
3090 let leaf_der = match certs.first() {
3091 Some(c) => c,
3092 None => return Vec::new(),
3093 };
3094
3095 let (_, cert) = match X509Certificate::from_der(leaf_der.as_ref()) {
3096 Ok(parsed) => parsed,
3097 Err(e) => {
3098 tracing::warn!("failed to parse leaf cert for SAN extraction: {}", e);
3099 return Vec::new();
3100 }
3101 };
3102
3103 let mut sans = Vec::new();
3104 if let Ok(Some(san_ext)) = cert.subject_alternative_name() {
3105 for name in &san_ext.value.general_names {
3106 match name {
3107 GeneralName::DNSName(dns) => {
3108 sans.push(SanIdentity::Dns(dns.to_string()));
3109 }
3110 GeneralName::IPAddress(bytes) => {
3111 let ip = match bytes.len() {
3112 4 => IpAddr::from(<[u8; 4]>::try_from(*bytes).unwrap()),
3113 16 => IpAddr::from(<[u8; 16]>::try_from(*bytes).unwrap()),
3114 _ => continue,
3115 };
3116 sans.push(SanIdentity::Ip(ip));
3117 }
3118 _ => {}
3119 }
3120 }
3121 }
3122
3123 sans
3124 }
3125
3126 fn pick_candidate(
3129 candidates: &[(String, SanIdentity)],
3130 cert_sans: &[SanIdentity],
3131 fallback: &str,
3132 ) -> String {
3133 if cert_sans.is_empty() {
3134 return fallback.to_string();
3135 }
3136 for (url_host, identity) in candidates {
3137 if cert_sans.iter().any(|san| san == identity) {
3138 return url_host.clone();
3139 }
3140 }
3141 fallback.to_string()
3142 }
3143
3144 #[cfg(test)]
3145 mod tests {
3146 use std::net::IpAddr;
3147 use std::net::Ipv4Addr;
3148 use std::net::Ipv6Addr;
3149
3150 use super::*;
3151
3152 #[test]
3153 fn cert_covers_hostname_only_picks_hostname() {
3154 let candidates = vec![
3155 ("myhost".to_string(), SanIdentity::Dns("myhost".to_string())),
3156 (
3157 "[::1]".to_string(),
3158 SanIdentity::Ip(IpAddr::V6(Ipv6Addr::LOCALHOST)),
3159 ),
3160 ];
3161 let sans = vec![SanIdentity::Dns("myhost".to_string())];
3162 assert_eq!(pick_candidate(&candidates, &sans, "fallback"), "myhost");
3163 }
3164
3165 #[test]
3166 fn cert_covers_ip_only_picks_ip() {
3167 let ip = IpAddr::V6("2803:6084:3894:2b36:b5d3:11ef:400:0".parse().unwrap());
3168 let candidates = vec![
3169 ("myhost".to_string(), SanIdentity::Dns("myhost".to_string())),
3170 (format!("[{}]", ip), SanIdentity::Ip(ip)),
3171 ];
3172 let sans = vec![SanIdentity::Ip(ip)];
3173 assert_eq!(
3174 pick_candidate(&candidates, &sans, "fallback"),
3175 format!("[{}]", ip)
3176 );
3177 }
3178
3179 #[test]
3180 fn cert_covers_both_prefers_hostname() {
3181 let ip = IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1));
3182 let candidates = vec![
3183 ("myhost".to_string(), SanIdentity::Dns("myhost".to_string())),
3184 ("10.0.0.1".to_string(), SanIdentity::Ip(ip)),
3185 ];
3186 let sans = vec![SanIdentity::Dns("myhost".to_string()), SanIdentity::Ip(ip)];
3187 assert_eq!(pick_candidate(&candidates, &sans, "fallback"), "myhost");
3188 }
3189
3190 #[test]
3191 fn no_sans_returns_fallback() {
3192 let candidates = vec![("myhost".to_string(), SanIdentity::Dns("myhost".to_string()))];
3193 assert_eq!(pick_candidate(&candidates, &[], "fallback"), "fallback");
3194 }
3195
3196 #[test]
3197 fn no_candidate_matches_returns_fallback() {
3198 let candidates = vec![("myhost".to_string(), SanIdentity::Dns("myhost".to_string()))];
3199 let sans = vec![SanIdentity::Dns("otherhost".to_string())];
3200 assert_eq!(pick_candidate(&candidates, &sans, "fallback"), "fallback");
3201 }
3202 }
3203}
3204
3205#[cfg(test)]
3206mod tests {
3207 use std::net::SocketAddr;
3208
3209 use hyperactor::channel::ChannelAddr;
3210 use hyperactor::id::Label;
3211 use hyperactor::testing::ids::test_proc_id_with_addr;
3212
3213 use super::*;
3214 use crate::mesh_id::ResourceId;
3215
3216 #[derive(Debug)]
3227 #[hyperactor::export(handlers = [])]
3228 struct TestIntrospectableActor;
3229 impl Actor for TestIntrospectableActor {}
3230
3231 #[test]
3236 fn test_build_root_payload() {
3237 let addr1: SocketAddr = "127.0.0.1:9001".parse().unwrap();
3238 let addr2: SocketAddr = "127.0.0.1:9002".parse().unwrap();
3239
3240 let proc1 = test_proc_id_with_addr(ChannelAddr::Tcp(addr1), "host1");
3241 let proc2 = test_proc_id_with_addr(ChannelAddr::Tcp(addr2), "host2");
3242
3243 let actor_id1 = proc1.actor_addr("mesh_agent");
3244 let actor_id2 = proc2.actor_addr("mesh_agent");
3245
3246 let ref1: ActorRef<HostAgent> = ActorRef::attest(actor_id1.clone());
3247 let ref2: ActorRef<HostAgent> = ActorRef::attest(actor_id2.clone());
3248
3249 let agent = MeshAdminAgent::new(
3250 vec![("host_a".to_string(), ref1), ("host_b".to_string(), ref2)],
3251 None,
3252 None,
3253 None,
3254 );
3255
3256 let payload = agent.build_root_payload();
3257 assert_eq!(payload.identity, crate::introspect::NodeRef::Root);
3258 assert_eq!(payload.parent, None);
3259 assert!(matches!(
3260 payload.properties,
3261 NodeProperties::Root { num_hosts: 2, .. }
3262 ));
3263 assert_eq!(payload.children.len(), 2);
3264 assert!(
3265 payload
3266 .children
3267 .contains(&crate::introspect::NodeRef::Host(actor_id1.clone()))
3268 );
3269 assert!(
3270 payload
3271 .children
3272 .contains(&crate::introspect::NodeRef::Host(actor_id2.clone()))
3273 );
3274
3275 match &payload.properties {
3277 NodeProperties::Root {
3278 num_hosts,
3279 started_by,
3280 system_children,
3281 ..
3282 } => {
3283 assert_eq!(*num_hosts, 2);
3284 assert!(!started_by.is_empty());
3285 assert!(
3287 system_children.is_empty(),
3288 "LC-1: root system_children must be empty"
3289 );
3290 }
3291 other => panic!("expected Root, got {:?}", other),
3292 }
3293 }
3294
3295 #[tokio::test]
3301 async fn test_resolve_reference_tree_walk() {
3302 use hyperactor::Proc;
3303 use hyperactor::channel::ChannelTransport;
3304
3305 use crate::host::Host;
3306 use crate::host::LocalProcManager;
3307 use crate::host_mesh::host_agent::ProcManagerSpawnFn;
3308 use crate::proc_agent::ProcAgent;
3309
3310 let spawn: ProcManagerSpawnFn =
3314 Box::new(|proc| Box::pin(std::future::ready(ProcAgent::boot_v1(proc, None))));
3315 let manager: LocalProcManager<ProcManagerSpawnFn> = LocalProcManager::new(spawn);
3316 let host: Host<LocalProcManager<ProcManagerSpawnFn>> =
3317 Host::new(manager, ChannelTransport::Unix.any())
3318 .await
3319 .unwrap();
3320 let host_addr = host.addr().clone();
3321 let system_proc = host.system_proc().clone();
3322 let host_agent_handle = system_proc.spawn_with_label(
3323 crate::host_mesh::host_agent::HOST_MESH_AGENT_ACTOR_NAME,
3324 HostAgent::new_local(host),
3325 );
3326 HostAgent::wait_initialized(&host_agent_handle)
3327 .await
3328 .unwrap();
3329 let host_agent_ref: ActorRef<HostAgent> = host_agent_handle.bind();
3330 let host_addr_str = host_addr.to_string();
3331
3332 let admin_proc = Proc::direct(ChannelTransport::Unix.any(), "admin".to_string()).unwrap();
3337 use hyperactor::testing::proc_supervison::ProcSupervisionCoordinator;
3340 let _supervision = ProcSupervisionCoordinator::set(&admin_proc).await.unwrap();
3341 let admin_handle = admin_proc.spawn_with_label(
3342 MESH_ADMIN_ACTOR_NAME,
3343 MeshAdminAgent::new(
3344 vec![(host_addr_str.clone(), host_agent_ref.clone())],
3345 None,
3346 Some("[::]:0".parse().unwrap()),
3347 None,
3348 ),
3349 );
3350 let admin_ref: ActorRef<MeshAdminAgent> = admin_handle.bind();
3351
3352 let client_proc = Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
3356 let client = client_proc.client("client");
3357
3358 let root_resp = admin_ref
3360 .resolve(&client, "root".to_string())
3361 .await
3362 .unwrap();
3363 let root = root_resp.0.unwrap();
3364 assert_eq!(root.identity, crate::introspect::NodeRef::Root);
3365 assert!(matches!(
3366 root.properties,
3367 NodeProperties::Root { num_hosts: 1, .. }
3368 ));
3369 assert_eq!(root.parent, None);
3370 assert_eq!(root.children.len(), 1); let expected_host_ref =
3374 crate::introspect::NodeRef::Host(host_agent_ref.actor_addr().clone());
3375 let host_child_ref = root
3376 .children
3377 .iter()
3378 .find(|c| **c == expected_host_ref)
3379 .expect("root children should contain the host agent (as Host ref)");
3380 let host_ref_string = host_child_ref.to_string();
3381 let host_resp = admin_ref.resolve(&client, host_ref_string).await.unwrap();
3382 let host_node = host_resp.0.unwrap();
3383 assert_eq!(host_node.identity, expected_host_ref);
3384 assert!(
3385 matches!(host_node.properties, NodeProperties::Host { .. }),
3386 "expected Host properties, got {:?}",
3387 host_node.properties
3388 );
3389 assert_eq!(host_node.parent, Some(crate::introspect::NodeRef::Root));
3390 assert!(
3391 !host_node.children.is_empty(),
3392 "host should have at least one proc child"
3393 );
3394 match &host_node.properties {
3396 NodeProperties::Host {
3397 system_children, ..
3398 } => {
3399 assert!(
3400 system_children.is_empty(),
3401 "LC-2: host system_children must be empty"
3402 );
3403 }
3404 other => panic!("expected Host, got {:?}", other),
3405 }
3406
3407 let proc_ref = &host_node.children[0];
3409 let proc_ref_str = proc_ref.to_string();
3410 let proc_resp = admin_ref.resolve(&client, proc_ref_str).await.unwrap();
3411 let proc_node = proc_resp.0.unwrap();
3412 assert!(
3413 matches!(proc_node.properties, NodeProperties::Proc { .. }),
3414 "expected Proc properties, got {:?}",
3415 proc_node.properties
3416 );
3417 assert_eq!(proc_node.parent, Some(expected_host_ref.clone()));
3418 assert!(
3420 !proc_node.children.is_empty(),
3421 "proc should have at least one actor child"
3422 );
3423
3424 let host_agent_node_ref =
3434 crate::introspect::NodeRef::Actor(host_agent_ref.actor_addr().clone());
3435 assert!(
3436 proc_node.children.contains(&host_agent_node_ref),
3437 "system proc children {:?} should contain the host agent {:?}",
3438 proc_node.children,
3439 host_agent_node_ref
3440 );
3441
3442 let xref_resp = admin_ref
3444 .resolve(&client, host_agent_ref.actor_addr().to_string())
3445 .await
3446 .unwrap();
3447 let xref_node = xref_resp.0.unwrap();
3448
3449 assert!(
3452 matches!(xref_node.properties, NodeProperties::Actor { .. }),
3453 "host agent child resolved as plain actor should be Actor, got {:?}",
3454 xref_node.properties
3455 );
3456 }
3457
3458 #[tokio::test]
3463 async fn test_proc_properties_for_all_procs() {
3464 use std::time::Duration;
3465
3466 use hyperactor::Proc;
3467 use hyperactor::channel::ChannelTransport;
3468 use hyperactor::id::Label;
3469
3470 use crate::host::Host;
3471 use crate::host::LocalProcManager;
3472 use crate::host_mesh::host_agent::ProcManagerSpawnFn;
3473 use crate::proc_agent::ProcAgent;
3474 use crate::resource;
3475 use crate::resource::ProcSpec;
3476 use crate::resource::Rank;
3477
3478 let spawn: ProcManagerSpawnFn =
3480 Box::new(|proc| Box::pin(std::future::ready(ProcAgent::boot_v1(proc, None))));
3481 let manager: LocalProcManager<ProcManagerSpawnFn> = LocalProcManager::new(spawn);
3482 let host: Host<LocalProcManager<ProcManagerSpawnFn>> =
3483 Host::new(manager, ChannelTransport::Unix.any())
3484 .await
3485 .unwrap();
3486 let host_addr = host.addr().clone();
3487 let system_proc_id: ProcAddr = host.system_proc().proc_addr().clone();
3488 let local_proc_id: ProcAddr = host.local_proc().proc_addr().clone();
3489 let system_proc = host.system_proc().clone();
3490 let host_agent_handle = system_proc.spawn_with_label(
3491 crate::host_mesh::host_agent::HOST_MESH_AGENT_ACTOR_NAME,
3492 HostAgent::new_local(host),
3493 );
3494 HostAgent::wait_initialized(&host_agent_handle)
3495 .await
3496 .unwrap();
3497 let host_agent_ref: ActorRef<HostAgent> = host_agent_handle.bind();
3498 let host_addr_str = host_addr.to_string();
3499
3500 let admin_proc = Proc::direct(ChannelTransport::Unix.any(), "admin".to_string()).unwrap();
3504 use hyperactor::testing::proc_supervison::ProcSupervisionCoordinator;
3505 let _supervision = ProcSupervisionCoordinator::set(&admin_proc).await.unwrap();
3506 let admin_handle = admin_proc.spawn_with_label(
3507 MESH_ADMIN_ACTOR_NAME,
3508 MeshAdminAgent::new(
3509 vec![(host_addr_str.clone(), host_agent_ref.clone())],
3510 None,
3511 Some("[::]:0".parse().unwrap()),
3512 None,
3513 ),
3514 );
3515 let admin_ref: ActorRef<MeshAdminAgent> = admin_handle.bind();
3516
3517 let client_proc = Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
3519 let client = client_proc.client("client");
3520
3521 let user_proc_name = ResourceId::instance(Label::new("user-proc").unwrap());
3523 host_agent_ref.post(
3524 &client,
3525 resource::CreateOrUpdate {
3526 id: user_proc_name.clone(),
3527 rank: Rank::new(0),
3528 spec: ProcSpec::default(),
3529 },
3530 );
3531
3532 tokio::time::sleep(Duration::from_secs(2)).await;
3534
3535 let host_ref_string =
3537 crate::introspect::NodeRef::Host(host_agent_ref.actor_addr().clone()).to_string();
3538 let host_resp = admin_ref.resolve(&client, host_ref_string).await.unwrap();
3539 let host_node = host_resp.0.unwrap();
3540
3541 assert!(
3544 host_node.children.len() >= 3,
3545 "expected at least 3 proc children (2 system + 1 user), got {}",
3546 host_node.children.len()
3547 );
3548
3549 let mut found_system = false;
3551 let mut found_user = false;
3552 for child_ref in &host_node.children {
3553 let resp = admin_ref
3554 .resolve(&client, child_ref.to_string())
3555 .await
3556 .unwrap();
3557 let node = resp.0.unwrap();
3558 if let NodeProperties::Proc { .. } = &node.properties {
3559 if matches!(
3560 child_ref,
3561 crate::introspect::NodeRef::Proc(proc_id)
3562 if *proc_id != system_proc_id && *proc_id != local_proc_id
3563 ) {
3564 found_user = true;
3565 } else {
3566 found_system = true;
3567 }
3568 } else {
3570 }
3572 }
3573 assert!(
3574 found_system,
3575 "should have resolved at least one system proc"
3576 );
3577 assert!(found_user, "should have resolved the user proc");
3578 }
3579
3580 #[test]
3584 fn test_build_root_payload_with_root_client() {
3585 let addr1: SocketAddr = "127.0.0.1:9001".parse().unwrap();
3586 let proc1 = ResourceId::proc_addr_from_name(ChannelAddr::Tcp(addr1), "host1");
3587 let actor_id1 = hyperactor::ActorAddr::root(proc1, Label::new("mesh_agent").unwrap());
3588 let ref1: ActorRef<HostAgent> = ActorRef::attest(actor_id1.clone());
3589
3590 let client_proc_id = ResourceId::proc_addr_from_name(ChannelAddr::Tcp(addr1), "local");
3591 let client_actor_id = client_proc_id.actor_addr("client");
3592
3593 let agent = MeshAdminAgent::new(
3594 vec![("host_a".to_string(), ref1)],
3595 Some(client_actor_id.clone()),
3596 None,
3597 None,
3598 );
3599
3600 let payload = agent.build_root_payload();
3601 assert!(matches!(
3602 payload.properties,
3603 NodeProperties::Root { num_hosts: 1, .. }
3604 ));
3605 assert_eq!(payload.children.len(), 1);
3607 assert!(
3608 payload
3609 .children
3610 .contains(&crate::introspect::NodeRef::Host(actor_id1.clone()))
3611 );
3612 }
3613
3614 #[tokio::test]
3618 async fn test_resolve_root_client_actor() {
3619 use hyperactor::channel::ChannelTransport;
3620
3621 use crate::host::Host;
3622 use crate::host::LocalProcManager;
3623 use crate::host_mesh::host_agent::ProcManagerSpawnFn;
3624 use crate::proc_agent::ProcAgent;
3625
3626 let spawn: ProcManagerSpawnFn =
3628 Box::new(|proc| Box::pin(std::future::ready(ProcAgent::boot_v1(proc, None))));
3629 let manager: LocalProcManager<ProcManagerSpawnFn> = LocalProcManager::new(spawn);
3630 let host: Host<LocalProcManager<ProcManagerSpawnFn>> =
3631 Host::new(manager, ChannelTransport::Unix.any())
3632 .await
3633 .unwrap();
3634 let host_addr = host.addr().clone();
3635 let system_proc = host.system_proc().clone();
3636
3637 let local_proc = host.local_proc();
3640 let local_proc_id = local_proc.proc_addr().clone();
3641 let root_client_handle = local_proc.spawn_with_label("client", TestIntrospectableActor);
3642 let root_client_ref: ActorRef<TestIntrospectableActor> = root_client_handle.bind();
3643 let root_client_actor_id = root_client_ref.actor_addr().clone();
3644
3645 let host_agent_handle = system_proc.spawn_with_label(
3646 crate::host_mesh::host_agent::HOST_MESH_AGENT_ACTOR_NAME,
3647 HostAgent::new_local(host),
3648 );
3649 HostAgent::wait_initialized(&host_agent_handle)
3650 .await
3651 .unwrap();
3652 let host_agent_ref: ActorRef<HostAgent> = host_agent_handle.bind();
3653 let host_addr_str = host_addr.to_string();
3654
3655 let admin_proc =
3660 hyperactor::Proc::direct(ChannelTransport::Unix.any(), "admin".to_string()).unwrap();
3661 use hyperactor::testing::proc_supervison::ProcSupervisionCoordinator;
3662 let _supervision = ProcSupervisionCoordinator::set(&admin_proc).await.unwrap();
3663 let admin_handle = admin_proc.spawn_with_label(
3664 MESH_ADMIN_ACTOR_NAME,
3665 MeshAdminAgent::new(
3666 vec![(host_addr_str.clone(), host_agent_ref.clone())],
3667 Some(root_client_actor_id.clone()),
3668 Some("[::]:0".parse().unwrap()),
3669 None,
3670 ),
3671 );
3672 let admin_ref: ActorRef<MeshAdminAgent> = admin_handle.bind();
3673
3674 let client_proc =
3676 hyperactor::Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
3677 let client = client_proc.client("client");
3678
3679 let root_resp = admin_ref
3681 .resolve(&client, "root".to_string())
3682 .await
3683 .unwrap();
3684 let root = root_resp.0.unwrap();
3685 let host_node_ref = crate::introspect::NodeRef::Host(host_agent_ref.actor_addr().clone());
3686 assert!(
3687 root.children.contains(&host_node_ref),
3688 "root children {:?} should contain host {:?}",
3689 root.children,
3690 host_node_ref
3691 );
3692
3693 let host_resp = admin_ref
3695 .resolve(&client, host_node_ref.to_string())
3696 .await
3697 .unwrap();
3698 let host_node = host_resp.0.unwrap();
3699 let local_proc_node_ref = crate::introspect::NodeRef::Proc(local_proc_id.clone());
3700 assert!(
3701 host_node.children.contains(&local_proc_node_ref),
3702 "host children {:?} should contain local proc {:?}",
3703 host_node.children,
3704 local_proc_node_ref
3705 );
3706
3707 let proc_resp = admin_ref
3709 .resolve(&client, local_proc_id.to_string())
3710 .await
3711 .unwrap();
3712 let proc_node = proc_resp.0.unwrap();
3713 assert!(
3714 matches!(proc_node.properties, NodeProperties::Proc { .. }),
3715 "expected Proc properties, got {:?}",
3716 proc_node.properties
3717 );
3718 let root_client_node_ref = crate::introspect::NodeRef::Actor(root_client_actor_id.clone());
3719 assert!(
3720 proc_node.children.contains(&root_client_node_ref),
3721 "local proc children {:?} should contain root client actor {:?}",
3722 proc_node.children,
3723 root_client_node_ref
3724 );
3725
3726 let client_resp = admin_ref
3728 .resolve(&client, root_client_actor_id.to_string())
3729 .await
3730 .unwrap();
3731 let client_node = client_resp.0.unwrap();
3732 assert!(
3733 matches!(client_node.properties, NodeProperties::Actor { .. }),
3734 "expected Actor properties, got {:?}",
3735 client_node.properties
3736 );
3737 assert_eq!(
3738 client_node.parent,
3739 Some(local_proc_node_ref),
3740 "root client parent should be the local proc"
3741 );
3742 }
3743
3744 #[test]
3748 fn test_skill_md_contains_canonical_strings() {
3749 let template = SKILL_MD_TEMPLATE;
3750 assert!(
3751 template.contains("GET {base}/v1/root"),
3752 "SKILL.md must document the root endpoint"
3753 );
3754 assert!(
3755 template.contains("GET {base}/v1/{reference}"),
3756 "SKILL.md must document the reference endpoint"
3757 );
3758 assert!(
3759 template.contains("NodePayload"),
3760 "SKILL.md must mention the NodePayload response type"
3761 );
3762 assert!(
3763 template.contains("GET {base}/SKILL.md"),
3764 "SKILL.md must document itself"
3765 );
3766 assert!(
3767 template.contains("{base}"),
3768 "SKILL.md must use {{base}} placeholder for interpolation"
3769 );
3770 }
3771
3772 #[tokio::test]
3781 async fn test_navigation_identity_invariant() {
3782 use hyperactor::Proc;
3783 use hyperactor::channel::ChannelTransport;
3784
3785 use crate::host::Host;
3786 use crate::host::LocalProcManager;
3787 use crate::host_mesh::host_agent::ProcManagerSpawnFn;
3788 use crate::proc_agent::ProcAgent;
3789
3790 let spawn: ProcManagerSpawnFn =
3792 Box::new(|proc| Box::pin(std::future::ready(ProcAgent::boot_v1(proc, None))));
3793 let manager: LocalProcManager<ProcManagerSpawnFn> = LocalProcManager::new(spawn);
3794 let host: Host<LocalProcManager<ProcManagerSpawnFn>> =
3795 Host::new(manager, ChannelTransport::Unix.any())
3796 .await
3797 .unwrap();
3798 let host_addr = host.addr().clone();
3799 let system_proc = host.system_proc().clone();
3800 let host_agent_handle = system_proc.spawn_with_label(
3801 crate::host_mesh::host_agent::HOST_MESH_AGENT_ACTOR_NAME,
3802 HostAgent::new_local(host),
3803 );
3804 HostAgent::wait_initialized(&host_agent_handle)
3805 .await
3806 .unwrap();
3807 let host_agent_ref: ActorRef<HostAgent> = host_agent_handle.bind();
3808 let host_addr_str = host_addr.to_string();
3809
3810 let admin_proc = Proc::direct(ChannelTransport::Unix.any(), "admin".to_string()).unwrap();
3814 use hyperactor::testing::proc_supervison::ProcSupervisionCoordinator;
3815 let _supervision = ProcSupervisionCoordinator::set(&admin_proc).await.unwrap();
3816 let admin_handle = admin_proc.spawn_with_label(
3817 MESH_ADMIN_ACTOR_NAME,
3818 MeshAdminAgent::new(
3819 vec![(host_addr_str, host_agent_ref)],
3820 None,
3821 Some("[::]:0".parse().unwrap()),
3822 None,
3823 ),
3824 );
3825 let admin_ref: ActorRef<MeshAdminAgent> = admin_handle.bind();
3826
3827 let client_proc = Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
3828 let client = client_proc.client("client");
3829
3830 let mut queue: std::collections::VecDeque<(String, Option<crate::introspect::NodeRef>)> =
3833 std::collections::VecDeque::new();
3834 queue.push_back(("root".to_string(), None));
3835
3836 let mut visited = std::collections::HashSet::new();
3837 while let Some((ref_str, expected_parent)) = queue.pop_front() {
3838 if !visited.insert(ref_str.clone()) {
3839 continue;
3840 }
3841
3842 let resp = admin_ref.resolve(&client, ref_str.clone()).await.unwrap();
3843 let node = resp.0.unwrap();
3844
3845 assert_eq!(
3847 node.identity.to_string(),
3848 ref_str,
3849 "identity mismatch: resolved '{}' but payload.identity = '{}'",
3850 ref_str,
3851 node.identity
3852 );
3853
3854 assert_eq!(
3856 node.parent, expected_parent,
3857 "parent mismatch for '{}': expected {:?}, got {:?}",
3858 ref_str, expected_parent, node.parent
3859 );
3860
3861 for child_ref in &node.children {
3864 let child_str = child_ref.to_string();
3865 if !visited.contains(&child_str) {
3866 queue.push_back((child_str, Some(node.identity.clone())));
3867 }
3868 }
3869 }
3870
3871 assert!(
3874 visited.len() >= 4,
3875 "expected at least 4 nodes in the tree, visited {}",
3876 visited.len()
3877 );
3878 }
3879
3880 #[tokio::test]
3882 async fn test_system_proc_identity() {
3883 use hyperactor::Proc;
3884 use hyperactor::channel::ChannelTransport;
3885
3886 use crate::host::Host;
3887 use crate::host::LocalProcManager;
3888 use crate::host_mesh::host_agent::ProcManagerSpawnFn;
3889 use crate::proc_agent::ProcAgent;
3890
3891 let spawn: ProcManagerSpawnFn =
3893 Box::new(|proc| Box::pin(std::future::ready(ProcAgent::boot_v1(proc, None))));
3894 let manager: LocalProcManager<ProcManagerSpawnFn> = LocalProcManager::new(spawn);
3895 let host: Host<LocalProcManager<ProcManagerSpawnFn>> =
3896 Host::new(manager, ChannelTransport::Unix.any())
3897 .await
3898 .unwrap();
3899 let host_addr = host.addr().clone();
3900 let system_proc = host.system_proc().clone();
3901 let system_proc_id = system_proc.proc_addr().clone();
3902 let host_agent_handle = system_proc.spawn_with_label(
3903 crate::host_mesh::host_agent::HOST_MESH_AGENT_ACTOR_NAME,
3904 HostAgent::new_local(host),
3905 );
3906 HostAgent::wait_initialized(&host_agent_handle)
3907 .await
3908 .unwrap();
3909 let host_agent_ref: ActorRef<HostAgent> = host_agent_handle.bind();
3910 let host_addr_str = host_addr.to_string();
3911
3912 let admin_proc = Proc::direct(ChannelTransport::Unix.any(), "admin".to_string()).unwrap();
3917 use hyperactor::testing::proc_supervison::ProcSupervisionCoordinator;
3918 let _supervision = ProcSupervisionCoordinator::set(&admin_proc).await.unwrap();
3919 let admin_handle = admin_proc.spawn_with_label(
3920 MESH_ADMIN_ACTOR_NAME,
3921 MeshAdminAgent::new(
3922 vec![(host_addr_str.clone(), host_agent_ref.clone())],
3923 None,
3924 Some("[::]:0".parse().unwrap()),
3925 None,
3926 ),
3927 );
3928 let admin_ref: ActorRef<MeshAdminAgent> = admin_handle.bind();
3929
3930 let client_proc = Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
3932 let client = client_proc.client("client");
3933
3934 let host_ref_str =
3936 crate::introspect::NodeRef::Host(host_agent_ref.actor_addr().clone()).to_string();
3937 let host_resp = admin_ref
3938 .resolve(&client, host_ref_str.clone())
3939 .await
3940 .unwrap();
3941 let host_node = host_resp.0.unwrap();
3942 assert!(
3943 !host_node.children.is_empty(),
3944 "host should have at least one proc child"
3945 );
3946
3947 let system_children = match &host_node.properties {
3949 NodeProperties::Host {
3950 system_children, ..
3951 } => system_children.clone(),
3952 other => panic!("expected Host properties, got {:?}", other),
3953 };
3954 assert!(
3956 system_children.is_empty(),
3957 "host system_children should be empty (procs are never system), got {:?}",
3958 system_children
3959 );
3960 assert!(
3962 matches!(&host_node.properties, NodeProperties::Host { .. }),
3963 "expected Host properties"
3964 );
3965
3966 let expected_system_ref = crate::introspect::NodeRef::Proc(system_proc_id.clone());
3968 assert!(
3969 host_node.children.contains(&expected_system_ref),
3970 "host children {:?} should contain the system proc ref {:?}",
3971 host_node.children,
3972 expected_system_ref
3973 );
3974
3975 let proc_child_ref = &host_node.children[0];
3977 let proc_resp = admin_ref
3978 .resolve(&client, proc_child_ref.to_string())
3979 .await
3980 .unwrap();
3981 let proc_node = proc_resp.0.unwrap();
3982
3983 assert_eq!(
3984 proc_node.identity, *proc_child_ref,
3985 "identity must match the proc ref from the host's children list"
3986 );
3987
3988 assert!(
3989 matches!(proc_node.properties, NodeProperties::Proc { .. }),
3990 "expected NodeProperties::Proc, got {:?}",
3991 proc_node.properties
3992 );
3993
3994 let host_node_ref = crate::introspect::NodeRef::Host(host_agent_ref.actor_addr().clone());
3995 assert_eq!(
3996 proc_node.parent,
3997 Some(host_node_ref),
3998 "proc parent should be the host reference"
3999 );
4000
4001 assert!(
4003 proc_node.as_of > std::time::UNIX_EPOCH,
4004 "as_of should be after the epoch"
4005 );
4006
4007 assert!(
4009 matches!(&proc_node.properties, NodeProperties::Proc { .. }),
4010 "expected Proc properties"
4011 );
4012 }
4013
4014 #[test]
4018 fn test_admin_handle_parse_https_url() {
4019 let h = super::AdminHandle::parse("https://myhost:1729");
4020 assert!(matches!(h, super::AdminHandle::Url(u) if u == "https://myhost:1729"));
4021 }
4022
4023 #[test]
4024 fn test_admin_handle_parse_bare_host_port() {
4025 let h = super::AdminHandle::parse("myhost:1729");
4027 assert!(
4028 matches!(h, super::AdminHandle::Url(ref u) if u == "https://myhost:1729"),
4029 "bare host:port should become https://host:port, got: {:?}",
4030 matches!(h, super::AdminHandle::Url(_))
4031 );
4032 }
4033
4034 #[test]
4035 fn test_admin_handle_parse_mast() {
4036 let h = super::AdminHandle::parse("mast_conda:///my-job");
4037 assert!(matches!(
4038 h,
4039 super::AdminHandle::Published(super::PublishedHandle::Mast(_))
4040 ));
4041 }
4042
4043 #[test]
4044 fn test_admin_handle_parse_unsupported() {
4045 let h = super::AdminHandle::parse("junk_hostname_no_port");
4047 assert!(matches!(h, super::AdminHandle::Unsupported(_)));
4048 }
4049
4050 #[tokio::test]
4051 async fn test_admin_handle_resolve_url_returns_url() {
4052 let h = super::AdminHandle::parse("https://myhost:1729");
4053 let result = h.resolve(None).await.unwrap();
4054 assert_eq!(result, "https://myhost:1729");
4055 }
4056
4057 #[tokio::test]
4058 async fn test_admin_handle_resolve_published_returns_error() {
4059 let h = super::AdminHandle::parse("mast_conda:///test-job");
4060 let err = format!("{:#}", h.resolve(Some(1729)).await.unwrap_err());
4061 assert!(
4062 err.contains("not yet implemented"),
4063 "expected 'not yet implemented' in error, got: {}",
4064 err
4065 );
4066 }
4067
4068 #[tokio::test]
4069 async fn test_admin_handle_resolve_unsupported_returns_error() {
4070 let h = super::AdminHandle::parse("junk_hostname_no_port");
4071 let err = format!("{:#}", h.resolve(None).await.unwrap_err());
4072 assert!(
4073 err.contains("unrecognized admin handle"),
4074 "expected 'unrecognized admin handle' in error, got: {}",
4075 err
4076 );
4077 }
4078
4079 #[tokio::test]
4082 async fn test_resolve_mast_handle_returns_not_yet_implemented_error() {
4083 let result = super::resolve_mast_handle("mast_conda:///test-job", Some(1729)).await;
4084 let err = format!("{:#}", result.unwrap_err());
4085 assert!(
4086 err.contains("not yet implemented"),
4087 "expected 'not yet implemented' in error, got: {}",
4088 err
4089 );
4090 }
4091
4092 #[test]
4096 fn test_admin_info_new_derives_host_from_url() {
4097 let info = super::AdminInfo::new(
4098 "actor".to_string(),
4099 "proc".to_string(),
4100 "https://myhost.example.com:1729".to_string(),
4101 )
4102 .unwrap();
4103 assert_eq!(info.host, "myhost.example.com");
4104 assert_eq!(info.url, "https://myhost.example.com:1729");
4105 }
4106
4107 #[test]
4109 fn test_admin_info_new_rejects_invalid_url() {
4110 let result = super::AdminInfo::new(
4111 "actor".to_string(),
4112 "proc".to_string(),
4113 "not a url".to_string(),
4114 );
4115 assert!(result.is_err(), "invalid URL must be rejected");
4116 }
4117
4118 #[test]
4120 fn test_admin_info_new_rejects_url_without_host() {
4121 let result = super::AdminInfo::new(
4123 "actor".to_string(),
4124 "proc".to_string(),
4125 "data:text/plain,hello".to_string(),
4126 );
4127 assert!(result.is_err(), "URL without host must be rejected");
4128 }
4129
4130 #[tokio::test]
4135 async fn test_spawn_admin_places_on_caller_proc() {
4136 use hyperactor::Proc;
4137 use hyperactor::channel::ChannelTransport;
4138 use hyperactor::testing::proc_supervison::ProcSupervisionCoordinator;
4139
4140 use crate::host_mesh::HostMesh;
4141
4142 let host_mesh = HostMesh::local().await.unwrap();
4144
4145 let caller_proc = Proc::direct(ChannelTransport::Unix.any(), "caller".to_string()).unwrap();
4147 let _supervision = ProcSupervisionCoordinator::set(&caller_proc).await.unwrap();
4148 let caller_cx = caller_proc.client("caller");
4149
4150 let admin_ref = crate::host_mesh::spawn_admin(
4152 [&host_mesh],
4153 &caller_cx,
4154 Some("[::]:0".parse().unwrap()),
4155 None,
4156 )
4157 .await
4158 .unwrap();
4159
4160 let admin_url = admin_ref
4164 .get_admin_addr(&caller_cx)
4165 .await
4166 .unwrap()
4167 .addr
4168 .expect("SA-5: admin must report an address");
4169 assert!(
4170 !admin_url.is_empty(),
4171 "spawn_admin ref must yield a non-empty URL"
4172 );
4173 }
4174
4175 #[tokio::test]
4193 async fn test_proc_children_reflect_directly_spawned_actors() {
4194 use hyperactor::Proc;
4195 use hyperactor::actor::ActorStatus;
4196 use hyperactor::channel::ChannelTransport;
4197 use hyperactor::testing::proc_supervison::ProcSupervisionCoordinator;
4198
4199 use crate::host::Host;
4200 use crate::host::LocalProcManager;
4201 use crate::host_mesh::host_agent::HOST_MESH_AGENT_ACTOR_NAME;
4202 use crate::host_mesh::host_agent::HostAgent;
4203 use crate::host_mesh::host_agent::ProcManagerSpawnFn;
4204 use crate::proc_agent::PROC_AGENT_ACTOR_NAME;
4205 use crate::proc_agent::ProcAgent;
4206
4207 let spawn_fn: ProcManagerSpawnFn =
4215 Box::new(|proc| Box::pin(std::future::ready(ProcAgent::boot_v1(proc, None))));
4216 let manager: LocalProcManager<ProcManagerSpawnFn> = LocalProcManager::new(spawn_fn);
4217 let host: Host<LocalProcManager<ProcManagerSpawnFn>> =
4218 Host::new(manager, ChannelTransport::Unix.any())
4219 .await
4220 .unwrap();
4221 let system_proc = host.system_proc().clone();
4222 let host_agent_handle =
4223 system_proc.spawn_with_label(HOST_MESH_AGENT_ACTOR_NAME, HostAgent::new_local(host));
4224 HostAgent::wait_initialized(&host_agent_handle)
4225 .await
4226 .unwrap();
4227 let host_agent_ref: ActorRef<HostAgent> = host_agent_handle.bind();
4228
4229 let user_proc =
4231 Proc::direct(ChannelTransport::Unix.any(), "user_proc".to_string()).unwrap();
4232 let user_proc_addr = user_proc.proc_addr().addr().to_string();
4233 let agent_handle = ProcAgent::boot_v1(user_proc.clone(), None).unwrap();
4234 agent_handle
4235 .status()
4236 .wait_for(|s| matches!(s, ActorStatus::Idle))
4237 .await
4238 .unwrap();
4239
4240 let admin_proc = Proc::direct(ChannelTransport::Unix.any(), "admin".to_string()).unwrap();
4246 let _supervision = ProcSupervisionCoordinator::set(&admin_proc).await.unwrap();
4247 let admin_handle = admin_proc.spawn_with_label(
4248 MESH_ADMIN_ACTOR_NAME,
4249 MeshAdminAgent::new(
4250 vec![(user_proc_addr, host_agent_ref.clone())],
4251 None,
4252 Some("[::]:0".parse().unwrap()),
4253 None,
4254 ),
4255 );
4256 let admin_ref: ActorRef<MeshAdminAgent> = admin_handle.bind();
4257
4258 let client_proc = Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
4259 let client = client_proc.client("client");
4260
4261 let user_proc_ref = user_proc.proc_addr().to_string();
4265 let resp = admin_ref
4266 .resolve(&client, user_proc_ref.clone())
4267 .await
4268 .unwrap();
4269 let node = resp.0.unwrap();
4270 assert!(
4271 matches!(node.properties, NodeProperties::Proc { .. }),
4272 "expected Proc, got {:?}",
4273 node.properties
4274 );
4275 let initial_count = node.children.len();
4276 assert!(
4277 node.children
4278 .iter()
4279 .any(|c| c.to_string().contains(PROC_AGENT_ACTOR_NAME)),
4280 "initial children {:?} should contain proc_agent",
4281 node.children
4282 );
4283
4284 user_proc.spawn_with_label("extra_actor", TestIntrospectableActor);
4287
4288 let resp2 = admin_ref
4291 .resolve(&client, user_proc_ref.clone())
4292 .await
4293 .unwrap();
4294 let node2 = resp2.0.unwrap();
4295 assert!(
4296 matches!(node2.properties, NodeProperties::Proc { .. }),
4297 "expected Proc, got {:?}",
4298 node2.properties
4299 );
4300 assert!(
4301 node2
4302 .children
4303 .iter()
4304 .any(|c| c.to_string().contains("extra_actor")),
4305 "after direct spawn, children {:?} should contain extra_actor",
4306 node2.children
4307 );
4308 assert!(
4309 node2.children.len() > initial_count,
4310 "expected at least {} children after direct spawn, got {:?}",
4311 initial_count + 1,
4312 node2.children
4313 );
4314 }
4315
4316 #[test]
4323 fn pyspy_parse_empty_reference() {
4324 let err = parse_proc_reference("").unwrap_err();
4326 assert_eq!(err.code, "bad_request");
4327 assert!(err.message.contains("empty"));
4328 }
4329
4330 #[test]
4331 fn pyspy_parse_slash_only() {
4332 let err = parse_proc_reference("/").unwrap_err();
4334 assert_eq!(err.code, "bad_request");
4335 assert!(err.message.contains("empty"));
4336 }
4337
4338 #[test]
4339 fn pyspy_parse_malformed_percent_encoding() {
4340 let err = parse_proc_reference("%FF%FE").unwrap_err();
4343 assert_eq!(err.code, "bad_request");
4344 assert!(err.message.contains("percent-encoding"));
4345 }
4346
4347 #[test]
4348 fn pyspy_parse_invalid_proc_id() {
4349 let err = parse_proc_reference("not-a-valid-proc-id").unwrap_err();
4351 assert_eq!(err.code, "bad_request");
4352 assert!(err.message.contains("invalid proc reference"));
4353 }
4354
4355 #[test]
4356 fn pyspy_parse_valid_proc_reference() {
4357 let addr: SocketAddr = "127.0.0.1:9000".parse().unwrap();
4359 let proc_id = test_proc_id_with_addr(ChannelAddr::Tcp(addr), "myproc");
4360 let proc_id_str = proc_id.to_string();
4361
4362 let (decoded, parsed) = parse_proc_reference(&proc_id_str).unwrap();
4363 assert_eq!(decoded, proc_id_str);
4364 assert_eq!(parsed, proc_id);
4365 }
4366
4367 #[test]
4368 fn pyspy_parse_strips_leading_slash() {
4369 let addr: SocketAddr = "127.0.0.1:9000".parse().unwrap();
4371 let proc_id = test_proc_id_with_addr(ChannelAddr::Tcp(addr), "myproc");
4372 let with_slash = format!("/{}", proc_id);
4373
4374 let (_, parsed) = parse_proc_reference(&with_slash).unwrap();
4375 assert_eq!(parsed, proc_id);
4376 }
4377
4378 #[test]
4380 fn route_proc_handler_service_proc_yields_host() {
4381 let addr: SocketAddr = "127.0.0.1:9000".parse().unwrap();
4382 let proc_id = ResourceId::proc_addr_from_name(ChannelAddr::Tcp(addr), SERVICE_PROC_NAME);
4383 let handler = route_proc_handler(&proc_id.to_string()).unwrap();
4384 assert!(
4385 matches!(handler, ResolvedProcHandler::Host(_)),
4386 "service proc should resolve to Host variant"
4387 );
4388 }
4389
4390 #[test]
4392 fn route_proc_handler_worker_proc_yields_proc() {
4393 let addr: SocketAddr = "127.0.0.1:9000".parse().unwrap();
4394 let proc_id = test_proc_id_with_addr(ChannelAddr::Tcp(addr), "worker_0");
4395 let handler = route_proc_handler(&proc_id.to_string()).unwrap();
4396 assert!(
4397 matches!(handler, ResolvedProcHandler::Proc(_)),
4398 "non-service proc should resolve to Proc variant"
4399 );
4400 }
4401
4402 #[test]
4404 fn route_proc_handler_service_instance_yields_proc() {
4405 let addr: SocketAddr = "127.0.0.1:9000".parse().unwrap();
4406 let proc_id =
4407 ResourceId::proc_addr_from_name(ChannelAddr::Tcp(addr), "service-deadbeefdeadbeef");
4408 let handler = route_proc_handler(&proc_id.to_string()).unwrap();
4409 assert!(
4410 matches!(handler, ResolvedProcHandler::Proc(_)),
4411 "service-labeled instance proc should resolve to Proc variant"
4412 );
4413 }
4414}