dex_sdk/
flow_result.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Sustainable Use 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-Sustainable-Use-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!(self.status, FlowStatus::Running)
112    }
113
114    /// Returns output-bearing completions in server collection order.
115    ///
116    /// Parallel Step order is not deterministic. Select by Step type or Step execution ID.
117    pub fn completions(&self) -> &[StepCompletion] {
118        &self.completions
119    }
120
121    /// Decodes the output when exactly one completion exists.
122    ///
123    /// # Errors
124    ///
125    /// Returns [`SdkError::InvalidArgument`] for a nonterminal result or zero or multiple outputs and
126    /// [`SdkError::ValueMapping`] when the output is incompatible with `Output`.
127    pub fn single_output<Output: Value>(&self) -> SdkResult<Output> {
128        if !self.is_terminal() {
129            return Err(SdkError::InvalidArgument {
130                message: "Flow result is not terminal".to_string(),
131            });
132        }
133        if self.completions().len() != 1 {
134            return Err(SdkError::InvalidArgument {
135                message: format!(
136                    "expected exactly one Step output, found {}",
137                    self.completions().len()
138                ),
139            });
140        }
141        self.completions[0].decode()
142    }
143}
144
145#[cfg(test)]
146mod tests {
147    use dex_protocol::dex::value;
148
149    use super::*;
150
151    #[test]
152    fn preserves_heterogeneous_completions_and_rejects_ambiguous_single_output() {
153        let result = FlowResult::new(
154            crate::FlowStatus::Completed,
155            None,
156            None,
157            vec![
158                completion(
159                    "First",
160                    "First-1",
161                    value::Kind::StringValue("one".to_string()),
162                ),
163                completion("Second", "Second-2", value::Kind::BoolValue(true)),
164            ],
165        );
166
167        assert_eq!(result.completions()[0].step_type, "First");
168        assert_eq!(result.completions()[1].step_execution_id, "Second-2");
169        assert_eq!(result.completions()[0].decode::<String>().unwrap(), "one");
170        assert!(result.completions()[1].decode::<bool>().unwrap());
171        assert!(matches!(
172            result.single_output::<String>(),
173            Err(SdkError::InvalidArgument { .. })
174        ));
175    }
176
177    #[test]
178    fn single_output_requires_exactly_one_completion() {
179        let single = FlowResult::new(
180            crate::FlowStatus::Completed,
181            None,
182            None,
183            vec![completion("Only", "Only-1", value::Kind::IntValue(7))],
184        );
185        assert_eq!(single.single_output::<i64>().unwrap(), 7);
186
187        let empty = FlowResult::new(crate::FlowStatus::Completed, None, None, Vec::new());
188        assert!(matches!(
189            empty.single_output::<String>(),
190            Err(SdkError::InvalidArgument { .. })
191        ));
192    }
193
194    fn completion(step_type: &str, step_execution_id: &str, kind: value::Kind) -> StepCompletion {
195        StepCompletion::new(
196            step_type.to_string(),
197            step_execution_id.to_string(),
198            ProtoValue { kind: Some(kind) },
199        )
200    }
201}
dex_sdk/flow_result.rs

dex_sdk/
flow_result.rs
