diff --git a/src/scheduler.py b/src/scheduler.py
index 4538bc8..ffc88b6 100644
--- a/src/scheduler.py
+++ b/src/scheduler.py
@@ -177,9 +177,21 @@ def run_goal_to_completion(db_path, target_folder_path, goal_text):
     run_task_lifecycle raises partway through -- needed to route
     CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError
     to task.status='blocked_pending_human' (see module docstring's "Second
-    gap" note). Any OTHER exception still propagates uncaught, same as
-    before -- run_scheduler_once's own try/except is what keeps one Goal's
-    unexpected failure from stopping the rest.
+    gap" note).
+
+    **2026-08-04**: any OTHER exception (e.g. work_graph_execute.
+    WorkGraphExecuteError from a real API hiccup -- found for real: a
+    "Response stalled mid-stream" error crashed the calling process
+    entirely) used to propagate uncaught, same as before this note --
+    `run_scheduler_once`'s own try/except only stops it from taking down
+    OTHER goals in the same batch, it does NOT apply the task-level
+    blocked_pending_human consequence either, so the task itself was left
+    stuck in limbo: not running, not blocked, `blocked_reason` still NULL,
+    nothing a human could `manual_resume` from the console. This mirrors
+    the exact fallback `task_execution._mark_blocked()` already has for its
+    own (Start-button) entry point -- both entry points now guarantee a
+    task always lands on a real terminal-ish status with a real error
+    detail, never a caller-crashes-and-the-task-is-just-stuck outcome.
     """
     conn = schema.connect(db_path)
     conn.execute("PRAGMA busy_timeout = {}".format(_BUSY_TIMEOUT_MS))
@@ -194,6 +206,9 @@ def run_goal_to_completion(db_path, target_folder_path, goal_text):
             blocked_reason = _blocked_reason_for(e)
             _block_task(conn, task_id, blocked_reason, detail=str(e))
             return {"task_id": task_id, "final_status": "blocked_pending_human", "blocked_reason": blocked_reason}
+        except Exception as e:
+            _block_task(conn, task_id, "execution_error", detail=str(e))
+            return {"task_id": task_id, "final_status": "blocked_pending_human", "blocked_reason": "execution_error"}
         return finish_task_lifecycle(conn, result, target_folder_path, goal_text)
     finally:
         conn.close()
diff --git a/src/test_scheduler.py b/src/test_scheduler.py
index bb5d8b3..fa7e639 100644
--- a/src/test_scheduler.py
+++ b/src/test_scheduler.py
@@ -350,10 +350,29 @@ class ConsistencyCheckFailedRoutingTest(RoutableExceptionTestBase):
         self._assert_task_blocked(result["task_id"], "prd_consistency_check_failed")
 
 
-class UnroutedExceptionStillPropagatesTest(RoutableExceptionTestBase):
-    def test_other_exceptions_are_not_swallowed(self):
-        with self.assertRaises(RuntimeError):
-            self._run_with_raising_lifecycle(RuntimeError("something else entirely"))
+class UnroutedExceptionStillBlocksTaskTest(RoutableExceptionTestBase):
+    def test_other_exceptions_block_the_task_instead_of_leaving_it_stuck(self):
+        # 2026-08-04 regression: this used to propagate uncaught, leaving
+        # the task neither running nor blocked_pending_human -- a real
+        # work_graph_execute.WorkGraphExecuteError (API "Response stalled
+        # mid-stream") crashed the calling process and left the task stuck
+        # in limbo, nothing a human could resume from the console. Mirrors
+        # task_execution._mark_blocked's same fallback for its own
+        # (Start-button) entry point.
+        result = self._run_with_raising_lifecycle(RuntimeError("something else entirely"))
+        self.assertEqual(result["final_status"], "blocked_pending_human")
+        self.assertEqual(result["blocked_reason"], "execution_error")
+        self._assert_task_blocked(result["task_id"], "execution_error")
+
+    def test_real_exception_message_is_persisted_for_diagnosis(self):
+        result = self._run_with_raising_lifecycle(RuntimeError("stalled mid-stream"))
+        conn = schema.connect(self.db_path)
+        payload = conn.execute(
+            "SELECT payload FROM execution_event WHERE task_id = ? ORDER BY rowid DESC LIMIT 1",
+            (result["task_id"],),
+        ).fetchone()[0]
+        conn.close()
+        self.assertEqual(json.loads(payload)["detail"], "stalled mid-stream")
 
 
 def _fix_loop_trigger(org_node, prompt, target_folder_path, timeout=300):