meerkat_core/
ops_lifecycle.rs

1//! Canonical async-operation lifecycle seam for shared child/background work.
2
3use serde::{Deserialize, Serialize};
4
5use std::future::Future;
6use std::pin::Pin;
7
8use crate::comms::{PeerAddress, PeerId, TrustedPeerDescriptor};
9use crate::lifecycle::{RunId, WaitRequestId};
10pub use crate::ops::{OperationId, OperationResult};
11use crate::runtime_epoch::EpochCursorState;
12use crate::types::SessionId;
13
14/// Default maximum number of completed operations to retain before eviction.
15pub const DEFAULT_MAX_COMPLETED: usize = 256;
16
17/// The kind of async operation tracked by the shared lifecycle registry.
18#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
19#[serde(rename_all = "snake_case")]
20pub enum OperationKind {
21    MobMemberChild,
22    BackgroundToolOp,
23    BackgroundToolCapacitySlot,
24}
25
26impl OperationKind {
27    /// Closed variant set mirrored from the generated MeerkatMachine named type.
28    pub const ALL: [Self; 3] = [
29        Self::MobMemberChild,
30        Self::BackgroundToolOp,
31        Self::BackgroundToolCapacitySlot,
32    ];
33
34    /// Generated named-type variant spelling used by drift ratchets.
35    pub const fn generated_variant(self) -> &'static str {
36        match self {
37            Self::MobMemberChild => "MobMemberChild",
38            Self::BackgroundToolOp => "BackgroundToolOp",
39            Self::BackgroundToolCapacitySlot => "BackgroundToolCapacitySlot",
40        }
41    }
42
43    /// Whether this kind can expose a peer-ready handoff.
44    pub fn expects_peer_channel(self) -> bool {
45        matches!(self, Self::MobMemberChild)
46    }
47}
48
49/// Generated-authority-owned source identity for an async operation.
50#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
51#[serde(tag = "kind", rename_all = "snake_case")]
52pub enum OperationSource {
53    SessionChild {
54        session_id: SessionId,
55    },
56    BackendPeer {
57        peer_id: PeerId,
58        address: PeerAddress,
59    },
60}
61
62impl OperationSource {
63    pub fn session_child(session_id: SessionId) -> Self {
64        Self::SessionChild { session_id }
65    }
66
67    pub fn backend_peer(peer_id: PeerId, address: PeerAddress) -> Self {
68        Self::BackendPeer { peer_id, address }
69    }
70}
71
72/// Lifecycle-relevant registration payload for an operation.
73#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
74pub struct OperationSpec {
75    pub id: OperationId,
76    pub kind: OperationKind,
77    /// Canonical owner bridge session binding for this operation.
78    ///
79    /// The serde field name `owner_session_id` is durable in
80    /// `PersistedOpsSnapshot`; renaming it is a persisted-shape change.
81    pub owner_session_id: SessionId,
82    pub display_name: String,
83    pub source_label: String,
84    #[serde(default, skip_serializing_if = "Option::is_none")]
85    pub operation_source: Option<OperationSource>,
86    pub child_session_id: Option<SessionId>,
87    pub expect_peer_channel: bool,
88}
89
90/// Peer-facing connection handoff surfaced once an operation is ready.
91#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
92pub struct OperationPeerHandle {
93    pub peer_name: crate::comms::PeerName,
94    pub trusted_peer: TrustedPeerDescriptor,
95}
96
97/// Progress update for a long-running async operation.
98#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
99pub struct OperationProgressUpdate {
100    pub message: String,
101    pub percent: Option<f32>,
102}
103
104/// Terminal lifecycle outcome recorded for an operation.
105///
106/// `Default` exists solely for generated machine-authority plumbing
107/// (`OptionValueExt::get` on `Option<OpTerminalPayload>` guard projections);
108/// the defaulted `Retired` value can never be admitted as truth because the
109/// generated guards reject any transition whose payload presence witness is
110/// `None` before the variant-match guard is consulted.
111#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
112#[serde(tag = "outcome_type", rename_all = "snake_case")]
113pub enum OperationTerminalOutcome {
114    Completed(OperationResult),
115    Failed {
116        error: String,
117    },
118    Aborted {
119        reason: Option<String>,
120    },
121    Cancelled {
122        reason: Option<String>,
123    },
124    #[default]
125    Retired,
126    Terminated {
127        reason: String,
128    },
129}
130
131/// Current lifecycle status for an operation.
132#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
133#[serde(rename_all = "snake_case")]
134pub enum OperationStatus {
135    Absent,
136    Provisioning,
137    Running,
138    Retiring,
139    Completed,
140    Failed,
141    Aborted,
142    Cancelled,
143    Retired,
144    Terminated,
145}
146
147impl OperationStatus {
148    /// Stable string representation for app-facing surfaces.
149    ///
150    /// Unlike `Debug` format, this is an explicit mapping that won't
151    /// produce uncontrolled strings when new variants are added.
152    pub fn as_str(self) -> &'static str {
153        match self {
154            Self::Absent => "absent",
155            Self::Provisioning => "provisioning",
156            Self::Running => "running",
157            Self::Retiring => "retiring",
158            Self::Completed => "completed",
159            Self::Failed => "failed",
160            Self::Aborted => "aborted",
161            Self::Cancelled => "cancelled",
162            Self::Retired => "retired",
163            Self::Terminated => "terminated",
164        }
165    }
166}
167
168/// Public result class for an operation lifecycle snapshot.
169///
170/// This is the typed domain value emitted by generated operation lifecycle
171/// authority before shell/tool surfaces project it into their presentation
172/// structs.
173#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
174#[serde(rename_all = "snake_case")]
175pub enum OperationPublicResultClass {
176    MissingAuthority,
177    Running,
178    Completed,
179    Failed,
180    Cancelled,
181}
182
183impl OperationPublicResultClass {
184    /// Closed variant set mirrored from the generated MeerkatMachine named type.
185    pub const ALL: [Self; 5] = [
186        Self::MissingAuthority,
187        Self::Running,
188        Self::Completed,
189        Self::Failed,
190        Self::Cancelled,
191    ];
192
193    /// Generated named-type variant spelling used by drift ratchets.
194    pub const fn generated_variant(self) -> &'static str {
195        match self {
196            Self::MissingAuthority => "MissingAuthority",
197            Self::Running => "Running",
198            Self::Completed => "Completed",
199            Self::Failed => "Failed",
200            Self::Cancelled => "Cancelled",
201        }
202    }
203}
204
205/// Generated classification for completion-feed entries that should wake the
206/// owning agent as detached background job completions.
207#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
208#[serde(rename_all = "snake_case")]
209pub enum OperationCompletionWakeClass {
210    Wake,
211    Ignore,
212}
213
214impl OperationCompletionWakeClass {
215    /// Closed variant set mirrored from the generated MeerkatMachine named type.
216    pub const ALL: [Self; 2] = [Self::Wake, Self::Ignore];
217
218    /// Generated named-type variant spelling used by drift ratchets.
219    pub const fn generated_variant(self) -> &'static str {
220        match self {
221            Self::Wake => "Wake",
222            Self::Ignore => "Ignore",
223        }
224    }
225}
226
227/// Operation lifecycle action observed by generated transition feedback.
228#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
229#[serde(rename_all = "snake_case")]
230pub enum OperationLifecycleAction {
231    Start,
232    Fail,
233    PeerReady,
234    ProgressReported,
235    Complete,
236    Abort,
237    Cancel,
238    RetireRequested,
239    RetireCompleted,
240    Terminate,
241}
242
243impl OperationLifecycleAction {
244    /// Closed variant set mirrored from the generated MeerkatMachine named type.
245    pub const ALL: [Self; 10] = [
246        Self::Start,
247        Self::Fail,
248        Self::PeerReady,
249        Self::ProgressReported,
250        Self::Complete,
251        Self::Abort,
252        Self::Cancel,
253        Self::RetireRequested,
254        Self::RetireCompleted,
255        Self::Terminate,
256    ];
257
258    /// Generated named-type variant spelling used by drift ratchets.
259    pub const fn generated_variant(self) -> &'static str {
260        match self {
261            Self::Start => "Start",
262            Self::Fail => "Fail",
263            Self::PeerReady => "PeerReady",
264            Self::ProgressReported => "ProgressReported",
265            Self::Complete => "Complete",
266            Self::Abort => "Abort",
267            Self::Cancel => "Cancel",
268            Self::RetireRequested => "RetireRequested",
269            Self::RetireCompleted => "RetireCompleted",
270            Self::Terminate => "Terminate",
271        }
272    }
273}
274
275/// Public snapshot of one operation's lifecycle state.
276#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
277pub struct OperationLifecycleSnapshot {
278    pub id: OperationId,
279    pub kind: OperationKind,
280    pub display_name: String,
281    #[serde(default, skip_serializing_if = "Option::is_none")]
282    pub operation_source: Option<OperationSource>,
283    pub status: OperationStatus,
284    /// Generated terminality classification for `status`.
285    pub terminal: bool,
286    /// Generated public result classification for `status`.
287    pub public_result_class: OperationPublicResultClass,
288    pub peer_ready: bool,
289    pub progress_count: u32,
290    pub watcher_count: u32,
291    pub terminal_outcome: Option<OperationTerminalOutcome>,
292    pub child_session_id: Option<SessionId>,
293    /// Peer handle info (exposed when peer_ready is true).
294    #[serde(skip_serializing_if = "Option::is_none", default)]
295    pub peer_handle: Option<OperationPeerHandle>,
296    /// Wall-clock epoch millis when the operation was registered.
297    #[serde(default)]
298    pub created_at_ms: u64,
299    /// Wall-clock epoch millis when provisioning succeeded (entered Running).
300    #[serde(skip_serializing_if = "Option::is_none", default)]
301    pub started_at_ms: Option<u64>,
302    /// Wall-clock epoch millis when the operation reached terminal state.
303    #[serde(skip_serializing_if = "Option::is_none", default)]
304    pub completed_at_ms: Option<u64>,
305    /// Monotonic elapsed millis from creation to terminal (computed from Instant).
306    #[serde(skip_serializing_if = "Option::is_none", default)]
307    pub elapsed_ms: Option<u64>,
308}
309
310/// One registry-owned watcher for a terminal lifecycle outcome.
311///
312/// This is a read-only waiter returned by [`OpsLifecycleRegistry`]. It does not
313/// expose terminal-outcome send authority to callers; registry implementations
314/// must resolve it only after generated operation lifecycle authority has
315/// accepted the terminal transition.
316pub type OperationCompletionWatch = Pin<
317    Box<
318        dyn Future<Output = Result<OperationTerminalOutcome, OperationCompletionWatchError>>
319            + Send
320            + 'static,
321    >,
322>;
323
324/// Mechanical failure while waiting for operation completion plumbing.
325///
326/// This is intentionally not an [`OperationTerminalOutcome`]: a dropped waiter
327/// channel is not async-operation terminal truth.
328#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
329pub enum OperationCompletionWatchError {
330    #[error("operation completion channel closed without an authorized terminal outcome")]
331    ChannelClosed,
332}
333
334/// Errors returned by the shared lifecycle registry.
335#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
336pub enum OpsLifecycleError {
337    #[error("operation already registered: {0}")]
338    AlreadyRegistered(OperationId),
339    #[error("operation not found: {0}")]
340    NotFound(OperationId),
341    #[error("invalid lifecycle transition for {id}: {status:?} -> {action}")]
342    InvalidTransition {
343        id: OperationId,
344        status: OperationStatus,
345        action: &'static str,
346    },
347    #[error("operation does not expect a peer handoff: {0}")]
348    PeerNotExpected(OperationId),
349    #[error("operation is already peer-ready: {0}")]
350    AlreadyPeerReady(OperationId),
351    #[error("max concurrent operations exceeded (limit: {limit}, active: {active})")]
352    MaxConcurrentExceeded { limit: usize, active: usize },
353    #[error("operation not supported: {0}")]
354    Unsupported(String),
355    #[error("wait_all already active")]
356    WaitAlreadyActive,
357    #[error("wait_all not active for request: {0}")]
358    WaitNotActive(WaitRequestId),
359    #[error("wait_all contains duplicate operation id: {0}")]
360    DuplicateWaitOperation(OperationId),
361    #[error("internal lifecycle registry error: {0}")]
362    Internal(String),
363}
364
365/// Authority-owned result of `wait_all()`.
366///
367/// Carries the per-operation outcomes alongside an authority-derived
368/// obligation token (`satisfied`). The obligation proves the authority owned
369/// the wait request lifecycle and emitted `WaitAllSatisfied` when the tracked
370/// barrier set became terminal.
371#[derive(Debug)]
372pub struct WaitAllResult {
373    /// Per-operation terminal outcomes.
374    pub outcomes: Vec<(OperationId, OperationTerminalOutcome)>,
375    /// Authority-validated obligation token for the ops_barrier_satisfaction protocol.
376    pub satisfied: WaitAllSatisfied,
377}
378
379/// Authority-owned obligation token emitted by the `WaitAllSatisfied` effect.
380///
381/// Created only by the `OpsLifecycleRegistry::wait_all()` implementation after
382/// the authority resolves an outstanding wait request. Core-owned so it can be
383/// consumed by the `protocol_ops_barrier_satisfaction` helper without crossing
384/// crate boundaries.
385#[derive(Debug)]
386pub struct WaitAllSatisfied {
387    /// The authority-owned wait request that reached satisfaction.
388    pub wait_request_id: WaitRequestId,
389    /// The run whose turn-state barrier was satisfied.
390    pub run_id: RunId,
391    /// The operation IDs validated as terminal by the authority.
392    pub operation_ids: Vec<OperationId>,
393}
394
395/// Completion-feed consumer cursor owned by generated machine authority.
396#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
397pub enum CompletionCursorConsumer {
398    /// Cursor advanced after the agent boundary has applied background notices.
399    AgentApplied,
400    /// Cursor advanced after the runtime loop has observed feed entries.
401    RuntimeObserved,
402    /// Cursor advanced after the runtime loop has injected detached continuation.
403    RuntimeInjected,
404}
405
406/// Shared async-operation lifecycle registry.
407pub trait OpsLifecycleRegistry: Send + Sync {
408    fn register_operation(&self, spec: OperationSpec) -> Result<(), OpsLifecycleError>;
409    fn provisioning_succeeded(&self, id: &OperationId) -> Result<(), OpsLifecycleError>;
410    fn provisioning_failed(&self, id: &OperationId, error: String)
411    -> Result<(), OpsLifecycleError>;
412    fn peer_ready(
413        &self,
414        id: &OperationId,
415        peer: OperationPeerHandle,
416    ) -> Result<(), OpsLifecycleError>;
417    fn register_watcher(
418        &self,
419        id: &OperationId,
420    ) -> Result<OperationCompletionWatch, OpsLifecycleError>;
421    fn report_progress(
422        &self,
423        id: &OperationId,
424        update: OperationProgressUpdate,
425    ) -> Result<(), OpsLifecycleError>;
426    fn complete_operation(
427        &self,
428        id: &OperationId,
429        result: OperationResult,
430    ) -> Result<(), OpsLifecycleError>;
431    fn fail_operation(&self, id: &OperationId, error: String) -> Result<(), OpsLifecycleError>;
432    fn abort_provisioning(
433        &self,
434        id: &OperationId,
435        reason: Option<String>,
436    ) -> Result<(), OpsLifecycleError>;
437    fn cancel_operation(
438        &self,
439        id: &OperationId,
440        reason: Option<String>,
441    ) -> Result<(), OpsLifecycleError>;
442    fn request_retire(&self, id: &OperationId) -> Result<(), OpsLifecycleError>;
443    fn mark_retired(&self, id: &OperationId) -> Result<(), OpsLifecycleError>;
444    fn snapshot(
445        &self,
446        id: &OperationId,
447    ) -> Result<Option<OperationLifecycleSnapshot>, OpsLifecycleError>;
448    fn list_operations(&self) -> Result<Vec<OperationLifecycleSnapshot>, OpsLifecycleError>;
449    fn terminate_owner(&self, reason: String) -> Result<(), OpsLifecycleError>;
450
451    /// Classify operation terminality through the registry's generated
452    /// lifecycle authority.
453    fn classify_operation_terminality(
454        &self,
455        _id: &OperationId,
456        _status: OperationStatus,
457    ) -> Result<bool, OpsLifecycleError> {
458        Err(OpsLifecycleError::Unsupported(
459            "classify_operation_terminality".into(),
460        ))
461    }
462
463    /// Classify operation public result projection through generated lifecycle
464    /// authority.
465    fn classify_operation_public_result(
466        &self,
467        _id: &OperationId,
468    ) -> Result<OperationPublicResultClass, OpsLifecycleError> {
469        Err(OpsLifecycleError::Unsupported(
470            "classify_operation_public_result".into(),
471        ))
472    }
473
474    /// Classify whether a completion-feed entry should wake the owning agent
475    /// as a detached background job completion.
476    fn classify_operation_completion_wake(
477        &self,
478        _id: &OperationId,
479        _kind: OperationKind,
480    ) -> Result<OperationCompletionWakeClass, OpsLifecycleError> {
481        Err(OpsLifecycleError::Unsupported(
482            "classify_operation_completion_wake".into(),
483        ))
484    }
485
486    /// Classify whether a generated invalid-transition rejection is an
487    /// idempotent success for the requested lifecycle action.
488    fn classify_operation_transition_idempotence(
489        &self,
490        _id: &OperationId,
491        _action: OperationLifecycleAction,
492    ) -> Result<bool, OpsLifecycleError> {
493        Err(OpsLifecycleError::Unsupported(
494            "classify_operation_transition_idempotence".into(),
495        ))
496    }
497
498    /// Register an operation while applying a caller-supplied generated
499    /// admission limit for this registration.
500    fn register_operation_with_admission_limit(
501        &self,
502        _spec: OperationSpec,
503        _max_concurrent: Option<usize>,
504    ) -> Result<(), OpsLifecycleError> {
505        Err(OpsLifecycleError::Unsupported(
506            "register_operation_with_admission_limit".into(),
507        ))
508    }
509
510    /// Drain all completed operations from the registry, returning their outcomes.
511    fn collect_completed(
512        &self,
513    ) -> Result<Vec<(OperationId, OperationTerminalOutcome)>, OpsLifecycleError> {
514        Err(OpsLifecycleError::Unsupported("collect_completed".into()))
515    }
516
517    /// Return the canonical completion feed, if this registry supports it.
518    ///
519    /// Runtime-backed registries return a feed handle that consumers (agent
520    /// boundary, idle wake) use for cursor-based completion delivery.
521    /// Returns `None` for registries that don't support the feed protocol.
522    fn completion_feed(
523        &self,
524    ) -> Option<std::sync::Arc<dyn crate::completion_feed::CompletionFeed>> {
525        None
526    }
527
528    /// Read the generated completion-consumer cursor for this registry.
529    ///
530    /// `Ok(None)` means this registry has no generated cursor authority.
531    /// `Err(OpsLifecycleError::Internal(_))` means the cursor authority is
532    /// corrupt (e.g. a poisoned registry lock) and must not be laundered into
533    /// the `None`/no-authority meaning by callers.
534    fn completion_cursor(
535        &self,
536        _consumer: CompletionCursorConsumer,
537    ) -> Result<Option<crate::completion_feed::CompletionSeq>, OpsLifecycleError> {
538        Ok(None)
539    }
540
541    /// Advance a completion-consumer cursor through generated authority.
542    ///
543    /// Runtime-backed registries update `projection` only after the generated
544    /// transition emits the matching cursor-advanced effect. The projection is
545    /// an epoch-local cache, never the source of cursor truth.
546    fn advance_completion_cursor(
547        &self,
548        _consumer: CompletionCursorConsumer,
549        _cursor: crate::completion_feed::CompletionSeq,
550        _projection: Option<&EpochCursorState>,
551    ) -> Result<crate::completion_feed::CompletionSeq, OpsLifecycleError> {
552        Err(OpsLifecycleError::Unsupported(
553            "advance_completion_cursor".into(),
554        ))
555    }
556
557    /// Register an authority-owned barrier wait and await its completion.
558    ///
559    /// Returns a [`WaitAllResult`] containing per-operation outcomes and an
560    /// authority-owned obligation token. The runtime may host the async future,
561    /// but wait completion truth comes from the registry authority emitting
562    /// `WaitAllSatisfied`, not from shell watcher timing alone.
563    fn wait_all(
564        &self,
565        _run_id: &RunId,
566        _ids: &[OperationId],
567    ) -> std::pin::Pin<
568        Box<dyn std::future::Future<Output = Result<WaitAllResult, OpsLifecycleError>> + Send + '_>,
569    > {
570        Box::pin(std::future::ready(Err(OpsLifecycleError::Unsupported(
571            "wait_all".into(),
572        ))))
573    }
574}
575
576#[cfg(test)]
577#[allow(clippy::unwrap_used, clippy::panic)]
578mod tests {
579    use super::*;
580
581    #[test]
582    fn operation_kind_peer_expectation_matches_contract() {
583        assert!(OperationKind::MobMemberChild.expects_peer_channel());
584        assert!(!OperationKind::BackgroundToolOp.expects_peer_channel());
585        assert!(!OperationKind::BackgroundToolCapacitySlot.expects_peer_channel());
586    }
587}
meerkat_core/ops_lifecycle.rs

meerkat_core/
ops_lifecycle.rs
