meerkat_mobkit/unified_runtime/
types.rs

1//! Error types, hook definitions, and report structures for the unified runtime.
2
3use std::fmt::{Display, Formatter};
4
5use serde::{Deserialize, Serialize};
6
7use crate::mob_handle_runtime::MobRuntimeError;
8use crate::runtime::{
9    NormalizationError, RuntimeRouteMutationError, RuntimeShutdownReport, ScheduleValidationError,
10    SubscribeError,
11};
12
13use super::edge_types::{DesiredPeerEdge, EdgeReconcileFailure};
14
15/// Report from dynamic edge reconciliation.
16///
17/// Best-effort: partial success is reported clearly. Apps decide whether
18/// to treat failures as fatal.
19#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
20pub struct UnifiedRuntimeReconcileEdgesReport {
21    pub desired_edges: Vec<DesiredPeerEdge>,
22    pub wired_edges: Vec<DesiredPeerEdge>,
23    pub unwired_edges: Vec<DesiredPeerEdge>,
24    pub retained_edges: Vec<DesiredPeerEdge>,
25    pub preexisting_edges: Vec<DesiredPeerEdge>,
26    pub skipped_missing_members: Vec<DesiredPeerEdge>,
27    pub pruned_stale_managed_edges: Vec<DesiredPeerEdge>,
28    #[serde(default)]
29    pub failures: Vec<EdgeReconcileFailure>,
30}
31
32impl UnifiedRuntimeReconcileEdgesReport {
33    /// True if all desired edges were successfully applied or retained.
34    pub fn is_complete(&self) -> bool {
35        self.failures.is_empty() && self.skipped_missing_members.is_empty()
36    }
37}
38
39#[derive(Debug)]
40pub enum UnifiedRuntimeBootstrapError {
41    Mob(MobRuntimeError),
42    Module(crate::runtime::MobkitRuntimeError),
43    ModuleStartupThreadPanicked,
44    ModuleStartupRollbackFailed {
45        startup_error: Box<UnifiedRuntimeBootstrapError>,
46        rollback_error: MobRuntimeError,
47    },
48    PreSpawnHook(String),
49    IdentityFirst(String),
50    Topology(String),
51}
52
53impl Display for UnifiedRuntimeBootstrapError {
54    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
55        match self {
56            Self::Mob(err) => write!(f, "failed to bootstrap mob runtime: {err}"),
57            Self::Module(err) => write!(f, "failed to bootstrap module runtime: {err:?}"),
58            Self::ModuleStartupThreadPanicked => {
59                write!(
60                    f,
61                    "failed to bootstrap module runtime: startup thread panicked"
62                )
63            }
64            Self::PreSpawnHook(err) => {
65                write!(f, "pre-spawn hook failed: {err}")
66            }
67            Self::IdentityFirst(err) => {
68                write!(f, "identity-first bootstrap failed: {err}")
69            }
70            Self::Topology(err) => write!(f, "topology-control bootstrap failed: {err}"),
71            Self::ModuleStartupRollbackFailed {
72                startup_error,
73                rollback_error,
74            } => {
75                write!(
76                    f,
77                    "failed to bootstrap unified runtime: startup error ({startup_error}) and rollback failed: {rollback_error}"
78                )
79            }
80        }
81    }
82}
83
84impl std::error::Error for UnifiedRuntimeBootstrapError {}
85
86#[derive(Debug, Clone, Copy, PartialEq, Eq)]
87pub enum UnifiedRuntimeBuilderField {
88    MobSpec,
89    ModuleConfig,
90    Timeout,
91}
92
93#[derive(Debug)]
94pub enum UnifiedRuntimeBuilderError {
95    MissingRequiredField(UnifiedRuntimeBuilderField),
96    Bootstrap(UnifiedRuntimeBootstrapError),
97    /// Failed to read a definition TOML file or create a state directory.
98    Io(String),
99    /// Failed to parse a mob definition TOML.
100    DefinitionLoad(String),
101    /// Conflicting builder configuration (e.g., persistent_state + scratch_dir).
102    ConflictingConfiguration(String),
103    /// Storage layout refusal (file-name twins in the state directory).
104    StorageLayout(crate::storage_layout::StorageLayoutError),
105    /// A storage provider failed to open the realm's store set, or the
106    /// fail-closed durability rule refused it (M4).
107    StorageProvider(crate::storage_provider::MobKitStorageProviderError),
108}
109
110impl From<crate::storage_layout::StorageLayoutError> for UnifiedRuntimeBuilderError {
111    fn from(error: crate::storage_layout::StorageLayoutError) -> Self {
112        Self::StorageLayout(error)
113    }
114}
115
116impl From<crate::storage_provider::MobKitStorageProviderError> for UnifiedRuntimeBuilderError {
117    fn from(error: crate::storage_provider::MobKitStorageProviderError) -> Self {
118        Self::StorageProvider(error)
119    }
120}
121
122impl Display for UnifiedRuntimeBuilderError {
123    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
124        match self {
125            Self::MissingRequiredField(UnifiedRuntimeBuilderField::MobSpec) => {
126                write!(f, "missing required builder field: mob_spec or definition")
127            }
128            Self::MissingRequiredField(UnifiedRuntimeBuilderField::ModuleConfig) => {
129                write!(f, "missing required builder field: module_config")
130            }
131            Self::MissingRequiredField(UnifiedRuntimeBuilderField::Timeout) => {
132                write!(f, "missing required builder field: timeout")
133            }
134            Self::Bootstrap(err) => write!(f, "{err}"),
135            Self::Io(msg) => write!(f, "{msg}"),
136            Self::DefinitionLoad(msg) => write!(f, "{msg}"),
137            Self::ConflictingConfiguration(msg) => write!(f, "conflicting configuration: {msg}"),
138            Self::StorageLayout(err) => write!(f, "{err}"),
139            Self::StorageProvider(err) => write!(f, "{err}"),
140        }
141    }
142}
143
144impl std::error::Error for UnifiedRuntimeBuilderError {}
145
146#[derive(Debug)]
147pub enum UnifiedRuntimeError {
148    Normalize(NormalizationError),
149    Subscribe(SubscribeError),
150    ScheduleValidation(ScheduleValidationError),
151    RuntimeShuttingDown,
152    ScheduleDispatchThreadPanicked,
153}
154
155impl Display for UnifiedRuntimeError {
156    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
157        match self {
158            Self::Normalize(err) => write!(f, "failed to normalize unified event: {err:?}"),
159            Self::Subscribe(err) => write!(f, "failed to subscribe to unified events: {err:?}"),
160            Self::ScheduleValidation(err) => {
161                write!(f, "failed to dispatch schedule tick: {err:?}")
162            }
163            Self::RuntimeShuttingDown => {
164                write!(
165                    f,
166                    "failed to dispatch schedule tick: unified runtime is shutting down"
167                )
168            }
169            Self::ScheduleDispatchThreadPanicked => {
170                write!(
171                    f,
172                    "failed to dispatch schedule tick: dispatch thread panicked"
173                )
174            }
175        }
176    }
177}
178
179impl std::error::Error for UnifiedRuntimeError {}
180
181impl From<NormalizationError> for UnifiedRuntimeError {
182    fn from(value: NormalizationError) -> Self {
183        Self::Normalize(value)
184    }
185}
186
187impl From<SubscribeError> for UnifiedRuntimeError {
188    fn from(value: SubscribeError) -> Self {
189        Self::Subscribe(value)
190    }
191}
192
193impl From<ScheduleValidationError> for UnifiedRuntimeError {
194    fn from(value: ScheduleValidationError) -> Self {
195        Self::ScheduleValidation(value)
196    }
197}
198
199/// Exact disposition of identity-first lease authority during shutdown.
200#[derive(Debug, Clone, PartialEq, Eq)]
201pub enum IdentityAuthorityReleaseOutcome {
202    /// The runtime did not have identity-first authority to release.
203    NotConfigured,
204    /// Every retained grant was released from the configured provider.
205    Released { grant_count: usize },
206    /// The provider rejected or failed the exact release operation.
207    Failed { error: String },
208    /// A reset-superseded session/member cleanup obligation remained after
209    /// physical shutdown retries, so provider grants were retained.
210    SkippedResetCleanupFailed { error: String },
211    /// Physical members did not quiesce, so their grants were deliberately retained.
212    SkippedMobStopFailed,
213}
214
215#[derive(Debug)]
216pub struct UnifiedRuntimeShutdownReport {
217    pub drain: ShutdownDrainReport,
218    pub module_shutdown: RuntimeShutdownReport,
219    pub mob_stop: Result<(), MobRuntimeError>,
220    pub identity_authority_release: IdentityAuthorityReleaseOutcome,
221}
222
223impl UnifiedRuntimeShutdownReport {
224    /// True only when every shutdown phase that owns external authority or
225    /// child-process state completed successfully.
226    pub fn cleanup_completed(&self) -> bool {
227        !self.drain.timed_out
228            && self.mob_stop.is_ok()
229            && matches!(
230                &self.identity_authority_release,
231                IdentityAuthorityReleaseOutcome::NotConfigured
232                    | IdentityAuthorityReleaseOutcome::Released { .. }
233            )
234            && self.module_shutdown.orphan_processes == 0
235    }
236}
237
238#[derive(Debug)]
239pub struct UnifiedRuntimeRunReport {
240    pub serve_result: std::io::Result<()>,
241    pub shutdown: UnifiedRuntimeShutdownReport,
242}
243
244/// Report from a rediscover operation (reset + re-run discovery + reconcile edges).
245#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
246pub struct RediscoverReport {
247    /// Number of members spawned by discovery.
248    pub spawned: Vec<String>,
249    /// Edge reconciliation report (if EdgeDiscovery is configured).
250    pub edges: UnifiedRuntimeReconcileEdgesReport,
251}
252
253#[derive(Debug, Clone, PartialEq, Eq)]
254pub struct UnifiedRuntimeReconcileRoutingReport {
255    pub router_module_loaded: bool,
256    pub active_members: Vec<String>,
257    pub added_route_keys: Vec<String>,
258    pub removed_route_keys: Vec<String>,
259}
260
261/// Per-identity reconcile failure — re-export of the canonical
262/// meerkat-contracts wire shape so SDK consumers see the same field
263/// names whether they go through `mob/reconcile` or `mobkit/reconcile`.
264pub use meerkat_contracts::MobReconcileFailureWire as MobReconcileFailure;
265
266/// Roster half of a reconcile pass — re-export of meerkat-contracts'
267/// canonical wire shape. `spawned: Vec<MobSpawnReceiptWire>` carries the
268/// server-resolved `WireMemberRef` per receipt, replacing the
269/// identity-string list mobkit projected before 0.6.
270pub use meerkat_contracts::MobReconcileReportWire as MobReconcileReport;
271
272/// Project meerkat's native `ReconcileReport` into the canonical wire shape.
273///
274/// Mirrors the `mob/reconcile` RPC handler's projection in
275/// `meerkat-rpc/src/handlers/mob.rs`, with one mobkit-specific step: the
276/// report's roster member ids are comms-safe encodings (meerkat 0.7
277/// `MemberCommsName`), and this is a projection boundary, so every id is
278/// decoded back to the public alias consoles/SDKs address members by.
279pub fn meerkat_reconcile_report_to_wire(
280    mob_id: &str,
281    report: meerkat_mob::runtime::reconcile::ReconcileReport,
282) -> MobReconcileReport {
283    use meerkat_contracts::{MobSpawnReceiptWire, WireMemberRef};
284    let alias_of =
285        |id: &str| -> String { crate::member_comms_id::runtime_alias_str(id).into_owned() };
286    MobReconcileReport {
287        desired: report
288            .desired
289            .into_iter()
290            .map(|id| alias_of(id.as_str()))
291            .collect(),
292        retained: report
293            .retained
294            .into_iter()
295            .map(|id| alias_of(id.as_str()))
296            .collect(),
297        spawned: report
298            .spawned
299            .into_iter()
300            .map(|receipt| {
301                let identity_str = alias_of(receipt.agent_identity.as_str());
302                MobSpawnReceiptWire {
303                    member_ref: WireMemberRef::encode(mob_id, &identity_str),
304                    agent_identity: identity_str,
305                }
306            })
307            .collect(),
308        retired: report
309            .retired
310            .into_iter()
311            .map(|id| alias_of(id.as_str()))
312            .collect(),
313        failures: report
314            .failures
315            .into_iter()
316            .map(|failure| MobReconcileFailure {
317                agent_identity: alias_of(failure.agent_identity.as_str()),
318                stage: match failure.stage {
319                    meerkat_mob::runtime::reconcile::ReconcileStage::Spawn => {
320                        meerkat_contracts::WireMobReconcileStage::Spawn
321                    }
322                    meerkat_mob::runtime::reconcile::ReconcileStage::Retire => {
323                        meerkat_contracts::WireMobReconcileStage::Retire
324                    }
325                },
326                error: meerkat_contracts::WireMobError {
327                    code: meerkat_mob::mob_error_wire_code(&failure.error),
328                    message: failure.error.to_string(),
329                },
330            })
331            .collect(),
332    }
333}
334
335// Eq is dropped because the canonical wire `MobReconcileReportWire` does
336// not implement `Eq` (its nested types are PartialEq only).
337#[derive(Debug, Clone, PartialEq)]
338pub struct UnifiedRuntimeReconcileReport {
339    pub mob: MobReconcileReport,
340    pub edges: UnifiedRuntimeReconcileEdgesReport,
341    pub routing: UnifiedRuntimeReconcileRoutingReport,
342}
343
344#[derive(Debug)]
345pub enum UnifiedRuntimeReconcileError {
346    Mob(MobRuntimeError),
347    RouteMutation(RuntimeRouteMutationError),
348    /// Meerkat 0.6's `MobHandle::reconcile` collects per-identity failures
349    /// into the returned report rather than returning `Err` on first failure.
350    /// `UnifiedRuntime::reconcile` re-lifts that into an error variant so
351    /// Rust callers using `?` still see failure propagation, while keeping
352    /// the full report available for inspection.
353    PartialFailure(Box<UnifiedRuntimeReconcileReport>),
354}
355
356impl Display for UnifiedRuntimeReconcileError {
357    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
358        match self {
359            Self::Mob(err) => write!(f, "failed to reconcile mob roster: {err}"),
360            Self::RouteMutation(err) => {
361                write!(f, "failed to reconcile routing wiring: {err:?}")
362            }
363            Self::PartialFailure(report) => {
364                write!(
365                    f,
366                    "reconcile completed with {} per-identity failure(s): {:?}",
367                    report.mob.failures.len(),
368                    report.mob.failures
369                )
370            }
371        }
372    }
373}
374
375impl std::error::Error for UnifiedRuntimeReconcileError {}
376
377#[derive(Debug)]
378pub struct ShutdownDrainReport {
379    pub drained_count: usize,
380    pub timed_out: bool,
381    pub drain_duration_ms: u64,
382}
383
384/// Operational error event for alerting.
385///
386/// Fired via the `on_error` hook when runtime operations fail. Apps
387/// match on variants to decide alerting (Slack, PagerDuty, log, etc.).
388///
389/// Marked `#[non_exhaustive]` — new variants can be added without
390/// breaking downstream match arms (use a `_` wildcard).
391///
392/// **Wired fire points:**
393/// - `SpawnFailure` — `mob_ops.rs` spawn error path
394/// - `ReconcileIncomplete` — `edge_reconcile.rs` after `reconcile_edges`
395/// - `RediscoverFailure` — `lifecycle.rs` rediscover error path
396/// - `HostLoopCrash` — `lifecycle.rs` detects `run_failed` agent events during drain
397/// - `CheckpointFailure` — via `run_periodic_gc_with_error_callback` in session store
398/// - `IdentityMaterializationFailure` — identity-first peer/fleet hydration skipped a member
399#[non_exhaustive]
400#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
401#[serde(tag = "category", rename_all = "snake_case")]
402pub enum ErrorEvent {
403    SpawnFailure {
404        member_id: String,
405        profile: String,
406        error: String,
407    },
408    ReconcileIncomplete {
409        failures: usize,
410        skipped: usize,
411    },
412    CheckpointFailure {
413        session_id: String,
414        error: String,
415    },
416    HostLoopCrash {
417        member_id: String,
418        error: String,
419    },
420    RediscoverFailure {
421        error: String,
422    },
423    EventLogFlushFailure {
424        error: String,
425    },
426    IdentityMaterializationFailure {
427        identity: String,
428        initiator: Option<String>,
429        operation: String,
430        error: String,
431    },
432}
433
434impl Display for ErrorEvent {
435    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
436        match self {
437            Self::SpawnFailure {
438                member_id, error, ..
439            } => {
440                write!(f, "spawn_failure: {member_id}: {error}")
441            }
442            Self::ReconcileIncomplete { failures, skipped } => {
443                write!(
444                    f,
445                    "reconcile_incomplete: {failures} failures, {skipped} skipped"
446                )
447            }
448            Self::CheckpointFailure { session_id, error } => {
449                write!(f, "checkpoint_failure: {session_id}: {error}")
450            }
451            Self::HostLoopCrash { member_id, error } => {
452                write!(f, "host_loop_crash: {member_id}: {error}")
453            }
454            Self::RediscoverFailure { error } => {
455                write!(f, "rediscover_failure: {error}")
456            }
457            Self::EventLogFlushFailure { error } => {
458                write!(f, "event_log_flush_failure: {error}")
459            }
460            Self::IdentityMaterializationFailure {
461                identity,
462                initiator,
463                operation,
464                error,
465            } => {
466                if let Some(initiator) = initiator {
467                    write!(
468                        f,
469                        "identity_materialization_failure: {identity} for {initiator} during {operation}: {error}"
470                    )
471                } else {
472                    write!(
473                        f,
474                        "identity_materialization_failure: {identity} during {operation}: {error}"
475                    )
476                }
477            }
478        }
479    }
480}
481
482#[cfg(test)]
483mod tests {
484    use super::*;
485
486    fn completed_shutdown_report() -> UnifiedRuntimeShutdownReport {
487        UnifiedRuntimeShutdownReport {
488            drain: ShutdownDrainReport {
489                drained_count: 1,
490                timed_out: false,
491                drain_duration_ms: 2,
492            },
493            module_shutdown: RuntimeShutdownReport {
494                terminated_modules: vec!["router".to_string()],
495                orphan_processes: 0,
496            },
497            mob_stop: Ok(()),
498            identity_authority_release: IdentityAuthorityReleaseOutcome::NotConfigured,
499        }
500    }
501
502    #[test]
503    fn shutdown_cleanup_attestation_requires_every_authority_boundary() {
504        let mut report = completed_shutdown_report();
505        assert!(report.cleanup_completed());
506
507        report.identity_authority_release =
508            IdentityAuthorityReleaseOutcome::Released { grant_count: 1 };
509        assert!(report.cleanup_completed());
510
511        let mut report = completed_shutdown_report();
512        report.drain.timed_out = true;
513        assert!(!report.cleanup_completed());
514
515        let mut report = completed_shutdown_report();
516        report.mob_stop = Err(MobRuntimeError::InvalidConfig(
517            "mob stop failed".to_string(),
518        ));
519        assert!(!report.cleanup_completed());
520
521        let mut report = completed_shutdown_report();
522        report.identity_authority_release = IdentityAuthorityReleaseOutcome::Failed {
523            error: "provider release failed".to_string(),
524        };
525        assert!(!report.cleanup_completed());
526
527        let mut report = completed_shutdown_report();
528        report.identity_authority_release = IdentityAuthorityReleaseOutcome::SkippedMobStopFailed;
529        assert!(!report.cleanup_completed());
530
531        let mut report = completed_shutdown_report();
532        report.identity_authority_release =
533            IdentityAuthorityReleaseOutcome::SkippedResetCleanupFailed {
534                error: "superseded member retained".to_string(),
535            };
536        assert!(!report.cleanup_completed());
537
538        let mut report = completed_shutdown_report();
539        report.module_shutdown.orphan_processes = 1;
540        assert!(!report.cleanup_completed());
541    }
542}
meerkat_mobkit/unified_runtime/types.rs

meerkat_mobkit/unified_runtime/
types.rs
