diff --git a/awxkit/awxkit/api/mixins/has_status.py b/awxkit/awxkit/api/mixins/has_status.py
index bd76400baa..db14874b6c 100644
--- a/awxkit/awxkit/api/mixins/has_status.py
+++ b/awxkit/awxkit/api/mixins/has_status.py
@@ -47,6 +47,13 @@ class HasStatus(object):
     def wait_until_started(self, interval=1, timeout=60):
         return self.wait_until_status(self.started_statuses, interval=interval, timeout=timeout)
 
+    def failure_output_details(self):
+        if getattr(self, 'result_stdout', ''):
+            output = bytes_to_str(self.result_stdout)
+            if output:
+                return '\nstdout:\n{}'.format(output)
+        return ''
+
     def assert_status(self, status_list, msg=None):
         if isinstance(status_list, str):
             status_list = [status_list]
@@ -65,10 +72,9 @@ class HasStatus(object):
             msg += '\njob_explanation: {}'.format(bytes_to_str(self.job_explanation))
         if getattr(self, 'result_traceback', ''):
             msg += '\nresult_traceback:\n{}'.format(bytes_to_str(self.result_traceback))
-        if getattr(self, 'result_stdout', ''):
-            output = bytes_to_str(self.result_stdout)
-            if output:
-                msg = msg + '\nstdout:\n{}'.format(output)
+
+        msg += self.failure_output_details()
+
         if getattr(self, 'job_explanation', '').startswith('Previous Task Failed'):
             try:
                 data = json.loads(self.job_explanation.replace('Previous Task Failed: ', ''))
diff --git a/awxkit/awxkit/api/pages/workflow_jobs.py b/awxkit/awxkit/api/pages/workflow_jobs.py
index d7fe487030..66ff73ac1e 100644
--- a/awxkit/awxkit/api/pages/workflow_jobs.py
+++ b/awxkit/awxkit/api/pages/workflow_jobs.py
@@ -13,11 +13,37 @@ class WorkflowJob(UnifiedJob):
         result = self.related.relaunch.post(payload)
         return self.walk(result.url)
 
+    def failure_output_details(self):
+        """Special implementation of this part of assert_status so that
+        workflow_job.assert_successful() will give a breakdown of failure
+        """
+        msg = '\nNode summary:'
+        node_list = self.related.workflow_nodes.get().results
+
+        for node in node_list:
+            msg += '\n{}:'.format(node.id)
+            if node.job:
+                msg += ' {}'.format(node.summary_fields.job)
+            else:
+                msg += ' None'
+            for rel in ('failure_nodes', 'always_nodes', 'success_nodes'):
+                val = getattr(node, rel, [])
+                if val:
+                    msg += ' {} {}'.format(rel, val)
+
+        msg += '\n\nUnhandled individual job failures:\n'
+        for node in node_list:
+            if node.job and not (node.failure_nodes or node.always_nodes):
+                job = node.related.job.get()
+                try:
+                    job.assert_successful()
+                except Exception as e:
+                    msg += str(e)
+        return msg
+
     @property
     def result_stdout(self):
         # workflow jobs do not have result_stdout
-        # which is problematic for the UnifiedJob.is_successful reliance on
-        # related stdout endpoint.
         if 'result_stdout' not in self.json:
             return 'Unprovided AWX field.'
         else: