diff --git a/salt/manager/tools/sbin/so-push-drainer b/salt/manager/tools/sbin/so-push-drainer
index 369fcc0fe..48c3e5b7e 100644
--- a/salt/manager/tools/sbin/so-push-drainer
+++ b/salt/manager/tools/sbin/so-push-drainer
@@ -52,7 +52,7 @@ RESULT_MAX_AGE = 7200
 RESULT_CHECKS_PER_PASS = 5
 TEXT_LIMIT = 500
 
-# salt-run --async reports the jid only in a log line on stderr.
+# salt-run --async reports the jid only in a log line (stderr by default).
 JID_RE = re.compile(r'salt/run/(\d{20})')
 
 
@@ -133,7 +133,7 @@ def _dispatch(actions, log):
     except Exception:
         log.exception('dispatch raised')
         return None
-    match = JID_RE.search(result.stderr or '')
+    match = JID_RE.search('{}\n{}'.format(result.stderr or '', result.stdout or ''))
     if not match:
         log.warning('dispatch accepted but no jid found, result will not be tracked: stderr=%s',
                     _trim(result.stderr))
@@ -197,26 +197,39 @@ def _minion_failure(minion_ret):
     return ''
 
 
+def _step_failures(step):
+    if not isinstance(step, dict) or step.get('result') is not False:
+        return []
+    failures = ['{}: {}'.format(step.get('__id__', step.get('name')), _trim(step.get('comment', '')))]
+    changes = step.get('changes')
+    minion_rets = changes.get('ret') if isinstance(changes, dict) else None
+    if isinstance(minion_rets, dict):
+        for minion, minion_ret in minion_rets.items():
+            text = _minion_failure(minion_ret)
+            if text:
+                failures.append('{}: {}'.format(minion, text))
+    return failures
+
+
 def _orch_failures(ret):
+    if not isinstance(ret, dict):
+        return [_trim(ret)]
     failures = []
     for job in ret.values():
         if not isinstance(job, dict):
             continue
         job_ret = job.get('return')
-        data = (job_ret.get('data') or {}) if isinstance(job_ret, dict) else {}
+        data = job_ret.get('data') if isinstance(job_ret, dict) else {}
+        if not isinstance(data, dict):
+            if data:
+                failures.append(_trim(data))
+            data = {}
         for steps in data.values():
             if not isinstance(steps, dict):
                 failures.append(_trim(steps))
                 continue
             for step in steps.values():
-                if not isinstance(step, dict) or step.get('result') is not False:
-                    continue
-                failures.append('{}: {}'.format(step.get('__id__', step.get('name')), _trim(step.get('comment', ''))))
-                minion_rets = (step.get('changes') or {}).get('ret') or {}
-                for minion, minion_ret in minion_rets.items():
-                    text = _minion_failure(minion_ret)
-                    if text:
-                        failures.append('{}: {}'.format(minion, text))
+                failures.extend(_step_failures(step))
         if job.get('success') is False and not failures:
             failures.append('orchestration reported failure: {}'.format(_trim(job.get('return'))))
     return failures
@@ -236,20 +249,32 @@ def _check_dispatched(log, now):
             continue
         checked += 1
         jid = record['jid']
-        paths = record.get('paths', [])
-        ret = _lookup_jid(jid, log)
-        if not ret:
-            if age > RESULT_MAX_AGE:
-                log.warning('no result for jid=%s after %ds, no longer tracking; paths=%s', jid, age, paths)
+        try:
+            if _report_result(record, age, log):
                 _unlink(path, log)
-            continue
-        failures = _orch_failures(ret)
-        if failures:
-            log.error('push failed jid=%s paths=%s; change will be applied at the next scheduled highstate: %s',
-                      jid, paths, ' | '.join(failures))
-        else:
-            log.info('push succeeded jid=%s paths=%s', jid, paths)
-        _unlink(path, log)
+        except Exception:
+            # Drop the record so one unreadable result can't fail every pass ahead of the drain.
+            log.exception('cannot evaluate result for jid=%s; no longer tracking', jid)
+            _unlink(path, log)
+
+
+def _report_result(record, age, log):
+    """Logs the outcome of a dispatched push. Returns True once the record is finished with."""
+    jid = record['jid']
+    paths = record.get('paths', [])
+    ret = _lookup_jid(jid, log)
+    if not ret:
+        if age > RESULT_MAX_AGE:
+            log.warning('no result for jid=%s after %ds, no longer tracking; paths=%s', jid, age, paths)
+            return True
+        return False
+    failures = _orch_failures(ret)
+    if failures:
+        log.error('push failed jid=%s paths=%s; change will be applied at the next scheduled highstate: %s',
+                  jid, paths, ' | '.join(failures))
+    else:
+        log.info('push succeeded jid=%s paths=%s', jid, paths)
+    return True
 
 
 def main():
