a3s_flow/model/
command.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3use serde_json::Value;
4use std::time::Duration;
5
6use crate::error::{FlowError, Result};
7
8use super::{ChildOperationReference, WorkflowProgress};
9
10/// JSON payload exchanged between the engine and runtimes.
11pub type JsonValue = Value;
12
13/// Runtime family used to execute workflow code.
14#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
15#[serde(rename_all = "snake_case")]
16pub enum RuntimeKind {
17    /// TypeScript compiled to a native executable through a native toolchain.
18    NativeTs,
19    /// Host-provided Rust runtime. Useful for tests and embedded deployments.
20    RustEmbedded,
21}
22
23/// Runtime metadata stored with a run so replay can happen on another process.
24#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
25pub struct RuntimeSpec {
26    pub kind: RuntimeKind,
27    pub entrypoint: String,
28    pub export_name: String,
29}
30
31impl RuntimeSpec {
32    pub fn native_ts(entrypoint: impl Into<String>, export_name: impl Into<String>) -> Self {
33        Self {
34            kind: RuntimeKind::NativeTs,
35            entrypoint: entrypoint.into(),
36            export_name: export_name.into(),
37        }
38    }
39
40    pub fn rust_embedded(entrypoint: impl Into<String>, export_name: impl Into<String>) -> Self {
41        Self {
42            kind: RuntimeKind::RustEmbedded,
43            entrypoint: entrypoint.into(),
44            export_name: export_name.into(),
45        }
46    }
47}
48
49/// Durable workflow definition.
50#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
51pub struct WorkflowSpec {
52    pub name: String,
53    pub version: String,
54    pub runtime: RuntimeSpec,
55}
56
57impl WorkflowSpec {
58    pub fn native_ts(
59        name: impl Into<String>,
60        version: impl Into<String>,
61        entrypoint: impl Into<String>,
62        export_name: impl Into<String>,
63    ) -> Self {
64        Self {
65            name: name.into(),
66            version: version.into(),
67            runtime: RuntimeSpec::native_ts(entrypoint, export_name),
68        }
69    }
70
71    pub fn rust_embedded(
72        name: impl Into<String>,
73        version: impl Into<String>,
74        entrypoint: impl Into<String>,
75        export_name: impl Into<String>,
76    ) -> Self {
77        Self {
78            name: name.into(),
79            version: version.into(),
80            runtime: RuntimeSpec::rust_embedded(entrypoint, export_name),
81        }
82    }
83
84    pub fn validate(&self) -> Result<()> {
85        if self.name.trim().is_empty() {
86            return Err(FlowError::InvalidWorkflow(
87                "workflow name must not be empty".to_string(),
88            ));
89        }
90        if self.version.trim().is_empty() {
91            return Err(FlowError::InvalidWorkflow(
92                "workflow version must not be empty".to_string(),
93            ));
94        }
95        if self.runtime.entrypoint.trim().is_empty() {
96            return Err(FlowError::InvalidWorkflow(
97                "runtime entrypoint must not be empty".to_string(),
98            ));
99        }
100        if self.runtime.export_name.trim().is_empty() {
101            return Err(FlowError::InvalidWorkflow(
102                "runtime export_name must not be empty".to_string(),
103            ));
104        }
105        Ok(())
106    }
107}
108
109/// What the engine should do after a step exhausts its retry attempts.
110#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
111#[serde(rename_all = "snake_case")]
112pub enum StepFailureAction {
113    /// Record `step_failed`, then fail the workflow run.
114    #[default]
115    FailRun,
116    /// Record `step_failed`, then replay the workflow so it can choose a
117    /// fallback, compensation, or explicit failure command.
118    ContinueWorkflow,
119}
120
121impl StepFailureAction {
122    pub fn is_fail_run(&self) -> bool {
123        matches!(self, Self::FailRun)
124    }
125}
126
127/// Retry behavior for a step command.
128#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
129pub struct RetryPolicy {
130    pub max_attempts: u32,
131    pub delay_ms: u64,
132    #[serde(default, skip_serializing_if = "StepFailureAction::is_fail_run")]
133    pub on_exhausted: StepFailureAction,
134}
135
136impl RetryPolicy {
137    pub fn none() -> Self {
138        Self {
139            max_attempts: 1,
140            delay_ms: 0,
141            on_exhausted: StepFailureAction::FailRun,
142        }
143    }
144
145    pub fn fixed(max_attempts: u32, delay: Duration) -> Self {
146        Self {
147            max_attempts: max_attempts.max(1),
148            delay_ms: delay.as_millis().min(u128::from(u64::MAX)) as u64,
149            on_exhausted: StepFailureAction::FailRun,
150        }
151    }
152
153    pub fn with_failure_action(mut self, action: StepFailureAction) -> Self {
154        self.on_exhausted = action;
155        self
156    }
157
158    pub fn continue_workflow_on_failure(self) -> Self {
159        self.with_failure_action(StepFailureAction::ContinueWorkflow)
160    }
161}
162
163impl Default for RetryPolicy {
164    fn default() -> Self {
165        Self {
166            max_attempts: 3,
167            delay_ms: 0,
168            on_exhausted: StepFailureAction::FailRun,
169        }
170    }
171}
172
173/// Command emitted by the workflow runtime after replay.
174#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
175#[serde(tag = "type", rename_all = "snake_case")]
176pub enum RuntimeCommand {
177    Complete {
178        output: JsonValue,
179    },
180    Fail {
181        error: String,
182    },
183    /// Finish a previously requested cleanup-aware cancellation.
184    Cancel,
185    /// Finish a run with a typed timeout outcome.
186    Timeout {
187        deadline: DateTime<Utc>,
188        #[serde(default, skip_serializing_if = "Option::is_none")]
189        reason: Option<String>,
190    },
191    /// Persist progress before replaying the workflow.
192    RecordProgress {
193        progress: WorkflowProgress,
194    },
195    /// Persist a parent-to-child operation reference before replaying.
196    LinkChildOperation {
197        child: ChildOperationReference,
198    },
199    ScheduleStep {
200        step_id: String,
201        step_name: String,
202        input: JsonValue,
203        #[serde(default)]
204        retry: RetryPolicy,
205    },
206    ScheduleSteps {
207        steps: Vec<StepCommand>,
208    },
209    WaitUntil {
210        wait_id: String,
211        resume_at: DateTime<Utc>,
212    },
213    CreateHook {
214        hook_id: String,
215        token: String,
216        #[serde(default)]
217        metadata: JsonValue,
218    },
219}
220
221/// Step definition returned by workflow replay.
222#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
223pub struct StepCommand {
224    pub step_id: String,
225    pub step_name: String,
226    pub input: JsonValue,
227    #[serde(default)]
228    pub retry: RetryPolicy,
229}
230
231impl StepCommand {
232    pub fn new(step_id: impl Into<String>, step_name: impl Into<String>, input: JsonValue) -> Self {
233        Self {
234            step_id: step_id.into(),
235            step_name: step_name.into(),
236            input,
237            retry: RetryPolicy::default(),
238        }
239    }
240
241    pub fn with_retry(mut self, retry: RetryPolicy) -> Self {
242        self.retry = retry;
243        self
244    }
245}
246
247impl RuntimeCommand {
248    pub fn schedule_step(
249        step_id: impl Into<String>,
250        step_name: impl Into<String>,
251        input: JsonValue,
252    ) -> Self {
253        Self::ScheduleStep {
254            step_id: step_id.into(),
255            step_name: step_name.into(),
256            input,
257            retry: RetryPolicy::default(),
258        }
259    }
260
261    pub fn schedule_steps(steps: Vec<StepCommand>) -> Self {
262        Self::ScheduleSteps { steps }
263    }
264}
a3s_flow/model/command.rs

a3s_flow/model/
command.rs
