actionqueue_core/mutation/
mod.rs

1//! Engine-facing mutation boundary contracts.
2//!
3//! This module defines the command-semantic boundary between engine intent and
4//! storage durability authority.
5//!
6//! Policy: callers propose typed mutation commands through
7//! [`MutationAuthority::submit_command`], while storage-owned implementations
8//! own WAL mapping, append discipline, durability sync, and projection apply.
9
10use crate::actor::ActorRegistration;
11use crate::budget::BudgetDimension;
12use crate::ids::{ActorId, AttemptId, RunId, TaskId, TenantId};
13use crate::platform::{Capability, LedgerEntry, Role, TenantRegistration};
14use crate::run::state::RunState;
15use crate::run::RunInstance;
16use crate::subscription::{EventFilter, SubscriptionId};
17use crate::task::task_spec::TaskSpec;
18
19/// Durability behavior requested for a submitted mutation command.
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub enum DurabilityPolicy {
22    /// Require an immediate durability sync after WAL append.
23    Immediate,
24    /// Allow deferred durability sync according to implementation policy.
25    Deferred,
26}
27
28/// Semantic mutation command proposed by an engine-facing caller.
29#[derive(Debug, Clone, PartialEq, Eq)]
30#[must_use = "mutation commands should be submitted to a MutationAuthority"]
31pub enum MutationCommand {
32    /// Request durable creation of a task specification.
33    TaskCreate(TaskCreateCommand),
34    /// Request durable creation of a run instance.
35    RunCreate(RunCreateCommand),
36    /// Request a validated run state transition.
37    RunStateTransition(RunStateTransitionCommand),
38    /// Request durable record of a run-attempt start.
39    AttemptStart(AttemptStartCommand),
40    /// Request durable record of a run-attempt finish.
41    AttemptFinish(AttemptFinishCommand),
42    /// Request durable record of lease acquisition for a run.
43    LeaseAcquire(LeaseAcquireCommand),
44    /// Request durable record of a lease heartbeat update.
45    LeaseHeartbeat(LeaseHeartbeatCommand),
46    /// Request durable record of a lease-expired event.
47    LeaseExpire(LeaseExpireCommand),
48    /// Request durable record of a lease-release event.
49    LeaseRelease(LeaseReleaseCommand),
50    /// Request durable record of engine pause intent.
51    EnginePause(EnginePauseCommand),
52    /// Request durable record of engine resume intent.
53    EngineResume(EngineResumeCommand),
54    /// Request durable record of task cancellation intent.
55    TaskCancel(TaskCancelCommand),
56    /// Request durable record of task dependency declarations.
57    DependencyDeclare(DependencyDeclareCommand),
58    /// Request durable record of a run suspension.
59    RunSuspend(RunSuspendCommand),
60    /// Request durable record of a run resumption.
61    RunResume(RunResumeCommand),
62    /// Request durable allocation of a budget for a task dimension.
63    BudgetAllocate(BudgetAllocateCommand),
64    /// Request durable record of resource consumption by a task.
65    BudgetConsume(BudgetConsumeCommand),
66    /// Request durable replenishment of an exhausted budget.
67    BudgetReplenish(BudgetReplenishCommand),
68    /// Request durable creation of an event subscription.
69    SubscriptionCreate(SubscriptionCreateCommand),
70    /// Request durable cancellation of an event subscription.
71    SubscriptionCancel(SubscriptionCancelCommand),
72    /// Request durable record of a subscription trigger.
73    SubscriptionTrigger(SubscriptionTriggerCommand),
74    /// Request durable registration of a remote actor.
75    ActorRegister(ActorRegisterCommand),
76    /// Request durable deregistration of a remote actor.
77    ActorDeregister(ActorDeregisterCommand),
78    /// Request durable record of an actor heartbeat.
79    ActorHeartbeat(ActorHeartbeatCommand),
80    /// Request durable creation of an organizational tenant.
81    TenantCreate(TenantCreateCommand),
82    /// Request durable role assignment for an actor within a tenant.
83    RoleAssign(RoleAssignCommand),
84    /// Request durable capability grant for an actor within a tenant.
85    CapabilityGrant(CapabilityGrantCommand),
86    /// Request durable revocation of a capability grant.
87    CapabilityRevoke(CapabilityRevokeCommand),
88    /// Request durable append of a ledger entry.
89    LedgerAppend(LedgerAppendCommand),
90}
91
92/// Semantic command for engine-pause control mutation.
93#[derive(Debug, Clone, Copy, PartialEq, Eq)]
94pub struct EnginePauseCommand {
95    sequence: u64,
96    timestamp: u64,
97}
98
99impl EnginePauseCommand {
100    /// Creates a new engine-pause command.
101    pub fn new(sequence: u64, timestamp: u64) -> Self {
102        Self { sequence, timestamp }
103    }
104
105    /// Returns the expected WAL sequence.
106    pub fn sequence(&self) -> u64 {
107        self.sequence
108    }
109
110    /// Returns the command timestamp.
111    pub fn timestamp(&self) -> u64 {
112        self.timestamp
113    }
114}
115
116/// Semantic command for engine-resume control mutation.
117#[derive(Debug, Clone, Copy, PartialEq, Eq)]
118pub struct EngineResumeCommand {
119    sequence: u64,
120    timestamp: u64,
121}
122
123impl EngineResumeCommand {
124    /// Creates a new engine-resume command.
125    pub fn new(sequence: u64, timestamp: u64) -> Self {
126        Self { sequence, timestamp }
127    }
128
129    /// Returns the expected WAL sequence.
130    pub fn sequence(&self) -> u64 {
131        self.sequence
132    }
133
134    /// Returns the command timestamp.
135    pub fn timestamp(&self) -> u64 {
136        self.timestamp
137    }
138}
139
140/// Semantic command for task creation.
141#[derive(Debug, Clone, PartialEq, Eq)]
142pub struct TaskCreateCommand {
143    sequence: u64,
144    task_spec: TaskSpec,
145    timestamp: u64,
146}
147
148impl TaskCreateCommand {
149    /// Creates a new task-creation command.
150    pub fn new(sequence: u64, task_spec: TaskSpec, timestamp: u64) -> Self {
151        Self { sequence, task_spec, timestamp }
152    }
153
154    /// Returns the expected WAL sequence.
155    pub fn sequence(&self) -> u64 {
156        self.sequence
157    }
158
159    /// Returns the task specification.
160    pub fn task_spec(&self) -> &TaskSpec {
161        &self.task_spec
162    }
163
164    /// Returns the command timestamp.
165    pub fn timestamp(&self) -> u64 {
166        self.timestamp
167    }
168}
169
170/// Semantic command for run creation.
171#[derive(Debug, Clone, PartialEq, Eq)]
172pub struct RunCreateCommand {
173    sequence: u64,
174    run_instance: RunInstance,
175}
176
177impl RunCreateCommand {
178    /// Creates a new run-creation command.
179    pub fn new(sequence: u64, run_instance: RunInstance) -> Self {
180        Self { sequence, run_instance }
181    }
182
183    /// Returns the expected WAL sequence.
184    pub fn sequence(&self) -> u64 {
185        self.sequence
186    }
187
188    /// Returns the run instance.
189    pub fn run_instance(&self) -> &RunInstance {
190        &self.run_instance
191    }
192}
193
194/// Semantic command for run lifecycle state transition mutation.
195#[derive(Debug, Clone, Copy, PartialEq, Eq)]
196pub struct RunStateTransitionCommand {
197    sequence: u64,
198    run_id: RunId,
199    previous_state: RunState,
200    new_state: RunState,
201    timestamp: u64,
202}
203
204impl RunStateTransitionCommand {
205    /// Creates a new run-state transition command.
206    pub fn new(
207        sequence: u64,
208        run_id: RunId,
209        previous_state: RunState,
210        new_state: RunState,
211        timestamp: u64,
212    ) -> Self {
213        Self { sequence, run_id, previous_state, new_state, timestamp }
214    }
215
216    /// Returns the expected WAL sequence.
217    pub fn sequence(&self) -> u64 {
218        self.sequence
219    }
220
221    /// Returns the targeted run identifier.
222    pub fn run_id(&self) -> RunId {
223        self.run_id
224    }
225
226    /// Returns the expected current run state.
227    pub fn previous_state(&self) -> RunState {
228        self.previous_state
229    }
230
231    /// Returns the requested target run state.
232    pub fn new_state(&self) -> RunState {
233        self.new_state
234    }
235
236    /// Returns the command timestamp.
237    pub fn timestamp(&self) -> u64 {
238        self.timestamp
239    }
240}
241
242/// Semantic command for attempt-start lifecycle mutation.
243#[derive(Debug, Clone, Copy, PartialEq, Eq)]
244pub struct AttemptStartCommand {
245    sequence: u64,
246    run_id: RunId,
247    attempt_id: AttemptId,
248    timestamp: u64,
249}
250
251impl AttemptStartCommand {
252    /// Creates a new attempt-start command.
253    pub fn new(sequence: u64, run_id: RunId, attempt_id: AttemptId, timestamp: u64) -> Self {
254        Self { sequence, run_id, attempt_id, timestamp }
255    }
256
257    /// Returns the expected WAL sequence.
258    pub fn sequence(&self) -> u64 {
259        self.sequence
260    }
261
262    /// Returns the targeted run identifier.
263    pub fn run_id(&self) -> RunId {
264        self.run_id
265    }
266
267    /// Returns the attempt identifier.
268    pub fn attempt_id(&self) -> AttemptId {
269        self.attempt_id
270    }
271
272    /// Returns the command timestamp.
273    pub fn timestamp(&self) -> u64 {
274        self.timestamp
275    }
276}
277
278/// Canonical durable attempt outcome taxonomy.
279#[derive(Debug, Clone, Copy, PartialEq, Eq)]
280#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
281pub enum AttemptResultKind {
282    /// Attempt completed successfully.
283    Success,
284    /// Attempt completed with a non-timeout failure.
285    Failure,
286    /// Attempt completed due to timeout classification.
287    Timeout,
288    /// Attempt was preempted (e.g. budget exhaustion) and the run is now Suspended.
289    /// Does not count toward the max_attempts retry cap.
290    Suspended,
291}
292
293/// Semantic grouping of attempt result kind, optional error detail, and optional
294/// opaque handler output.
295///
296/// The `result` and `error` fields are semantically coupled: the error message
297/// is only meaningful in the context of its result kind. The `output` field is
298/// an opaque byte payload that successful handlers can use to return structured
299/// data to the dispatch loop / projection (e.g., ContextStore references).
300#[derive(Debug, Clone, PartialEq, Eq)]
301#[must_use = "attempt outcome should be inspected for state transition decisions"]
302pub struct AttemptOutcome {
303    result: AttemptResultKind,
304    error: Option<String>,
305    /// Optional opaque output bytes produced by the handler.
306    ///
307    /// For successful attempts, the handler may return structured data (e.g.,
308    /// ContextStore keys, diagnostic payloads, external resource identifiers).
309    /// This output is threaded through to the WAL and projection so callers can
310    /// query it from the run's attempt history.
311    output: Option<Vec<u8>>,
312}
313
314/// Typed validation error for [`AttemptOutcome`] reconstruction from raw parts.
315#[derive(Debug, Clone, PartialEq, Eq)]
316pub enum AttemptOutcomeError {
317    /// A `Success` or `Suspended` outcome was provided with a non-None error field.
318    SuccessWithError {
319        /// The unexpected error detail.
320        error: String,
321    },
322    /// A `Failure` or `Timeout` outcome was provided without an error field.
323    NonSuccessWithoutError {
324        /// The result kind that requires an error detail.
325        result: AttemptResultKind,
326    },
327    /// A `Failure` or `Timeout` outcome was provided with output bytes.
328    NonSuccessWithOutput {
329        /// The result kind that cannot carry output.
330        result: AttemptResultKind,
331    },
332}
333
334impl std::fmt::Display for AttemptOutcomeError {
335    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
336        match self {
337            AttemptOutcomeError::SuccessWithError { error } => {
338                write!(f, "Success/Suspended outcome must not have an error detail, got: {error}")
339            }
340            AttemptOutcomeError::NonSuccessWithoutError { result } => {
341                write!(f, "{result:?} outcome must have an error detail")
342            }
343            AttemptOutcomeError::NonSuccessWithOutput { result } => {
344                write!(f, "non-success outcome {result:?} cannot carry output bytes")
345            }
346        }
347    }
348}
349
350impl std::error::Error for AttemptOutcomeError {}
351
352impl AttemptOutcome {
353    /// Creates a successful outcome with no error detail and no output.
354    pub fn success() -> Self {
355        Self { result: AttemptResultKind::Success, error: None, output: None }
356    }
357
358    /// Creates a successful outcome carrying opaque output bytes.
359    ///
360    /// An empty vec is normalized to `None` (no output).
361    pub fn success_with_output(output: Vec<u8>) -> Self {
362        let output = if output.is_empty() { None } else { Some(output) };
363        Self { result: AttemptResultKind::Success, error: None, output }
364    }
365
366    /// Creates a failure outcome with an error message.
367    pub fn failure(error: impl Into<String>) -> Self {
368        Self { result: AttemptResultKind::Failure, error: Some(error.into()), output: None }
369    }
370
371    /// Creates a timeout outcome with an error message.
372    pub fn timeout(error: impl Into<String>) -> Self {
373        Self { result: AttemptResultKind::Timeout, error: Some(error.into()), output: None }
374    }
375
376    /// Creates a suspended outcome with no output.
377    ///
378    /// Suspended attempts do not count toward the max_attempts retry cap.
379    pub fn suspended() -> Self {
380        Self { result: AttemptResultKind::Suspended, error: None, output: None }
381    }
382
383    /// Creates a suspended outcome carrying opaque partial-state bytes.
384    ///
385    /// An empty vec is normalized to `None` (no output). The handler may use
386    /// this to persist partial execution state for use when the run resumes.
387    pub fn suspended_with_output(output: Vec<u8>) -> Self {
388        let output = if output.is_empty() { None } else { Some(output) };
389        Self { result: AttemptResultKind::Suspended, error: None, output }
390    }
391
392    /// Reconstructs an outcome from raw parts with semantic validation.
393    ///
394    /// This is intended for WAL replay / deserialization paths where the result
395    /// kind, error, and output are stored separately. Validates that:
396    /// - `Success` / `Suspended` have `error == None` (output is optional)
397    /// - `Failure` / `Timeout` have `error == Some(_)` and `output == None`
398    ///
399    /// # Errors
400    ///
401    /// Returns [`AttemptOutcomeError`] if the result kind and error do not
402    /// satisfy the semantic coupling invariant.
403    pub fn from_raw_parts(
404        result: AttemptResultKind,
405        error: Option<String>,
406        output: Option<Vec<u8>>,
407    ) -> Result<Self, AttemptOutcomeError> {
408        match result {
409            AttemptResultKind::Success | AttemptResultKind::Suspended => {
410                if let Some(err) = error {
411                    return Err(AttemptOutcomeError::SuccessWithError { error: err });
412                }
413            }
414            AttemptResultKind::Failure | AttemptResultKind::Timeout => {
415                if error.is_none() {
416                    return Err(AttemptOutcomeError::NonSuccessWithoutError { result });
417                }
418                if output.is_some() {
419                    return Err(AttemptOutcomeError::NonSuccessWithOutput { result });
420                }
421            }
422        }
423        Ok(Self { result, error, output })
424    }
425
426    /// Returns the canonical attempt result kind.
427    pub fn result(&self) -> AttemptResultKind {
428        self.result
429    }
430
431    /// Returns the optional error detail.
432    pub fn error(&self) -> Option<&str> {
433        self.error.as_deref()
434    }
435
436    /// Returns the optional opaque output bytes produced by the handler.
437    pub fn output(&self) -> Option<&[u8]> {
438        self.output.as_deref()
439    }
440
441    /// Consumes the outcome, returning its parts for ownership transfer.
442    pub fn into_parts(self) -> (AttemptResultKind, Option<String>, Option<Vec<u8>>) {
443        (self.result, self.error, self.output)
444    }
445}
446
447/// Semantic command for attempt-finish lifecycle mutation.
448#[derive(Debug, Clone, PartialEq, Eq)]
449pub struct AttemptFinishCommand {
450    sequence: u64,
451    run_id: RunId,
452    attempt_id: AttemptId,
453    outcome: AttemptOutcome,
454    timestamp: u64,
455}
456
457impl AttemptFinishCommand {
458    /// Creates a new attempt-finish command.
459    pub fn new(
460        sequence: u64,
461        run_id: RunId,
462        attempt_id: AttemptId,
463        outcome: AttemptOutcome,
464        timestamp: u64,
465    ) -> Self {
466        Self { sequence, run_id, attempt_id, outcome, timestamp }
467    }
468
469    /// Returns the expected WAL sequence.
470    pub fn sequence(&self) -> u64 {
471        self.sequence
472    }
473
474    /// Returns the targeted run identifier.
475    pub fn run_id(&self) -> RunId {
476        self.run_id
477    }
478
479    /// Returns the attempt identifier.
480    pub fn attempt_id(&self) -> AttemptId {
481        self.attempt_id
482    }
483
484    /// Returns the attempt outcome.
485    pub fn outcome(&self) -> &AttemptOutcome {
486        &self.outcome
487    }
488
489    /// Returns the command timestamp.
490    pub fn timestamp(&self) -> u64 {
491        self.timestamp
492    }
493
494    /// Returns the canonical attempt result kind.
495    pub fn result(&self) -> AttemptResultKind {
496        self.outcome.result
497    }
498
499    /// Returns the optional error detail.
500    pub fn error(&self) -> Option<&str> {
501        self.outcome.error.as_deref()
502    }
503
504    /// Returns the optional opaque output bytes produced by the handler.
505    pub fn output(&self) -> Option<&[u8]> {
506        self.outcome.output.as_deref()
507    }
508}
509
510/// Shared accessor methods for lease command types.
511macro_rules! lease_command_accessors {
512    () => {
513        /// Returns the expected WAL sequence.
514        pub fn sequence(&self) -> u64 {
515            self.sequence
516        }
517
518        /// Returns the targeted run identifier.
519        pub fn run_id(&self) -> RunId {
520            self.run_id
521        }
522
523        /// Returns the lease owner identity.
524        pub fn owner(&self) -> &str {
525            &self.owner
526        }
527
528        /// Returns the lease expiry timestamp.
529        pub fn expiry(&self) -> u64 {
530            self.expiry
531        }
532
533        /// Returns the command timestamp.
534        pub fn timestamp(&self) -> u64 {
535            self.timestamp
536        }
537    };
538}
539
540/// Semantic command for lease-acquire lifecycle mutation.
541#[derive(Debug, Clone, PartialEq, Eq)]
542pub struct LeaseAcquireCommand {
543    sequence: u64,
544    run_id: RunId,
545    owner: String,
546    expiry: u64,
547    timestamp: u64,
548}
549
550impl LeaseAcquireCommand {
551    /// Creates a new lease-acquire command.
552    pub fn new(
553        sequence: u64,
554        run_id: RunId,
555        owner: impl Into<String>,
556        expiry: u64,
557        timestamp: u64,
558    ) -> Self {
559        let owner = owner.into();
560        debug_assert!(!owner.is_empty(), "lease owner must not be empty");
561        Self { sequence, run_id, owner, expiry, timestamp }
562    }
563
564    lease_command_accessors!();
565}
566
567/// Semantic command for lease-heartbeat lifecycle mutation.
568#[derive(Debug, Clone, PartialEq, Eq)]
569pub struct LeaseHeartbeatCommand {
570    sequence: u64,
571    run_id: RunId,
572    owner: String,
573    expiry: u64,
574    timestamp: u64,
575}
576
577impl LeaseHeartbeatCommand {
578    /// Creates a new lease-heartbeat command.
579    pub fn new(
580        sequence: u64,
581        run_id: RunId,
582        owner: impl Into<String>,
583        expiry: u64,
584        timestamp: u64,
585    ) -> Self {
586        let owner = owner.into();
587        debug_assert!(!owner.is_empty(), "lease owner must not be empty");
588        Self { sequence, run_id, owner, expiry, timestamp }
589    }
590
591    lease_command_accessors!();
592}
593
594/// Semantic command for lease-expire lifecycle mutation.
595#[derive(Debug, Clone, PartialEq, Eq)]
596pub struct LeaseExpireCommand {
597    sequence: u64,
598    run_id: RunId,
599    owner: String,
600    expiry: u64,
601    timestamp: u64,
602}
603
604impl LeaseExpireCommand {
605    /// Creates a new lease-expire command.
606    pub fn new(
607        sequence: u64,
608        run_id: RunId,
609        owner: impl Into<String>,
610        expiry: u64,
611        timestamp: u64,
612    ) -> Self {
613        let owner = owner.into();
614        debug_assert!(!owner.is_empty(), "lease owner must not be empty");
615        Self { sequence, run_id, owner, expiry, timestamp }
616    }
617
618    lease_command_accessors!();
619}
620
621/// Semantic command for lease-release lifecycle mutation.
622#[derive(Debug, Clone, PartialEq, Eq)]
623pub struct LeaseReleaseCommand {
624    sequence: u64,
625    run_id: RunId,
626    owner: String,
627    expiry: u64,
628    timestamp: u64,
629}
630
631impl LeaseReleaseCommand {
632    /// Creates a new lease-release command.
633    pub fn new(
634        sequence: u64,
635        run_id: RunId,
636        owner: impl Into<String>,
637        expiry: u64,
638        timestamp: u64,
639    ) -> Self {
640        let owner = owner.into();
641        debug_assert!(!owner.is_empty(), "lease owner must not be empty");
642        Self { sequence, run_id, owner, expiry, timestamp }
643    }
644
645    lease_command_accessors!();
646}
647
648/// Semantic command for task cancellation mutation.
649#[derive(Debug, Clone, Copy, PartialEq, Eq)]
650pub struct TaskCancelCommand {
651    sequence: u64,
652    task_id: TaskId,
653    timestamp: u64,
654}
655
656impl TaskCancelCommand {
657    /// Creates a new task-cancel command.
658    pub fn new(sequence: u64, task_id: TaskId, timestamp: u64) -> Self {
659        Self { sequence, task_id, timestamp }
660    }
661
662    /// Returns the expected WAL sequence.
663    pub fn sequence(&self) -> u64 {
664        self.sequence
665    }
666
667    /// Returns the targeted task identifier.
668    pub fn task_id(&self) -> TaskId {
669        self.task_id
670    }
671
672    /// Returns the command timestamp.
673    pub fn timestamp(&self) -> u64 {
674        self.timestamp
675    }
676}
677
678/// Semantic command for declaring task dependency relationships.
679#[derive(Debug, Clone, PartialEq, Eq)]
680pub struct DependencyDeclareCommand {
681    sequence: u64,
682    task_id: TaskId,
683    depends_on: Vec<TaskId>,
684    timestamp: u64,
685}
686
687impl DependencyDeclareCommand {
688    /// Creates a new dependency declaration command.
689    ///
690    /// # Panics (debug builds only)
691    ///
692    /// Panics if `depends_on` is empty. Empty declarations are rejected by the
693    /// mutation authority in all builds; the debug_assert guards against logic
694    /// errors in internal callers.
695    pub fn new(sequence: u64, task_id: TaskId, depends_on: Vec<TaskId>, timestamp: u64) -> Self {
696        debug_assert!(!depends_on.is_empty());
697        Self { sequence, task_id, depends_on, timestamp }
698    }
699
700    /// Returns the expected WAL sequence.
701    pub fn sequence(&self) -> u64 {
702        self.sequence
703    }
704
705    /// Returns the task whose promotion will be gated.
706    pub fn task_id(&self) -> TaskId {
707        self.task_id
708    }
709
710    /// Returns the prerequisite task identifiers.
711    pub fn depends_on(&self) -> &[TaskId] {
712        &self.depends_on
713    }
714
715    /// Returns the command timestamp.
716    pub fn timestamp(&self) -> u64 {
717        self.timestamp
718    }
719}
720
721/// Successful mutation command outcome metadata.
722#[derive(Debug, Clone, PartialEq, Eq)]
723#[must_use]
724pub struct MutationOutcome {
725    sequence: u64,
726    applied: AppliedMutation,
727}
728
729impl MutationOutcome {
730    /// Creates a new mutation outcome.
731    pub fn new(sequence: u64, applied: AppliedMutation) -> Self {
732        Self { sequence, applied }
733    }
734
735    /// Returns the WAL sequence assigned to the durable event.
736    pub fn sequence(&self) -> u64 {
737        self.sequence
738    }
739
740    /// Returns the semantic effect applied by the mutation.
741    pub fn applied(&self) -> &AppliedMutation {
742        &self.applied
743    }
744}
745
746/// Applied semantic mutation metadata.
747#[derive(Debug, Clone, PartialEq, Eq)]
748pub enum AppliedMutation {
749    /// Task specification was durably created.
750    TaskCreate {
751        /// Created task identifier.
752        task_id: TaskId,
753    },
754    /// Run instance was durably created.
755    RunCreate {
756        /// Created run identifier.
757        run_id: RunId,
758        /// Owning task identifier.
759        task_id: TaskId,
760    },
761    /// Run lifecycle transition was durably applied.
762    RunStateTransition {
763        /// Run receiving the transition.
764        run_id: RunId,
765        /// Transition source state.
766        previous_state: RunState,
767        /// Transition target state.
768        new_state: RunState,
769    },
770    /// Attempt start was durably applied.
771    AttemptStart {
772        /// Run that owns the attempt.
773        run_id: RunId,
774        /// Attempt that started.
775        attempt_id: AttemptId,
776    },
777    /// Attempt finish was durably applied.
778    AttemptFinish {
779        /// Run that owns the attempt.
780        run_id: RunId,
781        /// Attempt that finished.
782        attempt_id: AttemptId,
783        /// Attempt outcome (result kind + optional error detail).
784        outcome: AttemptOutcome,
785    },
786    /// Lease acquire was durably applied.
787    LeaseAcquire {
788        /// Run targeted by lease acquisition.
789        run_id: RunId,
790        /// Lease owner identity.
791        owner: String,
792        /// Lease expiry timestamp.
793        expiry: u64,
794    },
795    /// Lease heartbeat was durably applied.
796    LeaseHeartbeat {
797        /// Run targeted by lease heartbeat.
798        run_id: RunId,
799        /// Lease owner identity.
800        owner: String,
801        /// Lease expiry timestamp after heartbeat.
802        expiry: u64,
803    },
804    /// Lease expiry was durably applied.
805    LeaseExpire {
806        /// Run targeted by lease expiry.
807        run_id: RunId,
808        /// Lease owner identity.
809        owner: String,
810        /// Lease expiry timestamp being expired.
811        expiry: u64,
812    },
813    /// Lease release was durably applied.
814    LeaseRelease {
815        /// Run targeted by lease release.
816        run_id: RunId,
817        /// Lease owner identity.
818        owner: String,
819        /// Lease expiry timestamp at release time.
820        expiry: u64,
821    },
822    /// Engine pause intent was durably applied.
823    EnginePause,
824    /// Engine resume intent was durably applied.
825    EngineResume,
826    /// Task cancellation intent was durably applied.
827    TaskCancel {
828        /// Task targeted by cancellation intent.
829        task_id: TaskId,
830    },
831    /// Task dependency declarations were durably applied.
832    DependencyDeclare {
833        /// Task whose promotion is gated.
834        task_id: TaskId,
835        /// The prerequisite task identifiers.
836        depends_on: Vec<TaskId>,
837    },
838    /// Run suspension was durably applied.
839    RunSuspend {
840        /// The suspended run.
841        run_id: RunId,
842    },
843    /// Run resumption was durably applied.
844    RunResume {
845        /// The resumed run.
846        run_id: RunId,
847    },
848    /// Budget allocation was durably applied.
849    BudgetAllocate {
850        /// Task whose budget was allocated.
851        task_id: TaskId,
852        /// Dimension of the allocated budget.
853        dimension: BudgetDimension,
854        /// Allocated limit.
855        limit: u64,
856    },
857    /// Budget consumption was durably applied.
858    BudgetConsume {
859        /// Task whose budget was consumed.
860        task_id: TaskId,
861        /// Dimension consumed.
862        dimension: BudgetDimension,
863        /// Amount consumed.
864        amount: u64,
865    },
866    /// Budget replenishment was durably applied.
867    BudgetReplenish {
868        /// Task whose budget was replenished.
869        task_id: TaskId,
870        /// Dimension replenished.
871        dimension: BudgetDimension,
872        /// New limit after replenishment.
873        new_limit: u64,
874    },
875    /// Subscription creation was durably applied.
876    SubscriptionCreate {
877        /// The created subscription.
878        subscription_id: SubscriptionId,
879        /// The subscribing task.
880        task_id: TaskId,
881    },
882    /// Subscription cancellation was durably applied.
883    SubscriptionCancel {
884        /// The canceled subscription.
885        subscription_id: SubscriptionId,
886    },
887    /// Subscription trigger was durably applied.
888    SubscriptionTrigger {
889        /// The triggered subscription.
890        subscription_id: SubscriptionId,
891    },
892    /// No specific mutation outcome needed (actor/platform events).
893    NoOp,
894}
895
896/// Engine-facing command submission contract.
897///
898/// Implementors own mutation ordering and persistence semantics.
899pub trait MutationAuthority {
900    /// Typed implementation error.
901    type Error;
902
903    /// Submit a semantic mutation command through the authority lane.
904    fn submit_command(
905        &mut self,
906        command: MutationCommand,
907        durability: DurabilityPolicy,
908    ) -> Result<MutationOutcome, Self::Error>;
909}
910
911/// Semantic command for run suspension mutation.
912#[derive(Debug, Clone, PartialEq, Eq)]
913pub struct RunSuspendCommand {
914    sequence: u64,
915    run_id: RunId,
916    reason: Option<String>,
917    timestamp: u64,
918}
919
920impl RunSuspendCommand {
921    /// Creates a new run-suspend command.
922    pub fn new(sequence: u64, run_id: RunId, reason: Option<String>, timestamp: u64) -> Self {
923        Self { sequence, run_id, reason, timestamp }
924    }
925
926    /// Returns the expected WAL sequence.
927    pub fn sequence(&self) -> u64 {
928        self.sequence
929    }
930
931    /// Returns the targeted run identifier.
932    pub fn run_id(&self) -> RunId {
933        self.run_id
934    }
935
936    /// Returns the optional suspension reason.
937    pub fn reason(&self) -> Option<&str> {
938        self.reason.as_deref()
939    }
940
941    /// Returns the command timestamp.
942    pub fn timestamp(&self) -> u64 {
943        self.timestamp
944    }
945}
946
947/// Semantic command for run resumption mutation.
948#[derive(Debug, Clone, Copy, PartialEq, Eq)]
949pub struct RunResumeCommand {
950    sequence: u64,
951    run_id: RunId,
952    timestamp: u64,
953}
954
955impl RunResumeCommand {
956    /// Creates a new run-resume command.
957    pub fn new(sequence: u64, run_id: RunId, timestamp: u64) -> Self {
958        Self { sequence, run_id, timestamp }
959    }
960
961    /// Returns the expected WAL sequence.
962    pub fn sequence(&self) -> u64 {
963        self.sequence
964    }
965
966    /// Returns the targeted run identifier.
967    pub fn run_id(&self) -> RunId {
968        self.run_id
969    }
970
971    /// Returns the command timestamp.
972    pub fn timestamp(&self) -> u64 {
973        self.timestamp
974    }
975}
976
977/// Semantic command for budget allocation mutation.
978#[derive(Debug, Clone, Copy, PartialEq, Eq)]
979pub struct BudgetAllocateCommand {
980    sequence: u64,
981    task_id: TaskId,
982    dimension: BudgetDimension,
983    limit: u64,
984    timestamp: u64,
985}
986
987impl BudgetAllocateCommand {
988    /// Creates a new budget-allocate command.
989    pub fn new(
990        sequence: u64,
991        task_id: TaskId,
992        dimension: BudgetDimension,
993        limit: u64,
994        timestamp: u64,
995    ) -> Self {
996        debug_assert!(limit > 0, "budget limit must be greater than zero");
997        Self { sequence, task_id, dimension, limit, timestamp }
998    }
999
1000    /// Returns the expected WAL sequence.
1001    pub fn sequence(&self) -> u64 {
1002        self.sequence
1003    }
1004
1005    /// Returns the targeted task identifier.
1006    pub fn task_id(&self) -> TaskId {
1007        self.task_id
1008    }
1009
1010    /// Returns the budget dimension.
1011    pub fn dimension(&self) -> BudgetDimension {
1012        self.dimension
1013    }
1014
1015    /// Returns the budget limit.
1016    pub fn limit(&self) -> u64 {
1017        self.limit
1018    }
1019
1020    /// Returns the command timestamp.
1021    pub fn timestamp(&self) -> u64 {
1022        self.timestamp
1023    }
1024}
1025
1026/// Semantic command for budget consumption mutation.
1027#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1028pub struct BudgetConsumeCommand {
1029    sequence: u64,
1030    task_id: TaskId,
1031    dimension: BudgetDimension,
1032    amount: u64,
1033    timestamp: u64,
1034}
1035
1036impl BudgetConsumeCommand {
1037    /// Creates a new budget-consume command.
1038    pub fn new(
1039        sequence: u64,
1040        task_id: TaskId,
1041        dimension: BudgetDimension,
1042        amount: u64,
1043        timestamp: u64,
1044    ) -> Self {
1045        debug_assert!(amount > 0, "budget consume amount must be greater than zero");
1046        Self { sequence, task_id, dimension, amount, timestamp }
1047    }
1048
1049    /// Returns the expected WAL sequence.
1050    pub fn sequence(&self) -> u64 {
1051        self.sequence
1052    }
1053
1054    /// Returns the targeted task identifier.
1055    pub fn task_id(&self) -> TaskId {
1056        self.task_id
1057    }
1058
1059    /// Returns the budget dimension.
1060    pub fn dimension(&self) -> BudgetDimension {
1061        self.dimension
1062    }
1063
1064    /// Returns the amount consumed.
1065    pub fn amount(&self) -> u64 {
1066        self.amount
1067    }
1068
1069    /// Returns the command timestamp.
1070    pub fn timestamp(&self) -> u64 {
1071        self.timestamp
1072    }
1073}
1074
1075/// Semantic command for budget replenishment mutation.
1076#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1077pub struct BudgetReplenishCommand {
1078    sequence: u64,
1079    task_id: TaskId,
1080    dimension: BudgetDimension,
1081    new_limit: u64,
1082    timestamp: u64,
1083}
1084
1085impl BudgetReplenishCommand {
1086    /// Creates a new budget-replenish command.
1087    pub fn new(
1088        sequence: u64,
1089        task_id: TaskId,
1090        dimension: BudgetDimension,
1091        new_limit: u64,
1092        timestamp: u64,
1093    ) -> Self {
1094        debug_assert!(new_limit > 0, "budget replenish limit must be greater than zero");
1095        Self { sequence, task_id, dimension, new_limit, timestamp }
1096    }
1097
1098    /// Returns the expected WAL sequence.
1099    pub fn sequence(&self) -> u64 {
1100        self.sequence
1101    }
1102
1103    /// Returns the targeted task identifier.
1104    pub fn task_id(&self) -> TaskId {
1105        self.task_id
1106    }
1107
1108    /// Returns the budget dimension.
1109    pub fn dimension(&self) -> BudgetDimension {
1110        self.dimension
1111    }
1112
1113    /// Returns the new budget limit.
1114    pub fn new_limit(&self) -> u64 {
1115        self.new_limit
1116    }
1117
1118    /// Returns the command timestamp.
1119    pub fn timestamp(&self) -> u64 {
1120        self.timestamp
1121    }
1122}
1123
1124/// Semantic command for subscription creation mutation.
1125#[derive(Debug, Clone, PartialEq, Eq)]
1126pub struct SubscriptionCreateCommand {
1127    sequence: u64,
1128    subscription_id: SubscriptionId,
1129    task_id: TaskId,
1130    filter: EventFilter,
1131    timestamp: u64,
1132}
1133
1134impl SubscriptionCreateCommand {
1135    /// Creates a new subscription-create command.
1136    pub fn new(
1137        sequence: u64,
1138        subscription_id: SubscriptionId,
1139        task_id: TaskId,
1140        filter: EventFilter,
1141        timestamp: u64,
1142    ) -> Self {
1143        if let EventFilter::BudgetThreshold { threshold_pct, .. } = &filter {
1144            debug_assert!(
1145                *threshold_pct <= 100,
1146                "budget threshold percentage must be 0-100, got {threshold_pct}"
1147            );
1148        }
1149        if let EventFilter::Custom { key } = &filter {
1150            debug_assert!(!key.is_empty(), "custom event key must not be empty");
1151        }
1152        Self { sequence, subscription_id, task_id, filter, timestamp }
1153    }
1154
1155    /// Returns the expected WAL sequence.
1156    pub fn sequence(&self) -> u64 {
1157        self.sequence
1158    }
1159
1160    /// Returns the subscription identifier.
1161    pub fn subscription_id(&self) -> SubscriptionId {
1162        self.subscription_id
1163    }
1164
1165    /// Returns the subscribing task identifier.
1166    pub fn task_id(&self) -> TaskId {
1167        self.task_id
1168    }
1169
1170    /// Returns the event filter.
1171    pub fn filter(&self) -> &EventFilter {
1172        &self.filter
1173    }
1174
1175    /// Returns the command timestamp.
1176    pub fn timestamp(&self) -> u64 {
1177        self.timestamp
1178    }
1179}
1180
1181/// Semantic command for subscription cancellation mutation.
1182#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1183pub struct SubscriptionCancelCommand {
1184    sequence: u64,
1185    subscription_id: SubscriptionId,
1186    timestamp: u64,
1187}
1188
1189impl SubscriptionCancelCommand {
1190    /// Creates a new subscription-cancel command.
1191    pub fn new(sequence: u64, subscription_id: SubscriptionId, timestamp: u64) -> Self {
1192        Self { sequence, subscription_id, timestamp }
1193    }
1194
1195    /// Returns the expected WAL sequence.
1196    pub fn sequence(&self) -> u64 {
1197        self.sequence
1198    }
1199
1200    /// Returns the subscription identifier.
1201    pub fn subscription_id(&self) -> SubscriptionId {
1202        self.subscription_id
1203    }
1204
1205    /// Returns the command timestamp.
1206    pub fn timestamp(&self) -> u64 {
1207        self.timestamp
1208    }
1209}
1210
1211/// Semantic command for subscription trigger mutation.
1212#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1213pub struct SubscriptionTriggerCommand {
1214    sequence: u64,
1215    subscription_id: SubscriptionId,
1216    timestamp: u64,
1217}
1218
1219impl SubscriptionTriggerCommand {
1220    /// Creates a new subscription-trigger command.
1221    pub fn new(sequence: u64, subscription_id: SubscriptionId, timestamp: u64) -> Self {
1222        Self { sequence, subscription_id, timestamp }
1223    }
1224
1225    /// Returns the expected WAL sequence.
1226    pub fn sequence(&self) -> u64 {
1227        self.sequence
1228    }
1229
1230    /// Returns the subscription identifier.
1231    pub fn subscription_id(&self) -> SubscriptionId {
1232        self.subscription_id
1233    }
1234
1235    /// Returns the command timestamp.
1236    pub fn timestamp(&self) -> u64 {
1237        self.timestamp
1238    }
1239}
1240
1241// ── Sprint 4: Actor / Platform mutation commands ─────────────────────────────
1242
1243/// Semantic command for remote actor registration.
1244#[derive(Debug, Clone, PartialEq, Eq)]
1245pub struct ActorRegisterCommand {
1246    sequence: u64,
1247    registration: ActorRegistration,
1248    timestamp: u64,
1249}
1250
1251impl ActorRegisterCommand {
1252    /// Creates a new actor-register command.
1253    pub fn new(sequence: u64, registration: ActorRegistration, timestamp: u64) -> Self {
1254        Self { sequence, registration, timestamp }
1255    }
1256
1257    /// Returns the expected WAL sequence.
1258    pub fn sequence(&self) -> u64 {
1259        self.sequence
1260    }
1261    /// Returns the actor registration.
1262    pub fn registration(&self) -> &ActorRegistration {
1263        &self.registration
1264    }
1265    /// Returns the command timestamp.
1266    pub fn timestamp(&self) -> u64 {
1267        self.timestamp
1268    }
1269}
1270
1271/// Semantic command for remote actor deregistration.
1272#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1273pub struct ActorDeregisterCommand {
1274    sequence: u64,
1275    actor_id: ActorId,
1276    timestamp: u64,
1277}
1278
1279impl ActorDeregisterCommand {
1280    /// Creates a new actor-deregister command.
1281    pub fn new(sequence: u64, actor_id: ActorId, timestamp: u64) -> Self {
1282        Self { sequence, actor_id, timestamp }
1283    }
1284    /// Returns the expected WAL sequence.
1285    pub fn sequence(&self) -> u64 {
1286        self.sequence
1287    }
1288    /// Returns the actor identifier.
1289    pub fn actor_id(&self) -> ActorId {
1290        self.actor_id
1291    }
1292    /// Returns the command timestamp.
1293    pub fn timestamp(&self) -> u64 {
1294        self.timestamp
1295    }
1296}
1297
1298/// Semantic command for an actor heartbeat update.
1299#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1300pub struct ActorHeartbeatCommand {
1301    sequence: u64,
1302    actor_id: ActorId,
1303    timestamp: u64,
1304}
1305
1306impl ActorHeartbeatCommand {
1307    /// Creates a new actor-heartbeat command.
1308    pub fn new(sequence: u64, actor_id: ActorId, timestamp: u64) -> Self {
1309        Self { sequence, actor_id, timestamp }
1310    }
1311    /// Returns the expected WAL sequence.
1312    pub fn sequence(&self) -> u64 {
1313        self.sequence
1314    }
1315    /// Returns the actor identifier.
1316    pub fn actor_id(&self) -> ActorId {
1317        self.actor_id
1318    }
1319    /// Returns the command timestamp.
1320    pub fn timestamp(&self) -> u64 {
1321        self.timestamp
1322    }
1323}
1324
1325/// Semantic command for tenant creation.
1326#[derive(Debug, Clone, PartialEq, Eq)]
1327pub struct TenantCreateCommand {
1328    sequence: u64,
1329    registration: TenantRegistration,
1330    timestamp: u64,
1331}
1332
1333impl TenantCreateCommand {
1334    /// Creates a new tenant-create command.
1335    pub fn new(sequence: u64, registration: TenantRegistration, timestamp: u64) -> Self {
1336        Self { sequence, registration, timestamp }
1337    }
1338    /// Returns the expected WAL sequence.
1339    pub fn sequence(&self) -> u64 {
1340        self.sequence
1341    }
1342    /// Returns the tenant registration.
1343    pub fn registration(&self) -> &TenantRegistration {
1344        &self.registration
1345    }
1346    /// Returns the command timestamp.
1347    pub fn timestamp(&self) -> u64 {
1348        self.timestamp
1349    }
1350}
1351
1352/// Semantic command for role assignment.
1353#[derive(Debug, Clone, PartialEq, Eq)]
1354pub struct RoleAssignCommand {
1355    sequence: u64,
1356    actor_id: ActorId,
1357    role: Role,
1358    tenant_id: TenantId,
1359    timestamp: u64,
1360}
1361
1362impl RoleAssignCommand {
1363    /// Creates a new role-assign command.
1364    pub fn new(
1365        sequence: u64,
1366        actor_id: ActorId,
1367        role: Role,
1368        tenant_id: TenantId,
1369        timestamp: u64,
1370    ) -> Self {
1371        Self { sequence, actor_id, role, tenant_id, timestamp }
1372    }
1373    /// Returns the expected WAL sequence.
1374    pub fn sequence(&self) -> u64 {
1375        self.sequence
1376    }
1377    /// Returns the actor identifier.
1378    pub fn actor_id(&self) -> ActorId {
1379        self.actor_id
1380    }
1381    /// Returns the role to assign.
1382    pub fn role(&self) -> &Role {
1383        &self.role
1384    }
1385    /// Returns the tenant identifier.
1386    pub fn tenant_id(&self) -> TenantId {
1387        self.tenant_id
1388    }
1389    /// Returns the command timestamp.
1390    pub fn timestamp(&self) -> u64 {
1391        self.timestamp
1392    }
1393}
1394
1395/// Semantic command for capability grant.
1396#[derive(Debug, Clone, PartialEq, Eq)]
1397pub struct CapabilityGrantCommand {
1398    sequence: u64,
1399    actor_id: ActorId,
1400    capability: Capability,
1401    tenant_id: TenantId,
1402    timestamp: u64,
1403}
1404
1405impl CapabilityGrantCommand {
1406    /// Creates a new capability-grant command.
1407    pub fn new(
1408        sequence: u64,
1409        actor_id: ActorId,
1410        capability: Capability,
1411        tenant_id: TenantId,
1412        timestamp: u64,
1413    ) -> Self {
1414        Self { sequence, actor_id, capability, tenant_id, timestamp }
1415    }
1416    /// Returns the expected WAL sequence.
1417    pub fn sequence(&self) -> u64 {
1418        self.sequence
1419    }
1420    /// Returns the actor identifier.
1421    pub fn actor_id(&self) -> ActorId {
1422        self.actor_id
1423    }
1424    /// Returns the capability to grant.
1425    pub fn capability(&self) -> &Capability {
1426        &self.capability
1427    }
1428    /// Returns the tenant identifier.
1429    pub fn tenant_id(&self) -> TenantId {
1430        self.tenant_id
1431    }
1432    /// Returns the command timestamp.
1433    pub fn timestamp(&self) -> u64 {
1434        self.timestamp
1435    }
1436}
1437
1438/// Semantic command for capability revocation.
1439#[derive(Debug, Clone, PartialEq, Eq)]
1440pub struct CapabilityRevokeCommand {
1441    sequence: u64,
1442    actor_id: ActorId,
1443    capability: Capability,
1444    tenant_id: TenantId,
1445    timestamp: u64,
1446}
1447
1448impl CapabilityRevokeCommand {
1449    /// Creates a new capability-revoke command.
1450    pub fn new(
1451        sequence: u64,
1452        actor_id: ActorId,
1453        capability: Capability,
1454        tenant_id: TenantId,
1455        timestamp: u64,
1456    ) -> Self {
1457        Self { sequence, actor_id, capability, tenant_id, timestamp }
1458    }
1459    /// Returns the expected WAL sequence.
1460    pub fn sequence(&self) -> u64 {
1461        self.sequence
1462    }
1463    /// Returns the actor identifier.
1464    pub fn actor_id(&self) -> ActorId {
1465        self.actor_id
1466    }
1467    /// Returns the capability to revoke.
1468    pub fn capability(&self) -> &Capability {
1469        &self.capability
1470    }
1471    /// Returns the tenant identifier.
1472    pub fn tenant_id(&self) -> TenantId {
1473        self.tenant_id
1474    }
1475    /// Returns the command timestamp.
1476    pub fn timestamp(&self) -> u64 {
1477        self.timestamp
1478    }
1479}
1480
1481/// Semantic command for ledger entry append.
1482#[derive(Debug, Clone, PartialEq, Eq)]
1483pub struct LedgerAppendCommand {
1484    sequence: u64,
1485    entry: LedgerEntry,
1486    timestamp: u64,
1487}
1488
1489impl LedgerAppendCommand {
1490    /// Creates a new ledger-append command.
1491    pub fn new(sequence: u64, entry: LedgerEntry, timestamp: u64) -> Self {
1492        Self { sequence, entry, timestamp }
1493    }
1494    /// Returns the expected WAL sequence.
1495    pub fn sequence(&self) -> u64 {
1496        self.sequence
1497    }
1498    /// Returns the ledger entry.
1499    pub fn entry(&self) -> &LedgerEntry {
1500        &self.entry
1501    }
1502    /// Returns the command timestamp.
1503    pub fn timestamp(&self) -> u64 {
1504        self.timestamp
1505    }
1506}
actionqueue_core/mutation/mod.rs

actionqueue_core/mutation/
mod.rs
