feat(compiler): implicit_deps_by_order

- allow automatic implicit deps injection specified through subtasks

Author: LEXUGE

src/cascade/compiler/ast.py

@@ -183,6 +183,7 @@ class Goal(BaseTask):
     type: ClassVar[str] = "goal"
 
     subtasks: list[str]
+    implicit_deps_by_order: Optional[bool] = False
 
     @model_validator(mode="after")
     def validate_goal(self) -> Self:
@@ -230,6 +231,16 @@ def propogate_priority(self, ast: TaskAST):
         for t in self.subtasks:
             tasks_dict[t].priority *= self.priority
 
+    def inject_implicit_deps(self, ast: TaskAST):
+        """
+        Inject implicit dependencies specified by subtask ordering. This should be done before dependency graph checking
+        """
+        if self.implicit_deps_by_order:
+            for i in range(1, len(self.subtasks)):
+                ast.get_tasks_in_dict()[self.subtasks[i]].deps.after.add(
+                    self.subtasks[i - 1]
+                )
+
 
 def get_task_type(v: Any) -> str:
     if isinstance(v, dict):
@@ -376,6 +387,9 @@ def normalize_dependencies(self) -> list[Step]:
     def check(self) -> Self:
         self.check_refs()
         self.check_deadlines()
+        # NOTE: Implicit dependency injection should be done before deps checking
+        for goal in self.get_goals():
+            goal.inject_implicit_deps(self)
         self.check_deps_graph()
         return self
 

tests/compiler/test_deps.py

@@ -0,0 +1,32 @@
+from cascade.compiler import *
+from cascade.compiler.ast import CascadeConfig, Dependencies, Status
+from typing import List
+from zoneinfo import ZoneInfo
+
+DEFAULT_TZ = ZoneInfo("Europe/London")
+
+
+def test_implicit_deps():
+    tasks: List[Step | Goal] = [
+        Step(name="Task A", status=Status.todo, duration=DURATION_UNIT),
+        Step(
+            name="Task B",
+            status=Status.todo,
+            deps=Dependencies(after={"task-c"}),
+            duration=DURATION_UNIT,
+        ),
+        Step(name="Task C", status=Status.todo, duration=DURATION_UNIT),
+        Goal(
+            name="Goal A",
+            subtasks=["goal-b", "task-a"],
+            implicit_deps_by_order=True,
+        ),
+        Goal(name="Goal B", subtasks=["task-b"]),
+    ]
+    ast = TaskAST(config=CascadeConfig(default_tz=DEFAULT_TZ), tasks=tasks)
+    processed = ProcessedAST.from_raw_ast(ast)
+
+    assert len(processed.nodes) == 3
+    assert processed.nodes["task-a"].deps == {"task-b"}
+    assert processed.nodes["task-b"].deps == {"task-c"}
+    assert processed.nodes["task-c"].deps == set()