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

meerkat_mobkit/unified_runtime/
types.rs
