Skip to main content

hyperactor_mesh/
mesh_admin.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9//! Mesh-level admin surface for topology introspection and reference
10//! walking.
11//!
12//! This module defines `MeshAdminAgent`, an actor that exposes a
13//! uniform, reference-based HTTP API over an entire host mesh. Every
14//! addressable entity in the mesh is represented as a `NodePayload`
15//! and resolved via typed `NodeRef` references (parsed from HTTP
16//! path strings at the request boundary).
17//!
18//! Incoming HTTP requests are bridged into the actor message loop
19//! using `ResolveReferenceMessage`, ensuring that all topology
20//! resolution and data collection happens through actor messaging.
21//! The agent fans out to `HostAgent` instances to fetch host,
22//! proc, and actor details, then normalizes them into a single
23//! tree-shaped model (`NodeProperties` + children references)
24//! suitable for topology-agnostic clients such as the admin TUI.
25//!
26//! # Schema strategy
27//!
28//! The external API contract is schema-first: the JSON Schema
29//! (Draft 2020-12) served at `GET /v1/schema` is the
30//! authoritative definition of the response shape. The error
31//! envelope schema is at `GET /v1/schema/error`.
32//!
33//! Schema and OpenAPI are derived from the HTTP boundary DTO types
34//! in [`crate::introspect::dto`] (`NodePayloadDto`,
35//! `NodePropertiesDto`, `FailureInfoDto`) via
36//! `schemars::JsonSchema`. The domain types (`NodePayload`,
37//! `NodeProperties`, `FailureInfo`) do not carry `JsonSchema` —
38//! they own the typed internal model; the DTOs own the wire
39//! contract.
40//!
41//! This follows the "Admin Gateway Pattern" RFC
42//! ([doc](https://fburl.com/1dvah88uutaiyesebojouen2)):
43//! schema is the product; transports and tooling are projections.
44//!
45//! ## Schema generation pipeline
46//!
47//! 1. `#[derive(JsonSchema)]` on `NodePayloadDto`,
48//!    `NodePropertiesDto`, `FailureInfoDto`, `ApiError`,
49//!    `ApiErrorEnvelope`.
50//! 2. `schemars::schema_for!(T)` produces a `Schema` value at
51//!    runtime (Draft 2020-12).
52//! 3. The `serve_schema` / `serve_error_schema` handlers inject a
53//!    `$id` field (SC-4) and serve the result as JSON.
54//! 4. Snapshot tests in `introspect::tests` compare the raw
55//!    schemars output (without `$id`) against checked-in golden
56//!    files to detect drift (SC-2).
57//! 5. Validation tests construct domain payloads, convert to DTOs,
58//!    and confirm the serialized DTOs pass schema validation
59//!    (SC-3).
60//!
61//! ## Regenerating snapshots
62//!
63//! After intentional changes to the DTO types
64//! (`NodePayloadDto`, `NodePropertiesDto`, `FailureInfoDto`),
65//! `ApiError`, or `ApiErrorEnvelope`, regenerate the golden
66//! files:
67//!
68//! ```sh
69//! buck run fbcode//monarch/hyperactor_mesh:generate_api_artifacts \
70//!   @fbcode//mode/dev-nosan -- \
71//!   fbcode/monarch/hyperactor_mesh/src/testdata
72//! ```
73//!
74//! Or via cargo:
75//!
76//! ```sh
77//! cargo run -p hyperactor_mesh --bin generate_api_artifacts -- \
78//!   hyperactor_mesh/src/testdata
79//! ```
80//!
81//! Then re-run tests to confirm the new snapshot passes.
82//!
83//! ## Schema invariants (SC-*)
84//!
85//! - **SC-1 (schema-derived):** Schema is derived from the DTO
86//!   types via `schemars::JsonSchema`, not hand-written.
87//! - **SC-2 (schema-snapshot-stability):** Schema changes must
88//!   be explicit — a snapshot test catches unintentional drift.
89//! - **SC-3 (schema-payload-conformance):** Domain payloads
90//!   converted to DTOs validate against the generated schema.
91//! - **SC-4 (schema-version-identity):** Served schemas carry a
92//!   `$id` tied to the API version (e.g.
93//!   `https://monarch.meta.com/schemas/v1/node_payload`).
94//! - **SC-5 (route-precedence):** Literal schema routes are
95//!   matched by specificity before the `{*reference}` wildcard
96//!   (axum 0.8 specificity-based routing).
97//!
98//! Note on `ApiError.details`: the derived schema is maximally
99//! permissive for `details` (any valid JSON). This is intentional
100//! for v1 — `details` is a domain-specific escape hatch.
101//! Consumers must not assume a fixed shape.
102//!
103//! # Introspection visibility policy
104//!
105//! Admin tooling only displays **introspectable** nodes: entities
106//! that are reachable via actor messaging and respond to
107//! [`IntrospectMessage`]. Infrastructure procs that are
108//! **non-routable** are intentionally **opaque** to introspection and
109//! are omitted from the navigation graph.
110//!
111//! ## Definitions
112//!
113//! **Routable** — an entity is routable if the system can address it
114//! via the routing layer and successfully deliver a message to it
115//! using a `Addr` / `ActorAddr` (i.e., there exists a live mailbox
116//! sender reachable through normal routing). Practical test: "can I
117//! send `IntrospectMessage::Query` to it and get a reply?"
118//!
119//! **Non-routable** — an entity is non-routable if it has no
120//! externally reachable mailbox sender in the routing layer, so
121//! message delivery is impossible by construction (even if you know
122//! its name). Examples: `hyperactor_runtime[0]`, `mailbox_server[N]`,
123//! `local[N]` — these use `PanickingMailboxSender` and are never
124//! bound to the router.
125//!
126//! **Introspectable** — tooling can obtain a `NodePayload` for this
127//! node by sending `IntrospectMessage` to a routable actor.
128//!
129//! **Opaque** — the node exists but is not introspectable via
130//! messaging; tooling cannot observe it through the introspection
131//! protocol.
132//!
133//! ## Proc visibility
134//!
135//! A proc is not directly introspected; actors are. Tooling
136//! synthesizes proc-level nodes by grouping introspectable actors by
137//! `ProcAddr`.
138//!
139//! A proc is visible iff there exists at least one actor on that proc
140//! whose `ActorAddr` is deliverable via the routing layer (i.e., the
141//! actor has a bound mailbox sender reachable through normal routing)
142//! and responds to `IntrospectMessage`.
143//!
144//! The rule is: **if an entity is routable via the mesh routing layer
145//! (i.e., tooling can deliver `IntrospectMessage::Query` to one of its
146//! actors), then it is introspectable and appears in the admin graph.**
147//!
148//! ## Navigation identity invariants (NI-*)
149//!
150//! Every `NodePayload` in the topology tree satisfies:
151//!
152//! - **NI-1 (identity = reference):** A node's `identity: NodeRef`
153//!   must correspond to the reference used to resolve it. The
154//!   display form of `identity` round-trips through `NodeRef::from_str`.
155//!
156//! - **NI-2 (parent = containment parent):** A node's
157//!   `parent: Option<NodeRef>` records its canonical containment
158//!   parent, not the inverse of every navigation edge. Specifically:
159//!   root → `None`, host → `Root`, proc → `Host(…)`,
160//!   actor → `Proc(…)`. An actor's parent is always its owning proc,
161//!   even when the actor also appears as a child of another actor via
162//!   supervision.
163//!
164//! - **NI-3 (children = navigation graph):** A node's `children`
165//!   is the admin navigation graph. Actor-to-actor supervision links
166//!   coexist with proc→actor membership links without changing
167//!   `parent`. The same actor may therefore appear in `children` of
168//!   both its proc and its supervising actor.
169//!
170//! Together these ensure that the TUI can correlate responses to tree
171//! nodes, and that upward/downward navigation is consistent.
172//!
173//! ## Link-classification invariants (LC-*)
174//!
175//! These describe which nodes emit `system_children` and
176//! `stopped_children` classification sets, and what those sets
177//! contain.
178//!
179//! - **LC-1 (root system_children empty):** Root payloads always
180//!   emit `system_children: vec![]`. Root children are host nodes,
181//!   which are not classified as system.
182//!
183//! - **LC-2 (host system_children empty):** Host payloads always
184//!   emit `system_children: vec![]`. Host children are procs, which
185//!   are not classified as system — only actors carry the system
186//!   classification.
187//!
188//! - **LC-3 (proc system_children subset):** Proc payloads emit
189//!   `system_children ⊆ children`, containing only `NodeRef::Actor`
190//!   refs where `cell.is_system()` is true.
191//!
192//! - **LC-4 (proc stopped_children subset):** Proc payloads emit
193//!   `stopped_children ⊆ children`, containing only
194//!   `NodeRef::Actor` refs for terminated actors retained for
195//!   post-mortem inspection.
196//!
197//! - **LC-5 (actor/error no classification sets):** Actor and Error
198//!   payloads do not carry `system_children` or `stopped_children`.
199//!
200//! ## Proc-resolution invariants (SP-*)
201//!
202//! When a proc reference is resolved, the returned `NodePayload`
203//! satisfies:
204//!
205//! - **SP-1 (identity):** The identity matches the ProcAddr reference
206//!   from the parent's children list.
207//! - **SP-2 (properties):** The properties are `NodeProperties::Proc`.
208//! - **SP-3 (parent):** The parent is `NodeRef::Host(actor_id)`.
209//! - **SP-4 (as_of):** The `as_of` field is present and valid
210//!   (internally `SystemTime`; serialized as ISO 8601 string over
211//!   the HTTP JSON API per HB-1).
212//!
213//! Enforced by `test_system_proc_identity`.
214//!
215//! ## Proc-agent invariants (PA-*)
216//!
217//! - **PA-1 (live children):** Proc-node children used by admin/TUI
218//!   must be derived from live proc state at query time. No
219//!   additional publish event is required for a newly spawned actor
220//!   to appear.
221//!
222//! Enforced by `test_proc_children_reflect_directly_spawned_actors`.
223//!
224//! ## Robustness invariant (MA-R1)
225//!
226//! - **MA-R1 (no-crash):** `MeshAdminAgent` must never crash the OS
227//!   process it resides in. Every handler catches errors and converts
228//!   them into structured error payloads
229//!   (`ResolveReferenceResponse(Err(..))`, `NodeProperties::Error`,
230//!   etc.) rather than propagating panics or unwinding. Failed reply
231//!   sends (the caller went away) are silently swallowed.
232//!
233//! ## TLS transport invariant (MA-T1)
234//!
235//! - **MA-T1 (tls):** At Meta (`fbcode_build`), the admin HTTP
236//!   server **requires** mutual TLS. At startup it probes for
237//!   certificates via `try_tls_acceptor` with client cert
238//!   enforcement enabled. If no usable certificate bundle is found,
239//!   `init()` returns an error — no plain HTTP fallback. In OSS,
240//!   TLS is best-effort with plain HTTP fallback.
241//!
242//! - **MA-T2 (scheme-in-url):** The URL returned by `GetAdminAddr`
243//!   is always `https://host:port` or `http://host:port`, never a
244//!   bare `host:port`. All callers receive and use this full URL
245//!   directly.
246//!
247//! ## Client host invariants (CH-*)
248//!
249//! Let **A** denote the aggregated host set (the union of hosts
250//! from all meshes passed to [`host_mesh::spawn_admin`],
251//! deduplicated by `HostAgent` `ActorAddr` — see SA-3), and let
252//! **C** denote the process-global singleton client host mesh in
253//! the caller process (whose local proc hosts the root client
254//! actor).
255//!
256//! - **CH-1 (deduplication):** When C ∈ A, the client host appears
257//!   exactly once in the admin host list (deduplicated by `HostAgent`
258//!   `ActorAddr` identity). When C ∉ A, `spawn_admin` includes C
259//!   alongside A's hosts so the admin introspects C as a normal host
260//!   subtree, not as a standalone proc.
261//!
262//! - **CH-2 (reachability):** In both cases, the root client actor
263//!   is reachable through the standard host → proc → actor walk.
264//!
265//! - **CH-3 (ordering):** C must be initialized before
266//!   `spawn_admin` executes. In Rust, calling `context()` /
267//!   `this_host()` / `this_proc()` triggers `GLOBAL_CONTEXT`
268//!   bootstrap, which initializes C. In Python, `bootstrap_host()`
269//!   calls `register_client_host()` before any actor code runs.
270//!   Either path ensures C is available by the time `spawn_admin`
271//!   reads it via `try_this_host()`. Any refactor must preserve
272//!   this ordering.
273//!
274//! - **CH-4 (runtime-agnostic client-host discovery):** `spawn_admin`
275//!   discovers C via `try_this_host()`, which checks two sources
276//!   in order: the Rust `GLOBAL_CONTEXT` (initialized via
277//!   `context()` / `this_host()` / `this_proc()`) and the
278//!   externally registered client host (set by
279//!   `register_client_host()` from Python's `bootstrap_host()`).
280//!   Aggregation logic must not branch on which source provided C.
281//!
282//! **Mechanism:** [`host_mesh::spawn_admin`] aggregates hosts from
283//! all input meshes (SA-3), reads C from the caller process (via
284//! `try_this_host()`), merges it with the aggregated set (SA-6),
285//! deduplicates by `HostAgent` `ActorAddr`, and spawns the
286//! `MeshAdminAgent` on the caller's local proc via
287//! `cx.instance().proc().spawn(...)`. Placement now follows the
288//! caller context rather than mesh topology.
289//!
290//! ## Spawn/aggregation invariants (SA-*)
291//!
292//! [`host_mesh::spawn_admin`] aggregates hosts from one or more
293//! meshes into a single admin host set.
294//!
295//! - **SA-1 (non-empty mesh set):** The input must yield at least
296//!   one mesh.
297//! - **SA-2 (non-empty hosts):** Every input mesh must contain at
298//!   least one host.
299//! - **SA-3 (host-agent identity dedup):** The admin host set is
300//!   the ordered union of host agents from all input meshes,
301//!   deduplicated by `HostAgent` `ActorAddr` in first-seen order.
302//! - **SA-4 (single-mesh degeneracy):** `spawn_admin([mesh], ...)`
303//!   is behaviorally equivalent to the former `mesh.spawn_admin(...)`.
304//!   Established by existing single-mesh integration tests (e.g.
305//!   `dining_philosophers`); no dedicated unit test.
306//! - **SA-5 (caller-local placement):** The admin is spawned on the
307//!   caller's local proc — the `Proc` of the actor context passed to
308//!   `spawn_admin()`. In common remote launch flows, the caller is
309//!   typically the root client/control process.
310//! - **SA-6 (client-host merge after aggregation):** Client-host
311//!   inclusion/dedup (CH-1) operates on the already-aggregated host
312//!   set, not per-mesh independently.
313//!
314//! ## MAST resolution (disabled)
315//!
316//! `mast_conda:///` resolution is disabled. The old topology-based
317//! resolution assumed the admin lived on the first mesh head host,
318//! which is no longer true after SA-5 changed to caller-local
319//! placement. All resolution paths now return explicit errors.
320//! A publication-based discovery mechanism will replace this in a
321//! future change. Until then, discover the admin URL from
322//! startup output or another launch-time publication.
323//!
324//! ## Admin self-identification invariants (AI-*)
325//!
326//! - **AI-1 (live identity):** `GET /v1/admin` returns the live
327//!   admin actor identity as `AdminInfo`.
328//! - **AI-2 (reported proc):** `proc_id` reports the hosting proc.
329//!   Placement equality (SA-5) is proved by unit tests; integration
330//!   tests validate that `proc_id` is populated and well-formed.
331//! - **AI-3 (url consistency):** `url` matches `GetAdminAddr`.
332//!
333//! The relationship between `host` and `url` (formerly AI-4) is
334//! now a constructor guarantee of [`AdminInfo::new`] rather than a
335//! live invariant. It is not in this registry.
336
337use 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
393/// Send an `IntrospectMessage` to an actor and receive the reply.
394/// Encapsulates open_once_port + send + timeout + error handling.
395async 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
419/// Send an `IntrospectMessage::QueryChild` to an actor.
420async 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
444/// Actor name used when spawning the mesh admin agent.
445pub const MESH_ADMIN_ACTOR_NAME: &str = "mesh_admin";
446
447/// Actor name for the HTTP bridge client mailbox on the service proc.
448///
449/// Unlike `MESH_ADMIN_ACTOR_NAME`, this is not a full actor: it is a
450/// client-mode `Instance<()>` created via
451/// `Proc::introspectable_instance()` and driven by Axum's Tokio task
452/// pool rather than an actor message loop. A separate instance is
453/// required because `MeshAdminAgent`'s own `Instance<Self>` is only
454/// accessible inside its message loop and cannot be shared with
455/// external tasks. This instance gives the HTTP handlers a routable
456/// proc identity so they can open one-shot reply ports
457/// (`open_once_port`) to receive responses from `MeshAdminAgent`.
458///
459/// Unlike a plain `client()`, this uses
460/// `Proc::introspectable_instance()` so the bridge responds to
461/// `IntrospectMessage::Query` and appears as a navigable node in the
462/// mesh TUI rather than causing a 504 when selected.
463pub const MESH_ADMIN_BRIDGE_NAME: &str = "mesh_admin_bridge";
464
465/// Structured error response following the gateway RFC envelope
466/// pattern.
467#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
468pub struct ApiError {
469    /// Machine-readable error code (e.g. "not_found", "bad_request").
470    pub code: String,
471    /// Human-readable error message.
472    pub message: String,
473    /// Additional context about the error. Schema is permissive
474    /// (any valid JSON) — `details` is a domain-specific escape
475    /// hatch. Do not assume a fixed shape.
476    #[serde(skip_serializing_if = "Option::is_none")]
477    pub details: Option<Value>,
478}
479
480/// Wrapper for the structured error envelope.
481#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
482pub struct ApiErrorEnvelope {
483    pub error: ApiError,
484}
485
486impl ApiError {
487    /// Create a "not_found" error.
488    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    /// Create a "bad_request" error.
497    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/// Response payload for `MeshAdminMessage::GetAdminAddr`.
521///
522/// `addr` is `None` until the admin HTTP server has successfully
523/// bound a listening socket during `MeshAdminAgent::init`.
524#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Named)]
525pub struct MeshAdminAddrResponse {
526    pub addr: Option<String>,
527}
528wirevalue::register_type!(MeshAdminAddrResponse);
529
530/// Messages handled by the `MeshAdminAgent`.
531///
532/// These are mesh-admin control-plane queries (as opposed to topology
533/// resolution). They’re wirevalue-serializable and come with
534/// generated client/ref helpers via `HandleClient`/`RefClient`.
535#[derive(
536    Debug,
537    Clone,
538    PartialEq,
539    Serialize,
540    Deserialize,
541    Handler,
542    HandleClient,
543    RefClient,
544    Named
545)]
546pub enum MeshAdminMessage {
547    /// Return the HTTP admin server address that this agent bound in
548    /// `init`.
549    ///
550    /// The reply contains `None` if the server hasn't started yet.
551    GetAdminAddr {
552        #[reply]
553        reply: OncePortRef<MeshAdminAddrResponse>,
554    },
555}
556wirevalue::register_type!(MeshAdminMessage);
557
558/// Newtype wrapper around `Result<NodePayload, String>` for the
559/// resolve reply port (`OncePortRef` requires `Named`).
560#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Named)]
561pub struct ResolveReferenceResponse(pub Result<NodePayload, String>);
562wirevalue::register_type!(ResolveReferenceResponse);
563
564/// Message for resolving a reference (string from HTTP path) into a
565/// `NodePayload`.
566///
567/// This is the primary “navigation” request used by the admin HTTP
568/// bridge: the caller provides a reference (e.g. `"root"`, a `ProcAddr`
569/// string, or an `ActorAddr` string) and the `MeshAdminAgent` returns a
570/// uniformly shaped `NodePayload` plus child references to continue
571/// walking the topology.
572///
573/// The work happens inside the admin actor's message loop so
574/// resolution can:
575/// - parse and validate the reference format,
576/// - dispatch to the right host/proc/actor via existing admin
577///   queries, and
578/// - return a structured payload without blocking HTTP handlers on
579///   mesh logic.
580#[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 `reference_string` to a `NodePayload`.
593    ///
594    /// On success the reply contains `payload=Some(..), error=None`; on failure
595    /// it contains `payload=None, error=Some(..)`.
596    Resolve {
597        /// Addr string from the HTTP path, parsed into a typed
598        /// `NodeRef` at the resolve boundary.
599        reference_string: String,
600        /// Reply port receiving the resolution result.
601        #[reply]
602        reply: OncePortRef<ResolveReferenceResponse>,
603    },
604}
605wirevalue::register_type!(ResolveReferenceMessage);
606
607/// Actor that serves a mesh-level admin HTTP endpoint.
608///
609/// `MeshAdminAgent` is the mesh-wide aggregation point for
610/// introspection: it holds `ActorRef<HostAgent>` handles for each
611/// host, and answers admin queries by forwarding targeted requests to
612/// the appropriate host agent and assembling a uniform `NodePayload`
613/// response for the client.
614///
615/// The agent also exposes an HTTP server (spawned from `init`) and
616/// supports reference-based navigation (`GET /v1/{reference}`) by
617/// resolving HTTP path references into typed `NodePayload` values
618/// plus child references.
619#[hyperactor::export(handlers = [MeshAdminMessage, ResolveReferenceMessage])]
620pub struct MeshAdminAgent {
621    /// Map of host address string → `HostAgent` reference used to
622    /// fan out our target admin queries.
623    hosts: HashMap<String, ActorRef<HostAgent>>,
624
625    /// Reverse index: `HostAgent` `ActorAddr` → host address
626    /// string.
627    ///
628    /// The host agent itself is an actor that can appear in multiple
629    /// places (e.g., as a host node and as a child actor under a
630    /// system proc). This index lets reference resolution treat that
631    /// `ActorAddr` as a *Host* node (via `resolve_host_node`) rather
632    /// than a generic *Actor* node, avoiding cycles / dropped nodes
633    /// in clients like the TUI.
634    host_agents_by_actor_id: HashMap<hyperactor::ActorAddr, String>,
635
636    /// `ActorAddr` of the process-global root client (`client[0]` on
637    /// the singleton Host's `local_proc`), exposed as a first-class
638    /// child of the root node. Routable and introspectable via the
639    /// blanket `Handler<IntrospectMessage>`.
640    root_client_actor_id: Option<hyperactor::ActorAddr>,
641
642    /// This agent's own `ActorAddr`, captured during `init`. Used to
643    /// include the admin proc as a visible node in the introspection
644    /// tree (the principle: "if you can send it a message, you can
645    /// introspect it").
646    self_actor_id: Option<hyperactor::ActorAddr>,
647
648    // -- HTTP server address fields --
649    //
650    // The admin HTTP server has three address representations:
651    //
652    //   1. `admin_addr_override` — caller-supplied bind address
653    //      (constructor param). When `None`, `init` reads
654    //      `MESH_ADMIN_ADDR` from config instead.
655    //
656    //   2. `admin_addr` — the actual `SocketAddr` the OS assigned
657    //      after `TcpListener::bind`. Populated during `init`.
658    //
659    //   3. `admin_host` — human-friendly URL with the machine
660    //      hostname (not the raw IP) so it works with DNS and TLS
661    //      certificate validation. Returned via `GetAdminAddr`.
662    /// Caller-supplied bind address. When `None`, `init` reads
663    /// `MESH_ADMIN_ADDR` from config.
664    admin_addr_override: Option<std::net::SocketAddr>,
665
666    /// Actual bound address after `TcpListener::bind`, populated
667    /// during `init`.
668    admin_addr: Option<std::net::SocketAddr>,
669
670    /// Hostname-based URL (e.g. `"https://myhost.facebook.com:1729"`)
671    /// for the admin HTTP server. Returned via `GetAdminAddr`.
672    admin_host: Option<String>,
673
674    /// Base URL of the Monarch dashboard. Passed at construction.
675    /// Used by proxy routes that forward requests to the dashboard's
676    /// `/api/*` endpoints.
677    telemetry_url: Option<String>,
678
679    /// When the mesh was started (ISO-8601 timestamp).
680    started_at: String,
681
682    /// Username who started the mesh.
683    started_by: String,
684}
685
686impl MeshAdminAgent {
687    /// Construct a `MeshAdminAgent` from a list of `(host_addr,
688    /// host_agent_ref)` pairs and an optional root client `ActorAddr`.
689    ///
690    /// Builds both:
691    /// - `hosts`: the forward map used to route admin queries to the
692    ///   correct `HostAgent`, and
693    /// - `host_agents_by_actor_id`: a reverse index used during
694    ///   reference resolution to recognize host-agent `ActorAddr`s and
695    ///   resolve them as `NodeProperties::Host` rather than as
696    ///   generic actors.
697    ///
698    /// When `root_client_actor_id` is `Some`, the root client appears
699    /// as a first-class child of the root node in the introspection
700    /// tree.
701    ///
702    /// The HTTP listen address is initialized to `None` and populated
703    /// during `init()` after the server socket is bound.
704    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        // Capture start time and username
716        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/// Self-identification payload returned by `GET /v1/admin`.
752///
753/// Construct via [`AdminInfo::new`]. AI-1, AI-2, AI-3 are live
754/// invariants. The relationship between `host` and `url` is a
755/// constructor guarantee — `AdminInfo::new()` rejects URLs with no
756/// host, so `host` always derives from `url` at construction.
757#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
758pub struct AdminInfo {
759    /// Stringified `ActorAddr` of the `MeshAdminAgent`.
760    pub actor_id: String,
761    /// Stringified `ProcAddr` of the proc hosting `MeshAdminAgent`.
762    pub proc_id: String,
763    /// Hostname the admin HTTP server bound on (derived from `url`).
764    pub host: String,
765    /// Full admin URL (e.g. `"https://myhost.facebook.com:1729"`).
766    pub url: String,
767}
768
769impl AdminInfo {
770    /// Construct from identity components and a full admin URL.
771    ///
772    /// Parses `url` strictly using the `url` crate. Returns an error
773    /// if the URL is invalid or has no host component. `host` is
774    /// derived from the parsed URL — the relationship between `host`
775    /// and `url` holds by construction, not by test.
776    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
792/// Shared state for the reference-based `/v1/{*reference}` bridge
793/// route.
794///
795/// The HTTP handler itself is intentionally thin and does not perform
796/// any routing logic. Instead, it forwards each request into the
797/// `MeshAdminAgent` actor via `ResolveReferenceMessage`, ensuring
798/// resolution happens inside the actor message loop (with access to
799/// actor messaging, timeouts, and indices).
800struct BridgeState {
801    /// Addr to the `MeshAdminAgent` actor that performs
802    /// reference resolution.
803    admin_ref: ActorRef<MeshAdminAgent>,
804    /// Dedicated client mailbox on system_proc for HTTP bridge reply
805    /// ports. Using a separate `Instance<()>` avoids sharing the
806    /// actor's own mailbox with the HTTP bridge and ensures the
807    /// bridge context is routable via system_proc's frontend address.
808    // Previous approach used `this.clone_for_py()` which cloned the
809    // admin actor's Instance:
810    //   bridge_cx: Instance<MeshAdminAgent>,
811    bridge_cx: Instance<()>,
812    /// Limits the number of in-flight resolve requests to prevent
813    /// introspection queries from overwhelming the shared tokio
814    /// runtime and starving user actor workloads.
815    resolve_semaphore: tokio::sync::Semaphore,
816    /// Keep the handle alive so the bridge mailbox is not dropped.
817    _bridge_handle: ActorHandle<()>,
818    /// Base URL of the Monarch dashboard (e.g.
819    /// `"http://localhost:5000"`). Passed from `MeshAdminAgent` at
820    /// init time. Used by proxy routes that forward requests to the
821    /// dashboard's `/api/*` endpoints.
822    telemetry_url: Option<String>,
823    /// Shared HTTP client for outbound proxy requests to the
824    /// dashboard. Reuses connection pool across requests.
825    http_client: reqwest::Client,
826    /// Self-identification metadata, populated during admin init.
827    admin_info: AdminInfo,
828}
829
830/// Build an HTTP client for outbound proxy requests to the dashboard.
831///
832/// Loads the root CA via the same bundle-probing logic used by every
833/// other mesh-admin client so the reqwest client can verify the
834/// dashboard's TLS cert. Falls back to the default trust store (and
835/// a default client) when no bundle is available.
836fn 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
864/// A TCP listener that performs a TLS handshake on each accepted
865/// connection before handing it to axum.
866///
867/// Implements [`axum::serve::Listener`] so it can be passed directly
868/// to [`axum::serve`].  Per the trait contract, `accept` handles
869/// errors internally (logging + retrying) and never returns `Err`.
870struct 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    /// Initializes the mesh admin agent and its HTTP server.
907    ///
908    /// 1. Binds well-known handler ports (`proc.spawn_with_label()` does not
909    ///    call `bind()` — unlike `gspawn` — so the actor must do it
910    ///    itself before becoming reachable).
911    /// 2. Binds a TCP listener (ephemeral or fixed port).
912    /// 3. Builds a TLS acceptor (explicit env vars, then Meta default
913    ///    paths). At Meta (`fbcode_build`), mTLS is mandatory and
914    ///    init fails if no certs are found. In OSS, falls back to
915    ///    plain HTTP.
916    /// 4. Creates a dedicated `Instance<()>` client mailbox on
917    ///    system_proc for the HTTP bridge's reply ports, keeping
918    ///    bridge traffic off the actor's own mailbox.
919    /// 5. Spawns the axum server in a background task (HTTPS with
920    ///    mTLS at Meta, HTTPS or HTTP in OSS depending on step 3).
921    ///
922    /// The hostname-based listen address is stored in `admin_host` so
923    /// it can be returned via `GetAdminAddr`. The scheme (`https://`
924    /// or `http://`) is included so clients know which protocol to
925    /// use.
926    async fn init(&mut self, this: &Instance<Self>) -> Result<(), anyhow::Error> {
927        // Bind well-known ports before the HTTP server is spawned, so
928        // messages (including Undeliverable bounces) can be delivered
929        // as soon as the admin is reachable.
930        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        // At Meta: mTLS is mandatory — fail if no certs are found.
945        // In OSS: TLS is best-effort with plain HTTP fallback.
946        // See MA-T1 in module doc.
947        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        // Build the host portion of the admin URL.
965        //
966        // Explicit bind (loopback, specific IP): honour the caller's
967        // choice — they bound that address intentionally.
968        //
969        // Wildcard bind: choose an advertised host that the loaded
970        // TLS certificate actually authorizes. Extract SANs from the
971        // cert and pick the first candidate that matches. This avoids
972        // emitting a URL that fails TLS verification.
973        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        // Create a dedicated client mailbox on system_proc for the
988        // HTTP bridge's reply ports. This avoids sharing the admin
989        // actor's own mailbox with async HTTP handlers.
990        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            // OSS fallback: plain HTTP (only reachable when !fbcode_build).
1027            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    /// Swallow undeliverable message bounces instead of crashing.
1042    ///
1043    /// The admin agent sends `IntrospectMessage` to actors that may
1044    /// not have the introspection port bound (e.g. actors spawned
1045    /// via `cx.spawn()` whose `#[export]` list does not include it).
1046    /// When the message cannot be delivered, the routing layer
1047    /// bounces an `Undeliverable` back to the sender. The default
1048    /// delivery-failure handling would fail the actor, which
1049    /// would kill this admin agent and — via supervision cascade —
1050    /// take down the entire admin process with `exit(1)`.
1051    ///
1052    /// Since the admin agent is best-effort infrastructure, an
1053    /// undeliverable introspection probe is not a fatal error.
1054    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/// Manual Handler impl — swallows `reply.send()` failures so the
1108/// admin agent stays alive when the HTTP caller disconnects.
1109#[async_trait]
1110impl Handler<MeshAdminMessage> for MeshAdminAgent {
1111    /// Dispatches `MeshAdminMessage` variants.
1112    ///
1113    /// Reply-send failures are swallowed because a dropped receiver
1114    /// (e.g. the HTTP bridge timed out) is not an error — the caller
1115    /// simply went away. Propagating the failure would crash the admin
1116    /// agent and take down the entire process.
1117    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/// Manual Handler impl — swallows `reply.send()` failures so the
1135/// admin agent stays alive when the HTTP caller disconnects.
1136#[async_trait]
1137impl Handler<ResolveReferenceMessage> for MeshAdminAgent {
1138    /// Dispatches `ResolveReferenceMessage` variants.
1139    ///
1140    /// The inner `resolve_reference` call never returns `Err` to the
1141    /// handler — failures are captured in the response payload.
1142    /// Reply-send failures are swallowed for the same reason as
1143    /// `MeshAdminMessage`: a dropped receiver means the caller (HTTP
1144    /// bridge) went away, which must not crash the admin agent.
1145    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    /// Core resolver for the reference-based admin API.
1169    ///
1170    /// Parses the caller-provided `reference_string` (or handles the
1171    /// special `"root"` case), then dispatches to
1172    /// `resolve_host_node`, `resolve_proc_node`, or
1173    /// `resolve_actor_node` to assemble a fully-populated
1174    /// `NodePayload` (properties + child references).
1175    ///
1176    /// The returned payload satisfies the **navigation identity
1177    /// invariant** (see module docs): `payload.identity ==
1178    /// reference_string`, and `payload.parent` equals the identity of
1179    /// the node this one appears under.
1180    ///
1181    /// Note: this returns `Err` for internal use; the public
1182    /// `resolve` handler converts failures into
1183    /// `ResolveReferenceResponse(Err(..))` so the actor never crashes
1184    /// on
1185    /// lookup errors.
1186    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    /// Returns the known actors on standalone procs — procs not
1216    /// managed by any host but whose actors are routable and
1217    /// introspectable. Each proc appears as a root child; the
1218    /// actor is the "anchor" used to discover the proc's contents.
1219    ///
1220    /// The root client is no longer standalone: spawn_admin registers
1221    /// C (the bootstrap host) as a normal host entry (A/C invariant).
1222    fn standalone_proc_actors(&self) -> impl Iterator<Item = &hyperactor::ActorAddr> {
1223        std::iter::empty()
1224    }
1225
1226    /// If `proc_id` belongs to a standalone proc, return the anchor
1227    /// actor on that proc. Returns `None` for host-managed procs.
1228    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    /// Returns true if `actor_id` lives on a standalone proc.
1234    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    /// Construct the synthetic root node for the reference tree.
1240    ///
1241    /// The root is not a real actor/proc; it's a convenience node
1242    /// that anchors navigation. Its children are `NodeRef::Host`
1243    /// entries for each configured `HostAgent`.
1244    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(); // LC-1
1253        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    /// Resolve a `HostAgent` actor reference into a host-level
1272    /// `NodePayload`.
1273    ///
1274    /// Sends `IntrospectMessage::Query` directly to the
1275    /// `HostAgent`, which returns a `NodePayload` with
1276    /// `NodeProperties::Host` and the host's children. The resolver
1277    /// overrides `parent` to `"root"` since the host agent
1278    /// doesn't know its position in the navigation tree.
1279    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    /// Resolve a `ProcAddr` reference into a proc-level `NodePayload`.
1300    ///
1301    /// First tries `IntrospectMessage::QueryChild` against the owning
1302    /// `HostAgent` (which recognizes service and local procs). If
1303    /// that returns an error payload, falls back to `ProcAgent` for
1304    /// user procs by querying
1305    /// `QueryChild(hyperactor::Addr::Proc(proc_id))`
1306    /// on `<proc_id>/proc_agent[0]`.
1307    ///
1308    /// See PA-1 in module doc.
1309    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        // Try the host agent's QueryChild first.
1322        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        // If the host recognized the proc, normalize identity and parent.
1332        // The host's QueryChild returns IntrospectRef::Actor(self_id) as
1333        // parent, which lifts to NodeRef::Actor. We need NodeRef::Host.
1334        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        // Fall back to querying the ProcAgent directly (user procs).
1344        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    /// Resolve a standalone proc into a proc-level `NodePayload`.
1362    ///
1363    /// Standalone procs (e.g. the admin proc) are not managed by any
1364    /// `HostAgent`, so
1365    /// `resolve_proc_node` cannot resolve them. Instead, we query the
1366    /// anchor actor on the proc for its introspection data, collect
1367    /// its supervision children, and build a synthetic proc node.
1368    ///
1369    /// Special case: when the anchor actor is this agent itself, we
1370    /// build the children list directly (just `[self]`) to avoid a
1371    /// self-deadlock — the actor loop cannot process an
1372    /// `IntrospectMessage` it sends to itself while handling a
1373    /// resolve request.
1374    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    /// Resolve a non-host-agent `ActorAddr` reference into an
1472    /// actor-level `NodePayload`.
1473    ///
1474    /// Sends `IntrospectMessage::Query` directly to the target actor
1475    /// via `PortRef::attest_handler_port`. The blanket handler
1476    /// returns a `NodePayload` with `NodeProperties::Actor` (or a
1477    /// domain-specific override like `NodeProperties::Proc` for
1478    /// `ProcAgent`).
1479    ///
1480    /// The resolver sets `parent` based on the actor's position
1481    /// in the topology: if the actor lives in a system proc, the
1482    /// parent is the system proc ref; otherwise it's the proc's
1483    /// `ProcAddr` string.
1484    async fn resolve_actor_node(
1485        &self,
1486        cx: &Context<'_, Self>,
1487        actor_id: &hyperactor::ActorAddr,
1488    ) -> Result<NodePayload, anyhow::Error> {
1489        // Self-resolution: we cannot send IntrospectMessage to our
1490        // own actor loop while handling a resolve request (deadlock).
1491        // Use introspect_payload() to snapshot our own state
1492        // directly.
1493        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            // Standalone procs have no ProcAgent — query directly.
1497            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            // Check terminated snapshots first — fast, no ambiguity.
1507            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                    // Not terminated — query the live actor.
1527                    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
1565/// Build the Axum router for the mesh admin HTTP server.
1566///
1567/// Routes:
1568/// - `GET /v1/schema` — JSON Schema (Draft 2020-12) for `NodePayload`.
1569/// - `GET /v1/schema/error` — JSON Schema for `ApiErrorEnvelope`.
1570/// - `GET /v1/openapi.json` — OpenAPI 3.1 spec (embeds JSON Schemas).
1571/// - `GET /v1/tree` — ASCII topology dump.
1572/// - `POST /v1/query` — proxy SQL query to the dashboard server.
1573/// - `GET /v1/pyspy/{*proc_reference}` — py-spy stack dump for a proc.
1574/// - `POST /v1/pyspy_dump/{*proc_reference}` — py-spy dump + store in Datafusion.
1575/// - `POST /v1/pyspy_profile_svg/{*proc_reference}` — py-spy profile → SVG flamegraph.
1576/// - `GET /v1/config/{*proc_reference}` — config snapshot for a proc.
1577/// - `GET /v1/admin` — admin self-identification (`AdminInfo`).
1578/// - `GET /v1/{*reference}` — JSON `NodePayload` for a single reference.
1579/// - `GET /SKILL.md` — agent-facing API documentation (markdown).
1580fn create_mesh_admin_router(bridge_state: Arc<BridgeState>) -> Router {
1581    Router::new()
1582        .route("/SKILL.md", get(serve_skill_md))
1583        // Literal paths matched by specificity before wildcard (SC-5).
1584        .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
1605/// Raw markdown template for the SKILL.md API document.
1606const SKILL_MD_TEMPLATE: &str = include_str!("mesh_admin_skill.md");
1607
1608/// Extract base URL from request headers.
1609///
1610/// Defaults to `https` when `x-forwarded-proto` is absent — the
1611/// admin server uses TLS in production, so `http` is the wrong
1612/// default for direct connections.
1613fn 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
1625/// Self-identification endpoint: returns `AdminInfo` (AI-1..AI-3;
1626/// AI-4 is a constructor guarantee of `AdminInfo::new()`).
1627async 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
1633/// JSON Schema for `AdminInfo`.
1634async 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
1640/// Serves the self-describing API document with the base URL
1641/// interpolated so examples are copy-pasteable.
1642async 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
1654/// Build a JSON Schema value with a `$id` field.
1655fn 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
1668/// JSON Schema for the `NodePayload` response type.
1669async 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
1675/// JSON Schema for the `ApiErrorEnvelope` error response.
1676async 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
1682/// Hoist `$defs` from a schemars-generated schema into a shared
1683/// map and rewrite internal `$ref` pointers from `#/$defs/X` to
1684/// `#/components/schemas/X` so OpenAPI tools can resolve them.
1685fn 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        // Also remove $schema from embedded schemas — it's
1698        // only valid at the root of a JSON Schema document,
1699        // not inside an OpenAPI components/schemas entry.
1700        obj.remove("$schema");
1701    }
1702    rewrite_refs(schema);
1703}
1704
1705/// Recursively rewrite `$ref: "#/$defs/X"` →
1706/// `$ref: "#/components/schemas/X"`.
1707fn 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
1728/// Build the OpenAPI 3.1 spec, embedding schemars-derived JSON
1729/// Schemas into `components/schemas`.
1730pub 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    // Hoist $defs into a shared components/schemas map so
1750    // OpenAPI tools can resolve references.
1751    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    // Rewrite any remaining $defs refs in the hoisted component schemas.
1773    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    // Insert paths outside the json! macro to avoid hitting the
2028    // serde_json recursion limit.
2029    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
2086/// OpenAPI 3.1 spec for the mesh admin API.
2087async fn serve_openapi() -> Result<axum::response::Json<serde_json::Value>, ApiError> {
2088    Ok(axum::response::Json(build_openapi_spec()))
2089}
2090
2091/// Validate and parse a raw proc reference path segment into a
2092/// decoded reference string and `ProcAddr`. Extracted for testability.
2093fn 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
2112/// Probe whether an actor is reachable by sending a lightweight
2113/// introspect query bounded by `MESH_ADMIN_QUERY_CHILD_TIMEOUT`.
2114///
2115/// Returns `Ok(true)` if the actor responds, `Ok(false)` if the
2116/// actor is absent or unresponsive (timeout / recv error).
2117async 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
2152/// Core py-spy dump logic shared by `pyspy_bridge` and
2153/// `pyspy_dump_and_store`.
2154///
2155/// Typed proc-handler target. Private to this module. The single
2156/// minting point is `route_proc_handler` via `ActorRef::attest`.
2157/// After minting, all sends go through typed `ActorRef::send`.
2158enum 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
2261/// Parse + route + attest. No probe. The single `ActorRef::attest`
2262/// minting point. Used by `config_bridge` which intentionally skips
2263/// the probe (CFG-4).
2264fn 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
2279/// Parse + route + attest + probe (PS-13).
2280async 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
2318/// HTTP bridge for py-spy stack dump requests.
2319///
2320/// Parses the proc reference, routes to the appropriate actor
2321/// (ProcAgent on worker procs, HostAgent on the service proc),
2322/// probes for reachability, and sends `PySpyDump` directly.
2323/// See PS-12, PS-13 in `introspect` module doc.
2324async 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
2347/// HTTP bridge for py-spy profile SVG requests.
2348///
2349/// Accepts `PySpyProfileOpts` as JSON POST body, profiles the target
2350/// process, and returns raw SVG.
2351async 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/// Request body for `POST /v1/query`.
2425#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
2426pub struct QueryRequest {
2427    /// SQL query string.
2428    pub sql: String,
2429}
2430
2431/// Response body from `POST /v1/query`.
2432#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
2433pub struct QueryResponse {
2434    /// Query result rows.
2435    pub rows: serde_json::Value,
2436}
2437
2438/// Request body sent to the dashboard's `/api/pyspy_dump` endpoint.
2439#[derive(Debug, Serialize)]
2440struct StorePyspyDumpRequest {
2441    dump_id: String,
2442    proc_ref: String,
2443    pyspy_result_json: String,
2444}
2445
2446/// Response body from `POST /v1/pyspy_dump/{*proc_reference}`.
2447#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
2448pub struct PyspyDumpAndStoreResponse {
2449    /// Unique identifier for the stored dump.
2450    pub dump_id: String,
2451}
2452
2453/// Resolve the telemetry URL from bridge state, returning an
2454/// `ApiError` if not configured.
2455fn 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
2461/// Proxy SQL queries to the Monarch dashboard's `/api/query`
2462/// endpoint.
2463///
2464/// Requires `telemetry_url` to be set. The request body must
2465/// contain a `sql` field. The dashboard response rows are returned
2466/// verbatim.
2467async 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        // Try to extract error message from dashboard response.
2494        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
2519/// Trigger a py-spy dump and store the result in the dashboard's
2520/// DataFusion pyspy tables.
2521///
2522/// 1. Performs a py-spy dump via `do_pyspy_dump` (same as
2523///    `pyspy_bridge`).
2524/// 2. POSTs the serialized result to the dashboard's
2525///    `/api/pyspy_dump` endpoint for persistent storage.
2526/// 3. Returns the generated dump id.
2527async 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
2573/// HTTP bridge for config dump requests.
2574///
2575/// Config dump bridge. No preflight probe — the send + bridge
2576/// timeout handles both absent and busy actors correctly (CFG-4).
2577async 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
2588/// Resolve an opaque reference string to a `NodePayload` via the
2589/// actor-based resolver.
2590///
2591/// Implements `GET /v1/{*reference}` for the reference-walking client
2592/// (e.g. the TUI):
2593/// - Decodes the wildcard path segment into the original reference
2594///   string (Axum does not percent-decode `{*reference}` captures).
2595/// - Sends `ResolveReferenceMessage::Resolve` to `MeshAdminAgent` and
2596///   awaits the reply.
2597/// - Maps resolver failures into appropriate `ApiError`s
2598///   (`bad_request`, `not_found`, `gateway_timeout`, or
2599///   `internal_error`).
2600async fn resolve_reference_bridge(
2601    State(state): State<Arc<BridgeState>>,
2602    AxumPath(reference): AxumPath<String>,
2603) -> Result<Json<NodePayloadDto>, ApiError> {
2604    // Axum's wildcard may include a leading slash; strip it.
2605    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    // Limit concurrent resolves to avoid starving user workloads
2619    // that share this tokio runtime.
2620    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
2660// TODO: MESH_ADMIN_TREE_TIMEOUT is applied per-call, not as a total
2661// budget. On a mesh with N hosts and M procs, the worst case is
2662// N*(1+M) sequential calls each up to 10s. This should use a single
2663// deadline for the entire walk.
2664/// `GET /v1/tree` — ASCII topology dump.
2665///
2666/// Walks the reference graph starting from `"root"`, resolving each
2667/// host and its proc children, and formats the result as a
2668/// human-readable ASCII tree suitable for quick `curl` inspection.
2669/// Each line includes a clickable URL for drilling into that node via
2670/// the reference API. Built on top of the same
2671/// `ResolveReferenceMessage` protocol used by the TUI.
2672///
2673/// Output format:
2674/// ```text
2675/// unix:@hash  ->  https://host:port/v1/...  (or http:// in OSS)
2676/// ├── service  ->  https://host:port/v1/...
2677/// │   ├── agent[0]  ->  https://host:port/v1/...
2678/// │   └── client[0]  ->  https://host:port/v1/...
2679/// ├── local  ->  https://host:port/v1/...
2680/// └── philosophers_0  ->  https://host:port/v1/...
2681///     ├── agent[0]  ->  https://host:port/v1/...
2682///     └── philosopher[0]  ->  https://host:port/v1/...
2683/// ```
2684async fn tree_dump(
2685    State(state): State<Arc<BridgeState>>,
2686    headers: axum::http::header::HeaderMap,
2687) -> Result<String, ApiError> {
2688    // Limit concurrent resolves to avoid starving user workloads.
2689    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    // Build base URL from the Host header for clickable links.
2703    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    // Resolve root.
2714    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    // Resolve each root child. Hosts get the full host→proc→actor
2739    // subtree; non-host children (e.g. the root client actor) are
2740    // rendered as single leaf lines.
2741    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
2850/// Derive a short display label from a reference string for the ASCII
2851/// tree.
2852///
2853/// Extracts the proc name — the meaningful identifier for tree
2854/// display — from the various reference formats emitted by
2855/// `HostAgent`'s children list:
2856///
2857/// - System proc ref `"[system] unix:@hash,service"` → `"service"`
2858/// - ProcAgent ActorAddr `"unix:@hash,my_proc,agent[0]"` →
2859///   `"my_proc"`
2860/// - Bare ProcAddr `"unix:@hash,my_proc"` → `"my_proc"`
2861///
2862/// Note: `ActorAddr::Display` for `ProcAddr` uses commas as
2863/// separators (`proc_id,actor_name[idx]`), not slashes.
2864fn 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// -- Admin handle type discrimination --
2887
2888/// A handle scheme that requires a publication-based lookup to resolve
2889/// to a concrete admin URL.
2890///
2891/// Only `Mast` is defined today. The nested-enum shape allows future
2892/// scheduler-specific variants (Slurm, K8s, etc.) to be added without
2893/// changing `AdminHandle`.
2894#[non_exhaustive]
2895pub enum PublishedHandle {
2896    /// `mast_conda:///<job-name>` — requires publication-based discovery.
2897    Mast(String),
2898}
2899
2900impl PublishedHandle {
2901    /// Resolve a published handle to a concrete admin URL.
2902    ///
2903    /// All published-handle schemes return an explicit error today.
2904    /// When real publication lookup is implemented, dispatch by variant here.
2905    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/// A handle for locating the mesh admin server.
2916///
2917/// Parse a user-supplied address string with [`AdminHandle::parse`]
2918/// and resolve it to a concrete URL with [`AdminHandle::resolve`].
2919#[non_exhaustive]
2920pub enum AdminHandle {
2921    /// Already-resolved URL (e.g. `https://host:1729`).
2922    Url(String),
2923    /// Handle that requires a publication lookup. Currently unresolvable.
2924    Published(PublishedHandle),
2925    /// Scheme or format that is not recognized.
2926    Unsupported(String),
2927}
2928
2929impl AdminHandle {
2930    /// Parse an address string into an `AdminHandle`.
2931    ///
2932    /// Uses `url` crate parsing. Known publication-handle prefixes
2933    /// (`mast_conda:///`) are classified as `Published`. `http`/`https`
2934    /// scheme URLs are `Url`. Bare `host:port` inputs (no scheme) are
2935    /// inferred as `https://host:port` and classified as `Url` — this
2936    /// preserves existing TUI behavior where `--addr myhost:1729` is a
2937    /// valid input. Everything else is `Unsupported`.
2938    pub fn parse(addr: &str) -> Self {
2939        // Check known publication handle prefixes first.
2940        if addr.starts_with("mast_conda:///") {
2941            return AdminHandle::Published(PublishedHandle::Mast(addr.to_string()));
2942        }
2943        // Strict URL parse — only http/https accepted.
2944        if let Ok(parsed) = url::Url::parse(addr)
2945            && matches!(parsed.scheme(), "http" | "https")
2946        {
2947            return AdminHandle::Url(addr.to_string());
2948        }
2949        // Infer https:// for bare host:port inputs (e.g. "myhost:1729").
2950        // This preserves the TUI's documented --addr behavior.
2951        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    /// Resolve to a concrete admin base URL.
2962    ///
2963    /// `port_override` is retained to preserve existing call surfaces
2964    /// but is intentionally unused until real publication lookup is
2965    /// implemented.
2966    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
2978/// Resolve a `mast_conda:///<job-name>` handle into an admin base URL.
2979///
2980/// **Disabled.** Mesh admin placement has moved to the caller's local
2981/// proc. Delegates to [`AdminHandle::Published`] + [`PublishedHandle::resolve`].
2982/// Kept as a stable API shim; do not remove.
2983pub 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
2992/// Cert-aware advertised host selection for wildcard binds.
2993///
2994/// The server advertises one URL. For wildcard binds, this module
2995/// generates candidate hosts from environment sources and picks
2996/// the first candidate covered by the loaded server cert's SAN
2997/// set. This ensures the advertised URL is always consistent with
2998/// the certificate the server presents.
2999mod advertised_host {
3000    use std::net::IpAddr;
3001
3002    /// An identity that can appear as a cert SAN entry.
3003    #[derive(Debug, PartialEq, Eq)]
3004    pub(super) enum SanIdentity {
3005        Ip(IpAddr),
3006        Dns(String),
3007    }
3008
3009    /// Choose the advertised host for a wildcard-bind admin URL.
3010    ///
3011    /// Candidates (in preference order):
3012    /// 1. `hostname::get()` (preferred — human-readable, no
3013    ///    brackets in URLs)
3014    /// 2. `host_ipv6_address()` (Meta: TW metadata → fbwhoami →
3015    ///    local_ipv6)
3016    ///
3017    /// The first candidate whose identity is covered by the loaded
3018    /// server cert's SANs wins. If no cert is available or no
3019    /// candidate matches, falls back to hostname.
3020    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        // (url_display_form, identity_to_match_against_cert)
3027        // Prefer DNS names over IPs — more readable, no brackets
3028        // in URLs, more stable across container restarts.
3029        let mut candidates: Vec<(String, SanIdentity)> = Vec::new();
3030
3031        // Candidate 1: hostname (preferred if cert covers it).
3032        candidates.push((hostname.clone(), SanIdentity::Dns(hostname.clone())));
3033
3034        // Candidate 2: host IPv6 address (Meta environments).
3035        #[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    /// Extract SAN entries from the server cert PEM bundle.
3059    ///
3060    /// Loads the same cert bundle that `try_tls_acceptor` uses,
3061    /// parses the leaf cert with `x509_parser`, and returns SAN
3062    /// DNS names and IP addresses. Returns empty if no cert is
3063    /// available or parsing fails.
3064    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    /// Pick the first candidate covered by the given SAN set.
3127    /// Extracted from `from_cert_sans` for direct unit testing.
3128    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    // Integration tests that spawn MeshAdminAgent must pass
3217    // `Some("[::]:0".parse().unwrap())` as the admin_addr to get an
3218    // ephemeral port. The default (`None`) reads MESH_ADMIN_ADDR
3219    // config which is `[::]:1729` — a fixed port that causes bind
3220    // conflicts when tests run concurrently.
3221
3222    /// Minimal introspectable actor for tests. The `#[export]`
3223    /// attribute generates `Named + Referable + Binds` so that
3224    /// `handle.bind()` registers the `IntrospectMessage` port for
3225    /// remote delivery.
3226    #[derive(Debug)]
3227    #[hyperactor::export(handlers = [])]
3228    struct TestIntrospectableActor;
3229    impl Actor for TestIntrospectableActor {}
3230
3231    // Verifies that MeshAdminAgent::build_root_payload constructs the
3232    // expected root node: identity/root metadata, correct Root
3233    // properties (num_hosts), and child links populated with the
3234    // stringified IDs of the configured host mesh-agent ActorRefs.
3235    #[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        // Verify root properties derived from attrs.
3276        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                // LC-1: root system_children is always empty.
3286                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    // End-to-end smoke test for MeshAdminAgent::resolve that walks
3296    // the reference tree: root → host → system proc → host-agent
3297    // cross-reference. Verifies the reverse index routes the
3298    // HostAgent ActorAddr to NodeProperties::Host (not Actor),
3299    // preventing the TUI's cycle detection from dropping that node.
3300    #[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        // -- 1. Stand up a local in-process Host with a HostAgent --
3311        // Use Unix transport for all procs — Local transport does not
3312        // support cross-proc message routing.
3313        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        // -- 2. Spawn MeshAdminAgent on a dedicated test proc --
3333        // NOTE: This does not conform to SA-5 (caller-local placement).
3334        // Production uses host_mesh::spawn_admin(). This is a white-box
3335        // test of admin behavior, not placement.
3336        let admin_proc = Proc::direct(ChannelTransport::Unix.any(), "admin".to_string()).unwrap();
3337        // The admin proc has no supervision coordinator by default.
3338        // Without one, actor teardown triggers std::process::exit(1).
3339        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        // -- 3. Create a bare client instance for sending messages --
3353        // Only a mailbox is needed for reply ports — no actor message
3354        // loop required.
3355        let client_proc = Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
3356        let client = client_proc.client("client");
3357
3358        // -- 4. Resolve "root" --
3359        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); // host only (admin proc no longer standalone)
3371
3372        // -- 5. Resolve the host child --
3373        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        // LC-2: host system_children is always empty.
3395        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        // -- 6. Resolve a system proc child --
3408        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        // The system proc should have at least the "host_agent" actor.
3419        assert!(
3420            !proc_node.children.is_empty(),
3421            "proc should have at least one actor child"
3422        );
3423
3424        // -- 7. Cross-reference: system proc child is the host agent --
3425        //
3426        // The service proc's actor (agent[0]) IS the HostAgent, so
3427        // it appears both as a host node (from root, via NodeRef::Host)
3428        // and as an actor (from a proc's children list, via NodeRef::Actor).
3429        // NodeRef::Host in root children makes resolution unambiguous:
3430        // host refs get Entity view, plain actor refs get Actor view.
3431
3432        // The system proc must list the host agent among its children.
3433        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        // Resolve that child reference as a plain actor (no host: prefix).
3443        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        // When resolved as a plain actor reference, it must return
3450        // Actor properties (not Host), because it has no host: prefix.
3451        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    // Verifies MeshAdminAgent::resolve returns NodeProperties::Proc
3459    // for all proc children. Spawns a user proc via
3460    // CreateOrUpdate<ProcSpec>, resolves all host proc-children, and
3461    // asserts every proc returns Proc properties.
3462    #[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        // Stand up a local in-process Host with a HostAgent.
3479        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        // Spawn MeshAdminAgent on a dedicated test proc.
3501        // NOTE: Does not conform to SA-5 (caller-local placement).
3502        // Production uses host_mesh::spawn_admin(). White-box test setup.
3503        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        // Create a bare client instance for sending messages.
3518        let client_proc = Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
3519        let client = client_proc.client("client");
3520
3521        // Spawn a user proc via CreateOrUpdate<ProcSpec>.
3522        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        // Wait for the user proc to boot.
3533        tokio::time::sleep(Duration::from_secs(2)).await;
3534
3535        // Resolve the host to get its children (system + user procs).
3536        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        // The host should have at least 3 children: system proc,
3542        // local proc, and our user proc.
3543        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        // Resolve each proc child and verify it has Proc properties.
3550        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                // Properties derived from attrs — verified by derive_properties tests.
3569            } else {
3570                // Host agent cross-reference — skip.
3571            }
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    // Verifies that build_root_payload lists only the host as a
3581    // child. The root client is visible under its host's local proc,
3582    // not at root level.
3583    #[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        // Only the host; root client is under host → local proc.
3606        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    // Verifies that the root client actor is visible through the
3615    // host → local proc → actor path, not as a standalone child of
3616    // root.
3617    #[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        // Stand up a local in-process Host with a HostAgent.
3627        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        // Spawn the root client on the host's local proc (before
3638        // moving the host into HostAgent).
3639        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        // Spawn MeshAdminAgent on a dedicated test proc with the root
3656        // client ActorAddr. NOTE: Does not conform to SA-5 (caller-local
3657        // placement). Production uses host_mesh::spawn_admin().
3658        // White-box test of root-client visibility, not placement.
3659        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        // Client for sending messages.
3675        let client_proc =
3676            hyperactor::Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
3677        let client = client_proc.client("client");
3678
3679        // Resolve "root" — should contain only the host.
3680        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        // Resolve the host — should list the local proc in children.
3694        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        // Resolve the local proc — should contain the root client actor.
3708        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        // Resolve the root client actor — parent should be the local proc.
3727        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    // Verifies that the SKILL.md template contains the canonical
3745    // strings that agents and tests rely on. Prevents silent drift or
3746    // accidental removal.
3747    #[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    // Verifies the navigation identity invariant (see module docs):
3773    //
3774    // 1. payload.identity == reference_string used to resolve it.
3775    // 2. For each child reference C of a resolved node P,
3776    //    resolve(C).parent == P.identity.
3777    //
3778    // Walks the entire tree starting from root, checking both
3779    // properties at every reachable node.
3780    #[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        // Stand up a local host with a HostAgent.
3791        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        // Spawn MeshAdminAgent on a dedicated test proc.
3811        // NOTE: Does not conform to SA-5 (caller-local placement).
3812        // Production uses host_mesh::spawn_admin(). White-box test setup.
3813        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        // Walk the tree breadth-first, checking the invariant at every node.
3831        // Each entry is (reference_string, expected_parent_identity).
3832        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            // NI-1: identity display matches the reference used.
3846            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            // NI-2: parent matches the parent node's identity.
3855            assert_eq!(
3856                node.parent, expected_parent,
3857                "parent mismatch for '{}': expected {:?}, got {:?}",
3858                ref_str, expected_parent, node.parent
3859            );
3860
3861            // Enqueue children with this node's identity as their
3862            // expected parent.
3863            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        // Sanity: we should have visited at least root, host, a
3872        // proc, and an actor.
3873        assert!(
3874            visited.len() >= 4,
3875            "expected at least 4 nodes in the tree, visited {}",
3876            visited.len()
3877        );
3878    }
3879
3880    // Exercises SP-1..SP-4 for host/proc payloads.
3881    #[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        // -- 1. Stand up a local in-process Host with a HostAgent --
3892        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        // -- 2. Spawn MeshAdminAgent on a dedicated test proc --
3913        // NOTE: This does not conform to SA-5 (caller-local placement).
3914        // Production uses host_mesh::spawn_admin(). This is a white-box
3915        // test of admin behavior, not placement.
3916        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        // -- 3. Create a bare client instance for sending messages --
3931        let client_proc = Proc::direct(ChannelTransport::Unix.any(), "client".to_string()).unwrap();
3932        let client = client_proc.client("client");
3933
3934        // -- 4. Resolve the host to get its children --
3935        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        // -- 5. Find a system proc child via system_children --
3948        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        // Procs are never system — host system_children should be empty.
3955        assert!(
3956            system_children.is_empty(),
3957            "host system_children should be empty (procs are never system), got {:?}",
3958            system_children
3959        );
3960        // Verify host properties derived from attrs.
3961        assert!(
3962            matches!(&host_node.properties, NodeProperties::Host { .. }),
3963            "expected Host properties"
3964        );
3965
3966        // -- 6. Verify host children contain the system proc --
3967        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        // -- 7. Resolve a proc child --
3976        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        // as_of is a SystemTime — just verify it's not the epoch.
4002        assert!(
4003            proc_node.as_of > std::time::UNIX_EPOCH,
4004            "as_of should be after the epoch"
4005        );
4006
4007        // Verify proc properties derived from attrs.
4008        assert!(
4009            matches!(&proc_node.properties, NodeProperties::Proc { .. }),
4010            "expected Proc properties"
4011        );
4012    }
4013
4014    // -- AdminHandle / PublishedHandle tests --
4015
4016    // AdminHandle::parse — all four cases.
4017    #[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        // Bare host:port → inferred as https://host:port.
4026        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        // Bare hostname with no port → Unsupported (no port, scheme inference fails).
4046        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    // resolve_mast_handle delegates to PublishedHandle::resolve.
4080    // Error text changed from "disabled" to "not yet implemented".
4081    #[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    // -- AdminInfo::new() constructor tests --
4093
4094    // Constructor guarantee: valid https URL produces correct host.
4095    #[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    // Constructor guarantee: invalid URL is rejected.
4108    #[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    // Constructor guarantee: URL with no host is rejected.
4119    #[test]
4120    fn test_admin_info_new_rejects_url_without_host() {
4121        // data: URLs have no host component.
4122        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    // -- Placement test (SA-5) --
4131
4132    // Exercises the real public entrypoint and checks SA-5 via
4133    // ActorRef reachability on the caller proc.
4134    #[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        // 1. Stand up a local in-process host mesh.
4143        let host_mesh = HostMesh::local().await.unwrap();
4144
4145        // 2. Create a separate caller proc with an actor instance.
4146        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        // 3. Call the real public entrypoint.
4151        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        // 4. Prove the returned ActorRef is usable: fetch the URL
4161        //    via get_admin_addr. This also proves the admin is on
4162        //    caller_proc (undeliverable if not).
4163        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    // AI-1..AI-3: GET /v1/admin HTTP route test requires TLS certs
4176    // (fbcode_build enforces mTLS). Covered by integration tests in
4177    // fbcode//monarch/hyperactor_mesh/test/mesh_admin_integration.
4178    // AI-4 is a constructor guarantee tested via AdminInfo::new() above.
4179
4180    // Verifies that GET /v1/{proc_id} reflects actors spawned directly
4181    // on a proc — bypassing ProcAgent's gspawn message and therefore
4182    // never triggering publish_introspect_properties — so that the
4183    // resolved children list is always derived from live proc state.
4184    //
4185    // Regression guard for the bug introduced in 9a08d559: the switch
4186    // from a live handle_introspect to a cached publish model made
4187    // supervision-spawned actors (e.g. every sieve actor after
4188    // sieve[0]) invisible to the TUI.
4189    //
4190    // Exercises PA-1 (see module doc). See also
4191    // proc_agent::tests::test_query_child_proc_returns_live_children.
4192    #[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        // Stand up a HostMeshAgent. The user proc gets its own
4208        // ephemeral address; we register that address in
4209        // MeshAdminAgent so resolve_proc_node can look it up.
4210        // HostMeshAgent won't know the user proc (it wasn't spawned
4211        // through it), so QueryChild returns Error and resolve falls
4212        // back to querying proc_agent[0] via QueryChild(Proc) — the
4213        // path being tested.
4214        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        // User proc: own ephemeral Unix socket, own ProcAgent.
4230        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        // MeshAdminAgent: register the user proc's addr as a "host"
4241        // pointing to host_agent_ref. That agent doesn't know the
4242        // user proc, so QueryChild → Error → fallback to proc_agent.
4243        // NOTE: Does not conform to SA-5 (caller-local placement).
4244        // White-box test of proc-agent fallback, not placement.
4245        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        // Resolve the user proc via MeshAdminAgent. HostMeshAgent
4262        // returns Error for QueryChild → fallback to proc_agent[0]
4263        // QueryChild(Addr::Proc) → live NodeProperties::Proc.
4264        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        // Spawn an actor directly on the user proc, bypassing gspawn.
4285        // This simulates how sieve[0] spawns sieve[1], sieve[2], etc.
4286        user_proc.spawn_with_label("extra_actor", TestIntrospectableActor);
4287
4288        // Resolve again — the new actor must appear immediately
4289        // without any republish, proving PA-1 is satisfied.
4290        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    // -- pyspy bridge input validation tests --
4317    //
4318    // Tests for the v1 proc-reference strictness contract (see
4319    // introspect module doc): the py-spy bridge accepts only
4320    // ProcAddr-form references and rejects other forms as bad_request.
4321
4322    #[test]
4323    fn pyspy_parse_empty_reference() {
4324        // v1 contract: empty input → bad_request.
4325        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        // v1 contract: slash-only (axum wildcard artifact) → bad_request.
4333        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        // v1 contract: malformed encoding → bad_request.
4341        // %FF%FE is not valid UTF-8.
4342        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        // v1 contract: non-ProcAddr reference → bad_request.
4350        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        // v1 contract: valid ProcAddr → accepted.
4358        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        // v1 contract: leading slash from axum wildcard is stripped.
4370        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    /// PS-12: service proc routes to HostAgent.
4379    #[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    /// PS-12: non-service proc routes to ProcAgent.
4391    #[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    /// PS-12: a labeled instance named "service" is still a normal proc.
4403    #[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}