use indexmap::IndexMap;
use crate::{
EffectDisposition, EffectDispositionRule, EffectEmit, EnumSchema, Expr, FieldInit, FieldSchema,
Guard, HelperSchema, InitSchema, InputMatch, InvariantSchema, MachineSchema, Quantifier,
RustBinding, StateSchema, TransitionSchema, TypeRef, Update, VariantSchema,
};
pub fn flow_run_machine() -> MachineSchema {
MachineSchema {
machine: "FlowRunMachine".into(),
version: 1,
rust: RustBinding {
crate_name: "meerkat-mob".into(),
module: "generated::flow_run".into(),
},
state: StateSchema {
phase: EnumSchema {
name: "FlowRunStatus".into(),
variants: vec![
variant("Absent"),
variant("Pending"),
variant("Running"),
variant("Completed"),
variant("Failed"),
variant("Canceled"),
],
},
fields: vec![
field(
"tracked_steps",
TypeRef::Set(Box::new(TypeRef::Named("StepId".into()))),
),
field(
"ordered_steps",
TypeRef::Seq(Box::new(TypeRef::Named("StepId".into()))),
),
field(
"step_status",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Option(Box::new(TypeRef::Enum(
"StepRunStatus".into(),
)))),
),
),
field(
"output_recorded",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Bool),
),
),
field(
"step_condition_results",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Option(Box::new(TypeRef::Bool))),
),
),
field(
"step_has_conditions",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Bool),
),
),
field(
"step_dependencies",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Seq(Box::new(TypeRef::Named("StepId".into())))),
),
),
field(
"step_dependency_modes",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Enum("DependencyMode".into())),
),
),
field(
"step_branches",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Option(Box::new(TypeRef::Named("BranchId".into())))),
),
),
field(
"step_collection_policies",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Enum("CollectionPolicyKind".into())),
),
),
field(
"step_quorum_thresholds",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::U32),
),
),
field(
"step_target_counts",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::U32),
),
),
field(
"step_target_success_counts",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::U32),
),
),
field(
"step_target_terminal_failure_counts",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::U32),
),
),
field(
"target_retry_counts",
TypeRef::Map(Box::new(TypeRef::String), Box::new(TypeRef::U32)),
),
field("failure_count", TypeRef::U32),
field("consecutive_failure_count", TypeRef::U32),
field("escalation_threshold", TypeRef::U32),
field("max_step_retries", TypeRef::U32),
],
init: InitSchema {
phase: "Absent".into(),
fields: vec![
init("tracked_steps", Expr::EmptySet),
init("ordered_steps", Expr::SeqLiteral(vec![])),
init("step_status", Expr::EmptyMap),
init("output_recorded", Expr::EmptyMap),
init("step_condition_results", Expr::EmptyMap),
init("step_has_conditions", Expr::EmptyMap),
init("step_dependencies", Expr::EmptyMap),
init("step_dependency_modes", Expr::EmptyMap),
init("step_branches", Expr::EmptyMap),
init("step_collection_policies", Expr::EmptyMap),
init("step_quorum_thresholds", Expr::EmptyMap),
init("step_target_counts", Expr::EmptyMap),
init("step_target_success_counts", Expr::EmptyMap),
init("step_target_terminal_failure_counts", Expr::EmptyMap),
init("target_retry_counts", Expr::EmptyMap),
init("failure_count", Expr::U64(0)),
init("consecutive_failure_count", Expr::U64(0)),
init("escalation_threshold", Expr::U64(0)),
init("max_step_retries", Expr::U64(0)),
],
},
terminal_phases: vec!["Completed".into(), "Failed".into(), "Canceled".into()],
},
inputs: EnumSchema {
name: "FlowRunInput".into(),
variants: vec![
VariantSchema {
name: "CreateRun".into(),
fields: vec![
field(
"step_ids",
TypeRef::Seq(Box::new(TypeRef::Named("StepId".into()))),
),
field(
"ordered_steps",
TypeRef::Seq(Box::new(TypeRef::Named("StepId".into()))),
),
field(
"step_has_conditions",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Bool),
),
),
field(
"step_dependencies",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Seq(Box::new(TypeRef::Named("StepId".into())))),
),
),
field(
"step_dependency_modes",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Enum("DependencyMode".into())),
),
),
field(
"step_branches",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Option(Box::new(TypeRef::Named(
"BranchId".into(),
)))),
),
),
field(
"step_collection_policies",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::Enum("CollectionPolicyKind".into())),
),
),
field(
"step_quorum_thresholds",
TypeRef::Map(
Box::new(TypeRef::Named("StepId".into())),
Box::new(TypeRef::U32),
),
),
field("escalation_threshold", TypeRef::U32),
field("max_step_retries", TypeRef::U32),
],
},
variant("StartRun"),
VariantSchema {
name: "DispatchStep".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "CompleteStep".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "RecordStepOutput".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "ConditionPassed".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "ConditionRejected".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "FailStep".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "SkipStep".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "CancelStep".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "RegisterTargets".into(),
fields: vec![
field("step_id", TypeRef::Named("StepId".into())),
field("target_count", TypeRef::U32),
],
},
VariantSchema {
name: "RecordTargetSuccess".into(),
fields: vec![
field("step_id", TypeRef::Named("StepId".into())),
field("target_id", TypeRef::Named("MeerkatId".into())),
],
},
VariantSchema {
name: "RecordTargetTerminalFailure".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "RecordTargetCanceled".into(),
fields: vec![
field("step_id", TypeRef::Named("StepId".into())),
field("target_id", TypeRef::Named("MeerkatId".into())),
],
},
VariantSchema {
name: "RecordTargetFailure".into(),
fields: vec![
field("step_id", TypeRef::Named("StepId".into())),
field("target_id", TypeRef::Named("MeerkatId".into())),
field("retry_key", TypeRef::String),
],
},
variant("TerminalizeCompleted"),
variant("TerminalizeFailed"),
variant("TerminalizeCanceled"),
],
},
effects: EnumSchema {
name: "FlowRunEffect".into(),
variants: vec![
VariantSchema {
name: "EmitFlowRunNotice".into(),
fields: vec![field("run_status", TypeRef::Enum("FlowRunStatus".into()))],
},
VariantSchema {
name: "EmitStepNotice".into(),
fields: vec![
field("step_id", TypeRef::Named("StepId".into())),
field("step_status", TypeRef::Enum("StepRunStatus".into())),
],
},
VariantSchema {
name: "AppendFailureLedger".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "PersistStepOutput".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "AdmitStepWork".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "FlowTerminalized".into(),
fields: vec![field("run_status", TypeRef::Enum("FlowRunStatus".into()))],
},
VariantSchema {
name: "EscalateSupervisor".into(),
fields: vec![field("step_id", TypeRef::Named("StepId".into()))],
},
VariantSchema {
name: "ProjectTargetSuccess".into(),
fields: vec![
field("step_id", TypeRef::Named("StepId".into())),
field("target_id", TypeRef::Named("MeerkatId".into())),
],
},
VariantSchema {
name: "ProjectTargetFailure".into(),
fields: vec![
field("step_id", TypeRef::Named("StepId".into())),
field("target_id", TypeRef::Named("MeerkatId".into())),
],
},
VariantSchema {
name: "ProjectTargetCanceled".into(),
fields: vec![
field("step_id", TypeRef::Named("StepId".into())),
field("target_id", TypeRef::Named("MeerkatId".into())),
],
},
],
},
helpers: vec![
helper(
"RunIsTerminal",
vec![],
TypeRef::Bool,
Expr::Or(vec![
Expr::Eq(
Box::new(Expr::CurrentPhase),
Box::new(Expr::Phase("Completed".into())),
),
Expr::Eq(
Box::new(Expr::CurrentPhase),
Box::new(Expr::Phase("Failed".into())),
),
Expr::Eq(
Box::new(Expr::CurrentPhase),
Box::new(Expr::Phase("Canceled".into())),
),
]),
),
helper(
"StepIsTracked",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::Contains {
collection: Box::new(Expr::Field("tracked_steps".into())),
value: Box::new(Expr::Binding("step_id".into())),
},
),
helper(
"StepStatusIs",
vec![
field("step_id", TypeRef::Named("StepId".into())),
field("expected_status", TypeRef::Enum("StepRunStatus".into())),
],
TypeRef::Bool,
Expr::Eq(
Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_status".into())),
key: Box::new(Expr::Binding("step_id".into())),
}),
Box::new(Expr::Some(Box::new(Expr::Binding(
"expected_status".into(),
)))),
),
),
helper(
"StepOutputRecordedIs",
vec![
field("step_id", TypeRef::Named("StepId".into())),
field("expected", TypeRef::Bool),
],
TypeRef::Bool,
Expr::Eq(
Box::new(Expr::MapGet {
map: Box::new(Expr::Field("output_recorded".into())),
key: Box::new(Expr::Binding("step_id".into())),
}),
Box::new(Expr::Binding("expected".into())),
),
),
helper(
"StepConditionRecordedIs",
vec![
field("step_id", TypeRef::Named("StepId".into())),
field("expected", TypeRef::Option(Box::new(TypeRef::Bool))),
],
TypeRef::Bool,
Expr::Eq(
Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_condition_results".into())),
key: Box::new(Expr::Binding("step_id".into())),
}),
Box::new(Expr::Binding("expected".into())),
),
),
helper(
"StepConditionAllowsDispatch",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::Or(vec![
Expr::Eq(
Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_has_conditions".into())),
key: Box::new(Expr::Binding("step_id".into())),
}),
Box::new(Expr::Bool(false)),
),
Expr::Call {
helper: "StepConditionRecordedIs".into(),
args: vec![
Expr::Binding("step_id".into()),
Expr::Some(Box::new(Expr::Bool(true))),
],
},
]),
),
helper(
"AllTrackedStepsInAllowedStatuses",
vec![field(
"allowed_statuses",
TypeRef::Seq(Box::new(TypeRef::Option(Box::new(TypeRef::Enum(
"StepRunStatus".into(),
))))),
)],
TypeRef::Bool,
Expr::Quantified {
quantifier: Quantifier::All,
binding: "step_id".into(),
over: Box::new(Expr::Field("tracked_steps".into())),
body: Box::new(Expr::Contains {
collection: Box::new(Expr::Binding("allowed_statuses".into())),
value: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_status".into())),
key: Box::new(Expr::Binding("step_id".into())),
}),
}),
},
),
helper(
"NoTrackedStepInStatus",
vec![field("status", TypeRef::Enum("StepRunStatus".into()))],
TypeRef::Bool,
Expr::Quantified {
quantifier: Quantifier::All,
binding: "step_id".into(),
over: Box::new(Expr::Field("tracked_steps".into())),
body: Box::new(Expr::Neq(
Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_status".into())),
key: Box::new(Expr::Binding("step_id".into())),
}),
Box::new(Expr::Some(Box::new(Expr::Binding("status".into())))),
)),
},
),
helper(
"AnyTrackedStepInStatus",
vec![field("status", TypeRef::Enum("StepRunStatus".into()))],
TypeRef::Bool,
Expr::Quantified {
quantifier: Quantifier::Any,
binding: "step_id".into(),
over: Box::new(Expr::Field("tracked_steps".into())),
body: Box::new(Expr::Eq(
Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_status".into())),
key: Box::new(Expr::Binding("step_id".into())),
}),
Box::new(Expr::Some(Box::new(Expr::Binding("status".into())))),
)),
},
),
helper(
"StepHasDependencies",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::Gt(
Box::new(Expr::Len(Box::new(step_dependencies_for("step_id")))),
Box::new(Expr::U64(0)),
),
),
helper(
"AllDependenciesCompleted",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::Quantified {
quantifier: Quantifier::All,
binding: "dependency".into(),
over: Box::new(Expr::SeqElements(Box::new(step_dependencies_for(
"step_id",
)))),
body: Box::new(Expr::Call {
helper: "StepStatusIs".into(),
args: vec![
Expr::Binding("dependency".into()),
step_status(StepStatusVariant::Completed),
],
}),
},
),
helper(
"AllDependenciesSkipped",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::Quantified {
quantifier: Quantifier::All,
binding: "dependency".into(),
over: Box::new(Expr::SeqElements(Box::new(step_dependencies_for(
"step_id",
)))),
body: Box::new(Expr::Call {
helper: "StepStatusIs".into(),
args: vec![
Expr::Binding("dependency".into()),
step_status(StepStatusVariant::Skipped),
],
}),
},
),
helper(
"AnyDependencyCompleted",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::Quantified {
quantifier: Quantifier::Any,
binding: "dependency".into(),
over: Box::new(Expr::SeqElements(Box::new(step_dependencies_for(
"step_id",
)))),
body: Box::new(Expr::Call {
helper: "StepStatusIs".into(),
args: vec![
Expr::Binding("dependency".into()),
step_status(StepStatusVariant::Completed),
],
}),
},
),
helper(
"StepDependencyReady",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::IfElse {
condition: Box::new(Expr::Eq(
Box::new(step_dependency_mode_for("step_id")),
Box::new(dependency_mode(DependencyModeVariant::Any)),
)),
then_expr: Box::new(Expr::Call {
helper: "AnyDependencyCompleted".into(),
args: vec![Expr::Binding("step_id".into())],
}),
else_expr: Box::new(Expr::IfElse {
condition: Box::new(Expr::Not(Box::new(Expr::Call {
helper: "StepHasDependencies".into(),
args: vec![Expr::Binding("step_id".into())],
}))),
then_expr: Box::new(Expr::Bool(true)),
else_expr: Box::new(Expr::Call {
helper: "AllDependenciesCompleted".into(),
args: vec![Expr::Binding("step_id".into())],
}),
}),
},
),
helper(
"StepDependencyShouldSkip",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::And(vec![
Expr::Eq(
Box::new(step_dependency_mode_for("step_id")),
Box::new(dependency_mode(DependencyModeVariant::Any)),
),
Expr::Call {
helper: "StepHasDependencies".into(),
args: vec![Expr::Binding("step_id".into())],
},
Expr::Call {
helper: "AllDependenciesSkipped".into(),
args: vec![Expr::Binding("step_id".into())],
},
]),
),
helper(
"StepBranchBlocked",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::IfElse {
condition: Box::new(Expr::Eq(
Box::new(step_branch_for("step_id")),
Box::new(Expr::None),
)),
then_expr: Box::new(Expr::Bool(false)),
else_expr: Box::new(Expr::Quantified {
quantifier: Quantifier::Any,
binding: "candidate".into(),
over: Box::new(Expr::Field("tracked_steps".into())),
body: Box::new(Expr::And(vec![
Expr::Neq(
Box::new(Expr::Binding("candidate".into())),
Box::new(Expr::Binding("step_id".into())),
),
Expr::Eq(
Box::new(step_branch_for("candidate")),
Box::new(step_branch_for("step_id")),
),
Expr::Call {
helper: "StepStatusIs".into(),
args: vec![
Expr::Binding("candidate".into()),
step_status(StepStatusVariant::Completed),
],
},
])),
}),
},
),
helper(
"EscalationWillTrigger",
vec![],
TypeRef::Bool,
Expr::And(vec![
Expr::Gt(
Box::new(Expr::Field("escalation_threshold".into())),
Box::new(Expr::U64(0)),
),
Expr::Gte(
Box::new(Expr::Add(
Box::new(Expr::Field("consecutive_failure_count".into())),
Box::new(Expr::U64(1)),
)),
Box::new(Expr::Field("escalation_threshold".into())),
),
]),
),
helper(
"TargetRetryCount",
vec![field("retry_key", TypeRef::String)],
TypeRef::U32,
target_retry_count_for("retry_key"),
),
helper(
"TargetRetryAllowed",
vec![field("retry_key", TypeRef::String)],
TypeRef::Bool,
Expr::Lte(
Box::new(target_retry_count_for("retry_key")),
Box::new(Expr::Field("max_step_retries".into())),
),
),
helper(
"CollectionSatisfied",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::IfElse {
condition: Box::new(Expr::Eq(
Box::new(step_collection_policy_for("step_id")),
Box::new(collection_policy_kind(CollectionPolicyVariant::All)),
)),
then_expr: Box::new(Expr::Eq(
Box::new(step_target_success_count_for("step_id")),
Box::new(step_target_count_for("step_id")),
)),
else_expr: Box::new(Expr::IfElse {
condition: Box::new(Expr::Eq(
Box::new(step_collection_policy_for("step_id")),
Box::new(collection_policy_kind(CollectionPolicyVariant::Any)),
)),
then_expr: Box::new(Expr::Gte(
Box::new(step_target_success_count_for("step_id")),
Box::new(Expr::U64(1)),
)),
else_expr: Box::new(Expr::IfElse {
condition: Box::new(Expr::Eq(
Box::new(step_collection_policy_for("step_id")),
Box::new(collection_policy_kind(CollectionPolicyVariant::Quorum)),
)),
then_expr: Box::new(Expr::Gte(
Box::new(step_target_success_count_for("step_id")),
Box::new(step_quorum_threshold_for("step_id")),
)),
else_expr: Box::new(Expr::Bool(false)),
}),
}),
},
),
helper(
"CollectionFeasible",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::Bool,
Expr::IfElse {
condition: Box::new(Expr::Eq(
Box::new(step_collection_policy_for("step_id")),
Box::new(collection_policy_kind(CollectionPolicyVariant::All)),
)),
then_expr: Box::new(Expr::Eq(
Box::new(step_target_terminal_failure_count_for("step_id")),
Box::new(Expr::U64(0)),
)),
else_expr: Box::new(Expr::IfElse {
condition: Box::new(Expr::Eq(
Box::new(step_collection_policy_for("step_id")),
Box::new(collection_policy_kind(CollectionPolicyVariant::Any)),
)),
then_expr: Box::new(Expr::Or(vec![
Expr::Gte(
Box::new(step_target_success_count_for("step_id")),
Box::new(Expr::U64(1)),
),
Expr::Gt(
Box::new(remaining_target_count_for("step_id")),
Box::new(Expr::U64(0)),
),
])),
else_expr: Box::new(Expr::IfElse {
condition: Box::new(Expr::Eq(
Box::new(step_collection_policy_for("step_id")),
Box::new(collection_policy_kind(CollectionPolicyVariant::Quorum)),
)),
then_expr: Box::new(Expr::Gte(
Box::new(Expr::Add(
Box::new(step_target_success_count_for("step_id")),
Box::new(remaining_target_count_for("step_id")),
)),
Box::new(step_quorum_threshold_for("step_id")),
)),
else_expr: Box::new(Expr::Bool(false)),
}),
}),
},
),
helper(
"StepTargetCount",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::U32,
step_target_count_for("step_id"),
),
helper(
"StepTargetSuccessCount",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::U32,
step_target_success_count_for("step_id"),
),
helper(
"StepTargetTerminalFailureCount",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::U32,
step_target_terminal_failure_count_for("step_id"),
),
helper(
"RemainingTargetCount",
vec![field("step_id", TypeRef::Named("StepId".into()))],
TypeRef::U32,
remaining_target_count_for("step_id"),
),
],
derived: vec![],
invariants: vec![
InvariantSchema {
name: "output_only_follows_completed_steps".into(),
expr: Expr::Quantified {
quantifier: Quantifier::All,
binding: "step_id".into(),
over: Box::new(Expr::Field("tracked_steps".into())),
body: Box::new(Expr::Or(vec![
Expr::Not(Box::new(Expr::Call {
helper: "StepOutputRecordedIs".into(),
args: vec![Expr::Binding("step_id".into()), Expr::Bool(true)],
})),
Expr::Call {
helper: "StepStatusIs".into(),
args: vec![
Expr::Binding("step_id".into()),
step_status(StepStatusVariant::Completed),
],
},
])),
},
},
InvariantSchema {
name: "terminal_runs_have_no_dispatched_steps".into(),
expr: Expr::Or(vec![
Expr::Not(Box::new(Expr::Call {
helper: "RunIsTerminal".into(),
args: vec![],
})),
Expr::Call {
helper: "NoTrackedStepInStatus".into(),
args: vec![step_status(StepStatusVariant::Dispatched)],
},
]),
},
InvariantSchema {
name: "completed_runs_contain_only_completed_or_skipped_steps".into(),
expr: Expr::Or(vec![
Expr::Neq(
Box::new(Expr::CurrentPhase),
Box::new(Expr::Phase("Completed".into())),
),
Expr::Call {
helper: "AllTrackedStepsInAllowedStatuses".into(),
args: vec![Expr::SeqLiteral(vec![
Expr::Some(Box::new(step_status(StepStatusVariant::Completed))),
Expr::Some(Box::new(step_status(StepStatusVariant::Skipped))),
])],
},
]),
},
InvariantSchema {
name: "failed_step_presence_requires_failure_count".into(),
expr: Expr::Or(vec![
Expr::Not(Box::new(Expr::Call {
helper: "AnyTrackedStepInStatus".into(),
args: vec![step_status(StepStatusVariant::Failed)],
})),
Expr::Gte(
Box::new(Expr::Field("failure_count".into())),
Box::new(Expr::U64(1)),
),
]),
},
InvariantSchema {
name: "failed_run_has_failed_step_or_recorded_failure".into(),
expr: Expr::Or(vec![
Expr::Neq(
Box::new(Expr::CurrentPhase),
Box::new(Expr::Phase("Failed".into())),
),
Expr::Bool(true),
]),
},
],
transitions: vec![
TransitionSchema {
name: "CreateRun".into(),
from: vec!["Absent".into()],
on: InputMatch {
variant: "CreateRun".into(),
bindings: vec![
"step_ids".into(),
"ordered_steps".into(),
"step_has_conditions".into(),
"step_dependencies".into(),
"step_dependency_modes".into(),
"step_branches".into(),
"step_collection_policies".into(),
"step_quorum_thresholds".into(),
"escalation_threshold".into(),
"max_step_retries".into(),
],
},
guards: vec![
Guard {
name: "step_ids_are_non_empty".into(),
expr: Expr::Gt(
Box::new(Expr::Len(Box::new(Expr::Binding("step_ids".into())))),
Box::new(Expr::U64(0)),
),
},
Guard {
name: "ordered_steps_only_reference_step_ids".into(),
expr: sequence_members_are_in_binding("ordered_steps", "step_ids"),
},
Guard {
name: "step_ids_appear_in_ordered_steps".into(),
expr: sequence_members_are_in_binding("step_ids", "ordered_steps"),
},
map_keys_match_step_ids_guard("step_has_conditions"),
map_keys_match_step_ids_guard("step_dependencies"),
map_keys_match_step_ids_guard("step_dependency_modes"),
map_keys_match_step_ids_guard("step_branches"),
map_keys_match_step_ids_guard("step_collection_policies"),
map_keys_match_step_ids_guard("step_quorum_thresholds"),
Guard {
name: "step_dependencies_reference_known_steps".into(),
expr: Expr::Quantified {
quantifier: Quantifier::All,
binding: "step_id".into(),
over: Box::new(Expr::MapKeys(Box::new(Expr::Binding(
"step_dependencies".into(),
)))),
body: Box::new(Expr::Quantified {
quantifier: Quantifier::All,
binding: "dependency".into(),
over: Box::new(Expr::SeqElements(Box::new(Expr::MapGet {
map: Box::new(Expr::Binding("step_dependencies".into())),
key: Box::new(Expr::Binding("step_id".into())),
}))),
body: Box::new(Expr::Contains {
collection: Box::new(Expr::SeqElements(Box::new(
Expr::Binding("step_ids".into()),
))),
value: Box::new(Expr::Binding("dependency".into())),
}),
}),
},
},
],
updates: vec![
Update::Assign {
field: "tracked_steps".into(),
expr: Expr::EmptySet,
},
Update::Assign {
field: "ordered_steps".into(),
expr: Expr::Binding("ordered_steps".into()),
},
Update::Assign {
field: "step_status".into(),
expr: Expr::EmptyMap,
},
Update::Assign {
field: "output_recorded".into(),
expr: Expr::EmptyMap,
},
Update::Assign {
field: "step_condition_results".into(),
expr: Expr::EmptyMap,
},
Update::Assign {
field: "step_has_conditions".into(),
expr: Expr::Binding("step_has_conditions".into()),
},
Update::Assign {
field: "failure_count".into(),
expr: Expr::U64(0),
},
Update::Assign {
field: "consecutive_failure_count".into(),
expr: Expr::U64(0),
},
Update::Assign {
field: "step_dependencies".into(),
expr: Expr::Binding("step_dependencies".into()),
},
Update::Assign {
field: "step_dependency_modes".into(),
expr: Expr::Binding("step_dependency_modes".into()),
},
Update::Assign {
field: "step_branches".into(),
expr: Expr::Binding("step_branches".into()),
},
Update::Assign {
field: "step_collection_policies".into(),
expr: Expr::Binding("step_collection_policies".into()),
},
Update::Assign {
field: "step_quorum_thresholds".into(),
expr: Expr::Binding("step_quorum_thresholds".into()),
},
Update::Assign {
field: "step_target_counts".into(),
expr: Expr::EmptyMap,
},
Update::Assign {
field: "step_target_success_counts".into(),
expr: Expr::EmptyMap,
},
Update::Assign {
field: "step_target_terminal_failure_counts".into(),
expr: Expr::EmptyMap,
},
Update::Assign {
field: "target_retry_counts".into(),
expr: Expr::EmptyMap,
},
Update::Assign {
field: "escalation_threshold".into(),
expr: Expr::Binding("escalation_threshold".into()),
},
Update::Assign {
field: "max_step_retries".into(),
expr: Expr::Binding("max_step_retries".into()),
},
Update::ForEach {
binding: "step_id".into(),
over: Expr::Binding("step_ids".into()),
updates: vec![
Update::SetInsert {
field: "tracked_steps".into(),
value: Expr::Binding("step_id".into()),
},
Update::MapInsert {
field: "step_status".into(),
key: Expr::Binding("step_id".into()),
value: Expr::None,
},
Update::MapInsert {
field: "output_recorded".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Bool(false),
},
Update::MapInsert {
field: "step_condition_results".into(),
key: Expr::Binding("step_id".into()),
value: Expr::None,
},
Update::MapInsert {
field: "step_target_counts".into(),
key: Expr::Binding("step_id".into()),
value: Expr::U64(0),
},
Update::MapInsert {
field: "step_target_success_counts".into(),
key: Expr::Binding("step_id".into()),
value: Expr::U64(0),
},
Update::MapInsert {
field: "step_target_terminal_failure_counts".into(),
key: Expr::Binding("step_id".into()),
value: Expr::U64(0),
},
],
},
],
to: "Pending".into(),
emit: vec![flow_run_notice(FlowRunPhaseVariant::Pending)],
},
transition(
"StartRun",
&["Pending"],
"StartRun",
vec![],
vec![flow_run_notice(FlowRunPhaseVariant::Running)],
"Running",
),
step_transition(
"DispatchStep",
"DispatchStep",
vec![
tracked_step_guard("step_id"),
step_is_not_started_guard("step_id", "item_is_not_yet_dispatched"),
Guard {
name: "condition_allows_dispatch".into(),
expr: Expr::Call {
helper: "StepConditionAllowsDispatch".into(),
args: vec![Expr::Binding("step_id".into())],
},
},
Guard {
name: "dependencies_are_ready".into(),
expr: Expr::Call {
helper: "StepDependencyReady".into(),
args: vec![Expr::Binding("step_id".into())],
},
},
Guard {
name: "branch_is_not_blocked".into(),
expr: Expr::Not(Box::new(Expr::Call {
helper: "StepBranchBlocked".into(),
args: vec![Expr::Binding("step_id".into())],
})),
},
],
vec![Update::MapInsert {
field: "step_status".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Some(Box::new(step_status(StepStatusVariant::Dispatched))),
}],
vec![
step_notice("step_id", StepStatusVariant::Dispatched),
effect_with_step("AdmitStepWork", "step_id"),
],
),
step_transition(
"CompleteStep",
"CompleteStep",
vec![
tracked_step_guard("step_id"),
step_status_guard(
"step_id",
"step_is_dispatched",
StepStatusVariant::Dispatched,
),
],
vec![
Update::MapInsert {
field: "step_status".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Some(Box::new(step_status(StepStatusVariant::Completed))),
},
Update::Assign {
field: "consecutive_failure_count".into(),
expr: Expr::U64(0),
},
],
vec![step_notice("step_id", StepStatusVariant::Completed)],
),
step_transition(
"RecordStepOutput",
"RecordStepOutput",
vec![
tracked_step_guard("step_id"),
step_status_guard("step_id", "step_is_completed", StepStatusVariant::Completed),
Guard {
name: "output_not_yet_recorded".into(),
expr: Expr::Call {
helper: "StepOutputRecordedIs".into(),
args: vec![Expr::Binding("step_id".into()), Expr::Bool(false)],
},
},
],
vec![Update::MapInsert {
field: "output_recorded".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Bool(true),
}],
vec![effect_with_step("PersistStepOutput", "step_id")],
),
step_transition(
"ConditionPassed",
"ConditionPassed",
vec![
tracked_step_guard("step_id"),
step_is_not_started_guard("step_id", "step_is_not_started"),
],
vec![Update::MapInsert {
field: "step_condition_results".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Some(Box::new(Expr::Bool(true))),
}],
vec![],
),
step_transition(
"ConditionRejected",
"ConditionRejected",
vec![
tracked_step_guard("step_id"),
step_is_not_started_guard("step_id", "step_is_not_started"),
],
vec![
Update::MapInsert {
field: "step_condition_results".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Some(Box::new(Expr::Bool(false))),
},
Update::MapInsert {
field: "step_status".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Some(Box::new(step_status(StepStatusVariant::Skipped))),
},
],
vec![step_notice("step_id", StepStatusVariant::Skipped)],
),
step_transition(
"FailStepEscalating",
"FailStep",
vec![
tracked_step_guard("step_id"),
step_status_guard(
"step_id",
"step_is_dispatched",
StepStatusVariant::Dispatched,
),
Guard {
name: "escalation_will_trigger".into(),
expr: Expr::Call {
helper: "EscalationWillTrigger".into(),
args: vec![],
},
},
],
vec![
Update::MapInsert {
field: "step_status".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Some(Box::new(step_status(StepStatusVariant::Failed))),
},
Update::Increment {
field: "failure_count".into(),
amount: 1,
},
Update::Increment {
field: "consecutive_failure_count".into(),
amount: 1,
},
],
vec![
step_notice("step_id", StepStatusVariant::Failed),
effect_with_step("AppendFailureLedger", "step_id"),
effect_with_step("EscalateSupervisor", "step_id"),
],
),
step_transition(
"FailStep",
"FailStep",
vec![
tracked_step_guard("step_id"),
step_status_guard(
"step_id",
"step_is_dispatched",
StepStatusVariant::Dispatched,
),
Guard {
name: "escalation_does_not_trigger".into(),
expr: Expr::Not(Box::new(Expr::Call {
helper: "EscalationWillTrigger".into(),
args: vec![],
})),
},
],
vec![
Update::MapInsert {
field: "step_status".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Some(Box::new(step_status(StepStatusVariant::Failed))),
},
Update::Increment {
field: "failure_count".into(),
amount: 1,
},
Update::Increment {
field: "consecutive_failure_count".into(),
amount: 1,
},
],
vec![
step_notice("step_id", StepStatusVariant::Failed),
effect_with_step("AppendFailureLedger", "step_id"),
],
),
step_transition(
"SkipStep",
"SkipStep",
vec![
tracked_step_guard("step_id"),
step_is_not_started_guard("step_id", "step_is_not_started"),
],
vec![Update::MapInsert {
field: "step_status".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Some(Box::new(step_status(StepStatusVariant::Skipped))),
}],
vec![step_notice("step_id", StepStatusVariant::Skipped)],
),
step_transition(
"CancelStep",
"CancelStep",
vec![
tracked_step_guard("step_id"),
Guard {
name: "step_is_cancelable".into(),
expr: Expr::Or(vec![
step_is_not_started_expr("step_id"),
Expr::Call {
helper: "StepStatusIs".into(),
args: vec![
Expr::Binding("step_id".into()),
step_status(StepStatusVariant::Dispatched),
],
},
]),
},
],
vec![Update::MapInsert {
field: "step_status".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Some(Box::new(step_status(StepStatusVariant::Canceled))),
}],
vec![step_notice("step_id", StepStatusVariant::Canceled)],
),
TransitionSchema {
name: "RegisterTargets".into(),
from: vec!["Running".into()],
on: InputMatch {
variant: "RegisterTargets".into(),
bindings: vec!["step_id".into(), "target_count".into()],
},
guards: vec![
tracked_step_guard("step_id"),
step_is_not_started_guard("step_id", "step_is_not_started"),
],
updates: vec![
Update::MapInsert {
field: "step_target_counts".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Binding("target_count".into()),
},
Update::MapInsert {
field: "step_target_success_counts".into(),
key: Expr::Binding("step_id".into()),
value: Expr::U64(0),
},
Update::MapInsert {
field: "step_target_terminal_failure_counts".into(),
key: Expr::Binding("step_id".into()),
value: Expr::U64(0),
},
],
to: "Running".into(),
emit: vec![],
},
TransitionSchema {
name: "RecordTargetSuccess".into(),
from: vec!["Running".into()],
on: InputMatch {
variant: "RecordTargetSuccess".into(),
bindings: vec!["step_id".into(), "target_id".into()],
},
guards: vec![
tracked_step_guard("step_id"),
step_status_guard(
"step_id",
"step_is_dispatched",
StepStatusVariant::Dispatched,
),
],
updates: vec![Update::MapInsert {
field: "step_target_success_counts".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Add(
Box::new(step_target_success_count_for("step_id")),
Box::new(Expr::U64(1)),
),
}],
to: "Running".into(),
emit: vec![EffectEmit {
variant: "ProjectTargetSuccess".into(),
fields: IndexMap::from([
("step_id".into(), Expr::Binding("step_id".into())),
("target_id".into(), Expr::Binding("target_id".into())),
]),
}],
},
TransitionSchema {
name: "RecordTargetTerminalFailure".into(),
from: vec!["Running".into()],
on: InputMatch {
variant: "RecordTargetTerminalFailure".into(),
bindings: vec!["step_id".into()],
},
guards: vec![
tracked_step_guard("step_id"),
step_status_guard(
"step_id",
"step_is_dispatched",
StepStatusVariant::Dispatched,
),
],
updates: vec![Update::MapInsert {
field: "step_target_terminal_failure_counts".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Add(
Box::new(step_target_terminal_failure_count_for("step_id")),
Box::new(Expr::U64(1)),
),
}],
to: "Running".into(),
emit: vec![],
},
TransitionSchema {
name: "RecordTargetCanceled".into(),
from: vec!["Running".into()],
on: InputMatch {
variant: "RecordTargetCanceled".into(),
bindings: vec!["step_id".into(), "target_id".into()],
},
guards: vec![
tracked_step_guard("step_id"),
step_status_guard(
"step_id",
"step_is_dispatched",
StepStatusVariant::Dispatched,
),
],
updates: vec![Update::MapInsert {
field: "step_target_terminal_failure_counts".into(),
key: Expr::Binding("step_id".into()),
value: Expr::Add(
Box::new(step_target_terminal_failure_count_for("step_id")),
Box::new(Expr::U64(1)),
),
}],
to: "Running".into(),
emit: vec![EffectEmit {
variant: "ProjectTargetCanceled".into(),
fields: IndexMap::from([
("step_id".into(), Expr::Binding("step_id".into())),
("target_id".into(), Expr::Binding("target_id".into())),
]),
}],
},
TransitionSchema {
name: "RecordTargetFailure".into(),
from: vec!["Running".into()],
on: InputMatch {
variant: "RecordTargetFailure".into(),
bindings: vec!["step_id".into(), "target_id".into(), "retry_key".into()],
},
guards: vec![
tracked_step_guard("step_id"),
step_status_guard(
"step_id",
"step_is_dispatched",
StepStatusVariant::Dispatched,
),
],
updates: vec![Update::MapInsert {
field: "target_retry_counts".into(),
key: Expr::Binding("retry_key".into()),
value: Expr::Add(
Box::new(target_retry_count_for("retry_key")),
Box::new(Expr::U64(1)),
),
}],
to: "Running".into(),
emit: vec![
EffectEmit {
variant: "ProjectTargetFailure".into(),
fields: IndexMap::from([
("step_id".into(), Expr::Binding("step_id".into())),
("target_id".into(), Expr::Binding("target_id".into())),
]),
},
effect_with_step("AppendFailureLedger", "step_id"),
],
},
TransitionSchema {
name: "TerminalizeCompleted".into(),
from: vec!["Running".into()],
on: InputMatch {
variant: "TerminalizeCompleted".into(),
bindings: vec![],
},
guards: vec![Guard {
name: "all_steps_are_completed_or_skipped".into(),
expr: Expr::Call {
helper: "AllTrackedStepsInAllowedStatuses".into(),
args: vec![Expr::SeqLiteral(vec![
Expr::Some(Box::new(step_status(StepStatusVariant::Completed))),
Expr::Some(Box::new(step_status(StepStatusVariant::Skipped))),
])],
},
}],
updates: vec![],
to: "Completed".into(),
emit: vec![
flow_run_notice(FlowRunPhaseVariant::Completed),
EffectEmit {
variant: "FlowTerminalized".into(),
fields: IndexMap::from([(
"run_status".into(),
flow_run_status(FlowRunPhaseVariant::Completed),
)]),
},
],
},
TransitionSchema {
name: "TerminalizeFailed".into(),
from: vec!["Pending".into(), "Running".into()],
on: InputMatch {
variant: "TerminalizeFailed".into(),
bindings: vec![],
},
guards: vec![Guard {
name: "no_step_remains_dispatched".into(),
expr: Expr::Call {
helper: "NoTrackedStepInStatus".into(),
args: vec![step_status(StepStatusVariant::Dispatched)],
},
}],
updates: vec![],
to: "Failed".into(),
emit: vec![
flow_run_notice(FlowRunPhaseVariant::Failed),
EffectEmit {
variant: "FlowTerminalized".into(),
fields: IndexMap::from([(
"run_status".into(),
flow_run_status(FlowRunPhaseVariant::Failed),
)]),
},
],
},
TransitionSchema {
name: "TerminalizeCanceled".into(),
from: vec!["Pending".into(), "Running".into()],
on: InputMatch {
variant: "TerminalizeCanceled".into(),
bindings: vec![],
},
guards: vec![Guard {
name: "no_step_remains_dispatched".into(),
expr: Expr::Call {
helper: "NoTrackedStepInStatus".into(),
args: vec![step_status(StepStatusVariant::Dispatched)],
},
}],
updates: vec![],
to: "Canceled".into(),
emit: vec![
flow_run_notice(FlowRunPhaseVariant::Canceled),
EffectEmit {
variant: "FlowTerminalized".into(),
fields: IndexMap::from([(
"run_status".into(),
flow_run_status(FlowRunPhaseVariant::Canceled),
)]),
},
],
},
],
effect_dispositions: vec![
disposition("EmitFlowRunNotice", EffectDisposition::External),
disposition("EmitStepNotice", EffectDisposition::External),
disposition("AppendFailureLedger", EffectDisposition::Local),
disposition("PersistStepOutput", EffectDisposition::Local),
disposition(
"AdmitStepWork",
EffectDisposition::Routed {
consumer_machines: vec!["RuntimeControlMachine".into()],
},
),
disposition(
"FlowTerminalized",
EffectDisposition::Routed {
consumer_machines: vec!["MobOrchestratorMachine".into()],
},
),
disposition(
"EscalateSupervisor",
EffectDisposition::Routed {
consumer_machines: vec!["MobOrchestratorMachine".into()],
},
),
disposition("ProjectTargetSuccess", EffectDisposition::External),
disposition("ProjectTargetFailure", EffectDisposition::External),
disposition("ProjectTargetCanceled", EffectDisposition::External),
],
}
}
fn disposition(name: &str, d: EffectDisposition) -> EffectDispositionRule {
EffectDispositionRule {
effect_variant: name.into(),
disposition: d,
handoff_protocol: None,
}
}
fn variant(name: &str) -> VariantSchema {
VariantSchema {
name: name.into(),
fields: vec![],
}
}
fn field(name: &str, ty: TypeRef) -> FieldSchema {
FieldSchema {
name: name.into(),
ty,
}
}
fn init(field: &str, expr: Expr) -> FieldInit {
FieldInit {
field: field.into(),
expr,
}
}
fn helper(name: &str, params: Vec<FieldSchema>, returns: TypeRef, body: Expr) -> HelperSchema {
HelperSchema {
name: name.into(),
params,
returns,
body,
}
}
fn transition(
name: &str,
from: &[&str],
input_variant: &str,
guards: Vec<Guard>,
emit: Vec<EffectEmit>,
to: &str,
) -> TransitionSchema {
TransitionSchema {
name: name.into(),
from: from.iter().map(|phase| (*phase).to_owned()).collect(),
on: InputMatch {
variant: input_variant.into(),
bindings: vec![],
},
guards,
updates: vec![],
to: to.into(),
emit,
}
}
#[derive(Debug, Clone, Copy)]
enum FlowRunPhaseVariant {
Pending,
Running,
Completed,
Failed,
Canceled,
}
#[derive(Debug, Clone, Copy)]
enum StepStatusVariant {
Dispatched,
Completed,
Failed,
Skipped,
Canceled,
}
#[derive(Debug, Clone, Copy)]
enum DependencyModeVariant {
All,
Any,
}
#[derive(Debug, Clone, Copy)]
enum CollectionPolicyVariant {
All,
Any,
Quorum,
}
fn step_transition(
name: &str,
input_variant: &str,
guards: Vec<Guard>,
updates: Vec<Update>,
emit: Vec<EffectEmit>,
) -> TransitionSchema {
TransitionSchema {
name: name.into(),
from: vec![flow_run_phase_name(FlowRunPhaseVariant::Running)],
on: InputMatch {
variant: input_variant.into(),
bindings: vec!["step_id".into()],
},
guards,
updates,
to: flow_run_phase_name(FlowRunPhaseVariant::Running),
emit,
}
}
fn tracked_step_guard(binding: &str) -> Guard {
Guard {
name: "step_is_tracked".into(),
expr: Expr::Call {
helper: "StepIsTracked".into(),
args: vec![Expr::Binding(binding.into())],
},
}
}
fn step_status_guard(binding: &str, name: &str, status: StepStatusVariant) -> Guard {
Guard {
name: name.into(),
expr: Expr::Call {
helper: "StepStatusIs".into(),
args: vec![Expr::Binding(binding.into()), step_status(status)],
},
}
}
fn step_is_not_started_guard(binding: &str, name: &str) -> Guard {
Guard {
name: name.into(),
expr: step_is_not_started_expr(binding),
}
}
fn step_is_not_started_expr(binding: &str) -> Expr {
Expr::Eq(
Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_status".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
Box::new(Expr::None),
)
}
fn sequence_members_are_in_binding(seq_binding: &str, allowed_binding: &str) -> Expr {
Expr::Quantified {
quantifier: Quantifier::All,
binding: "value".into(),
over: Box::new(Expr::SeqElements(Box::new(Expr::Binding(
seq_binding.into(),
)))),
body: Box::new(Expr::Contains {
collection: Box::new(Expr::SeqElements(Box::new(Expr::Binding(
allowed_binding.into(),
)))),
value: Box::new(Expr::Binding("value".into())),
}),
}
}
fn map_keys_match_step_ids_guard(map_binding: &str) -> Guard {
Guard {
name: format!("{map_binding}_keys_match_step_ids"),
expr: Expr::And(vec![
Expr::Quantified {
quantifier: Quantifier::All,
binding: "step_key".into(),
over: Box::new(Expr::MapKeys(Box::new(Expr::Binding(map_binding.into())))),
body: Box::new(Expr::Contains {
collection: Box::new(Expr::SeqElements(Box::new(Expr::Binding(
"step_ids".into(),
)))),
value: Box::new(Expr::Binding("step_key".into())),
}),
},
Expr::Quantified {
quantifier: Quantifier::All,
binding: "step_id".into(),
over: Box::new(Expr::SeqElements(Box::new(Expr::Binding(
"step_ids".into(),
)))),
body: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Binding(
map_binding.into(),
)))),
value: Box::new(Expr::Binding("step_id".into())),
}),
},
]),
}
}
fn step_dependencies_for(binding: &str) -> Expr {
Expr::IfElse {
condition: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Field(
"step_dependencies".into(),
)))),
value: Box::new(Expr::Binding(binding.into())),
}),
then_expr: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_dependencies".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
else_expr: Box::new(Expr::SeqLiteral(vec![])),
}
}
fn step_dependency_mode_for(binding: &str) -> Expr {
Expr::IfElse {
condition: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Field(
"step_dependency_modes".into(),
)))),
value: Box::new(Expr::Binding(binding.into())),
}),
then_expr: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_dependency_modes".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
else_expr: Box::new(dependency_mode(DependencyModeVariant::All)),
}
}
fn step_branch_for(binding: &str) -> Expr {
Expr::IfElse {
condition: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Field("step_branches".into())))),
value: Box::new(Expr::Binding(binding.into())),
}),
then_expr: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_branches".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
else_expr: Box::new(Expr::None),
}
}
fn step_collection_policy_for(binding: &str) -> Expr {
Expr::IfElse {
condition: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Field(
"step_collection_policies".into(),
)))),
value: Box::new(Expr::Binding(binding.into())),
}),
then_expr: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_collection_policies".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
else_expr: Box::new(collection_policy_kind(CollectionPolicyVariant::All)),
}
}
fn step_quorum_threshold_for(binding: &str) -> Expr {
Expr::IfElse {
condition: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Field(
"step_quorum_thresholds".into(),
)))),
value: Box::new(Expr::Binding(binding.into())),
}),
then_expr: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_quorum_thresholds".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
else_expr: Box::new(Expr::U64(0)),
}
}
fn step_target_count_for(binding: &str) -> Expr {
Expr::IfElse {
condition: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Field(
"step_target_counts".into(),
)))),
value: Box::new(Expr::Binding(binding.into())),
}),
then_expr: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_target_counts".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
else_expr: Box::new(Expr::U64(0)),
}
}
fn step_target_success_count_for(binding: &str) -> Expr {
Expr::IfElse {
condition: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Field(
"step_target_success_counts".into(),
)))),
value: Box::new(Expr::Binding(binding.into())),
}),
then_expr: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_target_success_counts".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
else_expr: Box::new(Expr::U64(0)),
}
}
fn step_target_terminal_failure_count_for(binding: &str) -> Expr {
Expr::IfElse {
condition: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Field(
"step_target_terminal_failure_counts".into(),
)))),
value: Box::new(Expr::Binding(binding.into())),
}),
then_expr: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("step_target_terminal_failure_counts".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
else_expr: Box::new(Expr::U64(0)),
}
}
fn remaining_target_count_for(binding: &str) -> Expr {
Expr::Sub(
Box::new(step_target_count_for(binding)),
Box::new(step_target_terminal_failure_count_for(binding)),
)
}
fn target_retry_count_for(binding: &str) -> Expr {
Expr::IfElse {
condition: Box::new(Expr::Contains {
collection: Box::new(Expr::MapKeys(Box::new(Expr::Field(
"target_retry_counts".into(),
)))),
value: Box::new(Expr::Binding(binding.into())),
}),
then_expr: Box::new(Expr::MapGet {
map: Box::new(Expr::Field("target_retry_counts".into())),
key: Box::new(Expr::Binding(binding.into())),
}),
else_expr: Box::new(Expr::U64(0)),
}
}
fn flow_run_notice(status_variant: FlowRunPhaseVariant) -> EffectEmit {
EffectEmit {
variant: "EmitFlowRunNotice".into(),
fields: IndexMap::from([("run_status".into(), flow_run_status(status_variant))]),
}
}
fn step_notice(step_binding: &str, status_variant: StepStatusVariant) -> EffectEmit {
EffectEmit {
variant: "EmitStepNotice".into(),
fields: IndexMap::from([
("step_id".into(), Expr::Binding(step_binding.into())),
("step_status".into(), step_status(status_variant)),
]),
}
}
fn effect_with_step(variant: &str, step_binding: &str) -> EffectEmit {
EffectEmit {
variant: variant.into(),
fields: IndexMap::from([("step_id".into(), Expr::Binding(step_binding.into()))]),
}
}
fn flow_run_phase_name(variant: FlowRunPhaseVariant) -> String {
match variant {
FlowRunPhaseVariant::Pending => "Pending".into(),
FlowRunPhaseVariant::Running => "Running".into(),
FlowRunPhaseVariant::Completed => "Completed".into(),
FlowRunPhaseVariant::Failed => "Failed".into(),
FlowRunPhaseVariant::Canceled => "Canceled".into(),
}
}
fn flow_run_status(variant: FlowRunPhaseVariant) -> Expr {
Expr::NamedVariant {
enum_name: "FlowRunStatus".into(),
variant: flow_run_phase_name(variant),
}
}
fn step_status(variant: StepStatusVariant) -> Expr {
Expr::NamedVariant {
enum_name: "StepRunStatus".into(),
variant: match variant {
StepStatusVariant::Dispatched => "Dispatched".into(),
StepStatusVariant::Completed => "Completed".into(),
StepStatusVariant::Failed => "Failed".into(),
StepStatusVariant::Skipped => "Skipped".into(),
StepStatusVariant::Canceled => "Canceled".into(),
},
}
}
fn dependency_mode(variant: DependencyModeVariant) -> Expr {
Expr::NamedVariant {
enum_name: "DependencyMode".into(),
variant: match variant {
DependencyModeVariant::All => "All".into(),
DependencyModeVariant::Any => "Any".into(),
},
}
}
fn collection_policy_kind(variant: CollectionPolicyVariant) -> Expr {
Expr::NamedVariant {
enum_name: "CollectionPolicyKind".into(),
variant: match variant {
CollectionPolicyVariant::All => "All".into(),
CollectionPolicyVariant::Any => "Any".into(),
CollectionPolicyVariant::Quorum => "Quorum".into(),
},
}
}