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    /// Caller-supplied arguments handed to the workflow's query handler.
82    ///
83    /// Absent means "no arguments": the server materializes the JSON `null`
84    /// document so a handler always receives one well-formed input. Arguments
85    /// are read-only handler inputs and are never recorded in history.
86    #[prost(message, optional, tag = "5")]
87    pub arguments: Option<ProtoPayload>,
88}
89
90/// Proto representation of `QueryResponse`.
91#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
92pub struct ProtoQueryResponse {
93    /// Query result or typed wire error.
94    #[prost(oneof = "proto_query_response::Outcome", tags = "1, 2")]
95    pub outcome: Option<proto_query_response::Outcome>,
96}
97
98/// Types nested under [`ProtoQueryResponse`].
99pub mod proto_query_response {
100    /// Proto oneof for successful query payloads and typed failures.
101    #[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Oneof)]
102    pub enum Outcome {
103        /// Query result payload.
104        #[prost(message, tag = "1")]
105        Result(super::ProtoPayload),
106        /// Typed query error.
107        #[prost(message, tag = "2")]
108        Error(super::ProtoWireError),
109    }
110}
111
112/// Proto representation of `CancelRequest`.
113#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
114pub struct ProtoCancelRequest {
115    /// Namespace that scopes the operation.
116    #[prost(string, tag = "1")]
117    pub namespace: String,
118    /// Target workflow identifier.
119    #[prost(message, optional, tag = "2")]
120    pub workflow_id: Option<ProtoWorkflowId>,
121    /// Target run identifier.
122    #[prost(message, optional, tag = "3")]
123    pub run_id: Option<ProtoRunId>,
124    /// Human-readable cancellation reason.
125    #[prost(string, tag = "4")]
126    pub reason: String,
127}
128
129/// Proto representation of `CancelResponse`.
130#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
131pub struct ProtoCancelResponse {}
132
133/// Proto representation of `ReopenRequest`.
134///
135/// Mirrors [`ProtoCancelRequest`] without a `reason`: the reopen carries only a
136/// target. An absent `run_id` means the latest run.
137#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
138pub struct ProtoReopenRequest {
139    /// Namespace that scopes the operation.
140    #[prost(string, tag = "1")]
141    pub namespace: String,
142    /// Target workflow identifier.
143    #[prost(message, optional, tag = "2")]
144    pub workflow_id: Option<ProtoWorkflowId>,
145    /// Target run identifier (absent means the latest run).
146    #[prost(message, optional, tag = "3")]
147    pub run_id: Option<ProtoRunId>,
148}
149
150/// Proto representation of `ReopenResponse`.
151///
152/// Unlike [`ProtoCancelResponse`] (an empty ack) this returns the reopened run
153/// id and its projected status (Running) so the caller learns the run is live
154/// again.
155#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
156pub struct ProtoReopenResponse {
157    /// The reopened concrete run identifier.
158    #[prost(message, optional, tag = "1")]
159    pub run_id: Option<ProtoRunId>,
160    /// The projected workflow status after the reopen (Running).
161    #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
162    pub status: i32,
163}
164
165/// Proto representation of `PauseRequest` (#204).
166///
167/// Mirrors [`ProtoCancelRequest`]: a target plus an optional reason. An absent
168/// `run_id` means the latest run.
169#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
170pub struct ProtoPauseRequest {
171    /// Namespace that scopes the operation.
172    #[prost(string, tag = "1")]
173    pub namespace: String,
174    /// Target workflow identifier.
175    #[prost(message, optional, tag = "2")]
176    pub workflow_id: Option<ProtoWorkflowId>,
177    /// Target run identifier (absent means the latest run).
178    #[prost(message, optional, tag = "3")]
179    pub run_id: Option<ProtoRunId>,
180    /// Optional operator-supplied pause reason.
181    #[prost(string, tag = "4")]
182    pub reason: String,
183}
184
185/// Proto representation of `PauseResponse` (#204): the paused run and its status.
186#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
187pub struct ProtoPauseResponse {
188    /// The paused concrete run identifier.
189    #[prost(message, optional, tag = "1")]
190    pub run_id: Option<ProtoRunId>,
191    /// The projected workflow status after the pause (Paused).
192    #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
193    pub status: i32,
194}
195
196/// Proto representation of `ResumeRequest` (#204).
197///
198/// Mirrors [`ProtoReopenRequest`]: only a target. An absent `run_id` means the
199/// latest run.
200#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
201pub struct ProtoResumeRequest {
202    /// Namespace that scopes the operation.
203    #[prost(string, tag = "1")]
204    pub namespace: String,
205    /// Target workflow identifier.
206    #[prost(message, optional, tag = "2")]
207    pub workflow_id: Option<ProtoWorkflowId>,
208    /// Target run identifier (absent means the latest run).
209    #[prost(message, optional, tag = "3")]
210    pub run_id: Option<ProtoRunId>,
211}
212
213/// Proto representation of `ResumeResponse` (#204): the resumed run and status.
214#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
215pub struct ProtoResumeResponse {
216    /// The resumed concrete run identifier.
217    #[prost(message, optional, tag = "1")]
218    pub run_id: Option<ProtoRunId>,
219    /// The projected workflow status after the resume (Running).
220    #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
221    pub status: i32,
222}
223
224/// Proto representation of `ListWorkflowsRequest`.
225#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
226pub struct ProtoListWorkflowsRequest {
227    /// Namespace that scopes the operation.
228    #[prost(string, tag = "1")]
229    pub namespace: String,
230    /// Serde-encoded `aion_store::visibility::ListWorkflowsFilter` envelope.
231    #[prost(message, optional, tag = "2")]
232    pub filter: Option<WireEnvelope>,
233}
234
235/// Proto representation of `ListWorkflowsResponse`.
236#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
237pub struct ProtoListWorkflowsResponse {
238    /// Serde-encoded `aion_store::visibility::WorkflowSummary` envelopes.
239    #[prost(message, repeated, tag = "1")]
240    pub summaries: Vec<WireEnvelope>,
241}
242
243/// Proto representation of `CountWorkflowsRequest`.
244#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
245pub struct ProtoCountWorkflowsRequest {
246    /// Namespace that scopes the operation.
247    #[prost(string, tag = "1")]
248    pub namespace: String,
249    /// Serde-encoded `aion_store::visibility::ListWorkflowsFilter` envelope.
250    #[prost(message, optional, tag = "2")]
251    pub filter: Option<WireEnvelope>,
252}
253
254/// Proto representation of `CountWorkflowsResponse`.
255#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
256pub struct ProtoCountWorkflowsResponse {
257    /// Number of visibility summaries matching the filter.
258    #[prost(uint64, tag = "1")]
259    pub count: u64,
260}
261
262/// Proto representation of `DescribeWorkflowRequest`.
263#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
264pub struct ProtoDescribeWorkflowRequest {
265    /// Namespace that scopes the operation.
266    #[prost(string, tag = "1")]
267    pub namespace: String,
268    /// Target workflow identifier.
269    #[prost(message, optional, tag = "2")]
270    pub workflow_id: Option<ProtoWorkflowId>,
271    /// Target run identifier.
272    #[prost(message, optional, tag = "3")]
273    pub run_id: Option<ProtoRunId>,
274    /// Whether event history should be included in the response.
275    #[prost(bool, tag = "4")]
276    pub include_history: bool,
277}
278
279/// Proto representation of `DescribeWorkflowResponse`.
280#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
281pub struct ProtoDescribeWorkflowResponse {
282    /// Serde-encoded `aion_core::WorkflowSummary` envelope.
283    #[prost(message, optional, tag = "1")]
284    pub summary: Option<WireEnvelope>,
285    /// Optional serde-encoded `aion_core::Event` envelopes.
286    #[prost(message, repeated, tag = "2")]
287    pub history: Vec<WireEnvelope>,
288}
289
290#[cfg(test)]
291mod tests {
292    use std::collections::HashMap;
293
294    use aion_core::SearchAttributeValue;
295    use aion_store::visibility::{ListWorkflowsFilter, SearchAttributePredicate};
296    use chrono::{DateTime, Utc};
297    use prost::Message;
298    use serde::de::DeserializeOwned;
299    use serde_json::json;
300
301    use super::{
302        ProtoCountWorkflowsRequest, ProtoCountWorkflowsResponse, ProtoListWorkflowsRequest,
303        ProtoListWorkflowsResponse, ProtoQueryRequest, ProtoQueryResponse, ProtoReopenRequest,
304        ProtoReopenResponse, ProtoStartWorkflowRequest, ProtoStartWorkflowResponse,
305        proto_query_response,
306    };
307    use crate::convert::{
308        ProtoPayload, ProtoRunId, ProtoWorkflowId, decode_core_value, encode_core_value,
309    };
310    use crate::error::{ProtoWireError, WireError};
311
312    fn workflow_id() -> aion_core::WorkflowId {
313        aion_core::WorkflowId::new(uuid::Uuid::nil())
314    }
315
316    fn run_id() -> aion_core::RunId {
317        aion_core::RunId::new(uuid::Uuid::nil())
318    }
319
320    fn payload(label: &str) -> Result<ProtoPayload, aion_core::PayloadError> {
321        Ok(ProtoPayload::from(aion_core::Payload::from_json(
322            &json!({ "label": label }),
323        )?))
324    }
325
326    fn recorded_at() -> Result<DateTime<Utc>, chrono::ParseError> {
327        Ok(DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")?.with_timezone(&Utc))
328    }
329
330    fn assert_json_round_trip<T>(value: &T) -> Result<(), serde_json::Error>
331    where
332        T: Clone + PartialEq + serde::Serialize + DeserializeOwned,
333    {
334        let encoded = serde_json::to_string(value)?;
335        let decoded = serde_json::from_str::<T>(&encoded)?;
336        assert!(decoded == *value);
337        Ok(())
338    }
339
340    fn assert_proto_round_trip<T>(value: &T) -> Result<(), Box<dyn std::error::Error>>
341    where
342        T: Clone + PartialEq + Message + Default,
343    {
344        let mut bytes = Vec::new();
345        value.encode(&mut bytes)?;
346        let decoded = T::decode(bytes.as_slice())?;
347        assert!(decoded == *value);
348        Ok(())
349    }
350
351    #[test]
352    fn start_workflow_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
353        let request = ProtoStartWorkflowRequest {
354            namespace: String::from("tenant-a"),
355            workflow_type: String::from("checkout"),
356            input: Some(payload("input")?),
357            routing_key: Some(String::from("tenant-a/order-1")),
358            task_queue: Some(String::from("gpu")),
359        };
360        let response = ProtoStartWorkflowResponse {
361            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
362            run_id: Some(ProtoRunId::from(run_id())),
363        };
364
365        assert_json_round_trip(&request)?;
366        assert_proto_round_trip(&request)?;
367        assert_json_round_trip(&response)?;
368        assert_proto_round_trip(&response)?;
369        Ok(())
370    }
371
372    #[test]
373    fn list_workflows_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
374        let filter = ListWorkflowsFilter {
375            workflow_type: Some(String::from("checkout")),
376            status: Some(aion_core::WorkflowStatus::Running),
377            search_attributes: vec![SearchAttributePredicate::Equals {
378                name: String::from("customer_id"),
379                value: SearchAttributeValue::String(String::from("12345")),
380            }],
381            limit: Some(10),
382            offset: Some(5),
383            ..ListWorkflowsFilter::default()
384        };
385        let summary = aion_store::visibility::WorkflowSummary {
386            workflow_id: workflow_id(),
387            run_id: run_id(),
388            workflow_type: String::from("checkout"),
389            status: aion_core::WorkflowStatus::Running,
390            start_time: recorded_at()?,
391            close_time: None,
392            failed_step: None,
393            failure_reason: None,
394            search_attributes: HashMap::from([(
395                String::from("customer_id"),
396                SearchAttributeValue::String(String::from("12345")),
397            )]),
398        };
399        let filter_envelope = encode_core_value("tenant-a", Some(String::from("r1")), &filter)?;
400        let summary_envelope = encode_core_value("tenant-a", None, &summary)?;
401        let request = ProtoListWorkflowsRequest {
402            namespace: String::from("tenant-a"),
403            filter: Some(filter_envelope.clone()),
404        };
405        let response = ProtoListWorkflowsResponse {
406            summaries: vec![summary_envelope.clone()],
407        };
408        let count_request = ProtoCountWorkflowsRequest {
409            namespace: String::from("tenant-a"),
410            filter: Some(filter_envelope.clone()),
411        };
412        let count_response = ProtoCountWorkflowsResponse { count: 1 };
413
414        assert_json_round_trip(&request)?;
415        assert_proto_round_trip(&request)?;
416        assert_json_round_trip(&response)?;
417        assert_proto_round_trip(&response)?;
418        assert_json_round_trip(&count_request)?;
419        assert_proto_round_trip(&count_request)?;
420        assert_json_round_trip(&count_response)?;
421        assert_proto_round_trip(&count_response)?;
422        assert_eq!(
423            decode_core_value::<ListWorkflowsFilter>(&filter_envelope)?,
424            filter
425        );
426        assert_eq!(
427            decode_core_value::<aion_store::visibility::WorkflowSummary>(&summary_envelope)?,
428            summary
429        );
430        Ok(())
431    }
432
433    #[test]
434    fn query_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
435        let request = ProtoQueryRequest {
436            namespace: String::from("tenant-a"),
437            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
438            run_id: Some(ProtoRunId::from(run_id())),
439            query_name: String::from("state"),
440            arguments: Some(payload("arguments")?),
441        };
442        // A caller that supplies no arguments is a distinct encoded shape, not
443        // an error: the field is optional on the wire and the server supplies
444        // the canonical `null` document in its place.
445        let no_arguments_request = ProtoQueryRequest {
446            arguments: None,
447            ..request.clone()
448        };
449        let result_response = ProtoQueryResponse {
450            outcome: Some(proto_query_response::Outcome::Result(payload("result")?)),
451        };
452        let error_response = ProtoQueryResponse {
453            outcome: Some(proto_query_response::Outcome::Error(ProtoWireError::from(
454                WireError::unknown_query("state query is not registered"),
455            ))),
456        };
457
458        assert_json_round_trip(&request)?;
459        assert_proto_round_trip(&request)?;
460        assert_json_round_trip(&no_arguments_request)?;
461        assert_proto_round_trip(&no_arguments_request)?;
462        assert_json_round_trip(&result_response)?;
463        assert_proto_round_trip(&result_response)?;
464        assert_json_round_trip(&error_response)?;
465        assert_proto_round_trip(&error_response)?;
466        // The two shapes stay distinguishable across a proto round trip: a
467        // decoder can tell "no arguments supplied" from any supplied document.
468        assert_ne!(request, no_arguments_request);
469        Ok(())
470    }
471
472    #[test]
473    fn reopen_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
474        let request = ProtoReopenRequest {
475            namespace: String::from("tenant-a"),
476            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
477            run_id: Some(ProtoRunId::from(run_id())),
478        };
479        let response = ProtoReopenResponse {
480            run_id: Some(ProtoRunId::from(run_id())),
481            status: crate::convert::ProtoWorkflowStatus::Running as i32,
482        };
483
484        assert_json_round_trip(&request)?;
485        assert_proto_round_trip(&request)?;
486        assert_json_round_trip(&response)?;
487        assert_proto_round_trip(&response)?;
488        Ok(())
489    }
490}
aion_proto/workflow.rs

aion_proto/
workflow.rs
