a3s_flow/model/
child_workflow.rs

1use chrono::{DateTime, Utc};
2use serde::de::DeserializeOwned;
3use serde::{Deserialize, Serialize};
4
5use crate::error::{FlowError, Result};
6
7use super::{validate_run_id, JsonValue, WorkflowSpec, WorkflowTerminalOutcome};
8
9/// Action applied to an open child when its parent enters cancellation.
10#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
11#[non_exhaustive]
12#[serde(rename_all = "snake_case")]
13pub enum ChildWorkflowCancellationPolicy {
14    /// Request cleanup-aware cancellation and keep the parent cancelling until
15    /// the child reaches a durable terminal outcome.
16    #[default]
17    RequestCancellation,
18    /// Leave the child independent and allow the parent to finish cancelling.
19    Abandon,
20}
21
22/// Parent-owned projection of one first-class child workflow execution.
23#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
24#[non_exhaustive]
25pub struct ChildWorkflowSnapshot {
26    /// Replay-stable parent-local identity of the child.
27    pub child_id: String,
28    /// Globally addressable Flow run identifier assigned to the child.
29    pub run_id: String,
30    /// Immutable workflow definition used to create the child.
31    pub spec: WorkflowSpec,
32    /// Initial JSON input supplied to the child.
33    pub input: JsonValue,
34    /// Policy applied when the parent is cancelled or terminated.
35    pub cancellation_policy: ChildWorkflowCancellationPolicy,
36    /// UTC time at which the request was persisted.
37    pub requested_at: DateTime<Utc>,
38    /// Event sequence that recorded the request.
39    pub requested_sequence: u64,
40    /// Terminal child outcome observed by the parent.
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub outcome: Option<WorkflowTerminalOutcome>,
43    /// UTC time at which the terminal outcome was recorded.
44    #[serde(default, skip_serializing_if = "Option::is_none")]
45    pub resolved_at: Option<DateTime<Utc>>,
46    /// Event sequence that recorded the terminal outcome.
47    #[serde(default, skip_serializing_if = "Option::is_none")]
48    pub resolved_sequence: Option<u64>,
49}
50
51impl ChildWorkflowSnapshot {
52    /// Returns whether the child has no durable terminal outcome yet.
53    pub fn is_open(&self) -> bool {
54        self.outcome.is_none()
55    }
56
57    /// Decode a completed child output into a host-defined serde type.
58    pub fn output_as<T>(&self) -> Result<Option<T>>
59    where
60        T: DeserializeOwned,
61    {
62        match &self.outcome {
63            Some(WorkflowTerminalOutcome::Completed { output }) => {
64                serde_json::from_value(output.clone())
65                    .map(Some)
66                    .map_err(FlowError::from)
67            }
68            _ => Ok(None),
69        }
70    }
71
72    pub(crate) fn validate_request(&self) -> Result<()> {
73        validate_child_workflow_request(
74            &self.child_id,
75            &self.run_id,
76            &self.spec,
77            self.requested_sequence,
78        )
79    }
80}
81
82pub(crate) fn validate_child_workflow_command(child_id: &str, spec: &WorkflowSpec) -> Result<()> {
83    if child_id.trim().is_empty() {
84        return Err(FlowError::InvalidTransition(
85            "child workflow id must not be empty".to_string(),
86        ));
87    }
88    spec.validate()
89}
90
91fn validate_child_workflow_request(
92    child_id: &str,
93    run_id: &str,
94    spec: &WorkflowSpec,
95    requested_sequence: u64,
96) -> Result<()> {
97    validate_child_workflow_command(child_id, spec)?;
98    validate_run_id(run_id)?;
99    if requested_sequence == 0 {
100        return Err(FlowError::InvalidTransition(format!(
101            "child workflow {child_id} requested sequence must be positive"
102        )));
103    }
104    Ok(())
105}
a3s_flow/model/child_workflow.rs

a3s_flow/model/
child_workflow.rs
