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    /// Immutable owner tuple from the registered operation spec.
281    pub owner_session_id: SessionId,
282    pub display_name: String,
283    pub source_label: String,
284    #[serde(default, skip_serializing_if = "Option::is_none")]
285    pub operation_source: Option<OperationSource>,
286    pub status: OperationStatus,
287    /// Generated terminality classification for `status`.
288    pub terminal: bool,
289    /// Generated public result classification for `status`.
290    pub public_result_class: OperationPublicResultClass,
291    pub peer_ready: bool,
292    pub progress_count: u32,
293    pub watcher_count: u32,
294    pub terminal_outcome: Option<OperationTerminalOutcome>,
295    pub child_session_id: Option<SessionId>,
296    pub expect_peer_channel: bool,
297    /// Peer handle info (exposed when peer_ready is true).
298    #[serde(skip_serializing_if = "Option::is_none", default)]
299    pub peer_handle: Option<OperationPeerHandle>,
300    /// Wall-clock epoch millis when the operation was registered.
301    #[serde(default)]
302    pub created_at_ms: u64,
303    /// Wall-clock epoch millis when provisioning succeeded (entered Running).
304    #[serde(skip_serializing_if = "Option::is_none", default)]
305    pub started_at_ms: Option<u64>,
306    /// Wall-clock epoch millis when the operation reached terminal state.
307    #[serde(skip_serializing_if = "Option::is_none", default)]
308    pub completed_at_ms: Option<u64>,
309    /// Monotonic elapsed millis from creation to terminal (computed from Instant).
310    #[serde(skip_serializing_if = "Option::is_none", default)]
311    pub elapsed_ms: Option<u64>,
312}
313
314/// One registry-owned watcher for a terminal lifecycle outcome.
315///
316/// This is a read-only waiter returned by [`OpsLifecycleRegistry`]. It does not
317/// expose terminal-outcome send authority to callers; registry implementations
318/// must resolve it only after generated operation lifecycle authority has
319/// accepted the terminal transition.
320pub type OperationCompletionWatch = Pin<
321    Box<
322        dyn Future<Output = Result<OperationTerminalOutcome, OperationCompletionWatchError>>
323            + Send
324            + 'static,
325    >,
326>;
327
328/// Mechanical failure while waiting for operation completion plumbing.
329///
330/// This is intentionally not an [`OperationTerminalOutcome`]: a dropped waiter
331/// channel is not async-operation terminal truth.
332#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
333pub enum OperationCompletionWatchError {
334    #[error("operation completion channel closed without an authorized terminal outcome")]
335    ChannelClosed,
336}
337
338/// Errors returned by the shared lifecycle registry.
339#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
340pub enum OpsLifecycleError {
341    #[error("operation already registered: {0}")]
342    AlreadyRegistered(OperationId),
343    #[error("operation not found: {0}")]
344    NotFound(OperationId),
345    #[error("invalid lifecycle transition for {id}: {status:?} -> {action}")]
346    InvalidTransition {
347        id: OperationId,
348        status: OperationStatus,
349        action: &'static str,
350    },
351    #[error("operation does not expect a peer handoff: {0}")]
352    PeerNotExpected(OperationId),
353    #[error("operation is already peer-ready: {0}")]
354    AlreadyPeerReady(OperationId),
355    #[error("max concurrent operations exceeded (limit: {limit}, active: {active})")]
356    MaxConcurrentExceeded { limit: usize, active: usize },
357    #[error("operation not supported: {0}")]
358    Unsupported(String),
359    #[error("wait_all already active")]
360    WaitAlreadyActive,
361    #[error("wait_all not active for request: {0}")]
362    WaitNotActive(WaitRequestId),
363    #[error("wait_all contains duplicate operation id: {0}")]
364    DuplicateWaitOperation(OperationId),
365    #[error("operation owner is retired")]
366    OwnerRetired,
367    #[error("internal lifecycle registry error: {0}")]
368    Internal(String),
369}
370
371/// Authority-owned result of `wait_all()`.
372///
373/// Carries the per-operation outcomes alongside an authority-derived
374/// obligation token (`satisfied`). The obligation proves the authority owned
375/// the wait request lifecycle and emitted `WaitAllSatisfied` when the tracked
376/// barrier set became terminal.
377#[derive(Debug)]
378pub struct WaitAllResult {
379    /// Per-operation terminal outcomes.
380    pub outcomes: Vec<(OperationId, OperationTerminalOutcome)>,
381    /// Authority-validated obligation token for the ops_barrier_satisfaction protocol.
382    pub satisfied: WaitAllSatisfied,
383}
384
385/// Authority-owned obligation token emitted by the `WaitAllSatisfied` effect.
386///
387/// Created only by the `OpsLifecycleRegistry::wait_all()` implementation after
388/// the authority resolves an outstanding wait request. Core-owned so it can be
389/// consumed by the `protocol_ops_barrier_satisfaction` helper without crossing
390/// crate boundaries.
391#[derive(Debug)]
392pub struct WaitAllSatisfied {
393    /// The authority-owned wait request that reached satisfaction.
394    pub wait_request_id: WaitRequestId,
395    /// The run whose turn-state barrier was satisfied.
396    pub run_id: RunId,
397    /// The operation IDs validated as terminal by the authority.
398    pub operation_ids: Vec<OperationId>,
399}
400
401/// Completion-feed consumer cursor owned by generated machine authority.
402#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
403pub enum CompletionCursorConsumer {
404    /// Cursor advanced after the agent boundary has applied background notices.
405    AgentApplied,
406    /// Cursor advanced after the runtime loop has observed feed entries.
407    RuntimeObserved,
408    /// Cursor advanced after the runtime loop has injected detached continuation.
409    RuntimeInjected,
410}
411
412/// Shared async-operation lifecycle registry.
413pub trait OpsLifecycleRegistry: Send + Sync {
414    fn register_operation(&self, spec: OperationSpec) -> Result<(), OpsLifecycleError>;
415    fn provisioning_succeeded(&self, id: &OperationId) -> Result<(), OpsLifecycleError>;
416    fn provisioning_failed(&self, id: &OperationId, error: String)
417    -> Result<(), OpsLifecycleError>;
418    fn peer_ready(
419        &self,
420        id: &OperationId,
421        peer: OperationPeerHandle,
422    ) -> Result<(), OpsLifecycleError>;
423    fn register_watcher(
424        &self,
425        id: &OperationId,
426    ) -> Result<OperationCompletionWatch, OpsLifecycleError>;
427    fn report_progress(
428        &self,
429        id: &OperationId,
430        update: OperationProgressUpdate,
431    ) -> Result<(), OpsLifecycleError>;
432    fn complete_operation(
433        &self,
434        id: &OperationId,
435        result: OperationResult,
436    ) -> Result<(), OpsLifecycleError>;
437    fn fail_operation(&self, id: &OperationId, error: String) -> Result<(), OpsLifecycleError>;
438    fn abort_provisioning(
439        &self,
440        id: &OperationId,
441        reason: Option<String>,
442    ) -> Result<(), OpsLifecycleError>;
443    fn cancel_operation(
444        &self,
445        id: &OperationId,
446        reason: Option<String>,
447    ) -> Result<(), OpsLifecycleError>;
448    fn request_retire(&self, id: &OperationId) -> Result<(), OpsLifecycleError>;
449    fn mark_retired(&self, id: &OperationId) -> Result<(), OpsLifecycleError>;
450    fn snapshot(
451        &self,
452        id: &OperationId,
453    ) -> Result<Option<OperationLifecycleSnapshot>, OpsLifecycleError>;
454    fn list_operations(&self) -> Result<Vec<OperationLifecycleSnapshot>, OpsLifecycleError>;
455    fn terminate_owner(&self, reason: String) -> Result<(), OpsLifecycleError>;
456
457    /// Classify operation terminality through the registry's generated
458    /// lifecycle authority.
459    fn classify_operation_terminality(
460        &self,
461        _id: &OperationId,
462        _status: OperationStatus,
463    ) -> Result<bool, OpsLifecycleError> {
464        Err(OpsLifecycleError::Unsupported(
465            "classify_operation_terminality".into(),
466        ))
467    }
468
469    /// Classify operation public result projection through generated lifecycle
470    /// authority.
471    fn classify_operation_public_result(
472        &self,
473        _id: &OperationId,
474    ) -> Result<OperationPublicResultClass, OpsLifecycleError> {
475        Err(OpsLifecycleError::Unsupported(
476            "classify_operation_public_result".into(),
477        ))
478    }
479
480    /// Classify whether a completion-feed entry should wake the owning agent
481    /// as a detached background job completion.
482    fn classify_operation_completion_wake(
483        &self,
484        _id: &OperationId,
485        _kind: OperationKind,
486    ) -> Result<OperationCompletionWakeClass, OpsLifecycleError> {
487        Err(OpsLifecycleError::Unsupported(
488            "classify_operation_completion_wake".into(),
489        ))
490    }
491
492    /// Classify whether a generated invalid-transition rejection is an
493    /// idempotent success for the requested lifecycle action.
494    fn classify_operation_transition_idempotence(
495        &self,
496        _id: &OperationId,
497        _action: OperationLifecycleAction,
498    ) -> Result<bool, OpsLifecycleError> {
499        Err(OpsLifecycleError::Unsupported(
500            "classify_operation_transition_idempotence".into(),
501        ))
502    }
503
504    /// Register an operation while applying a caller-supplied generated
505    /// admission limit for this registration.
506    fn register_operation_with_admission_limit(
507        &self,
508        _spec: OperationSpec,
509        _max_concurrent: Option<usize>,
510    ) -> Result<(), OpsLifecycleError> {
511        Err(OpsLifecycleError::Unsupported(
512            "register_operation_with_admission_limit".into(),
513        ))
514    }
515
516    /// Drain all completed operations from the registry, returning their outcomes.
517    fn collect_completed(
518        &self,
519    ) -> Result<Vec<(OperationId, OperationTerminalOutcome)>, OpsLifecycleError> {
520        Err(OpsLifecycleError::Unsupported("collect_completed".into()))
521    }
522
523    /// Return the canonical completion feed, if this registry supports it.
524    ///
525    /// Runtime-backed registries return a feed handle that consumers (agent
526    /// boundary, idle wake) use for cursor-based completion delivery.
527    /// Returns `None` for registries that don't support the feed protocol.
528    fn completion_feed(
529        &self,
530    ) -> Option<std::sync::Arc<dyn crate::completion_feed::CompletionFeed>> {
531        None
532    }
533
534    /// Read the generated completion-consumer cursor for this registry.
535    ///
536    /// `Ok(None)` means this registry has no generated cursor authority.
537    /// `Err(OpsLifecycleError::Internal(_))` means the cursor authority is
538    /// corrupt (e.g. a poisoned registry lock) and must not be laundered into
539    /// the `None`/no-authority meaning by callers.
540    fn completion_cursor(
541        &self,
542        _consumer: CompletionCursorConsumer,
543    ) -> Result<Option<crate::completion_feed::CompletionSeq>, OpsLifecycleError> {
544        Ok(None)
545    }
546
547    /// Advance a completion-consumer cursor through generated authority.
548    ///
549    /// Runtime-backed registries update `projection` only after the generated
550    /// transition emits the matching cursor-advanced effect. The projection is
551    /// an epoch-local cache, never the source of cursor truth.
552    fn advance_completion_cursor(
553        &self,
554        _consumer: CompletionCursorConsumer,
555        _cursor: crate::completion_feed::CompletionSeq,
556        _projection: Option<&EpochCursorState>,
557    ) -> Result<crate::completion_feed::CompletionSeq, OpsLifecycleError> {
558        Err(OpsLifecycleError::Unsupported(
559            "advance_completion_cursor".into(),
560        ))
561    }
562
563    /// Register an authority-owned barrier wait and await its completion.
564    ///
565    /// Returns a [`WaitAllResult`] containing per-operation outcomes and an
566    /// authority-owned obligation token. The runtime may host the async future,
567    /// but wait completion truth comes from the registry authority emitting
568    /// `WaitAllSatisfied`, not from shell watcher timing alone.
569    fn wait_all(
570        &self,
571        _run_id: &RunId,
572        _ids: &[OperationId],
573    ) -> std::pin::Pin<
574        Box<dyn std::future::Future<Output = Result<WaitAllResult, OpsLifecycleError>> + Send + '_>,
575    > {
576        Box::pin(std::future::ready(Err(OpsLifecycleError::Unsupported(
577            "wait_all".into(),
578        ))))
579    }
580}
581
582#[cfg(test)]
583#[allow(clippy::unwrap_used, clippy::panic)]
584mod tests {
585    use super::*;
586
587    #[test]
588    fn operation_kind_peer_expectation_matches_contract() {
589        assert!(OperationKind::MobMemberChild.expects_peer_channel());
590        assert!(!OperationKind::BackgroundToolOp.expects_peer_channel());
591        assert!(!OperationKind::BackgroundToolCapacitySlot.expects_peer_channel());
592    }
593}
meerkat_core/ops_lifecycle.rs

meerkat_core/
ops_lifecycle.rs
