ledgence_orchestration_api/
workflow_children.rs

1//! Owned workflow composition, distinct from an individual controller attempt.
2use crate::*;
3use ledgence_worker_api::validate_wire_value;
4
5/// Root depth is zero. This bounds cancellation paths without a tree-wide lock.
6pub const WORKFLOW_MAX_DEPTH: u32 = 16;
7/// Limits simultaneous owned subworkflows, not retained historical child keys.
8pub const WORKFLOW_MAX_LIVE_SUBWORKFLOWS: u32 = 64;
9
10#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
11#[serde(rename_all = "snake_case")]
12pub enum WorkflowChildKind {
13    #[default]
14    Task,
15    Workflow,
16}
17impl WorkflowChildKind {
18    pub fn is_task(&self) -> bool {
19        *self == Self::Task
20    }
21}
22
23/// Legacy task inputs retain their wire shape. Workflow inputs have an explicit
24/// kind and workflow outcome. Both variants reject mixed or unknown fields.
25#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
26#[serde(untagged)]
27pub enum WorkflowChildResult {
28    Task(WorkflowTaskResult),
29    Workflow(WorkflowSubworkflowResult),
30}
31#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
32#[serde(deny_unknown_fields)]
33pub struct WorkflowTaskResult {
34    pub task_id: String,
35    pub state: TaskState,
36    pub outcome: TaskOutcome,
37}
38#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
39#[serde(deny_unknown_fields)]
40pub struct WorkflowSubworkflowResult {
41    pub kind: WorkflowChildKind,
42    pub workflow_id: String,
43    pub state: WorkflowState,
44    pub outcome: WorkflowOutcome,
45}
46impl WorkflowChildResult {
47    pub fn validate(&self) -> Result<()> {
48        match self {
49            Self::Task(child) => {
50                validate_text(&child.task_id, 128)?;
51                match (&child.state, &child.outcome) {
52                    (
53                        TaskState::Succeeded,
54                        TaskOutcome::Succeeded {
55                            output,
56                            attempt_id,
57                            execution_may_have_started: true,
58                            ..
59                        },
60                    ) => {
61                        validate_text(attempt_id, 128)?;
62                        validate_wire_value(output)?;
63                    }
64                    (
65                        TaskState::Failed,
66                        TaskOutcome::Failed {
67                            attempt_id,
68                            quiescence,
69                            execution_may_have_started,
70                            failure,
71                        },
72                    ) => {
73                        validate_text(attempt_id, 128)?;
74                        if (matches!(failure, TaskFailure::AttemptLost {})
75                            && *quiescence != Quiescence::Unconfirmed)
76                            || (matches!(failure, TaskFailure::Application { .. })
77                                && !execution_may_have_started)
78                        {
79                            return Err(invalid("inconsistent failed task input"));
80                        }
81                    }
82                    (TaskState::Cancelled, TaskOutcome::Cancelled {}) => {}
83                    _ => return Err(invalid("inconsistent terminal task input")),
84                }
85            }
86            Self::Workflow(child) => {
87                if child.kind != WorkflowChildKind::Workflow {
88                    return Err(invalid("workflow input requires workflow kind"));
89                }
90                validate_text(&child.workflow_id, 128)?;
91                match (&child.state, &child.outcome) {
92                    (WorkflowState::Succeeded, WorkflowOutcome::Succeeded { output }) => {
93                        validate_wire_value(output)?
94                    }
95                    (WorkflowState::Failed, WorkflowOutcome::Failed { error }) => {
96                        validate_workflow_error(error)?
97                    }
98                    (WorkflowState::Cancelled, WorkflowOutcome::Cancelled {}) => {}
99                    _ => return Err(invalid("inconsistent terminal workflow input")),
100                }
101            }
102        }
103        Ok(())
104    }
105}
106
107/// A controller task finishing is not its workflow finishing. Non-completion
108/// work carries no fabricated public task identity.
109#[derive(Debug, Clone, PartialEq, Eq)]
110pub enum WorkflowWorkSource {
111    TaskTerminal { task_id: String, activation: bool },
112    WorkflowTerminal { workflow_id: String },
113    Drain,
114    CancelOwned,
115    Wait,
116}
117
118/// Nested lineage is paired and immutable. Roots/legacy runs omit both fields.
119pub fn validate_workflow_lineage(
120    workflow_id: Option<&str>,
121    parent: Option<&str>,
122    root: Option<&str>,
123) -> Result<()> {
124    if let Some(id) = workflow_id {
125        validate_text(id, 128)?;
126    }
127    match (workflow_id, parent, root) {
128        (_, None, None) => Ok(()),
129        (Some(id), Some(parent), Some(root)) => {
130            validate_text(parent, 128)?;
131            validate_text(root, 128)?;
132            if parent == id || root == id {
133                return Err(invalid("workflow cannot be its own ancestor"));
134            }
135            Ok(())
136        }
137        _ => Err(invalid(
138            "workflow parent and root identities must be paired",
139        )),
140    }
141}
142fn invalid(message: &str) -> ContractError {
143    ContractError::InvalidInput(message.into())
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149    use serde_json::{Value, json};
150
151    fn child() -> Value {
152        json!({"key":"child","program":{"id":"program","version":"1.0.0"},"queue":"queue","data":null,"retry_policy":RetryPolicy::default(),"attempt_timeout_ms":300000})
153    }
154    fn decision(commands: Value) -> Value {
155        json!({"v":1,"activation_id":"activation","revision":0,"kind":"suspend","state":null,"continuation":"after","commands":commands,"until":["child"]})
156    }
157    fn context(input: Value) -> Value {
158        json!({"v":1,"workflow_id":"parent","activation_id":"activation","revision":1,"continuation":"after","state":null,"inputs":{"child":input},"local_steps":[]})
159    }
160    fn workflow_result() -> Value {
161        json!({"kind":"workflow","workflow_id":"nested","state":"succeeded","outcome":{"kind":"succeeded","output":null}})
162    }
163    #[test]
164    fn command_kind_preserves_legacy_shape_and_shared_key_namespace() {
165        let value = decision(json!([child()]));
166        let decoded = WorkflowDecision::decode(&value).unwrap();
167        assert_eq!(decoded.commands()[0].kind, WorkflowChildKind::Task);
168        assert_eq!(serde_json::to_value(decoded).unwrap(), value);
169        let mut nested = child();
170        nested["kind"] = json!("workflow");
171        let decoded = WorkflowDecision::decode(&decision(json!([nested]))).unwrap();
172        assert_eq!(decoded.commands()[0].kind, WorkflowChildKind::Workflow);
173        assert!(WorkflowDecision::decode(&decision(json!([child(), nested]))).is_err());
174        for kind in [json!("unknown"), Value::Null, json!(1)] {
175            let mut wrong = child();
176            wrong["kind"] = kind;
177            assert!(WorkflowDecision::decode(&decision(json!([wrong]))).is_err());
178        }
179    }
180    #[test]
181    fn workflow_inputs_have_their_own_terminal_identity_and_outcome() {
182        let valid = context(workflow_result());
183        let decoded: WorkflowActivationContext = serde_json::from_value(valid.clone()).unwrap();
184        decoded.validate().unwrap();
185        assert_eq!(serde_json::to_value(decoded).unwrap(), valid);
186        for change in 0..7 {
187            let mut input = workflow_result();
188            match change {
189                0 => input["task_id"] = json!("controller"),
190                1 => {
191                    input.as_object_mut().unwrap().remove("kind");
192                }
193                2 => input["kind"] = json!("task"),
194                3 => input["state"] = json!("waiting"),
195                4 => {
196                    input["outcome"] =
197                        json!({"kind":"failed","error":{"kind":"business","message":"failed"}})
198                }
199                5 => {
200                    input["outcome"].as_object_mut().unwrap().remove("output");
201                }
202                _ => input["workflow_id"] = json!(""),
203            }
204            assert!(
205                serde_json::from_value::<WorkflowActivationContext>(context(input))
206                    .map_or(true, |v| v.validate().is_err()),
207                "case {change}"
208            );
209        }
210        for (state, outcome) in [
211            (
212                "failed",
213                json!({"kind":"failed","error":{"kind":"business","message":"failed"}}),
214            ),
215            ("cancelled", json!({"kind":"cancelled"})),
216        ] {
217            let mut value = workflow_result();
218            value["state"] = json!(state);
219            value["outcome"] = outcome;
220            serde_json::from_value::<WorkflowActivationContext>(context(value))
221                .unwrap()
222                .validate()
223                .unwrap();
224        }
225    }
226    #[test]
227    fn legacy_and_workflow_inputs_share_the_existing_byte_and_member_budget() {
228        let mut value = context(workflow_result());
229        let legacy = json!({"task_id":"task","state":"succeeded","outcome":{"kind":"succeeded","attempt_id":"attempt","execution_may_have_started":true,"quiescence":"confirmed","output":null}});
230        value["inputs"]["task"] = legacy.clone();
231        let decoded: WorkflowActivationContext = serde_json::from_value(value.clone()).unwrap();
232        decoded.validate().unwrap();
233        assert_eq!(serde_json::to_value(decoded).unwrap(), value);
234        for index in 0..WORKFLOW_MAX_COMMANDS {
235            value["inputs"][format!("extra_{index}")] = legacy.clone();
236        }
237        assert!(
238            serde_json::from_value::<WorkflowActivationContext>(value)
239                .unwrap()
240                .validate()
241                .is_err()
242        );
243        let mut large = workflow_result();
244        large["outcome"]["output"] = json!("x".repeat(WORKFLOW_INPUTS_MAX_BYTES));
245        assert!(
246            serde_json::from_value::<WorkflowActivationContext>(context(large))
247                .unwrap()
248                .validate()
249                .is_err()
250        );
251    }
252    #[test]
253    fn failed_task_inputs_preserve_execution_and_cleanup_evidence() {
254        let mut input = json!({"task_id":"task","state":"failed","outcome":{
255            "kind":"failed","attempt_id":"attempt","quiescence":"unconfirmed",
256            "execution_may_have_started":true,"failure":{"kind":"attempt_lost"}}});
257        let validate = |input| {
258            serde_json::from_value::<WorkflowActivationContext>(context(input))
259                .unwrap()
260                .validate()
261        };
262        validate(input.clone()).unwrap();
263        input["outcome"]["quiescence"] = json!("confirmed");
264        assert!(validate(input.clone()).is_err());
265        input["outcome"]["failure"] =
266            json!({"kind":"application","error":{"kind":"business","message":"failed"}});
267        validate(input.clone()).unwrap();
268        input["outcome"]["execution_may_have_started"] = json!(false);
269        assert!(validate(input.clone()).is_err());
270        input["outcome"]["attempt_id"] = json!("");
271        assert!(validate(input).is_err());
272    }
273    #[test]
274    fn lineage_is_paired_and_cannot_claim_self_ancestry() {
275        assert!(validate_workflow_lineage(None, None, None).is_ok());
276        assert!(validate_workflow_lineage(Some("root"), None, None).is_ok());
277        assert!(validate_workflow_lineage(Some("child"), Some("root"), Some("root")).is_ok());
278        assert!(validate_workflow_lineage(Some("grandchild"), Some("child"), Some("root")).is_ok());
279        for (id, parent, root) in [
280            (None, Some("parent"), Some("root")),
281            (Some("child"), None, Some("root")),
282            (Some("child"), Some("parent"), None),
283            (Some("child"), Some("child"), Some("root")),
284            (Some("child"), Some("parent"), Some("child")),
285            (Some("child"), Some(""), Some("root")),
286        ] {
287            assert!(validate_workflow_lineage(id, parent, root).is_err());
288        }
289    }
290}
ledgence_orchestration_api/workflow_children.rs

ledgence_orchestration_api/
workflow_children.rs
