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}
51
52impl Display for UnifiedRuntimeBootstrapError {
53    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
54        match self {
55            Self::Mob(err) => write!(f, "failed to bootstrap mob runtime: {err}"),
56            Self::Module(err) => write!(f, "failed to bootstrap module runtime: {err:?}"),
57            Self::ModuleStartupThreadPanicked => {
58                write!(
59                    f,
60                    "failed to bootstrap module runtime: startup thread panicked"
61                )
62            }
63            Self::PreSpawnHook(err) => {
64                write!(f, "pre-spawn hook failed: {err}")
65            }
66            Self::IdentityFirst(err) => {
67                write!(f, "identity-first bootstrap failed: {err}")
68            }
69            Self::ModuleStartupRollbackFailed {
70                startup_error,
71                rollback_error,
72            } => {
73                write!(
74                    f,
75                    "failed to bootstrap unified runtime: startup error ({startup_error}) and rollback failed: {rollback_error}"
76                )
77            }
78        }
79    }
80}
81
82impl std::error::Error for UnifiedRuntimeBootstrapError {}
83
84#[derive(Debug, Clone, Copy, PartialEq, Eq)]
85pub enum UnifiedRuntimeBuilderField {
86    MobSpec,
87    ModuleConfig,
88    Timeout,
89}
90
91#[derive(Debug)]
92pub enum UnifiedRuntimeBuilderError {
93    MissingRequiredField(UnifiedRuntimeBuilderField),
94    Bootstrap(UnifiedRuntimeBootstrapError),
95    /// Failed to read a definition TOML file or create a state directory.
96    Io(String),
97    /// Failed to parse a mob definition TOML.
98    DefinitionLoad(String),
99    /// Conflicting builder configuration (e.g., persistent_state + continuity_store).
100    ConflictingConfiguration(String),
101}
102
103impl Display for UnifiedRuntimeBuilderError {
104    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
105        match self {
106            Self::MissingRequiredField(UnifiedRuntimeBuilderField::MobSpec) => {
107                write!(f, "missing required builder field: mob_spec or definition")
108            }
109            Self::MissingRequiredField(UnifiedRuntimeBuilderField::ModuleConfig) => {
110                write!(f, "missing required builder field: module_config")
111            }
112            Self::MissingRequiredField(UnifiedRuntimeBuilderField::Timeout) => {
113                write!(f, "missing required builder field: timeout")
114            }
115            Self::Bootstrap(err) => write!(f, "{err}"),
116            Self::Io(msg) => write!(f, "{msg}"),
117            Self::DefinitionLoad(msg) => write!(f, "{msg}"),
118            Self::ConflictingConfiguration(msg) => write!(f, "conflicting configuration: {msg}"),
119        }
120    }
121}
122
123impl std::error::Error for UnifiedRuntimeBuilderError {}
124
125#[derive(Debug)]
126pub enum UnifiedRuntimeError {
127    Normalize(NormalizationError),
128    Subscribe(SubscribeError),
129    ScheduleValidation(ScheduleValidationError),
130    RuntimeShuttingDown,
131    ScheduleDispatchThreadPanicked,
132}
133
134impl Display for UnifiedRuntimeError {
135    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
136        match self {
137            Self::Normalize(err) => write!(f, "failed to normalize unified event: {err:?}"),
138            Self::Subscribe(err) => write!(f, "failed to subscribe to unified events: {err:?}"),
139            Self::ScheduleValidation(err) => {
140                write!(f, "failed to dispatch schedule tick: {err:?}")
141            }
142            Self::RuntimeShuttingDown => {
143                write!(
144                    f,
145                    "failed to dispatch schedule tick: unified runtime is shutting down"
146                )
147            }
148            Self::ScheduleDispatchThreadPanicked => {
149                write!(
150                    f,
151                    "failed to dispatch schedule tick: dispatch thread panicked"
152                )
153            }
154        }
155    }
156}
157
158impl std::error::Error for UnifiedRuntimeError {}
159
160impl From<NormalizationError> for UnifiedRuntimeError {
161    fn from(value: NormalizationError) -> Self {
162        Self::Normalize(value)
163    }
164}
165
166impl From<SubscribeError> for UnifiedRuntimeError {
167    fn from(value: SubscribeError) -> Self {
168        Self::Subscribe(value)
169    }
170}
171
172impl From<ScheduleValidationError> for UnifiedRuntimeError {
173    fn from(value: ScheduleValidationError) -> Self {
174        Self::ScheduleValidation(value)
175    }
176}
177
178#[derive(Debug)]
179pub struct UnifiedRuntimeShutdownReport {
180    pub drain: ShutdownDrainReport,
181    pub module_shutdown: RuntimeShutdownReport,
182    pub mob_stop: Result<(), MobRuntimeError>,
183}
184
185#[derive(Debug)]
186pub struct UnifiedRuntimeRunReport {
187    pub serve_result: std::io::Result<()>,
188    pub shutdown: UnifiedRuntimeShutdownReport,
189}
190
191/// Report from a rediscover operation (reset + re-run discovery + reconcile edges).
192#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
193pub struct RediscoverReport {
194    /// Number of members spawned by discovery.
195    pub spawned: Vec<String>,
196    /// Edge reconciliation report (if EdgeDiscovery is configured).
197    pub edges: UnifiedRuntimeReconcileEdgesReport,
198}
199
200#[derive(Debug, Clone, PartialEq, Eq)]
201pub struct UnifiedRuntimeReconcileRoutingReport {
202    pub router_module_loaded: bool,
203    pub active_members: Vec<String>,
204    pub added_route_keys: Vec<String>,
205    pub removed_route_keys: Vec<String>,
206}
207
208/// Per-identity reconcile failure — re-export of the canonical
209/// meerkat-contracts wire shape so SDK consumers see the same field
210/// names whether they go through `mob/reconcile` or `mobkit/reconcile`.
211pub use meerkat_contracts::MobReconcileFailureWire as MobReconcileFailure;
212
213/// Roster half of a reconcile pass — re-export of meerkat-contracts'
214/// canonical wire shape. `spawned: Vec<MobSpawnReceiptWire>` carries the
215/// server-resolved `WireMemberRef` per receipt, replacing the
216/// identity-string list mobkit projected before 0.6.
217pub use meerkat_contracts::MobReconcileReportWire as MobReconcileReport;
218
219/// Project meerkat's native `ReconcileReport` into the canonical wire shape.
220///
221/// Mirrors the `mob/reconcile` RPC handler's projection in
222/// `meerkat-rpc/src/handlers/mob.rs`, with one mobkit-specific step: the
223/// report's roster member ids are comms-safe encodings (meerkat 0.7
224/// `MemberCommsName`), and this is a projection boundary, so every id is
225/// decoded back to the public alias consoles/SDKs address members by.
226pub fn meerkat_reconcile_report_to_wire(
227    mob_id: &str,
228    report: meerkat_mob::runtime::reconcile::ReconcileReport,
229) -> MobReconcileReport {
230    use meerkat_contracts::{MobSpawnReceiptWire, WireMemberRef};
231    let alias_of =
232        |id: &str| -> String { crate::member_comms_id::runtime_alias_str(id).into_owned() };
233    MobReconcileReport {
234        desired: report
235            .desired
236            .into_iter()
237            .map(|id| alias_of(id.as_str()))
238            .collect(),
239        retained: report
240            .retained
241            .into_iter()
242            .map(|id| alias_of(id.as_str()))
243            .collect(),
244        spawned: report
245            .spawned
246            .into_iter()
247            .map(|receipt| {
248                let identity_str = alias_of(receipt.agent_identity.as_str());
249                MobSpawnReceiptWire {
250                    member_ref: WireMemberRef::encode(mob_id, &identity_str),
251                    agent_identity: identity_str,
252                }
253            })
254            .collect(),
255        retired: report
256            .retired
257            .into_iter()
258            .map(|id| alias_of(id.as_str()))
259            .collect(),
260        failures: report
261            .failures
262            .into_iter()
263            .map(|failure| MobReconcileFailure {
264                agent_identity: alias_of(failure.agent_identity.as_str()),
265                stage: match failure.stage {
266                    meerkat_mob::runtime::reconcile::ReconcileStage::Spawn => {
267                        meerkat_contracts::WireMobReconcileStage::Spawn
268                    }
269                    meerkat_mob::runtime::reconcile::ReconcileStage::Retire => {
270                        meerkat_contracts::WireMobReconcileStage::Retire
271                    }
272                },
273                error: meerkat_contracts::WireMobError {
274                    code: meerkat_mob::mob_error_wire_code(&failure.error),
275                    message: failure.error.to_string(),
276                },
277            })
278            .collect(),
279    }
280}
281
282// Eq is dropped because the canonical wire `MobReconcileReportWire` does
283// not implement `Eq` (its nested types are PartialEq only).
284#[derive(Debug, Clone, PartialEq)]
285pub struct UnifiedRuntimeReconcileReport {
286    pub mob: MobReconcileReport,
287    pub edges: UnifiedRuntimeReconcileEdgesReport,
288    pub routing: UnifiedRuntimeReconcileRoutingReport,
289}
290
291#[derive(Debug)]
292pub enum UnifiedRuntimeReconcileError {
293    Mob(MobRuntimeError),
294    RouteMutation(RuntimeRouteMutationError),
295    /// Meerkat 0.6's `MobHandle::reconcile` collects per-identity failures
296    /// into the returned report rather than returning `Err` on first failure.
297    /// `UnifiedRuntime::reconcile` re-lifts that into an error variant so
298    /// Rust callers using `?` still see failure propagation, while keeping
299    /// the full report available for inspection.
300    PartialFailure(Box<UnifiedRuntimeReconcileReport>),
301}
302
303impl Display for UnifiedRuntimeReconcileError {
304    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
305        match self {
306            Self::Mob(err) => write!(f, "failed to reconcile mob roster: {err}"),
307            Self::RouteMutation(err) => {
308                write!(f, "failed to reconcile routing wiring: {err:?}")
309            }
310            Self::PartialFailure(report) => {
311                write!(
312                    f,
313                    "reconcile completed with {} per-identity failure(s): {:?}",
314                    report.mob.failures.len(),
315                    report.mob.failures
316                )
317            }
318        }
319    }
320}
321
322impl std::error::Error for UnifiedRuntimeReconcileError {}
323
324#[derive(Debug)]
325pub struct ShutdownDrainReport {
326    pub drained_count: usize,
327    pub timed_out: bool,
328    pub drain_duration_ms: u64,
329}
330
331/// Operational error event for alerting.
332///
333/// Fired via the `on_error` hook when runtime operations fail. Apps
334/// match on variants to decide alerting (Slack, PagerDuty, log, etc.).
335///
336/// Marked `#[non_exhaustive]` — new variants can be added without
337/// breaking downstream match arms (use a `_` wildcard).
338///
339/// **Wired fire points:**
340/// - `SpawnFailure` — `mob_ops.rs` spawn error path
341/// - `ReconcileIncomplete` — `edge_reconcile.rs` after `reconcile_edges`
342/// - `RediscoverFailure` — `lifecycle.rs` rediscover error path
343/// - `HostLoopCrash` — `lifecycle.rs` detects `run_failed` agent events during drain
344/// - `CheckpointFailure` — via `run_periodic_gc_with_error_callback` in session store
345/// - `IdentityMaterializationFailure` — identity-first peer/fleet hydration skipped a member
346#[non_exhaustive]
347#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
348#[serde(tag = "category", rename_all = "snake_case")]
349pub enum ErrorEvent {
350    SpawnFailure {
351        member_id: String,
352        profile: String,
353        error: String,
354    },
355    ReconcileIncomplete {
356        failures: usize,
357        skipped: usize,
358    },
359    CheckpointFailure {
360        session_id: String,
361        error: String,
362    },
363    HostLoopCrash {
364        member_id: String,
365        error: String,
366    },
367    RediscoverFailure {
368        error: String,
369    },
370    EventLogFlushFailure {
371        error: String,
372    },
373    IdentityMaterializationFailure {
374        identity: String,
375        initiator: Option<String>,
376        operation: String,
377        error: String,
378    },
379}
380
381impl Display for ErrorEvent {
382    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
383        match self {
384            Self::SpawnFailure {
385                member_id, error, ..
386            } => {
387                write!(f, "spawn_failure: {member_id}: {error}")
388            }
389            Self::ReconcileIncomplete { failures, skipped } => {
390                write!(
391                    f,
392                    "reconcile_incomplete: {failures} failures, {skipped} skipped"
393                )
394            }
395            Self::CheckpointFailure { session_id, error } => {
396                write!(f, "checkpoint_failure: {session_id}: {error}")
397            }
398            Self::HostLoopCrash { member_id, error } => {
399                write!(f, "host_loop_crash: {member_id}: {error}")
400            }
401            Self::RediscoverFailure { error } => {
402                write!(f, "rediscover_failure: {error}")
403            }
404            Self::EventLogFlushFailure { error } => {
405                write!(f, "event_log_flush_failure: {error}")
406            }
407            Self::IdentityMaterializationFailure {
408                identity,
409                initiator,
410                operation,
411                error,
412            } => {
413                if let Some(initiator) = initiator {
414                    write!(
415                        f,
416                        "identity_materialization_failure: {identity} for {initiator} during {operation}: {error}"
417                    )
418                } else {
419                    write!(
420                        f,
421                        "identity_materialization_failure: {identity} during {operation}: {error}"
422                    )
423                }
424            }
425        }
426    }
427}
meerkat_mobkit/unified_runtime/types.rs

meerkat_mobkit/unified_runtime/
types.rs
