a3s_flow/model/
operation.rs

1use chrono::{DateTime, Utc};
2use serde::de::DeserializeOwned;
3use serde::{Deserialize, Serialize};
4
5use crate::error::{FlowError, Result};
6
7use super::JsonValue;
8
9/// Durable link from a closed history segment to its successor run.
10///
11/// The successor inherits the predecessor's complete [`super::WorkflowSpec`].
12/// Only input changes across the boundary, so replay-code admission and patch
13/// markers cannot drift while an execution is segmented.
14#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
15#[non_exhaustive]
16pub struct WorkflowContinuation {
17    /// Run identifier assigned to the successor history segment.
18    pub successor_run_id: String,
19    /// Initial JSON input supplied to the successor.
20    pub input: JsonValue,
21}
22
23impl WorkflowContinuation {
24    /// Decode the successor input into a host-defined serde type.
25    pub fn input_as<T>(&self) -> Result<T>
26    where
27        T: DeserializeOwned,
28    {
29        serde_json::from_value(self.input.clone()).map_err(FlowError::from)
30    }
31}
32
33/// Durable request for a workflow to stop through its cleanup-aware path.
34#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
35#[non_exhaustive]
36pub struct CancellationRequest {
37    /// Optional operator- or application-supplied reason.
38    #[serde(default, skip_serializing_if = "Option::is_none")]
39    pub reason: Option<String>,
40}
41
42impl CancellationRequest {
43    /// Creates a cancellation request with an optional reason.
44    pub fn new(reason: Option<String>) -> Self {
45        Self { reason }
46    }
47}
48
49/// Projected cancellation request with its durable event position.
50#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
51#[non_exhaustive]
52pub struct CancellationRequestSnapshot {
53    /// Immutable request delivered during cleanup replay.
54    pub request: CancellationRequest,
55    /// UTC time at which the request was persisted.
56    pub requested_at: DateTime<Utc>,
57    /// Event sequence that introduced the request.
58    pub sequence: u64,
59}
60
61/// A durable, idempotently identified progress update.
62#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
63#[non_exhaustive]
64pub struct WorkflowProgress {
65    /// Caller-chosen idempotency identity for this update.
66    pub progress_id: String,
67    /// Number of completed work units.
68    pub completed: u64,
69    /// Optional total number of work units.
70    #[serde(default, skip_serializing_if = "Option::is_none")]
71    pub total: Option<u64>,
72    /// Optional human-readable progress message.
73    #[serde(default, skip_serializing_if = "Option::is_none")]
74    pub message: Option<String>,
75    /// Application-defined structured progress details.
76    #[serde(default, skip_serializing_if = "JsonValue::is_null")]
77    pub details: JsonValue,
78}
79
80impl WorkflowProgress {
81    /// Creates a progress update without a total, message, or details.
82    pub fn new(progress_id: impl Into<String>, completed: u64) -> Self {
83        Self {
84            progress_id: progress_id.into(),
85            completed,
86            total: None,
87            message: None,
88            details: JsonValue::Null,
89        }
90    }
91
92    /// Sets the total number of work units.
93    pub fn with_total(mut self, total: u64) -> Self {
94        self.total = Some(total);
95        self
96    }
97
98    /// Sets a human-readable progress message.
99    pub fn with_message(mut self, message: impl Into<String>) -> Self {
100        self.message = Some(message.into());
101        self
102    }
103
104    /// Sets application-defined structured details.
105    pub fn with_details(mut self, details: JsonValue) -> Self {
106        self.details = details;
107        self
108    }
109
110    pub(crate) fn validate(&self) -> Result<()> {
111        if self.progress_id.trim().is_empty() {
112            return Err(FlowError::InvalidTransition(
113                "workflow progress id must not be empty".to_string(),
114            ));
115        }
116        if self
117            .total
118            .is_some_and(|total| total == 0 || self.completed > total)
119        {
120            return Err(FlowError::InvalidTransition(format!(
121                "workflow progress {} must satisfy completed <= total and total > 0",
122                self.progress_id
123            )));
124        }
125        Ok(())
126    }
127}
128
129/// Durable reference from a parent workflow to a child operation.
130///
131/// `flow_run_id` is set only when the child is another A3S Flow run. The
132/// reference itself does not imply automatic cancellation; the parent
133/// workflow owns propagation through durable, idempotent cleanup steps.
134#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
135#[non_exhaustive]
136pub struct ChildOperationReference {
137    /// Replay-stable parent-local identity of the reference.
138    pub reference_id: String,
139    /// Application-defined operation kind.
140    pub kind: String,
141    /// Identifier assigned by the child operation's owner.
142    pub operation_id: String,
143    /// Linked A3S Flow run identifier, when the child is a workflow run.
144    #[serde(default, skip_serializing_if = "Option::is_none")]
145    pub flow_run_id: Option<String>,
146    /// Application-defined structured metadata.
147    #[serde(default, skip_serializing_if = "JsonValue::is_null")]
148    pub metadata: JsonValue,
149}
150
151impl ChildOperationReference {
152    /// Creates a child-operation reference without Flow ownership or metadata.
153    pub fn new(
154        reference_id: impl Into<String>,
155        kind: impl Into<String>,
156        operation_id: impl Into<String>,
157    ) -> Self {
158        Self {
159            reference_id: reference_id.into(),
160            kind: kind.into(),
161            operation_id: operation_id.into(),
162            flow_run_id: None,
163            metadata: JsonValue::Null,
164        }
165    }
166
167    /// Associates the operation with an A3S Flow run.
168    pub fn with_flow_run_id(mut self, flow_run_id: impl Into<String>) -> Self {
169        self.flow_run_id = Some(flow_run_id.into());
170        self
171    }
172
173    /// Sets application-defined structured metadata.
174    pub fn with_metadata(mut self, metadata: JsonValue) -> Self {
175        self.metadata = metadata;
176        self
177    }
178
179    pub(crate) fn validate(&self) -> Result<()> {
180        if self.reference_id.trim().is_empty() {
181            return Err(FlowError::InvalidTransition(
182                "child operation reference id must not be empty".to_string(),
183            ));
184        }
185        if self.kind.trim().is_empty() {
186            return Err(FlowError::InvalidTransition(format!(
187                "child operation {} kind must not be empty",
188                self.reference_id
189            )));
190        }
191        if self.operation_id.trim().is_empty() {
192            return Err(FlowError::InvalidTransition(format!(
193                "child operation {} operation id must not be empty",
194                self.reference_id
195            )));
196        }
197        if self
198            .flow_run_id
199            .as_deref()
200            .is_some_and(|run_id| run_id.trim().is_empty())
201        {
202            return Err(FlowError::InvalidTransition(format!(
203                "child operation {} Flow run id must not be empty",
204                self.reference_id
205            )));
206        }
207        Ok(())
208    }
209}
210
211/// Typed terminal result projected from the final run event.
212#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
213#[non_exhaustive]
214#[serde(tag = "type", rename_all = "snake_case")]
215pub enum WorkflowTerminalOutcome {
216    /// The workflow returned a successful output.
217    Completed {
218        /// Final JSON value returned by the workflow.
219        output: JsonValue,
220    },
221    /// The workflow terminated with an application or runtime error.
222    Failed {
223        /// Human-readable failure description.
224        error: String,
225    },
226    /// The workflow completed cancellation.
227    Cancelled {
228        /// Optional operator- or application-supplied cancellation reason.
229        #[serde(default, skip_serializing_if = "Option::is_none")]
230        reason: Option<String>,
231    },
232    /// The workflow exceeded its deadline.
233    TimedOut {
234        /// UTC deadline that caused the timeout.
235        deadline: DateTime<Utc>,
236        /// Optional context for the timeout decision.
237        #[serde(default, skip_serializing_if = "Option::is_none")]
238        reason: Option<String>,
239    },
240    /// A step exhausted every permitted attempt.
241    RetryExhausted {
242        /// Stable identifier of the exhausted step.
243        step_id: String,
244        /// Final attempt number that failed.
245        attempt: u32,
246        /// Error returned by the final attempt.
247        error: String,
248    },
249    /// The owning host terminated the run during shutdown.
250    HostShutdown {
251        /// Optional host shutdown reason.
252        #[serde(default, skip_serializing_if = "Option::is_none")]
253        reason: Option<String>,
254    },
255    /// This history segment closed after creating a successor.
256    ContinuedAsNew {
257        /// Identifier assigned to the successor run.
258        successor_run_id: String,
259    },
260}
a3s_flow/model/operation.rs

a3s_flow/model/
operation.rs
