diff --git a/awx/api/generics.py b/awx/api/generics.py
index 4c4247b23d..e474a1bafc 100644
--- a/awx/api/generics.py
+++ b/awx/api/generics.py
@@ -370,6 +370,9 @@ class SubListCreateAttachDetachAPIView(SubListCreateAPIView):
     # Base class for a sublist view that allows for creating subobjects and
     # attaching/detaching them from the parent.
 
+    def is_valid_relation(self, parent, sub, created=False):
+        return None
+
     def get_description_context(self):
         d = super(SubListCreateAttachDetachAPIView, self).get_description_context()
         d.update({
@@ -406,6 +409,13 @@ class SubListCreateAttachDetachAPIView(SubListCreateAPIView):
                                        skip_sub_obj_read_check=created):
             raise PermissionDenied()
 
+        # Verify that the relationship to be added is valid.
+        attach_errors = self.is_valid_relation(parent, sub, created=created)
+        if attach_errors is not None:
+            if created:
+                sub.delete()
+            return Response(attach_errors, status=status.HTTP_400_BAD_REQUEST)
+
         # Attach the object to the collection.
         if sub not in relationship.all():
             relationship.add(sub)
diff --git a/awx/api/views.py b/awx/api/views.py
index 6bec6d9980..9b1b4e0cd5 100644
--- a/awx/api/views.py
+++ b/awx/api/views.py
@@ -2655,6 +2655,46 @@ class WorkflowJobTemplateNodeChildrenBaseList(EnforceParentRelationshipMixin, Su
         self.check_parent_access(parent)
         return getattr(parent, self.relationship).all()
 
+    def is_valid_relation(self, parent, sub, created=False):
+        mutex_list = ('success_nodes', 'failure_nodes') if self.relationship == 'always_nodes' else ('always_nodes',)
+        for relation in mutex_list:
+            if getattr(parent, relation).all().exists():
+                return {'Error': 'Cannot associate {0} when {1} have been associated.'.format(self.relationship, relation)}
+
+        if created:
+            return None
+
+        workflow_nodes = parent.workflow_job_template.workflow_job_template_nodes.all().\
+            prefetch_related('success_nodes', 'failure_nodes', 'always_nodes')
+        graph = {}
+        for workflow_node in workflow_nodes:
+            graph[workflow_node.pk] = dict(node_object=workflow_node, metadata={'parent': None, 'traversed': False})
+
+        find = False
+        for node_type in ['success_nodes', 'failure_nodes', 'always_nodes']:
+            for workflow_node in workflow_nodes:
+                parent_node = graph[workflow_node.pk]
+                related_nodes = getattr(parent_node['node_object'], node_type).all()
+                for related_node in related_nodes:
+                    sub_node = graph[related_node.pk]
+                    sub_node['metadata']['parent'] = parent_node
+                    if not find and parent == workflow_node and sub == related_node and self.relationship == node_type:
+                        find = True
+        if not find:
+            sub_node = graph[sub.pk]
+            parent_node = graph[parent.pk]
+            if sub_node['metadata']['parent'] is not None:
+                return {"Error": "Multiple parent relationship not allowed."}
+            sub_node['metadata']['parent'] = parent_node
+            iter_node = sub_node
+            while iter_node is not None:
+                if iter_node['metadata']['traversed']:
+                    return {"Error": "Cycle detected."}
+                iter_node['metadata']['traversed'] = True
+                iter_node = iter_node['metadata']['parent']
+
+        return None
+
 class WorkflowJobTemplateNodeSuccessNodesList(WorkflowJobTemplateNodeChildrenBaseList):
     relationship = 'success_nodes'
 
diff --git a/awx/main/scheduler/dag_simple.py b/awx/main/scheduler/dag_simple.py
index aeb0ff759e..c6aa6247c0 100644
--- a/awx/main/scheduler/dag_simple.py
+++ b/awx/main/scheduler/dag_simple.py
@@ -137,4 +137,3 @@ class SimpleDAG(object):
             if len(self.get_dependents(n['node_object'])) < 1:
                 roots.append(n)
         return roots
-
diff --git a/awx/main/tests/functional/models/test_workflow.py b/awx/main/tests/functional/models/test_workflow.py
index 4f96a4db0c..1ead52fb0d 100644
--- a/awx/main/tests/functional/models/test_workflow.py
+++ b/awx/main/tests/functional/models/test_workflow.py
@@ -118,6 +118,22 @@ class TestWorkflowJobTemplate:
         assert len(parent_qs) == 1
         assert parent_qs[0] == wfjt.workflow_job_template_nodes.all()[1]
 
+    def test_topology_validator(self, wfjt):
+        from awx.api.views import WorkflowJobTemplateNodeChildrenBaseList
+        test_view = WorkflowJobTemplateNodeChildrenBaseList()
+        nodes = wfjt.workflow_job_template_nodes.all()
+        node_assoc = WorkflowJobTemplateNode.objects.create(workflow_job_template=wfjt)
+        nodes[2].always_nodes.add(node_assoc)
+        # test cycle validation
+        assert test_view.is_valid_relation(node_assoc, nodes[0]) == {'Error': 'Cycle detected!'}
+        # test multi-ancestor validation
+        assert test_view.is_valid_relation(node_assoc, nodes[1]) == {'Error': 'Multiple ancestor detected!'}
+        # test mutex validation
+        test_view.relationship = 'failure_nodes'
+        node_assoc_1 = WorkflowJobTemplateNode.objects.create(workflow_job_template=wfjt)
+        assert (test_view.is_valid_relation(nodes[2], node_assoc_1) ==
+                {'Error': 'Cannot associate failure_nodes when always_nodes have been associated.'})
+
 @pytest.mark.django_db
 class TestWorkflowJobFailure:
     """