af_workflow/
host.rs

1//! Validated multi-branch workflow host.
2
3use std::sync::Arc;
4
5use crate::executor::CompiledBranch;
6use crate::{
7    validate, CompileError, Event, NodeRegistry, RunOutcome, Spec, SpecError, State, Violation,
8    WorkflowContext,
9};
10
11/// Conventional id of the root branch.
12pub const ROOT_BRANCH: &str = "__root__";
13
14/// Why a spec could not be hosted.
15#[derive(Debug, thiserror::Error)]
16pub enum HostError {
17    /// Workflow spec failed to parse.
18    #[error("workflow spec failed to parse: {0}")]
19    Parse(#[from] SpecError),
20    /// Workflow spec '`spec_id`' has no branches.
21    #[error("workflow spec '{spec_id}' has no branches")]
22    NoBranch {
23        /// Spec id.
24        spec_id: String,
25    },
26    /// Workflow spec '`spec_id`' violates safety rules: `violations`.
27    #[error("workflow spec '{spec_id}' violates safety rules: {violations:?}")]
28    Safety {
29        /// Spec that failed.
30        spec_id: String,
31        /// Safety rules violated.
32        violations: Vec<Violation>,
33    },
34    /// Workflow spec '`spec_id`' failed to compile: `source`.
35    #[error("workflow spec '{spec_id}' failed to compile: {source}")]
36    Compile {
37        /// Spec that failed.
38        spec_id: String,
39        /// Compile error.
40        #[source]
41        source: CompileError,
42    },
43    /// A step performs an external effect; durable spec execution runs pure/read steps only.
44    #[error("workflow spec '{spec_id}' step '{node_id}' ({node_type}) performs an external effect; durable spec execution runs pure/read steps only, register a WorkflowActionProvider for it")]
45    UnsupportedDurableStep {
46        /// Spec that failed.
47        spec_id: String,
48        /// Node that cannot run durably.
49        node_id: String,
50        /// Its node type.
51        node_type: String,
52    },
53    /// Workflow spec '`spec_id`' has an invalid schedule: `reason`.
54    #[error("workflow spec '{spec_id}' has an invalid schedule: {reason}")]
55    Schedule {
56        /// Spec id.
57        spec_id: String,
58        /// Why the schedule is invalid.
59        reason: String,
60    },
61}
62
63struct BranchEntry {
64    id: String,
65    branch: CompiledBranch,
66}
67
68/// A parsed, validated and compiled workflow. Assembly is the only constructor,
69/// so a long-running host cannot accidentally skip static validation.
70pub struct WorkflowHost {
71    spec_id: String,
72    branches: Vec<BranchEntry>,
73}
74
75impl WorkflowHost {
76    /// Parse, validate and compile a spec from JSON.
77    pub fn assemble(spec_json: &str, registry: &NodeRegistry) -> Result<Self, HostError> {
78        Self::from_spec(&Spec::from_json(spec_json)?, registry)
79    }
80
81    /// Validate and compile every branch of `spec`.
82    pub fn from_spec(spec: &Spec, registry: &NodeRegistry) -> Result<Self, HostError> {
83        if spec.branches.is_empty() {
84            return Err(HostError::NoBranch {
85                spec_id: spec.spec_id.clone(),
86            });
87        }
88        if let Err(violations) = validate(spec, registry) {
89            return Err(HostError::Safety {
90                spec_id: spec.spec_id.clone(),
91                violations,
92            });
93        }
94
95        let branches = spec
96            .branches
97            .iter()
98            .map(|branch| {
99                Ok(BranchEntry {
100                    id: branch.branch_id.clone(),
101                    branch: CompiledBranch::compile(branch, registry).map_err(|source| {
102                        HostError::Compile {
103                            spec_id: spec.spec_id.clone(),
104                            source,
105                        }
106                    })?,
107                })
108            })
109            .collect::<Result<Vec<_>, HostError>>()?;
110        Ok(Self {
111            spec_id: spec.spec_id.clone(),
112            branches,
113        })
114    }
115
116    /// Spec id.
117    pub fn spec_id(&self) -> &str {
118        &self.spec_id
119    }
120
121    /// Id of the first (root) branch.
122    pub fn branch_id(&self) -> &str {
123        &self.branches[0].id
124    }
125
126    /// Every branch id.
127    pub fn branch_ids(&self) -> impl Iterator<Item = &str> {
128        self.branches.iter().map(|branch| branch.id.as_str())
129    }
130
131    /// Number of branches.
132    pub fn branch_count(&self) -> usize {
133        self.branches.len()
134    }
135
136    /// Context for the root branch over `state`.
137    pub fn context(&self, state: Arc<dyn State>) -> WorkflowContext {
138        self.context_for(self.branch_id(), state)
139    }
140
141    /// Context for `branch_id` over `state`.
142    pub fn context_for(&self, branch_id: &str, state: Arc<dyn State>) -> WorkflowContext {
143        WorkflowContext::new(branch_id, state)
144    }
145
146    /// Drive one event through the context's branch.
147    pub async fn run_event(&self, ctx: &WorkflowContext, event: Event) -> Option<RunOutcome> {
148        self.run_event_on(&ctx.branch_id, ctx, event).await
149    }
150
151    /// Drive one event through `branch_id`; `None` when the branch does not exist.
152    pub async fn run_event_on(
153        &self,
154        branch_id: &str,
155        ctx: &WorkflowContext,
156        event: Event,
157    ) -> Option<RunOutcome> {
158        let branch = self.branches.iter().find(|branch| branch.id == branch_id)?;
159        Some(branch.branch.run_event(ctx, event).await)
160    }
161}
af_workflow/host.rs

af_workflow/
host.rs
