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

meerkat_core/
ops_lifecycle.rs
