fix: also ensure it's connected

sky/dag.py

@@ -242,9 +242,15 @@ def is_chain(self) -> bool:
                 (all(degree <= 1 for degree in out_degrees) and
                  sum(degree == 0 for degree in out_degrees) == 1))
 
-    def is_dag(self) -> bool:
-        """Check if the DAG is a DAG."""
-        return nx.is_directed_acyclic_graph(self.graph)
+    def is_connected_dag(self) -> bool:
+        """Check if the graph is a connected directed acyclic graph (DAG).
+
+        Returns:
+            True if the graph is a connected DAG (weakly connected,
+            directed and acyclic), False otherwise.
+        """
+        return (nx.is_directed_acyclic_graph(self.graph) and
+                nx.is_weakly_connected(self.graph))
 
 
 class _DagContext(threading.local):

sky/jobs/core.py

@@ -59,7 +59,7 @@ def launch(
     dag = dag_utils.convert_entrypoint_to_dag(entrypoint)
     dag, mutated_user_config = admin_policy_utils.apply(
         dag, use_mutated_config_in_current_request=False)
-    if not dag.is_dag():
+    if not dag.is_connected_dag():
         with ux_utils.print_exception_no_traceback():
             raise ValueError(f'Only DAG is allowed for job_launch. Dag: {dag}')