hyperactor/introspect.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//! Introspection protocol for hyperactor actors.
10//!
11//! Every actor has a dedicated introspect task that handles
12//! [`IntrospectMessage`] by reading [`InstanceCell`] state directly,
13//! without going through the actor's message loop. This means:
14//!
15//! - Stuck actors can be introspected (the task runs independently).
16//! - Introspection does not perturb observed state (no Heisenberg).
17//! - Live status is reported accurately.
18//!
19//! Infrastructure actors publish domain-specific metadata via
20//! `publish_attrs()`, which the introspect task reads for Entity-view
21//! queries. Non-addressable children (e.g., system procs) are
22//! resolved via a callback registered on [`InstanceCell`].
23//!
24//! Callers navigate topology by fetching an [`IntrospectResult`] and
25//! following its `children` references.
26//!
27//! # Design Invariants
28//!
29//! The introspection subsystem maintains twelve invariants (S1--S12).
30//! Each is documented at the code site that enforces it.
31//!
32//! - **S1.** Introspection must not depend on actor responsiveness --
33//! a wedged actor can still be introspected (runtime task, not
34//! actor loop).
35//! - **S2.** Introspection must not perturb observed state -- reading
36//! `InstanceCell` never sets `last_message_handler` to
37//! `IntrospectMessage`.
38//! - **S3.** Sender routing is unchanged -- senders target the same
39//! control `PortId` across processes.
40//! - **S4.** `IntrospectMessage` never produces a `WorkCell` --
41//! pre-registration via `bind_control_port` gives the introspect
42//! port its own channel, independent of the actor's work queue.
43//! - **S5.** Replies never use `PanickingMailboxSender` -- the
44//! introspect task replies via `Mailbox::serialize_and_send_once`.
45//! - **S6.** View semantics are stable -- Actor view uses live
46//! structural state + supervision children; Entity view uses
47//! published properties + domain children.
48//! - **S7.** `QueryChild` must work without actor handlers -- system
49//! procs are resolved via a per-actor callback on `InstanceCell`.
50//! - **S8.** Published properties are constrained -- actors cannot
51//! publish `Root` or `Error` payloads (only `Host` and `Proc`
52//! variants).
53//! - **S9.** Port binding is single source of truth -- the introspect
54//! port is bound exactly once via `bind_handler_port()` in
55//! `Instance::new()`.
56//! - **S10.** Introspect receiver lifecycle -- created in
57//! `Instance::new()`, spawned in `start()`, dropped in
58//! `child_instance()`.
59//! - **S11.** Terminated snapshots do not keep actors resolvable --
60//! `store_terminated_snapshot` writes to the proc's snapshot map,
61//! not the instances map. `resolve_actor_ref` checks terminal
62//! status independently and is unaffected by snapshot storage.
63//! - **S12.** Introspection must not impair actor liveness --
64//! introspection queries (including DashMap reads for actor
65//! enumeration) must not cause convoy starvation or scheduling
66//! delays that stall concurrent actor spawn/stop operations.
67//!
68//! ## Introspection key invariants (IK-*)
69//!
70//! - **IK-1 (metadata completeness):** Every actor-runtime
71//! introspection key must carry `@meta(INTROSPECT = ...)` with
72//! non-empty `name` and `desc`.
73//! - **IK-2 (short-name uniqueness):** No two introspection keys may
74//! share the same `IntrospectAttr.name`. Duplicates would break the
75//! FQ-to-short HTTP remap and schema output.
76//!
77//! ## Failure introspection invariants (FI-*)
78//!
79//! The FailureInfo presentation type lives in
80//! `hyperactor_mesh::introspect`; these invariants are documented
81//! here because the enforcement sites are in hyperactor (`proc.rs`
82//! `serve()`, `live_actor_payload`).
83//!
84//! - **FI-1 (event-before-status):** All `InstanceCell` state that
85//! `live_actor_payload` reads must be written BEFORE
86//! `change_status()` transitions to terminal.
87//! - **FI-2 (write-once):** `InstanceCellState::supervision_event` is
88//! written at most once per actor lifetime.
89//! - **FI-3 (failure attrs <-> status):** Failure attrs are present
90//! iff status is `"failed"`.
91//! - **FI-4 (is_propagated <-> root_cause_actor):**
92//! `failure_is_propagated == true` iff `failure_root_cause_actor !=
93//! this_actor_id`.
94//! - **FI-5 (is_poisoned <-> failed_actor_count):** `is_poisoned ==
95//! true` iff `failed_actor_count > 0`.
96//! - **FI-6 (clean stop leaves no failure artifacts):** When an
97//! actor stops cleanly through the serving loop,
98//! `supervision_event` holds a non-error terminal event
99//! (`actor_status` is `Stopped`, so `is_error()` is `false`),
100//! failure attrs are absent, and the actor does not contribute to
101//! `failed_actor_count`. The field records the terminal supervision
102//! event, not just failures; see the `supervision_event` accessor
103//! for when the field is `None`.
104//! - **FI-7 (propagated-stopped-root-cause):** When a failed actor's
105//! supervision chain bottoms out in a `Stopped` child event,
106//! structured failure metadata must still name the stopped child as
107//! `failure_root_cause_actor`.
108//! - **FI-8 (propagation-classification):** `failure_is_propagated`
109//! is derived from root-cause actor identity; a parent that failed
110//! due to a child's event must report `failure_is_propagated ==
111//! true`.
112//! - **FI-9 (stored-terminal vs delivered-zombie):** For an actor
113//! marked `Zombie` during teardown that then reaches a terminal
114//! status, `InstanceCell::supervision_event` records the true
115//! terminal event (`Stopped`/`Failed`, for introspection) while
116//! the supervision event delivered to the parent/proc remains the
117//! non-error `Zombie` verdict (for supervision policy). The two
118//! intentionally diverge; enforced in `proc.rs` `serve()`, where
119//! the stored `event` and the delivered `event_to_deliver` are
120//! computed separately.
121//!
122//! ## Attrs view invariants (AV-*)
123//!
124//! These govern the typed view layer (`ActorAttrsView`). The full
125//! AV-* / DP-* family is documented in `hyperactor_mesh::introspect`;
126//! the subset relevant to this crate:
127//!
128//! - **AV-1 (view-roundtrip):** For each view V,
129//! `V::from_attrs(&v.to_attrs()) == Ok(v)`.
130//! - **AV-2 (required-key-strictness):** `from_attrs` fails iff
131//! required keys for that view are missing.
132//! - **AV-3 (unknown-key-tolerance):** Unknown attrs keys must not
133//! affect successful decode outcome.
134//!
135//! ## Inbound ordering exposure invariants (IO-*)
136//!
137//! - **IO-1 (inbound-ordering state semantics):**
138//! `ActorAttrsView::inbound_ordering` carries four meaningful states
139//! that consumers (DTO, TUI, agents) MUST distinguish:
140//! * `None` -- no snapshot callback was installed. In current code
141//! this means structural absence: an `InstanceCellState` not built
142//! through `Instance::new` (hand-built test fixtures, or any future
143//! code path that bypasses the constructor). Live actors built via
144//! `Instance::new` always install Some, and terminated-actor
145//! payloads -- which still go through `live_actor_payload(&cell)`
146//! while the cell exists -- inherit that Some.
147//! * `Some({enabled: false, ...})` -- ordered path exists but reorder
148//! buffering is disabled; `sessions` is empty regardless of traffic.
149//! Messages bypass receiver-local sequencing.
150//! * `Some({enabled: true, skipped_session_count: 1, sessions: []})`
151//! -- the receiver-local sequencing mutex was busy; session state
152//! was unavailable at this observation.
153//! * `Some({enabled: true, skipped_session_count: 0, ...})` -- a
154//! complete snapshot; `sessions` is meaningful.
155//!
156//! `None` is NOT equivalent to `Some({enabled: false, ...})`.
157//! - **IO-2 (inbound-ordering reflects publish-time state):** Snapshot
158//! atomicity follows lock granularity: one mutex protects the entire
159//! receiver session map, so acquiring it observes every session and
160//! failing to acquire it observes none. When present, the snapshot is
161//! computed at `build_actor_attrs` invocation time via the sequenced
162//! receiver's snapshot handle. `last_released_seq` etc. are
163//! point-in-time. A busy mutex yields no session rows and
164//! `skipped_session_count == 1`; acquiring it yields every session and
165//! `skipped_session_count == 0`. The count is an aggregate availability
166//! marker, not a per-session count.
167//! - **IO-3 (queue-depth and inbound-ordering are independent
168//! diagnostics, no arithmetic contract):**
169//! * `ACTOR_QUEUE_DEPTH` (per PD-5a/PD-5b in `proc.rs`): accepted
170//! handler work not yet dequeued by the actor loop.
171//! * `INBOUND_ORDERING.sessions[*].buffered_count`: messages held by
172//! receiver-local sequencing waiting for a seq gap to fill.
173//!
174//! These are two independent point-in-time diagnostics. No
175//! arithmetic or ordering relationship between them is part of the
176//! API contract; the accounting paths are free to change. Consumers
177//! must not derive one from the other.
178//!
179//! ## Actor-attrs snapshot invariants (AS-*)
180//!
181//! These govern the generic per-actor introspection-attrs snapshot
182//! seam: an actor may install a `Fn() -> Attrs` callback via
183//! `Instance::set_attrs_snapshot`, which the introspect task invokes in
184//! `build_actor_attrs` and merges into the Actor view. The seam is the
185//! runtime-agnostic transport for actor-supplied introspection data
186//! (e.g. a Python actor reporting in-flight handler execution); core
187//! interprets none of it.
188//!
189//! - **AS-1 (snapshot-opacity):** actor-supplied attrs are merged into
190//! the Actor view verbatim. Core neither interprets nor validates the
191//! keys -- any key the actor sets is transported as-is. This is what
192//! keeps the seam runtime-agnostic (no Rust-vs-Python knowledge, no
193//! "endpoint" concept in core).
194//! - **AS-2 (core-precedence):** on a key collision, the core/runtime
195//! key wins over the actor-supplied value. This is the Actor-view
196//! analog of IA-2, which governs the published-attrs path; the two
197//! sources of actor-supplied keys (published attrs, snapshot seam) are
198//! distinct, but both yield to core keys.
199//! - **AS-3 (snapshot-non-fatal):** an absent, empty, or panicking
200//! snapshot degrades to the core-only Actor view. The callback is
201//! `catch_unwind`-guarded and the merge falls back to the core attrs,
202//! so a bad snapshot can never empty or invalidate `attrs` (with IA-1,
203//! IA-5).
204
205use std::fmt;
206use std::str::FromStr;
207use std::time::SystemTime;
208
209use hyperactor_config::Attrs;
210use hyperactor_config::INTROSPECT;
211use hyperactor_config::IntrospectAttr;
212use hyperactor_config::declare_attrs;
213use serde::Deserialize;
214use serde::Serialize;
215use typeuri::Named;
216
217use crate::ActorAddr;
218use crate::Addr;
219use crate::AddrParseError;
220use crate::InstanceCell;
221use crate::OncePortRef;
222use crate::ProcAddr;
223/// Typed reference to an introspectable entity.
224///
225/// This is the generic hyperactor layer — it knows about procs and
226/// actors, not mesh-specific concepts like root or host.
227///
228/// Port references are intentionally excluded — introspection
229/// does not address individual ports.
230#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, Named)]
231pub enum IntrospectRef {
232 /// A proc reference.
233 Proc(ProcAddr),
234 /// An actor reference.
235 Actor(ActorAddr),
236}
237hyperactor_config::impl_attrvalue!(IntrospectRef);
238
239/// Error returned when parsing an [`IntrospectRef`].
240#[derive(Debug, thiserror::Error)]
241pub enum IntrospectRefParseError {
242 /// The address text could not be parsed.
243 #[error(transparent)]
244 Addr(#[from] AddrParseError),
245 /// Port references are not introspectable.
246 #[error("port references are not valid introspection references")]
247 PortNotAllowed,
248}
249
250impl fmt::Display for IntrospectRef {
251 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
252 match self {
253 Self::Proc(id) => fmt::Display::fmt(id, f),
254 Self::Actor(id) => fmt::Display::fmt(id, f),
255 }
256 }
257}
258
259impl FromStr for IntrospectRef {
260 type Err = IntrospectRefParseError;
261
262 fn from_str(s: &str) -> Result<Self, Self::Err> {
263 let r: Addr = s.parse()?;
264 match r {
265 Addr::Proc(id) => Ok(Self::Proc(id)),
266 Addr::Actor(id) => Ok(Self::Actor(id)),
267 Addr::Port(_) => Err(IntrospectRefParseError::PortNotAllowed),
268 }
269 }
270}
271
272impl From<ProcAddr> for IntrospectRef {
273 fn from(id: ProcAddr) -> Self {
274 Self::Proc(id)
275 }
276}
277
278impl From<ActorAddr> for IntrospectRef {
279 fn from(id: ActorAddr) -> Self {
280 Self::Actor(id)
281 }
282}
283
284// Introspection attr keys — actor-runtime concepts.
285//
286// These keys are populated by the introspect handler from
287// InstanceCell data. Mesh-topology keys (node_type, addr, num_procs,
288// etc.) are declared in hyperactor_mesh::introspect.
289//
290// Naming convention:
291//
292// - Attr names are node-type-agnostic. The `node_type` attr (from the
293// mesh layer) identifies what kind of node it is; individual attr
294// names don't repeat that. So `status`, not `actor_status`.
295// - Related attrs share a prefix to form a group. The `failure_*`
296// keys decompose failure info into flat attrs — the `failure_`
297// prefix groups them semantically.
298// - `actor_type` is an exception: the `actor_` prefix disambiguates
299// it from `node_type` (mesh-layer concept). `actor_type` is the
300// Rust actor type name; `node_type` is the topology role.
301// - Use real types where possible (e.g. SystemTime for timestamps),
302// not String. Serialization format is a presentation concern.
303// - Internal key names are fully-qualified by `declare_attrs!`
304// (module_path + attr constant), e.g.
305// `hyperactor::introspect::status`.
306// - HTTP/schema public key names come from `@meta(INTROSPECT =
307// IntrospectAttr { name, desc })`. Keep `name` explicit so API
308// stability is decoupled from internal refactors.
309//
310// See IK-1 (metadata completeness) and IK-2 (short-name uniqueness)
311// in module doc.
312declare_attrs! {
313 /// Actor lifecycle status: "running", "stopped", "failed".
314 ///
315 /// Together with `STATUS_REASON`, these two attrs replace the
316 /// former `actor_status` prefix protocol (`"stopped:reason"`,
317 /// `"failed:reason"`) with structured fields, eliminating string
318 /// prefix parsing in consumers.
319 @meta(INTROSPECT = IntrospectAttr {
320 name: "status".into(),
321 desc: "Actor lifecycle status: running, stopped, failed".into(),
322 })
323 pub attr STATUS: String;
324
325 /// Reason for stop/failure (absent when running).
326 @meta(INTROSPECT = IntrospectAttr {
327 name: "status_reason".into(),
328 desc: "Reason for stop/failure (absent when running)".into(),
329 })
330 pub attr STATUS_REASON: String;
331
332 /// Fully-qualified actor type name.
333 @meta(INTROSPECT = IntrospectAttr {
334 name: "actor_type".into(),
335 desc: "Fully-qualified actor type name".into(),
336 })
337 pub attr ACTOR_TYPE: String;
338
339 /// Number of messages processed by this actor.
340 @meta(INTROSPECT = IntrospectAttr {
341 name: "messages_processed".into(),
342 desc: "Number of messages processed by this actor".into(),
343 })
344 pub attr MESSAGES_PROCESSED: u64 = 0;
345
346 /// Timestamp when this actor was created.
347 @meta(INTROSPECT = IntrospectAttr {
348 name: "created_at".into(),
349 desc: "Timestamp when this actor was created".into(),
350 })
351 pub attr CREATED_AT: SystemTime;
352
353 /// Name of the last message handler invoked.
354 @meta(INTROSPECT = IntrospectAttr {
355 name: "last_handler".into(),
356 desc: "Name of the last message handler invoked".into(),
357 })
358 pub attr LAST_HANDLER: String;
359
360 /// Total CPU time in message handlers (microseconds).
361 @meta(INTROSPECT = IntrospectAttr {
362 name: "total_processing_time_us".into(),
363 desc: "Total CPU time in message handlers (microseconds)".into(),
364 })
365 pub attr TOTAL_PROCESSING_TIME_US: u64 = 0;
366
367 /// Flight recorder JSON (recent trace events).
368 @meta(INTROSPECT = IntrospectAttr {
369 name: "flight_recorder".into(),
370 desc: "Flight recorder JSON (recent trace events)".into(),
371 })
372 pub attr FLIGHT_RECORDER: String;
373
374 /// Whether this actor is infrastructure/system.
375 @meta(INTROSPECT = IntrospectAttr {
376 name: "is_system".into(),
377 desc: "Whether this actor is infrastructure/system".into(),
378 })
379 pub attr IS_SYSTEM: bool = false;
380
381 /// Child references for tree navigation. Published by
382 /// infrastructure actors (HostMeshAgent, ProcAgent) so the
383 /// Entity view can return children without parsing mesh-layer keys.
384 @meta(INTROSPECT = IntrospectAttr {
385 name: "children".into(),
386 desc: "Child references for tree navigation".into(),
387 })
388 pub attr CHILDREN: Vec<IntrospectRef>;
389
390 /// Machine-readable error code for error nodes.
391 @meta(INTROSPECT = IntrospectAttr {
392 name: "error_code".into(),
393 desc: "Machine-readable error code (e.g. not_found)".into(),
394 })
395 pub attr ERROR_CODE: String;
396
397 /// Human-readable error message for error nodes.
398 @meta(INTROSPECT = IntrospectAttr {
399 name: "error_message".into(),
400 desc: "Human-readable error message".into(),
401 })
402 pub attr ERROR_MESSAGE: String;
403
404 // Failure attrs — decomposition of FailureInfo into flat attrs.
405 //
406 // - **FI-A1 (presence):** failure_* attrs are present iff
407 // status == "failed"; absent otherwise. (Attr-level restatement
408 // of FI-3.)
409 // - **FI-A2 (propagation):** failure_is_propagated == true iff
410 // failure_root_cause_actor != this actor's id. (Attr-level
411 // restatement of FI-4.)
412 // FI-1, FI-2 (write ordering) are enforced in proc.rs serve()
413 // and are unaffected by the representation change.
414 // FI-5, FI-6 are proc/mesh-level and unaffected.
415
416 /// Failure error message.
417 @meta(INTROSPECT = IntrospectAttr {
418 name: "failure_error_message".into(),
419 desc: "Failure error message".into(),
420 })
421 pub attr FAILURE_ERROR_MESSAGE: String;
422
423 /// Actor that caused the failure (root cause).
424 @meta(INTROSPECT = IntrospectAttr {
425 name: "failure_root_cause_actor".into(),
426 desc: "Actor that caused the failure (root cause)".into(),
427 })
428 pub attr FAILURE_ROOT_CAUSE_ACTOR: ActorAddr;
429
430 /// Name of root cause actor.
431 @meta(INTROSPECT = IntrospectAttr {
432 name: "failure_root_cause_name".into(),
433 desc: "Name of root cause actor".into(),
434 })
435 pub attr FAILURE_ROOT_CAUSE_NAME: String;
436
437 /// Timestamp when failure occurred.
438 @meta(INTROSPECT = IntrospectAttr {
439 name: "failure_occurred_at".into(),
440 desc: "Timestamp when failure occurred".into(),
441 })
442 pub attr FAILURE_OCCURRED_AT: SystemTime;
443
444 /// Whether the failure was propagated from a child.
445 @meta(INTROSPECT = IntrospectAttr {
446 name: "failure_is_propagated".into(),
447 desc: "Whether the failure was propagated from a child".into(),
448 })
449 pub attr FAILURE_IS_PROPAGATED: bool = false;
450
451 /// Stable per-instance identifier (`Uuid::now_v7`) assigned at
452 /// `Instance::new`.
453 @meta(INTROSPECT = IntrospectAttr {
454 name: "instance_id".into(),
455 desc: "Stable per-instance Uuid::now_v7() identity assigned at Instance::new".into(),
456 })
457 pub attr INSTANCE_ID: String;
458
459 /// Accepted handler work not yet dequeued by the actor loop (per
460 /// `proc.rs` PD-5a/PD-5b). Independent of `INBOUND_ORDERING`:
461 /// no arithmetic or ordering relationship between the two is part
462 /// of the API contract. See IO-3.
463 @meta(INTROSPECT = IntrospectAttr {
464 name: "queue_depth".into(),
465 desc: "Accepted handler work not yet dequeued by the actor loop (PD-5a/b). Independent of inbound_ordering; no arithmetic contract -- see IO-3.".into(),
466 })
467 pub attr ACTOR_QUEUE_DEPTH: u64 = 0;
468
469 /// Per-session reorder state from the sequenced receiver snapshot.
470 /// `sessions[*].buffered_count` reports messages held by
471 /// receiver-local sequencing waiting for a seq gap to fill. Independent
472 /// diagnostic from `queue_depth`; no arithmetic contract -- see
473 /// IO-3.
474 ///
475 /// Absence (`None` on `ActorAttrsView`) means structural absence
476 /// (no snapshot callback installed). `Some({enabled: false, ...})`
477 /// means the path exists but buffering is disabled. See IO-1.
478 @meta(INTROSPECT = IntrospectAttr {
479 name: "inbound_ordering".into(),
480 desc: "Per-session reorder-buffer state from receiver-local sequencing. Independent diagnostic from queue_depth; no arithmetic contract -- see IO-3. Absence vs Some({enabled: false}) is meaningful -- see IO-1.".into(),
481 })
482 pub attr INBOUND_ORDERING: crate::ordering::OrderingSnapshot;
483}
484
485// See FI-1 through FI-8 in module doc.
486
487/// Error from decoding an `Attrs` bag into a typed view.
488#[derive(Debug, Clone, PartialEq)]
489pub enum AttrsViewError {
490 /// A required key was absent (and has no default).
491 MissingKey {
492 /// The attr key that was absent.
493 key: &'static str,
494 },
495 /// A cross-field coherence check failed.
496 InvariantViolation {
497 /// Invariant label (e.g. "IA-4").
498 label: &'static str,
499 /// Human-readable description of the violation.
500 detail: String,
501 },
502}
503
504impl fmt::Display for AttrsViewError {
505 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
506 match self {
507 Self::MissingKey { key } => write!(f, "missing required key: {key}"),
508 Self::InvariantViolation { label, detail } => {
509 write!(f, "invariant {label} violated: {detail}")
510 }
511 }
512 }
513}
514
515impl std::error::Error for AttrsViewError {}
516
517impl AttrsViewError {
518 /// Convenience constructor for a missing required key.
519 pub fn missing(key: &'static str) -> Self {
520 Self::MissingKey { key }
521 }
522
523 /// Convenience constructor for an invariant violation.
524 pub fn invariant(label: &'static str, detail: String) -> Self {
525 Self::InvariantViolation { label, detail }
526 }
527}
528
529/// Structured failure fields decoded from `FAILURE_*` attrs.
530#[derive(Debug, Clone, PartialEq)]
531pub struct FailureAttrs {
532 /// Error message describing the failure.
533 pub error_message: String,
534 /// Actor that caused the failure (root cause).
535 pub root_cause_actor: ActorAddr,
536 /// Display name of the root-cause actor, if available.
537 pub root_cause_name: Option<String>,
538 /// When the failure occurred.
539 pub occurred_at: SystemTime,
540 /// Whether this failure was propagated from a child.
541 pub is_propagated: bool,
542}
543
544/// Typed view over attrs for an actor node.
545#[derive(Debug, Clone, PartialEq)]
546pub struct ActorAttrsView {
547 /// Lifecycle status: "running", "stopped", "failed".
548 pub status: String,
549 /// Reason for stop/failure, if any.
550 pub status_reason: Option<String>,
551 /// Fully-qualified actor type name.
552 pub actor_type: String,
553 /// Stable per-instance identifier (`Uuid::now_v7`) as a string.
554 pub instance_id: String,
555 /// Number of messages processed.
556 pub messages_processed: u64,
557 /// When this actor was created.
558 pub created_at: Option<SystemTime>,
559 /// Name of the last message handler invoked.
560 pub last_handler: Option<String>,
561 /// Total CPU time in message handlers (microseconds).
562 pub total_processing_time_us: u64,
563 /// Accepted handler work not yet dequeued by the actor loop
564 /// (PD-5a/b). Independent diagnostic from `inbound_ordering`;
565 /// no arithmetic contract between the two -- see IO-3. Defaults
566 /// to 0 when the attr is absent.
567 pub queue_depth: u64,
568 /// Flight recorder JSON, if available.
569 pub flight_recorder: Option<String>,
570 /// Whether this is a system/infrastructure actor.
571 pub is_system: bool,
572 /// Per-session reorder state. `None` means no snapshot callback was
573 /// installed (structural absence per IO-1); `Some({enabled: false, ..})`
574 /// means buffering is disabled; `Some({enabled: true, ..})` means active.
575 /// Consumers must distinguish all three states.
576 pub inbound_ordering: Option<crate::ordering::OrderingSnapshot>,
577 /// Failure details, present iff status == "failed".
578 pub failure: Option<FailureAttrs>,
579}
580
581impl ActorAttrsView {
582 /// Decode from an `Attrs` bag (AV-2, AV-3). Requires `STATUS`
583 /// and `ACTOR_TYPE`. Enforces IA-3 (status_reason must not be
584 /// present for non-terminal status), IA-4 (failure attrs iff
585 /// failed), and failure completeness (if any required failure
586 /// key is present, all three required keys must be).
587 pub fn from_attrs(attrs: &Attrs) -> Result<Self, AttrsViewError> {
588 let status = attrs
589 .get(STATUS)
590 .ok_or_else(|| AttrsViewError::missing("status"))?
591 .clone();
592 let status_reason = attrs.get(STATUS_REASON).cloned();
593 let actor_type = attrs
594 .get(ACTOR_TYPE)
595 .ok_or_else(|| AttrsViewError::missing("actor_type"))?
596 .clone();
597 let instance_id = attrs
598 .get(INSTANCE_ID)
599 .ok_or_else(|| AttrsViewError::missing("instance_id"))?
600 .clone();
601 let messages_processed = *attrs.get(MESSAGES_PROCESSED).unwrap_or(&0);
602 let created_at = attrs.get(CREATED_AT).copied();
603 let last_handler = attrs.get(LAST_HANDLER).cloned();
604 let total_processing_time_us = *attrs.get(TOTAL_PROCESSING_TIME_US).unwrap_or(&0);
605 let queue_depth = *attrs.get(ACTOR_QUEUE_DEPTH).unwrap_or(&0);
606 let flight_recorder = attrs.get(FLIGHT_RECORDER).cloned();
607 let is_system = *attrs.get(IS_SYSTEM).unwrap_or(&false);
608 let inbound_ordering = attrs.get(INBOUND_ORDERING).cloned();
609
610 // IA-3 (one-sided): status_reason must not be present for
611 // non-terminal status. The converse is not enforced —
612 // terminal status without a reason is valid (clean shutdown).
613 let is_terminal = status == "stopped" || status == "failed";
614 if status_reason.is_some() && !is_terminal {
615 return Err(AttrsViewError::invariant(
616 "IA-3",
617 format!(
618 "status_reason present but status is '{status}' (expected stopped or failed)"
619 ),
620 ));
621 }
622
623 // Decode failure attrs. If any of the three required
624 // failure keys is present, require all three.
625 // FAILURE_IS_PROPAGATED has a declare_attrs! default of
626 // false, so it always resolves via attrs.get() and needs
627 // no explicit presence check. FAILURE_ROOT_CAUSE_NAME is
628 // genuinely optional.
629 let has_any_failure = attrs.get(FAILURE_ERROR_MESSAGE).is_some()
630 || attrs.get(FAILURE_ROOT_CAUSE_ACTOR).is_some()
631 || attrs.get(FAILURE_OCCURRED_AT).is_some();
632
633 let failure = if has_any_failure {
634 let error_message = attrs
635 .get(FAILURE_ERROR_MESSAGE)
636 .ok_or_else(|| AttrsViewError::missing("failure_error_message"))?
637 .clone();
638 let root_cause_actor = attrs
639 .get(FAILURE_ROOT_CAUSE_ACTOR)
640 .ok_or_else(|| AttrsViewError::missing("failure_root_cause_actor"))?
641 .clone();
642 let root_cause_name = attrs.get(FAILURE_ROOT_CAUSE_NAME).cloned();
643 let occurred_at = *attrs
644 .get(FAILURE_OCCURRED_AT)
645 .ok_or_else(|| AttrsViewError::missing("failure_occurred_at"))?;
646 // Default false: failure originated at this actor.
647 let is_propagated = *attrs.get(FAILURE_IS_PROPAGATED).unwrap_or(&false);
648 Some(FailureAttrs {
649 error_message,
650 root_cause_actor,
651 root_cause_name,
652 occurred_at,
653 is_propagated,
654 })
655 } else {
656 None
657 };
658
659 // IA-4: failure attrs present iff status == "failed".
660 if status == "failed" && failure.is_none() {
661 return Err(AttrsViewError::invariant(
662 "IA-4",
663 "status is 'failed' but no failure_* attrs present".to_string(),
664 ));
665 }
666 if status != "failed" && failure.is_some() {
667 return Err(AttrsViewError::invariant(
668 "IA-4",
669 format!("status is '{status}' but failure_* attrs are present"),
670 ));
671 }
672
673 Ok(Self {
674 status,
675 status_reason,
676 actor_type,
677 instance_id,
678 messages_processed,
679 created_at,
680 last_handler,
681 total_processing_time_us,
682 queue_depth,
683 flight_recorder,
684 is_system,
685 inbound_ordering,
686 failure,
687 })
688 }
689
690 /// Encode into an `Attrs` bag (AV-1 round-trip producer).
691 pub fn to_attrs(&self) -> Attrs {
692 let mut attrs = Attrs::new();
693 attrs.set(STATUS, self.status.clone());
694 if let Some(reason) = &self.status_reason {
695 attrs.set(STATUS_REASON, reason.clone());
696 }
697 attrs.set(ACTOR_TYPE, self.actor_type.clone());
698 attrs.set(INSTANCE_ID, self.instance_id.clone());
699 attrs.set(MESSAGES_PROCESSED, self.messages_processed);
700 if let Some(t) = self.created_at {
701 attrs.set(CREATED_AT, t);
702 }
703 if let Some(handler) = &self.last_handler {
704 attrs.set(LAST_HANDLER, handler.clone());
705 }
706 attrs.set(TOTAL_PROCESSING_TIME_US, self.total_processing_time_us);
707 attrs.set(ACTOR_QUEUE_DEPTH, self.queue_depth);
708 if let Some(fr) = &self.flight_recorder {
709 attrs.set(FLIGHT_RECORDER, fr.clone());
710 }
711 attrs.set(IS_SYSTEM, self.is_system);
712 if let Some(snapshot) = &self.inbound_ordering {
713 attrs.set(INBOUND_ORDERING, snapshot.clone());
714 }
715 if let Some(fi) = &self.failure {
716 attrs.set(FAILURE_ERROR_MESSAGE, fi.error_message.clone());
717 attrs.set(FAILURE_ROOT_CAUSE_ACTOR, fi.root_cause_actor.clone());
718 if let Some(name) = &fi.root_cause_name {
719 attrs.set(FAILURE_ROOT_CAUSE_NAME, name.clone());
720 }
721 attrs.set(FAILURE_OCCURRED_AT, fi.occurred_at);
722 attrs.set(FAILURE_IS_PROPAGATED, fi.is_propagated);
723 }
724 attrs
725 }
726}
727
728/// Internal introspection result. Carries attrs as a JSON string.
729/// The mesh layer constructs the API-facing `NodePayload` (with
730/// `properties`) from this via `derive_properties`.
731///
732/// This is the internal wire type — it travels over handler ports
733/// via `IntrospectMessage`. The presentation-layer `NodePayload`
734/// (with `NodeProperties`) lives in `hyperactor_mesh::introspect`.
735#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Named)]
736pub struct IntrospectResult {
737 /// Addr identifying this node.
738 pub identity: IntrospectRef,
739 /// JSON-serialized `Attrs` bag containing introspection attributes.
740 pub attrs: String,
741 /// Child references the client can follow to descend the tree.
742 pub children: Vec<IntrospectRef>,
743 /// Parent reference for upward navigation.
744 pub parent: Option<IntrospectRef>,
745 /// When this data was captured.
746 pub as_of: SystemTime,
747}
748wirevalue::register_type!(IntrospectResult);
749
750/// Context for introspection query - what aspect of the actor to
751/// describe.
752///
753/// Infrastructure actors (e.g., ProcAgent, HostAgent)
754/// have dual nature: they manage entities (Proc, Host) while also
755/// being actors themselves. IntrospectView allows callers to
756/// specify which aspect to query.
757// TODO(monarch-introspection): IntrospectView currently uses
758// Entity/Actor naming. Consider renaming to runtime-neutral query
759// modes (e.g. Published/Runtime) to avoid mesh-domain wording in
760// hyperactor while preserving behavior and wire compatibility.
761#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Named)]
762pub enum IntrospectView {
763 /// Return managed-entity properties (Proc, Host, etc.) for
764 /// infrastructure actors.
765 Entity,
766 /// Return standard actor properties (status, messages_processed,
767 /// flight_recorder).
768 Actor,
769}
770wirevalue::register_type!(IntrospectView);
771
772/// Introspection query sent to any actor.
773///
774/// `Query` asks the actor to describe itself. `QueryChild` asks the
775/// actor to describe one of its non-addressable children — an entity
776/// that appears in the navigation tree but has no mailbox of its own
777/// (e.g. a system proc owned by a host). The parent actor answers on
778/// the child's behalf.
779#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Named)]
780pub enum IntrospectMessage {
781 /// "Describe yourself."
782 Query {
783 /// View context - Entity or Actor.
784 view: IntrospectView,
785 /// Reply port receiving the actor's self-description.
786 reply: OncePortRef<IntrospectResult>,
787 },
788 /// "Describe one of your children."
789 QueryChild {
790 /// Addr identifying the child to describe.
791 child_ref: Addr,
792 /// Reply port receiving the child's description.
793 reply: OncePortRef<IntrospectResult>,
794 },
795}
796wirevalue::register_type!(IntrospectMessage);
797
798/// Structured tracing event from the actor-local flight recorder.
799///
800/// Deserialization target for the `FLIGHT_RECORDER` attrs JSON string.
801#[derive(Debug, Clone, Serialize, Deserialize)]
802pub struct RecordedEvent {
803 /// ISO 8601 timestamp of the event.
804 pub timestamp: String,
805 /// Monotonic sequence number for ordering.
806 #[serde(default)]
807 pub seq: usize,
808 /// Event level (INFO, DEBUG, etc.).
809 pub level: String,
810 /// Event target (module path).
811 #[serde(default)]
812 pub target: String,
813 /// Event name.
814 pub name: String,
815 /// Event fields as JSON.
816 pub fields: serde_json::Value,
817}
818
819/// Format a [`SystemTime`] as an ISO 8601 timestamp with millisecond
820/// precision.
821pub fn format_timestamp(time: SystemTime) -> String {
822 humantime::format_rfc3339_millis(time).to_string()
823}
824
825/// Build a JSON-serialized `Attrs` string from values already
826/// computed by `live_actor_payload`. Reuses the same data — no
827/// redundant reads from `InstanceCell`.
828///
829/// Populates actor-runtime keys (STATUS, ACTOR_TYPE, etc.),
830/// decomposes the status prefix protocol into STATUS + STATUS_REASON,
831/// and decomposes failure fields into individual FAILURE_* attrs.
832///
833/// Starts from a fresh `Attrs` bag — published attrs (node_type,
834/// addr, etc.) are NOT included. This ensures the Actor view
835/// produces actor-only data; the Entity view handles published
836/// attrs separately.
837/// Failure fields extracted from a supervision event.
838struct FailureSnapshot {
839 error_message: String,
840 root_cause_actor: ActorAddr,
841 root_cause_name: Option<String>,
842 occurred_at: SystemTime,
843 is_propagated: bool,
844}
845
846/// Pre-computed actor state for building the attrs JSON string.
847/// Avoids redundant InstanceCell reads — `live_actor_payload`
848/// computes these once and passes them in.
849struct ActorSnapshot {
850 status_str: String,
851 is_system: bool,
852 last_handler: Option<String>,
853 flight_recorder: Option<String>,
854 failure: Option<FailureSnapshot>,
855}
856
857fn build_actor_attrs(cell: &crate::InstanceCell, snap: &ActorSnapshot) -> String {
858 // Actor view builds a clean attrs bag with only actor-runtime
859 // keys. Published attrs (node_type, addr, etc.) belong to the
860 // Entity view — they are NOT merged here. This ensures that
861 // e.g. a HostMeshAgent resolved via Actor view produces Actor
862 // properties, not Host properties.
863 let mut attrs = hyperactor_config::Attrs::new();
864
865 // IA-3: status_reason present iff status carries a reason.
866 if let Some(reason) = snap.status_str.strip_prefix("stopped:") {
867 attrs.set(STATUS, "stopped".to_string());
868 attrs.set(STATUS_REASON, reason.trim().to_string());
869 } else if let Some(reason) = snap.status_str.strip_prefix("failed:") {
870 attrs.set(STATUS, "failed".to_string());
871 attrs.set(STATUS_REASON, reason.trim().to_string());
872 } else {
873 attrs.set(STATUS, snap.status_str.clone());
874 // IA-3: no status_reason for non-terminal states —
875 // guaranteed by fresh Attrs bag.
876 }
877
878 attrs.set(ACTOR_TYPE, cell.actor_type_name().to_string());
879 attrs.set(MESSAGES_PROCESSED, cell.num_processed_messages());
880 attrs.set(CREATED_AT, cell.created_at());
881 attrs.set(TOTAL_PROCESSING_TIME_US, cell.total_processing_time_us());
882 attrs.set(IS_SYSTEM, snap.is_system);
883 attrs.set(INSTANCE_ID, cell.instance_id().to_string());
884 attrs.set(ACTOR_QUEUE_DEPTH, cell.queue_depth());
885
886 if let Some(snapshot) = cell.inbound_ordering_snapshot() {
887 // TODO: truncation / filtering of long session lists belongs at
888 // the API/DTO layer, not here. `build_actor_attrs` returns the
889 // full per-actor snapshot verbatim so consumers can decide.
890 attrs.set(INBOUND_ORDERING, snapshot);
891 }
892
893 if let Some(handler) = &snap.last_handler {
894 attrs.set(LAST_HANDLER, handler.clone());
895 }
896 if let Some(fr) = &snap.flight_recorder {
897 attrs.set(FLIGHT_RECORDER, fr.clone());
898 }
899
900 // IA-4 / FI-A1: failure attrs present iff status == "failed".
901 if let Some(fi) = &snap.failure {
902 attrs.set(FAILURE_ERROR_MESSAGE, fi.error_message.clone());
903 attrs.set(FAILURE_ROOT_CAUSE_ACTOR, fi.root_cause_actor.clone());
904 if let Some(name) = &fi.root_cause_name {
905 attrs.set(FAILURE_ROOT_CAUSE_NAME, name.clone());
906 }
907 attrs.set(FAILURE_OCCURRED_AT, fi.occurred_at);
908 attrs.set(FAILURE_IS_PROPAGATED, fi.is_propagated);
909 }
910 // IA-4: failure attrs absent when not failed — guaranteed by
911 // starting from a fresh Attrs bag (no stale keys possible).
912
913 let core_json = serde_json::to_string(&attrs).unwrap_or_else(|_| "{}".to_string());
914
915 // Merge actor-supplied attrs (the generic seam, AS-1). `Attrs::merge`
916 // overwrites the receiver's keys with the argument's, so merging the
917 // core bag *onto* the actor snapshot makes core keys win on collision
918 // (AS-2, the Actor-view analog of IA-2). No snapshot → the core bag is
919 // returned unchanged; if the merged bag somehow fails to serialize,
920 // fall back to the core-only JSON so a bad snapshot can never empty or
921 // invalidate the actor view (AS-3).
922 match cell.actor_attrs_snapshot() {
923 None => core_json,
924 Some(mut merged) => {
925 merged.merge(attrs);
926 serde_json::to_string(&merged).unwrap_or(core_json)
927 }
928 }
929}
930
931/// Build an [`IntrospectResult`] from live [`InstanceCell`] state.
932///
933/// Reads the current live status and last handler directly from
934/// the cell. Used by the introspect task (which runs outside
935/// the actor's message loop) and by `Instance::introspect_payload`.
936pub fn live_actor_payload(cell: &InstanceCell) -> IntrospectResult {
937 let status = cell.status().borrow().clone();
938 live_actor_payload_with_status(cell, &status)
939}
940
941fn live_actor_payload_with_status(
942 cell: &InstanceCell,
943 status: &crate::actor::ActorStatus,
944) -> IntrospectResult {
945 let actor_id = cell.actor_addr();
946 let last_handler = cell.last_message_handler();
947
948 let children: Vec<IntrospectRef> = cell
949 .child_actor_ids()
950 .into_iter()
951 .map(IntrospectRef::Actor)
952 .collect();
953
954 let events = cell.recording().tail();
955 let flight_recorder_events: Vec<RecordedEvent> = events
956 .into_iter()
957 .map(|event| RecordedEvent {
958 timestamp: format_timestamp(event.time),
959 seq: event.seq,
960 level: event.metadata.level().to_string(),
961 target: event.metadata.target().to_string(),
962 name: event.metadata.name().to_string(),
963 fields: event.json_value(),
964 })
965 .collect();
966
967 let flight_recorder = if flight_recorder_events.is_empty() {
968 None
969 } else {
970 serde_json::to_string(&flight_recorder_events).ok()
971 };
972
973 let supervisor = cell
974 .parent()
975 .map(|p| IntrospectRef::Actor(p.actor_addr().clone()));
976
977 // FI-3: failure_info is computed from the same status value as
978 // actor_status, ensuring they agree on whether the actor failed.
979 let failure = if status.is_failed() {
980 cell.supervision_event().and_then(|event| {
981 let root = event.actually_failing_actor()?;
982 Some(FailureSnapshot {
983 error_message: event.actor_status.to_string(),
984 root_cause_actor: root.actor_id.clone(),
985 root_cause_name: root.display_name.clone(),
986 occurred_at: event.occurred_at,
987 is_propagated: root.actor_id != actor_id.clone(),
988 })
989 })
990 } else {
991 None
992 };
993
994 let snap = ActorSnapshot {
995 status_str: status.to_string(),
996 is_system: cell.is_system(),
997 last_handler: last_handler.map(|info| info.to_string()),
998 flight_recorder,
999 failure,
1000 };
1001
1002 let attrs = build_actor_attrs(cell, &snap);
1003
1004 IntrospectResult {
1005 identity: IntrospectRef::Actor(actor_id.clone()),
1006 attrs,
1007 children,
1008 parent: supervisor,
1009 as_of: SystemTime::now(),
1010 }
1011}
1012
1013/// Serve introspection for one actor.
1014///
1015/// This runs on a dedicated Tokio task owned by the actor runtime. It
1016/// handles [`IntrospectMessage`] by reading [`InstanceCell`] directly
1017/// and replying through the owning [`Proc`](crate::Proc). The actor's
1018/// message loop never sees these messages, so a stuck actor can still
1019/// be introspected.
1020///
1021/// The task's lifetime is controlled by `shutdown`, not by terminal
1022/// [`ActorStatus`](crate::actor::ActorStatus). Runtime teardown sends
1023/// the final status through `shutdown`; this task then builds and
1024/// stores the terminated snapshot with that status and exits. The
1025/// actor's serving loop, or the proc-managed lifecycle for detached
1026/// instances, joins this task before publishing terminal status, so
1027/// terminal status remains the single authoritative signal that the
1028/// actor's full runtime, including introspection, has shut down.
1029///
1030/// If the introspect receiver closes before shutdown, the task stops
1031/// accepting queries but remains alive until runtime shutdown. This
1032/// preserves the shutdown path that stores the post-mortem snapshot and
1033/// breaks the `InstanceCell` reference cycle.
1034///
1035/// # Invariants exercised
1036///
1037/// Exercises S1, S2, S4, S5, S6, S11 (see module doc).
1038pub(crate) async fn serve_introspect(
1039 cell: InstanceCell,
1040 mut receiver: crate::mailbox::PortReceiver<IntrospectMessage>,
1041 mut shutdown: tokio::sync::oneshot::Receiver<crate::actor::ActorStatus>,
1042) {
1043 use crate::mailbox::PortSender as _;
1044
1045 // Runtime shutdown, not terminal status, owns this task's lifetime.
1046 // Terminal status is published only after this task snapshots and exits.
1047 let mut receiver_open = true;
1048
1049 loop {
1050 let msg = tokio::select! {
1051 msg = receiver.recv(), if receiver_open => {
1052 match msg {
1053 Ok(msg) => msg,
1054 Err(_) => {
1055 receiver_open = false;
1056 continue;
1057 }
1058 }
1059 }
1060 terminal_status = &mut shutdown => {
1061 let Ok(terminal_status) = terminal_status else {
1062 break;
1063 };
1064 let snapshot = live_actor_payload_with_status(&cell, &terminal_status);
1065 cell.store_terminated_snapshot(snapshot);
1066 break;
1067 }
1068 };
1069
1070 let result = match msg {
1071 IntrospectMessage::Query { view, reply } => {
1072 let payload = match view {
1073 IntrospectView::Entity => match cell.published_attrs() {
1074 Some(published) => {
1075 let attrs_json =
1076 serde_json::to_string(&published).unwrap_or_else(|_| "{}".into());
1077 let children: Vec<IntrospectRef> =
1078 published.get(CHILDREN).cloned().unwrap_or_default();
1079 IntrospectResult {
1080 identity: IntrospectRef::Actor(cell.actor_addr().clone()),
1081 attrs: attrs_json,
1082 children,
1083 parent: cell
1084 .parent()
1085 .map(|p| IntrospectRef::Actor(p.actor_addr().clone())),
1086 as_of: SystemTime::now(),
1087 }
1088 }
1089 None => live_actor_payload(&cell),
1090 },
1091 IntrospectView::Actor => live_actor_payload(&cell),
1092 };
1093 cell.proc().serialize_and_send_once(
1094 reply,
1095 payload,
1096 crate::mailbox::monitored_return_handle(),
1097 )
1098 }
1099 IntrospectMessage::QueryChild { child_ref, reply } => {
1100 let child_ref_: Addr = child_ref.clone();
1101 let payload = cell.query_child(&child_ref_).unwrap_or_else(|| {
1102 let mut error_attrs = hyperactor_config::Attrs::new();
1103 error_attrs.set(ERROR_CODE, "not_found".to_string());
1104 error_attrs.set(
1105 ERROR_MESSAGE,
1106 format!("child {} not found (no callback registered)", child_ref),
1107 );
1108 // Use the queried child_ref as identity for the error node.
1109 let identity = match &child_ref {
1110 Addr::Proc(id) => IntrospectRef::Proc(id.clone()),
1111 Addr::Actor(id) => IntrospectRef::Actor(id.clone()),
1112 Addr::Port(id) => IntrospectRef::Actor(id.actor_addr()),
1113 };
1114 IntrospectResult {
1115 identity,
1116 attrs: serde_json::to_string(&error_attrs)
1117 .unwrap_or_else(|_| "{}".to_string()),
1118 children: Vec::new(),
1119 parent: None,
1120 as_of: SystemTime::now(),
1121 }
1122 });
1123 cell.proc().serialize_and_send_once(
1124 reply,
1125 payload,
1126 crate::mailbox::monitored_return_handle(),
1127 )
1128 }
1129 };
1130 if let Err(e) = result {
1131 tracing::debug!("introspect reply failed: {e}");
1132 }
1133 }
1134 tracing::debug!(
1135 actor_id = %cell.actor_addr(),
1136 "introspect task exiting"
1137 );
1138}
1139
1140#[cfg(test)]
1141mod tests {
1142 use super::*;
1143 use crate::ActorAddr;
1144 use crate::ProcAddr;
1145 use crate::actor::ActorErrorKind;
1146 use crate::actor::ActorStatus;
1147 use crate::channel::ChannelAddr;
1148 use crate::supervision::ActorSupervisionEvent;
1149
1150 /// Exercises IK-1 (see module doc).
1151 #[test]
1152 fn test_introspect_keys_are_tagged() {
1153 let cases = vec![
1154 ("status", STATUS.attrs()),
1155 ("status_reason", STATUS_REASON.attrs()),
1156 ("actor_type", ACTOR_TYPE.attrs()),
1157 ("messages_processed", MESSAGES_PROCESSED.attrs()),
1158 ("created_at", CREATED_AT.attrs()),
1159 ("last_handler", LAST_HANDLER.attrs()),
1160 ("total_processing_time_us", TOTAL_PROCESSING_TIME_US.attrs()),
1161 ("flight_recorder", FLIGHT_RECORDER.attrs()),
1162 ("is_system", IS_SYSTEM.attrs()),
1163 ("children", CHILDREN.attrs()),
1164 ("error_code", ERROR_CODE.attrs()),
1165 ("error_message", ERROR_MESSAGE.attrs()),
1166 ("failure_error_message", FAILURE_ERROR_MESSAGE.attrs()),
1167 ("failure_root_cause_actor", FAILURE_ROOT_CAUSE_ACTOR.attrs()),
1168 ("failure_root_cause_name", FAILURE_ROOT_CAUSE_NAME.attrs()),
1169 ("failure_occurred_at", FAILURE_OCCURRED_AT.attrs()),
1170 ("failure_is_propagated", FAILURE_IS_PROPAGATED.attrs()),
1171 ("instance_id", INSTANCE_ID.attrs()),
1172 ("queue_depth", ACTOR_QUEUE_DEPTH.attrs()),
1173 ("inbound_ordering", INBOUND_ORDERING.attrs()),
1174 ];
1175
1176 for (expected_name, meta) in &cases {
1177 // IK-1: see module doc.
1178 let introspect = meta
1179 .get(INTROSPECT)
1180 .unwrap_or_else(|| panic!("{expected_name}: missing INTROSPECT meta-attr"));
1181 assert_eq!(
1182 introspect.name, *expected_name,
1183 "short name mismatch for {expected_name}"
1184 );
1185 assert!(
1186 !introspect.desc.is_empty(),
1187 "{expected_name}: desc should not be empty"
1188 );
1189 }
1190
1191 // Exhaustiveness: verify cases covers all INTROSPECT-tagged
1192 // keys declared in this module.
1193 use hyperactor_config::attrs::AttrKeyInfo;
1194 let registry_count = inventory::iter::<AttrKeyInfo>()
1195 .filter(|info| {
1196 info.name.starts_with("hyperactor::introspect::")
1197 && info.meta.get(INTROSPECT).is_some()
1198 })
1199 .count();
1200 assert_eq!(
1201 cases.len(),
1202 registry_count,
1203 "test must cover all INTROSPECT-tagged keys in this module"
1204 );
1205 }
1206
1207 /// Exercises IK-2 (see module doc).
1208 #[test]
1209 fn test_introspect_short_names_are_globally_unique() {
1210 use hyperactor_config::attrs::AttrKeyInfo;
1211
1212 let mut seen = std::collections::HashMap::new();
1213 for info in inventory::iter::<AttrKeyInfo>() {
1214 let Some(introspect) = info.meta.get(INTROSPECT) else {
1215 continue;
1216 };
1217 // Metadata quality: every tagged key must have
1218 // non-empty name and desc.
1219 assert!(
1220 !introspect.name.is_empty(),
1221 "INTROSPECT key {:?} has empty name",
1222 info.name
1223 );
1224 assert!(
1225 !introspect.desc.is_empty(),
1226 "INTROSPECT key {:?} has empty desc",
1227 info.name
1228 );
1229 if let Some(prev_fq) = seen.insert(introspect.name.clone(), info.name) {
1230 panic!(
1231 "IK-2 violation: duplicate short name {:?} declared by both {:?} and {:?}",
1232 introspect.name, prev_fq, info.name
1233 );
1234 }
1235 }
1236 }
1237
1238 // IA-1 tests require spawning actors and live in actor.rs
1239 // where #[hyperactor::export] and test infrastructure are
1240 // available. IA-3 and IA-4 are tested below at the view level.
1241
1242 fn running_actor_attrs() -> Attrs {
1243 let mut attrs = Attrs::new();
1244 attrs.set(STATUS, "running".to_string());
1245 attrs.set(ACTOR_TYPE, "MyActor".to_string());
1246 attrs.set(INSTANCE_ID, uuid::Uuid::from_u128(0xfeed_face).to_string());
1247 attrs.set(MESSAGES_PROCESSED, 42u64);
1248 attrs.set(CREATED_AT, SystemTime::UNIX_EPOCH);
1249 attrs.set(IS_SYSTEM, false);
1250 attrs
1251 }
1252
1253 fn test_actor_id(proc_name: &str, actor_name: &str) -> ActorAddr {
1254 ProcAddr::singleton(ChannelAddr::Local(0), proc_name).actor_addr(actor_name)
1255 }
1256
1257 fn failed_actor_attrs() -> Attrs {
1258 let mut attrs = running_actor_attrs();
1259 attrs.set(STATUS, "failed".to_string());
1260 attrs.set(STATUS_REASON, "something broke".to_string());
1261 attrs.set(FAILURE_ERROR_MESSAGE, "boom".to_string());
1262 attrs.set(FAILURE_ROOT_CAUSE_ACTOR, test_actor_id("proc", "other"));
1263 attrs.set(FAILURE_ROOT_CAUSE_NAME, "OtherActor".to_string());
1264 attrs.set(FAILURE_OCCURRED_AT, SystemTime::UNIX_EPOCH);
1265 attrs.set(FAILURE_IS_PROPAGATED, true);
1266 attrs
1267 }
1268
1269 /// AV-1: from_attrs(to_attrs(v)) == v.
1270 #[test]
1271 fn test_actor_view_round_trip_running() {
1272 let view = ActorAttrsView::from_attrs(&running_actor_attrs()).unwrap();
1273 assert_eq!(view.status, "running");
1274 assert_eq!(view.actor_type, "MyActor");
1275 assert_eq!(view.messages_processed, 42);
1276 // Default values for the new fields when not set in the
1277 // running_actor_attrs() fixture.
1278 assert_eq!(view.queue_depth, 0);
1279 assert!(view.inbound_ordering.is_none());
1280 assert!(view.failure.is_none());
1281
1282 let round_tripped = ActorAttrsView::from_attrs(&view.to_attrs()).unwrap();
1283 assert_eq!(round_tripped, view);
1284 }
1285
1286 /// AV-1.
1287 #[test]
1288 fn test_actor_view_round_trip_failed() {
1289 let view = ActorAttrsView::from_attrs(&failed_actor_attrs()).unwrap();
1290 assert_eq!(view.status, "failed");
1291 let fi = view.failure.as_ref().unwrap();
1292 assert_eq!(fi.error_message, "boom");
1293 assert!(fi.is_propagated);
1294
1295 let round_tripped = ActorAttrsView::from_attrs(&view.to_attrs()).unwrap();
1296 assert_eq!(round_tripped, view);
1297 }
1298
1299 /// AV-2: missing required key rejected.
1300 #[test]
1301 fn test_actor_view_missing_status() {
1302 let mut attrs = Attrs::new();
1303 attrs.set(ACTOR_TYPE, "X".to_string());
1304 let err = ActorAttrsView::from_attrs(&attrs).unwrap_err();
1305 assert_eq!(err, AttrsViewError::MissingKey { key: "status" });
1306 }
1307
1308 /// AV-2.
1309 #[test]
1310 fn test_actor_view_missing_actor_type() {
1311 let mut attrs = Attrs::new();
1312 attrs.set(STATUS, "running".to_string());
1313 let err = ActorAttrsView::from_attrs(&attrs).unwrap_err();
1314 assert_eq!(err, AttrsViewError::MissingKey { key: "actor_type" });
1315 }
1316
1317 /// AV-2: `instance_id` is a required key on the actor view --
1318 /// every live actor's attrs bag carries one.
1319 #[test]
1320 fn test_actor_view_missing_instance_id() {
1321 let mut attrs = Attrs::new();
1322 attrs.set(STATUS, "running".to_string());
1323 attrs.set(ACTOR_TYPE, "X".to_string());
1324 let err = ActorAttrsView::from_attrs(&attrs).unwrap_err();
1325 assert_eq!(err, AttrsViewError::MissingKey { key: "instance_id" });
1326 }
1327
1328 /// AV-1 + IO-1: round-trip with inbound_ordering = Some(...).
1329 /// Pins that the typed `OrderingSnapshot` survives the
1330 /// Attrs encode/decode boundary.
1331 #[test]
1332 fn test_actor_view_round_trip_with_inbound_ordering() {
1333 use crate::ordering::OrderingSessionSnapshot;
1334 use crate::ordering::OrderingSnapshot;
1335
1336 let session_addr = test_actor_id("sender_proc", "sender_actor");
1337 let snapshot = OrderingSnapshot {
1338 enabled: true,
1339 sessions: vec![OrderingSessionSnapshot {
1340 session_id: uuid::Uuid::from_u128(7),
1341 sender: Some(session_addr),
1342 last_released_seq: 3,
1343 expected_next_seq: 4,
1344 buffered_count: 2,
1345 oldest_buffered_seq: Some(5),
1346 newest_buffered_seq: Some(6),
1347 }],
1348 skipped_session_count: 0,
1349 };
1350
1351 let mut attrs = running_actor_attrs();
1352 attrs.set(ACTOR_QUEUE_DEPTH, 7u64);
1353 attrs.set(INBOUND_ORDERING, snapshot.clone());
1354
1355 let view = ActorAttrsView::from_attrs(&attrs).unwrap();
1356 assert_eq!(view.queue_depth, 7);
1357 assert_eq!(view.inbound_ordering.as_ref(), Some(&snapshot));
1358
1359 let round_tripped = ActorAttrsView::from_attrs(&view.to_attrs()).unwrap();
1360 assert_eq!(round_tripped, view);
1361 }
1362
1363 /// AV-1 + IO-1: round-trip with inbound_ordering = None survives
1364 /// cleanly. Pins the "structural absence" semantics: round-tripping
1365 /// the view does not invent a Some({enabled: false}) value.
1366 #[test]
1367 fn test_actor_view_round_trip_without_inbound_ordering() {
1368 let view = ActorAttrsView::from_attrs(&running_actor_attrs()).unwrap();
1369 assert!(view.inbound_ordering.is_none());
1370
1371 let round_tripped = ActorAttrsView::from_attrs(&view.to_attrs()).unwrap();
1372 assert!(round_tripped.inbound_ordering.is_none());
1373 assert_eq!(round_tripped, view);
1374 }
1375
1376 #[test]
1377 fn test_actor_view_ia3_rejects_reason_on_running() {
1378 let mut attrs = running_actor_attrs();
1379 attrs.set(STATUS_REASON, "should not be here".to_string());
1380 let err = ActorAttrsView::from_attrs(&attrs).unwrap_err();
1381 assert!(matches!(
1382 err,
1383 AttrsViewError::InvariantViolation { label: "IA-3", .. }
1384 ));
1385 }
1386
1387 #[test]
1388 fn test_actor_view_ia3_allows_terminal_without_reason() {
1389 let mut attrs = running_actor_attrs();
1390 attrs.set(STATUS, "stopped".to_string());
1391 // No status_reason — should be fine.
1392 let view = ActorAttrsView::from_attrs(&attrs).unwrap();
1393 assert_eq!(view.status, "stopped");
1394 assert!(view.status_reason.is_none());
1395 }
1396
1397 #[test]
1398 fn test_actor_view_ia4_rejects_failed_without_failure_attrs() {
1399 let mut attrs = running_actor_attrs();
1400 attrs.set(STATUS, "failed".to_string());
1401 // No failure_* keys.
1402 let err = ActorAttrsView::from_attrs(&attrs).unwrap_err();
1403 assert!(matches!(
1404 err,
1405 AttrsViewError::InvariantViolation { label: "IA-4", .. }
1406 ));
1407 }
1408
1409 #[test]
1410 fn test_actor_view_ia4_rejects_failure_attrs_on_running() {
1411 let mut attrs = running_actor_attrs();
1412 attrs.set(FAILURE_ERROR_MESSAGE, "boom".to_string());
1413 attrs.set(FAILURE_ROOT_CAUSE_ACTOR, test_actor_id("proc", "x"));
1414 attrs.set(FAILURE_OCCURRED_AT, SystemTime::UNIX_EPOCH);
1415 let err = ActorAttrsView::from_attrs(&attrs).unwrap_err();
1416 assert!(matches!(
1417 err,
1418 AttrsViewError::InvariantViolation { label: "IA-4", .. }
1419 ));
1420 }
1421
1422 /// AV-2: partial failure set → missing key.
1423 #[test]
1424 fn test_actor_view_partial_failure_attrs_rejected() {
1425 let mut attrs = running_actor_attrs();
1426 attrs.set(STATUS, "failed".to_string());
1427 // Only one of the three required failure keys.
1428 attrs.set(FAILURE_ERROR_MESSAGE, "boom".to_string());
1429 let err = ActorAttrsView::from_attrs(&attrs).unwrap_err();
1430 assert_eq!(
1431 err,
1432 AttrsViewError::MissingKey {
1433 key: "failure_root_cause_actor"
1434 }
1435 );
1436 }
1437
1438 /// Exercises FI-7 and FI-8 (see module doc): when a parent fails
1439 /// due to an unhandled Stopped child event, structured failure
1440 /// attrs must name the stopped child as
1441 /// `failure_root_cause_actor` (FI-7) and report
1442 /// `failure_is_propagated == true` (FI-8).
1443 ///
1444 /// Partially white-box: re-creates `FailureSnapshot` construction
1445 /// from `live_actor_payload` because that function requires an
1446 /// `InstanceCell`. This test will fail if
1447 /// `actually_failing_actor()` regresses, because that helper is
1448 /// the shared decision point for root-cause attribution. See
1449 /// `test_propagated_failure_info` in `proc.rs` for end-to-end
1450 /// integration coverage.
1451 #[test]
1452 fn test_fi7_fi8_propagated_stopped_child() {
1453 let proc_id = ProcAddr::singleton(ChannelAddr::Local(0), "test_proc");
1454 let child_id = proc_id.actor_addr("proc_agent");
1455 let parent_id = proc_id.actor_addr("mesh_actor");
1456
1457 let child_event = ActorSupervisionEvent::new(
1458 child_id.clone(),
1459 Some("proc_agent".into()),
1460 ActorStatus::Stopped("host died".into()),
1461 None,
1462 );
1463 let parent_event = ActorSupervisionEvent::new(
1464 parent_id.clone(),
1465 Some("mesh_actor".into()),
1466 ActorStatus::Failed(ActorErrorKind::UnhandledSupervisionEvent(Box::new(
1467 child_event,
1468 ))),
1469 None,
1470 );
1471
1472 // -- reproduce FailureSnapshot construction (same logic as
1473 // live_actor_payload) --
1474 let root = parent_event
1475 .actually_failing_actor()
1476 .expect("parent_event is a failure");
1477 let snap = FailureSnapshot {
1478 error_message: parent_event.actor_status.to_string(),
1479 root_cause_actor: root.actor_id.clone(),
1480 root_cause_name: root.display_name.clone(),
1481 occurred_at: parent_event.occurred_at,
1482 is_propagated: root.actor_id != parent_id,
1483 };
1484
1485 // FI-7: failure_root_cause_actor is the stopped child.
1486 assert_eq!(snap.root_cause_actor, child_id);
1487 // FI-8: failure_is_propagated is true.
1488 assert!(snap.is_propagated);
1489 // root_cause_name pinned before round-trip.
1490 assert_eq!(snap.root_cause_name.as_deref(), Some("proc_agent"));
1491
1492 // -- attrs round-trip through ActorAttrsView --
1493 let mut attrs = failed_actor_attrs();
1494 attrs.set(FAILURE_ERROR_MESSAGE, snap.error_message);
1495 attrs.set(FAILURE_ROOT_CAUSE_ACTOR, snap.root_cause_actor.clone());
1496 if let Some(name) = &snap.root_cause_name {
1497 attrs.set(FAILURE_ROOT_CAUSE_NAME, name.clone());
1498 }
1499 attrs.set(FAILURE_OCCURRED_AT, snap.occurred_at);
1500 attrs.set(FAILURE_IS_PROPAGATED, snap.is_propagated);
1501
1502 let view = ActorAttrsView::from_attrs(&attrs).unwrap();
1503 assert_eq!(view.status, "failed");
1504 let fi = view.failure.as_ref().expect("failure_info must be present");
1505 // FI-7: failure_root_cause_actor survives attrs round-trip.
1506 assert_eq!(fi.root_cause_actor, child_id);
1507 // FI-8: failure_is_propagated survives attrs round-trip.
1508 assert!(fi.is_propagated);
1509 // root_cause_name also survives.
1510 assert_eq!(fi.root_cause_name.as_deref(), Some("proc_agent"));
1511 }
1512}