[Job] Support DAG execution by replacing is_chain with is_dag check

sky/dag.py

@@ -242,6 +242,22 @@ 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_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.
+        """
+
+        # A graph is weakly connected if replacing all directed edges with
+        # undirected edges produces a connected graph, i.e., any two nodes
+        # can reach each other ignoring edge directions.
+        if not nx.is_weakly_connected(self.graph):
+            return False
+
+        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,12 @@ 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_connected_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 connected DAG is allowed for job_launch. If your dag '
+                f'contains multiple subgraph that is a connected dag, please '
+                f'separate them into multiple dag. Get: {dag}')
 
     dag_utils.maybe_infer_and_fill_dag_and_task_names(dag)