Add cycle detection to AI flow config (#23589)

* Add cycles detection in flow config

* Add two edge-cases tests

* Remove _detect_cycle tests and add one for disjoined graphs

* Find all possible cycles in _detect_cycles

* Changes in _detect_cycles: change comment, add cycle limit and tighten test

Author: luisorofino

ddev/src/ddev/ai/phases/config.py

@@ -2,6 +2,8 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
+from __future__ import annotations
+
 from pathlib import Path
 
 import yaml
@@ -14,14 +16,49 @@ class FlowConfigError(Exception):
     """Wraps Pydantic ValidationError or YAML errors with a user-friendly message."""
 
 
+def _detect_cycles(
+    dependency_map: dict[str, list[str]],
+    limit: int = 50,
+) -> tuple[list[list[str]], bool]:
+    """Return every simple cycle in the dependency graph, each as an ordered list of phase IDs."""
+    # Enumerate every simple cycle exactly once: from each node, DFS only through
+    # higher-ranked nodes, so each cycle is reported only when started from its
+    # lowest-ranked member. (Tiernan-style enumeration with rank canonicalization.)
+    rank = {n: i for i, n in enumerate(dependency_map)}
+    cycles: list[list[str]] = []
+
+    class _LimitReached(Exception):
+        """Raised when the cycle limit is reached."""
+
+        pass
+
+    def dfs(start: str, current: str, path: list[str], on_path: set[str]):
+        for dep in dependency_map.get(current, []):
+            if dep == start:
+                cycles.append(path + [start])
+                if len(cycles) >= limit:
+                    raise _LimitReached
+            elif dep in rank and rank[dep] > rank[start] and dep not in on_path:
+                on_path.add(dep)
+                dfs(start, dep, path + [dep], on_path)
+                on_path.discard(dep)
+
+    try:
+        for start in dependency_map:
+            dfs(start, start, [start], {start})
+    except _LimitReached:
+        return cycles, True
+    return cycles, False
+
+
 class TaskConfig(BaseModel):
     model_config = ConfigDict(extra="forbid")
     name: str
     prompt_path: Path | None = None
     prompt: str | None = None
 
     @model_validator(mode="after")
-    def exactly_one_source(self) -> "TaskConfig":
+    def exactly_one_source(self) -> TaskConfig:
         if (self.prompt_path is None) == (self.prompt is None):
             raise ValueError("Exactly one of 'prompt_path' or 'prompt' must be set")
         return self
@@ -35,7 +72,7 @@ class CheckpointConfig(BaseModel):
     memory_prompt_path: Path | None = None
 
     @model_validator(mode="after")
-    def exactly_one_source(self) -> "CheckpointConfig":
+    def exactly_one_source(self) -> CheckpointConfig:
         if (self.memory_prompt is None) == (self.memory_prompt_path is None):
             raise ValueError("Exactly one of 'memory_prompt' or 'memory_prompt_path' must be set")
         return self
@@ -86,7 +123,7 @@ class FlowConfig(BaseModel):
     flow: list[FlowEntry]
 
     @model_validator(mode="after")
-    def cross_references(self) -> "FlowConfig":
+    def cross_references(self) -> FlowConfig:
         """Validate all cross-references between agents, phases, and dependencies."""
         scheduled = {entry.phase for entry in self.flow}
         seen: set[str] = set()
@@ -105,10 +142,18 @@ def cross_references(self) -> "FlowConfig":
         for phase_id, phase in self.phases.items():
             if phase.agent not in self.agents:
                 raise ValueError(f"Phase {phase_id!r} references unknown agent: {phase.agent!r}")
+
+        dependency_map = {entry.phase: entry.dependencies for entry in self.flow}
+        cycles, truncated = _detect_cycles(dependency_map)
+        if cycles:
+            formatted = "\n  ".join(" → ".join(c) for c in cycles)
+            suffix = f"\n  (showing first {len(cycles)}; more cycles exist)" if truncated else ""
+            raise ValueError(f"Cycle(s) detected in flow:\n  {formatted}{suffix}")
+
         return self
 
     @classmethod
-    def from_yaml(cls, path: Path, config_dir: Path) -> "FlowConfig":
+    def from_yaml(cls, path: Path, config_dir: Path) -> FlowConfig:
         """Load, parse, and validate flow.yaml. Raises FlowConfigError on any problem."""
         try:
             raw = yaml.safe_load(path.read_text())

ddev/tests/ai/phases/test_config.py

@@ -170,6 +170,13 @@ def test_flow_config_dependency_not_scheduled_in_flow():
         FlowConfig.model_validate(raw)
 
 
+def test_flow_config_duplicate_phase_raises():
+    raw = _minimal_config()
+    raw["flow"] = [{"phase": "p1"}, {"phase": "p1"}]
+    with pytest.raises(ValidationError, match="Duplicate phase"):
+        FlowConfig.model_validate(raw)
+
+
 def test_flow_config_unknown_agent_in_phase():
     raw = _minimal_config()
     raw["phases"]["p1"]["agent"] = "nonexistent"
@@ -295,3 +302,109 @@ def test_from_yaml_invalid_yaml(tmp_path):
 def test_from_yaml_missing_file(tmp_path):
     with pytest.raises(FlowConfigError, match="Failed to load"):
         FlowConfig.from_yaml(tmp_path / "nonexistent.yaml", tmp_path)
+
+
+# ---------------------------------------------------------------------------
+# FlowConfig cycle detection via model_validate
+# ---------------------------------------------------------------------------
+
+
+def _three_phase_config() -> dict:
+    agent = {"tools": []}
+    task = {"name": "t", "prompt": "Do it."}
+    return {
+        "agents": {"writer": agent},
+        "phases": {
+            "p1": {"agent": "writer", "tasks": [task]},
+            "p2": {"agent": "writer", "tasks": [task]},
+            "p3": {"agent": "writer", "tasks": [task]},
+        },
+    }
+
+
+def test_flow_config_direct_cycle_raises():
+    raw = _three_phase_config()
+    raw["flow"] = [
+        {"phase": "p1", "dependencies": ["p2"]},
+        {"phase": "p2", "dependencies": ["p1"]},
+    ]
+    with pytest.raises(ValidationError, match="Cycle"):
+        FlowConfig.model_validate(raw)
+
+
+def test_flow_config_three_node_cycle_raises():
+    raw = _three_phase_config()
+    raw["flow"] = [
+        {"phase": "p1", "dependencies": ["p3"]},
+        {"phase": "p2", "dependencies": ["p1"]},
+        {"phase": "p3", "dependencies": ["p2"]},
+    ]
+    with pytest.raises(ValidationError, match="Cycle"):
+        FlowConfig.model_validate(raw)
+
+
+def test_flow_config_acyclic_chain_ok():
+    raw = _three_phase_config()
+    raw["flow"] = [
+        {"phase": "p1"},
+        {"phase": "p2", "dependencies": ["p1"]},
+        {"phase": "p3", "dependencies": ["p1"]},
+    ]
+    config = FlowConfig.model_validate(raw)
+    assert len(config.flow) == 3
+
+
+def test_flow_disjoined_graphs_ok():
+    agent = {"tools": []}
+    task = {"name": "t", "prompt": "Do it."}
+    raw = {
+        "agents": {"writer": agent},
+        "phases": {
+            "p1": {"agent": "writer", "tasks": [task]},
+            "p2": {"agent": "writer", "tasks": [task]},
+            "p3": {"agent": "writer", "tasks": [task]},
+            "p4": {"agent": "writer", "tasks": [task]},
+        },
+        "flow": [
+            {"phase": "p1"},
+            {"phase": "p2", "dependencies": ["p1"]},
+            {"phase": "p3"},
+            {"phase": "p4", "dependencies": ["p3"]},
+        ],
+    }
+    config = FlowConfig.model_validate(raw)
+    assert len(config.flow) == 4
+
+
+def test_flow_config_self_dependency_raises():
+    raw = _minimal_config()
+    raw["flow"] = [{"phase": "p1", "dependencies": ["p1"]}]
+    with pytest.raises(ValidationError, match="Cycle"):
+        FlowConfig.model_validate(raw)
+
+
+def test_flow_config_two_independent_cycles_reports_both():
+    agent = {"tools": []}
+    task = {"name": "t", "prompt": "Do it."}
+    raw = {
+        "agents": {"writer": agent},
+        "phases": {
+            "p1": {"agent": "writer", "tasks": [task]},
+            "p2": {"agent": "writer", "tasks": [task]},
+            "p3": {"agent": "writer", "tasks": [task]},
+            "p4": {"agent": "writer", "tasks": [task]},
+        },
+        "flow": [
+            # dependency edges: p1→p3→p2→p1 and p1→p4→p2→p1
+            {"phase": "p1", "dependencies": ["p3", "p4"]},
+            {"phase": "p2", "dependencies": ["p1"]},
+            {"phase": "p3", "dependencies": ["p2"]},
+            {"phase": "p4", "dependencies": ["p2"]},
+        ],
+    }
+    with pytest.raises(ValidationError) as exc_info:
+        FlowConfig.model_validate(raw)
+    error = str(exc_info.value)
+    assert "Cycle" in error
+    assert "p1 → p3 → p2 → p1" in error
+    assert "p1 → p4 → p2 → p1" in error