feat: check is dag

sky/dag.py

@@ -242,6 +242,10 @@ 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)
+
 
 class _DagContext(threading.local):
     """A thread-local stack of Dags."""

sky/jobs/core.py

@@ -59,10 +59,9 @@ 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_chain():
+    if not dag.is_dag():
         with ux_utils.print_exception_no_traceback():
-            raise ValueError('Only single-task or chain DAG is '
-                             f'allowed for job_launch. Dag: {dag}')
+            raise ValueError(f'Only DAG is allowed for job_launch. Dag: {dag}')
 
     dag_utils.maybe_infer_and_fill_dag_and_task_names(dag)