dex_sdk/
flow_result.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Super Durable Source License 1.0.
4// You may not use this file except in compliance with the License.
5// See the LICENSE file in the repository root.
6//
7// SPDX-License-Identifier: LicenseRef-Super-Durable-1.0
8
9use dex_protocol::dex::{FlowResult as ProtoFlowResult, Value as ProtoValue};
10
11use crate::{FlowErrorType, FlowStatus, SdkError, SdkResult, Value, value_mapper};
12
13#[derive(Clone, Debug)]
14/// Contains one output-bearing Step completion returned by [`crate::Client::wait_for_flow`].
15pub struct StepCompletion {
16    /// Registered Step type that produced this output.
17    pub step_type: String,
18    /// Exact server Step execution identity that produced this output.
19    pub step_execution_id: String,
20    output: ProtoValue,
21}
22
23impl StepCompletion {
24    pub(crate) fn new(step_type: String, step_execution_id: String, output: ProtoValue) -> Self {
25        Self {
26            step_type,
27            step_execution_id,
28            output,
29        }
30    }
31
32    /// Decodes this already hydrated Step output as `Output`.
33    ///
34    /// # Errors
35    ///
36    /// Returns [`SdkError::ValueMapping`] when the wire value is incompatible with `Output`.
37    pub fn decode<Output: Value>(&self) -> SdkResult<Output> {
38        value_mapper::decode(&self.output)
39    }
40}
41
42#[derive(Clone, Debug)]
43/// Describes an observed Flow status and its output-bearing completions.
44///
45/// Client waits return terminal results. A SubFlow AnyOf loser can return a Running snapshot,
46/// which does not guarantee that the independently running Flow remains active at read time.
47pub struct FlowResult {
48    status: FlowStatus,
49    error_type: Option<FlowErrorType>,
50    error_message: Option<String>,
51    completions: Vec<StepCompletion>,
52}
53
54impl FlowResult {
55    pub(crate) fn new(
56        status: FlowStatus,
57        error_type: Option<FlowErrorType>,
58        error_message: Option<String>,
59        completions: Vec<StepCompletion>,
60    ) -> Self {
61        Self {
62            status,
63            error_type,
64            error_message,
65            completions,
66        }
67    }
68
69    pub(crate) fn from_proto(result: &ProtoFlowResult) -> SdkResult<Self> {
70        let status = crate::client::map_flow_status(result.flow_status)?;
71        let completions = result
72            .results
73            .iter()
74            .map(|completion| {
75                Ok(StepCompletion::new(
76                    completion.completed_step_type.clone(),
77                    completion.completed_step_execution_id.clone(),
78                    completion.completed_step_output.clone().ok_or_else(|| {
79                        SdkError::ValueMapping {
80                            message: "Step completion output is required".to_string(),
81                        }
82                    })?,
83                ))
84            })
85            .collect::<SdkResult<Vec<_>>>()?;
86        Ok(Self::new(
87            status,
88            crate::client::map_flow_error_type(result.error_type),
89            (!result.error_message.is_empty()).then(|| result.error_message.clone()),
90            completions,
91        ))
92    }
93
94    /// Returns the observed Flow lifecycle state.
95    pub fn status(&self) -> FlowStatus {
96        self.status
97    }
98
99    /// Returns the Dex failure category when one was reported.
100    pub fn error_type(&self) -> Option<FlowErrorType> {
101        self.error_type
102    }
103
104    /// Returns the server failure detail when one was reported.
105    pub fn error_message(&self) -> Option<&str> {
106        self.error_message.as_deref()
107    }
108
109    /// Returns whether the observed run can no longer execute.
110    pub fn is_terminal(&self) -> bool {
111        !matches!(
112            self.status,
113            FlowStatus::Running | FlowStatus::ContinuedAsNew
114        )
115    }
116
117    /// Returns output-bearing completions in server collection order.
118    ///
119    /// Parallel Step order is not deterministic. Select by Step type or Step execution ID.
120    pub fn completions(&self) -> &[StepCompletion] {
121        &self.completions
122    }
123
124    /// Decodes the output when exactly one completion exists.
125    ///
126    /// # Errors
127    ///
128    /// Returns [`SdkError::InvalidArgument`] for a nonterminal result or zero or multiple outputs and
129    /// [`SdkError::ValueMapping`] when the output is incompatible with `Output`.
130    pub fn single_output<Output: Value>(&self) -> SdkResult<Output> {
131        if !self.is_terminal() {
132            return Err(SdkError::InvalidArgument {
133                message: "Flow result is not terminal".to_string(),
134            });
135        }
136        if self.completions().len() != 1 {
137            return Err(SdkError::InvalidArgument {
138                message: format!(
139                    "expected exactly one Step output, found {}",
140                    self.completions().len()
141                ),
142            });
143        }
144        self.completions[0].decode()
145    }
146}
147
148#[cfg(test)]
149mod tests {
150    use dex_protocol::dex::value;
151
152    use super::*;
153
154    #[test]
155    fn preserves_heterogeneous_completions_and_rejects_ambiguous_single_output() {
156        let result = FlowResult::new(
157            crate::FlowStatus::Completed,
158            None,
159            None,
160            vec![
161                completion(
162                    "First",
163                    "First-1",
164                    value::Kind::StringValue("one".to_string()),
165                ),
166                completion("Second", "Second-2", value::Kind::BoolValue(true)),
167            ],
168        );
169
170        assert_eq!(result.completions()[0].step_type, "First");
171        assert_eq!(result.completions()[1].step_execution_id, "Second-2");
172        assert_eq!(result.completions()[0].decode::<String>().unwrap(), "one");
173        assert!(result.completions()[1].decode::<bool>().unwrap());
174        assert!(matches!(
175            result.single_output::<String>(),
176            Err(SdkError::InvalidArgument { .. })
177        ));
178    }
179
180    #[test]
181    fn single_output_requires_exactly_one_completion() {
182        let single = FlowResult::new(
183            crate::FlowStatus::Completed,
184            None,
185            None,
186            vec![completion("Only", "Only-1", value::Kind::IntValue(7))],
187        );
188        assert_eq!(single.single_output::<i64>().unwrap(), 7);
189
190        let empty = FlowResult::new(crate::FlowStatus::Completed, None, None, Vec::new());
191        assert!(matches!(
192            empty.single_output::<String>(),
193            Err(SdkError::InvalidArgument { .. })
194        ));
195    }
196
197    fn completion(step_type: &str, step_execution_id: &str, kind: value::Kind) -> StepCompletion {
198        StepCompletion::new(
199            step_type.to_string(),
200            step_execution_id.to_string(),
201            ProtoValue { kind: Some(kind) },
202        )
203    }
204}
dex_sdk/flow_result.rs

dex_sdk/
flow_result.rs