diff --git a/salt/manager/tools/sbin/so-push-drainer_test.py b/salt/manager/tools/sbin/so-push-drainer_test.py
index f758b6107..0c415710e 100644
--- a/salt/manager/tools/sbin/so-push-drainer_test.py
+++ b/salt/manager/tools/sbin/so-push-drainer_test.py
@@ -126,7 +126,13 @@ class TestHelpers(DrainerTestCase):
 
     def test_make_logger_adds_handler_once(self):
         logger = logging.getLogger('so-push-drainer')
-        self.addCleanup(logger.handlers.clear)
+
+        def close_handlers():
+            for handler in logger.handlers:
+                handler.close()
+            logger.handlers.clear()
+
+        self.addCleanup(close_handlers)
         logger.handlers.clear()
         self.assertIs(drainer._make_logger(), logger)
         drainer._make_logger()
@@ -194,6 +200,10 @@ class TestDispatch(DrainerTestCase):
         self.assertEqual(cmd[:3], ['salt-run', 'state.orchestrate', 'orch.push_batch'])
         self.assertIn('--async', cmd)
 
+    def test_jid_parsed_from_stdout(self):
+        jid, _ = self.run_dispatch(return_value=MagicMock(stdout=ASYNC_STDERR, stderr=None))
+        self.assertEqual(jid, JID)
+
     def test_no_jid(self):
         jid, _ = self.run_dispatch(return_value=MagicMock(stdout='', stderr=None))
         self.assertEqual(jid, '')
@@ -260,6 +270,21 @@ class TestResults(DrainerTestCase):
         ret = {MASTER: {'return': {'data': {MASTER: ['Rendering SLS failed']}}, 'success': False}}
         self.assertEqual(drainer._orch_failures(ret), ['["Rendering SLS failed"]'])
 
+    def test_orch_failures_not_a_dict(self):
+        self.assertEqual(drainer._orch_failures(['No minions matched']), ['["No minions matched"]'])
+        self.assertEqual(drainer._orch_failures('Runner error'), ['Runner error'])
+
+    def test_orch_failures_data_not_a_dict(self):
+        ret = {MASTER: {'return': {'data': ["Rendering SLS 'orch.push_batch' failed"]}, 'success': False}}
+        self.assertEqual(drainer._orch_failures(ret), ['["Rendering SLS \'orch.push_batch\' failed"]'])
+
+    def test_orch_failures_odd_changes(self):
+        for changes in ('Run failed', {'ret': ['manager_standalone']}):
+            ret = _orch_ret({'salt_|-apply_soc_1_|-apply_soc_1_|-state': {
+                '__id__': 'apply_soc_1', 'result': False, 'changes': changes, 'comment': 'Run failed on minions',
+            }}, success=False)
+            self.assertEqual(drainer._orch_failures(ret), ['apply_soc_1: Run failed on minions'])
+
     def test_orch_failures_unparsed(self):
         self.assertEqual(drainer._orch_failures({MASTER: 'odd'}), [])
         ret = {MASTER: {'return': 'Exception occurred', 'success': False}}
@@ -295,6 +320,25 @@ class TestResults(DrainerTestCase):
         self.assertIn('push succeeded jid=2_ok', self.logged('info'))
         self.assertIn('no result for jid=4_expired', self.logged('warning'))
 
+    def test_check_dispatched_survives_bad_result(self):
+        now = time.time()
+        bad = self.record('1_bad', 60, now)
+        good = self.record('2_ok', 60, now)
+
+        def orch_failures(ret):
+            if ret == 'boom':
+                raise ValueError('unexpected shape')
+            return []
+
+        with patch.object(drainer, '_lookup_jid', side_effect=lambda jid, log: 'boom' if jid == '1_bad' else SUCCESS_RET), \
+                patch.object(drainer, '_orch_failures', side_effect=orch_failures):
+            drainer._check_dispatched(self.log, now)
+        self.assertFalse(os.path.exists(bad))
+        self.assertFalse(os.path.exists(good))
+        self.log.exception.assert_called_once()
+        self.assertIn('jid=1_bad', self.log.exception.call_args[0][0] % self.log.exception.call_args[0][1:])
+        self.assertIn('push succeeded jid=2_ok', self.logged('info'))
+
     def test_check_dispatched_limit(self):
         now = time.time()
         for i in range(drainer.RESULT_CHECKS_PER_PASS + 2):