aion_proto/
workflow.rs

1//! Workflow-management serde/prost wire types.
2
3use crate::convert::{ProtoPayload, ProtoRunId, ProtoWorkflowId, WireEnvelope};
4use crate::error::ProtoWireError;
5
6/// Proto representation of `StartWorkflowRequest`.
7#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
8pub struct ProtoStartWorkflowRequest {
9    /// Namespace that scopes the operation.
10    #[prost(string, tag = "1")]
11    pub namespace: String,
12    /// Workflow type name registered with the engine.
13    #[prost(string, tag = "2")]
14    pub workflow_type: String,
15    /// Workflow start input payload.
16    #[prost(message, optional, tag = "3")]
17    pub input: Option<ProtoPayload>,
18    /// R-4 steered-start routing key. When set, the start is steered to
19    /// `shard_for(routing_key)`'s owner (forwarded there when this node is not the
20    /// owner). `None`/empty keeps the unsteered R-1 remint behaviour.
21    #[prost(string, optional, tag = "4")]
22    pub routing_key: Option<String>,
23    /// Optional task queue this workflow defaults its activities to (the
24    /// namespace × `task_queue` targeting story). When set, the server records it
25    /// durably on the start so it survives replay/failover. `None`/empty keeps
26    /// the namespace's default queue.
27    #[prost(string, optional, tag = "5")]
28    pub task_queue: Option<String>,
29}
30
31/// Proto representation of `StartWorkflowResponse`.
32#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
33pub struct ProtoStartWorkflowResponse {
34    /// Assigned workflow identifier.
35    #[prost(message, optional, tag = "1")]
36    pub workflow_id: Option<ProtoWorkflowId>,
37    /// Assigned concrete run identifier.
38    #[prost(message, optional, tag = "2")]
39    pub run_id: Option<ProtoRunId>,
40}
41
42/// Proto representation of `SignalRequest`.
43#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
44pub struct ProtoSignalRequest {
45    /// Namespace that scopes the operation.
46    #[prost(string, tag = "1")]
47    pub namespace: String,
48    /// Target workflow identifier.
49    #[prost(message, optional, tag = "2")]
50    pub workflow_id: Option<ProtoWorkflowId>,
51    /// Target run identifier.
52    #[prost(message, optional, tag = "3")]
53    pub run_id: Option<ProtoRunId>,
54    /// Signal name registered by workflow code.
55    #[prost(string, tag = "4")]
56    pub signal_name: String,
57    /// Signal payload.
58    #[prost(message, optional, tag = "5")]
59    pub payload: Option<ProtoPayload>,
60}
61
62/// Proto representation of `SignalResponse`.
63#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
64pub struct ProtoSignalResponse {}
65
66/// Proto representation of `QueryRequest`.
67#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
68pub struct ProtoQueryRequest {
69    /// Namespace that scopes the operation.
70    #[prost(string, tag = "1")]
71    pub namespace: String,
72    /// Target workflow identifier.
73    #[prost(message, optional, tag = "2")]
74    pub workflow_id: Option<ProtoWorkflowId>,
75    /// Target run identifier.
76    #[prost(message, optional, tag = "3")]
77    pub run_id: Option<ProtoRunId>,
78    /// Query name registered by workflow code.
79    #[prost(string, tag = "4")]
80    pub query_name: String,
81}
82
83/// Proto representation of `QueryResponse`.
84#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
85pub struct ProtoQueryResponse {
86    /// Query result or typed wire error.
87    #[prost(oneof = "proto_query_response::Outcome", tags = "1, 2")]
88    pub outcome: Option<proto_query_response::Outcome>,
89}
90
91/// Types nested under [`ProtoQueryResponse`].
92pub mod proto_query_response {
93    /// Proto oneof for successful query payloads and typed failures.
94    #[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Oneof)]
95    pub enum Outcome {
96        /// Query result payload.
97        #[prost(message, tag = "1")]
98        Result(super::ProtoPayload),
99        /// Typed query error.
100        #[prost(message, tag = "2")]
101        Error(super::ProtoWireError),
102    }
103}
104
105/// Proto representation of `CancelRequest`.
106#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
107pub struct ProtoCancelRequest {
108    /// Namespace that scopes the operation.
109    #[prost(string, tag = "1")]
110    pub namespace: String,
111    /// Target workflow identifier.
112    #[prost(message, optional, tag = "2")]
113    pub workflow_id: Option<ProtoWorkflowId>,
114    /// Target run identifier.
115    #[prost(message, optional, tag = "3")]
116    pub run_id: Option<ProtoRunId>,
117    /// Human-readable cancellation reason.
118    #[prost(string, tag = "4")]
119    pub reason: String,
120}
121
122/// Proto representation of `CancelResponse`.
123#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
124pub struct ProtoCancelResponse {}
125
126/// Proto representation of `ReopenRequest`.
127///
128/// Mirrors [`ProtoCancelRequest`] without a `reason`: the reopen carries only a
129/// target. An absent `run_id` means the latest run.
130#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
131pub struct ProtoReopenRequest {
132    /// Namespace that scopes the operation.
133    #[prost(string, tag = "1")]
134    pub namespace: String,
135    /// Target workflow identifier.
136    #[prost(message, optional, tag = "2")]
137    pub workflow_id: Option<ProtoWorkflowId>,
138    /// Target run identifier (absent means the latest run).
139    #[prost(message, optional, tag = "3")]
140    pub run_id: Option<ProtoRunId>,
141}
142
143/// Proto representation of `ReopenResponse`.
144///
145/// Unlike [`ProtoCancelResponse`] (an empty ack) this returns the reopened run
146/// id and its projected status (Running) so the caller learns the run is live
147/// again.
148#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
149pub struct ProtoReopenResponse {
150    /// The reopened concrete run identifier.
151    #[prost(message, optional, tag = "1")]
152    pub run_id: Option<ProtoRunId>,
153    /// The projected workflow status after the reopen (Running).
154    #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
155    pub status: i32,
156}
157
158/// Proto representation of `PauseRequest` (#204).
159///
160/// Mirrors [`ProtoCancelRequest`]: a target plus an optional reason. An absent
161/// `run_id` means the latest run.
162#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
163pub struct ProtoPauseRequest {
164    /// Namespace that scopes the operation.
165    #[prost(string, tag = "1")]
166    pub namespace: String,
167    /// Target workflow identifier.
168    #[prost(message, optional, tag = "2")]
169    pub workflow_id: Option<ProtoWorkflowId>,
170    /// Target run identifier (absent means the latest run).
171    #[prost(message, optional, tag = "3")]
172    pub run_id: Option<ProtoRunId>,
173    /// Optional operator-supplied pause reason.
174    #[prost(string, tag = "4")]
175    pub reason: String,
176}
177
178/// Proto representation of `PauseResponse` (#204): the paused run and its status.
179#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
180pub struct ProtoPauseResponse {
181    /// The paused concrete run identifier.
182    #[prost(message, optional, tag = "1")]
183    pub run_id: Option<ProtoRunId>,
184    /// The projected workflow status after the pause (Paused).
185    #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
186    pub status: i32,
187}
188
189/// Proto representation of `ResumeRequest` (#204).
190///
191/// Mirrors [`ProtoReopenRequest`]: only a target. An absent `run_id` means the
192/// latest run.
193#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
194pub struct ProtoResumeRequest {
195    /// Namespace that scopes the operation.
196    #[prost(string, tag = "1")]
197    pub namespace: String,
198    /// Target workflow identifier.
199    #[prost(message, optional, tag = "2")]
200    pub workflow_id: Option<ProtoWorkflowId>,
201    /// Target run identifier (absent means the latest run).
202    #[prost(message, optional, tag = "3")]
203    pub run_id: Option<ProtoRunId>,
204}
205
206/// Proto representation of `ResumeResponse` (#204): the resumed run and status.
207#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
208pub struct ProtoResumeResponse {
209    /// The resumed concrete run identifier.
210    #[prost(message, optional, tag = "1")]
211    pub run_id: Option<ProtoRunId>,
212    /// The projected workflow status after the resume (Running).
213    #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
214    pub status: i32,
215}
216
217/// Proto representation of `ListWorkflowsRequest`.
218#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
219pub struct ProtoListWorkflowsRequest {
220    /// Namespace that scopes the operation.
221    #[prost(string, tag = "1")]
222    pub namespace: String,
223    /// Serde-encoded `aion_store::visibility::ListWorkflowsFilter` envelope.
224    #[prost(message, optional, tag = "2")]
225    pub filter: Option<WireEnvelope>,
226}
227
228/// Proto representation of `ListWorkflowsResponse`.
229#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
230pub struct ProtoListWorkflowsResponse {
231    /// Serde-encoded `aion_store::visibility::WorkflowSummary` envelopes.
232    #[prost(message, repeated, tag = "1")]
233    pub summaries: Vec<WireEnvelope>,
234}
235
236/// Proto representation of `CountWorkflowsRequest`.
237#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
238pub struct ProtoCountWorkflowsRequest {
239    /// Namespace that scopes the operation.
240    #[prost(string, tag = "1")]
241    pub namespace: String,
242    /// Serde-encoded `aion_store::visibility::ListWorkflowsFilter` envelope.
243    #[prost(message, optional, tag = "2")]
244    pub filter: Option<WireEnvelope>,
245}
246
247/// Proto representation of `CountWorkflowsResponse`.
248#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
249pub struct ProtoCountWorkflowsResponse {
250    /// Number of visibility summaries matching the filter.
251    #[prost(uint64, tag = "1")]
252    pub count: u64,
253}
254
255/// Proto representation of `DescribeWorkflowRequest`.
256#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
257pub struct ProtoDescribeWorkflowRequest {
258    /// Namespace that scopes the operation.
259    #[prost(string, tag = "1")]
260    pub namespace: String,
261    /// Target workflow identifier.
262    #[prost(message, optional, tag = "2")]
263    pub workflow_id: Option<ProtoWorkflowId>,
264    /// Target run identifier.
265    #[prost(message, optional, tag = "3")]
266    pub run_id: Option<ProtoRunId>,
267    /// Whether event history should be included in the response.
268    #[prost(bool, tag = "4")]
269    pub include_history: bool,
270}
271
272/// Proto representation of `DescribeWorkflowResponse`.
273#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
274pub struct ProtoDescribeWorkflowResponse {
275    /// Serde-encoded `aion_core::WorkflowSummary` envelope.
276    #[prost(message, optional, tag = "1")]
277    pub summary: Option<WireEnvelope>,
278    /// Optional serde-encoded `aion_core::Event` envelopes.
279    #[prost(message, repeated, tag = "2")]
280    pub history: Vec<WireEnvelope>,
281}
282
283#[cfg(test)]
284mod tests {
285    use std::collections::HashMap;
286
287    use aion_core::SearchAttributeValue;
288    use aion_store::visibility::{ListWorkflowsFilter, SearchAttributePredicate};
289    use chrono::{DateTime, Utc};
290    use prost::Message;
291    use serde::de::DeserializeOwned;
292    use serde_json::json;
293
294    use super::{
295        ProtoCountWorkflowsRequest, ProtoCountWorkflowsResponse, ProtoListWorkflowsRequest,
296        ProtoListWorkflowsResponse, ProtoQueryRequest, ProtoQueryResponse, ProtoReopenRequest,
297        ProtoReopenResponse, ProtoStartWorkflowRequest, ProtoStartWorkflowResponse,
298        proto_query_response,
299    };
300    use crate::convert::{
301        ProtoPayload, ProtoRunId, ProtoWorkflowId, decode_core_value, encode_core_value,
302    };
303    use crate::error::{ProtoWireError, WireError};
304
305    fn workflow_id() -> aion_core::WorkflowId {
306        aion_core::WorkflowId::new(uuid::Uuid::nil())
307    }
308
309    fn run_id() -> aion_core::RunId {
310        aion_core::RunId::new(uuid::Uuid::nil())
311    }
312
313    fn payload(label: &str) -> Result<ProtoPayload, aion_core::PayloadError> {
314        Ok(ProtoPayload::from(aion_core::Payload::from_json(
315            &json!({ "label": label }),
316        )?))
317    }
318
319    fn recorded_at() -> Result<DateTime<Utc>, chrono::ParseError> {
320        Ok(DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")?.with_timezone(&Utc))
321    }
322
323    fn assert_json_round_trip<T>(value: &T) -> Result<(), serde_json::Error>
324    where
325        T: Clone + PartialEq + serde::Serialize + DeserializeOwned,
326    {
327        let encoded = serde_json::to_string(value)?;
328        let decoded = serde_json::from_str::<T>(&encoded)?;
329        assert!(decoded == *value);
330        Ok(())
331    }
332
333    fn assert_proto_round_trip<T>(value: &T) -> Result<(), Box<dyn std::error::Error>>
334    where
335        T: Clone + PartialEq + Message + Default,
336    {
337        let mut bytes = Vec::new();
338        value.encode(&mut bytes)?;
339        let decoded = T::decode(bytes.as_slice())?;
340        assert!(decoded == *value);
341        Ok(())
342    }
343
344    #[test]
345    fn start_workflow_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
346        let request = ProtoStartWorkflowRequest {
347            namespace: String::from("tenant-a"),
348            workflow_type: String::from("checkout"),
349            input: Some(payload("input")?),
350            routing_key: Some(String::from("tenant-a/order-1")),
351            task_queue: Some(String::from("gpu")),
352        };
353        let response = ProtoStartWorkflowResponse {
354            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
355            run_id: Some(ProtoRunId::from(run_id())),
356        };
357
358        assert_json_round_trip(&request)?;
359        assert_proto_round_trip(&request)?;
360        assert_json_round_trip(&response)?;
361        assert_proto_round_trip(&response)?;
362        Ok(())
363    }
364
365    #[test]
366    fn list_workflows_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
367        let filter = ListWorkflowsFilter {
368            workflow_type: Some(String::from("checkout")),
369            status: Some(aion_core::WorkflowStatus::Running),
370            search_attributes: vec![SearchAttributePredicate::Equals {
371                name: String::from("customer_id"),
372                value: SearchAttributeValue::String(String::from("12345")),
373            }],
374            limit: Some(10),
375            offset: Some(5),
376            ..ListWorkflowsFilter::default()
377        };
378        let summary = aion_store::visibility::WorkflowSummary {
379            workflow_id: workflow_id(),
380            run_id: run_id(),
381            workflow_type: String::from("checkout"),
382            status: aion_core::WorkflowStatus::Running,
383            start_time: recorded_at()?,
384            close_time: None,
385            failed_step: None,
386            failure_reason: None,
387            search_attributes: HashMap::from([(
388                String::from("customer_id"),
389                SearchAttributeValue::String(String::from("12345")),
390            )]),
391        };
392        let filter_envelope = encode_core_value("tenant-a", Some(String::from("r1")), &filter)?;
393        let summary_envelope = encode_core_value("tenant-a", None, &summary)?;
394        let request = ProtoListWorkflowsRequest {
395            namespace: String::from("tenant-a"),
396            filter: Some(filter_envelope.clone()),
397        };
398        let response = ProtoListWorkflowsResponse {
399            summaries: vec![summary_envelope.clone()],
400        };
401        let count_request = ProtoCountWorkflowsRequest {
402            namespace: String::from("tenant-a"),
403            filter: Some(filter_envelope.clone()),
404        };
405        let count_response = ProtoCountWorkflowsResponse { count: 1 };
406
407        assert_json_round_trip(&request)?;
408        assert_proto_round_trip(&request)?;
409        assert_json_round_trip(&response)?;
410        assert_proto_round_trip(&response)?;
411        assert_json_round_trip(&count_request)?;
412        assert_proto_round_trip(&count_request)?;
413        assert_json_round_trip(&count_response)?;
414        assert_proto_round_trip(&count_response)?;
415        assert_eq!(
416            decode_core_value::<ListWorkflowsFilter>(&filter_envelope)?,
417            filter
418        );
419        assert_eq!(
420            decode_core_value::<aion_store::visibility::WorkflowSummary>(&summary_envelope)?,
421            summary
422        );
423        Ok(())
424    }
425
426    #[test]
427    fn query_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
428        let request = ProtoQueryRequest {
429            namespace: String::from("tenant-a"),
430            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
431            run_id: Some(ProtoRunId::from(run_id())),
432            query_name: String::from("state"),
433        };
434        let result_response = ProtoQueryResponse {
435            outcome: Some(proto_query_response::Outcome::Result(payload("result")?)),
436        };
437        let error_response = ProtoQueryResponse {
438            outcome: Some(proto_query_response::Outcome::Error(ProtoWireError::from(
439                WireError::unknown_query("state query is not registered"),
440            ))),
441        };
442
443        assert_json_round_trip(&request)?;
444        assert_proto_round_trip(&request)?;
445        assert_json_round_trip(&result_response)?;
446        assert_proto_round_trip(&result_response)?;
447        assert_json_round_trip(&error_response)?;
448        assert_proto_round_trip(&error_response)?;
449        Ok(())
450    }
451
452    #[test]
453    fn reopen_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
454        let request = ProtoReopenRequest {
455            namespace: String::from("tenant-a"),
456            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
457            run_id: Some(ProtoRunId::from(run_id())),
458        };
459        let response = ProtoReopenResponse {
460            run_id: Some(ProtoRunId::from(run_id())),
461            status: crate::convert::ProtoWorkflowStatus::Running as i32,
462        };
463
464        assert_json_round_trip(&request)?;
465        assert_proto_round_trip(&request)?;
466        assert_json_round_trip(&response)?;
467        assert_proto_round_trip(&response)?;
468        Ok(())
469    }
470}
aion_proto/workflow.rs

aion_proto/
workflow.rs
