test: validate multica workflow reachability

backend-fastapi/scripts/seed_multica_org_agents.py

@@ -46,6 +46,59 @@ def build_agent_payload(agent_fixture: Dict[str, Any], workflow_id: str | None)
     return payload
 
 
+def _find_parallel_merge(
+    node_by_id: Dict[str, Dict[str, Any]],
+    outgoing_edges: Dict[str, list[str]],
+    branch_starts: list[str],
+) -> str | None:
+    branch_merges: list[str] = []
+    for start in branch_starts:
+        current = start
+        seen: set[str] = set()
+        while current and current not in seen:
+            seen.add(current)
+            next_nodes = outgoing_edges.get(current, [])
+            if not next_nodes or len(next_nodes) > 1:
+                break
+            current = next_nodes[0]
+            if (node_by_id.get(current) or {}).get("type") == "merge":
+                branch_merges.append(current)
+                break
+    unique_merges = set(branch_merges)
+    return branch_merges[0] if len(unique_merges) == 1 and len(branch_merges) == len(branch_starts) else None
+
+
+def _walk_workflow_nodes(
+    node_by_id: Dict[str, Dict[str, Any]],
+    outgoing_edges: Dict[str, list[str]],
+    start_id: str,
+) -> set[str]:
+    visited: set[str] = set()
+    current = start_id
+    while current and current not in visited:
+        visited.add(current)
+        node_type = (node_by_id.get(current) or {}).get("type")
+        if node_type == "end":
+            break
+        if node_type == "parallel":
+            branch_starts = outgoing_edges.get(current, [])
+            for branch_start in branch_starts:
+                branch_current = branch_start
+                branch_seen: set[str] = set()
+                while branch_current and branch_current not in branch_seen:
+                    if (node_by_id.get(branch_current) or {}).get("type") == "merge":
+                        break
+                    visited.add(branch_current)
+                    branch_seen.add(branch_current)
+                    branch_next_nodes = outgoing_edges.get(branch_current, [])
+                    branch_current = branch_next_nodes[0] if branch_next_nodes else None
+            current = _find_parallel_merge(node_by_id, outgoing_edges, branch_starts)
+            continue
+        next_nodes = outgoing_edges.get(current, [])
+        current = next_nodes[0] if next_nodes else None
+    return visited
+
+
 def validate_fixture(fixture: Dict[str, Any]) -> Dict[str, int]:
     agents = fixture.get("agents", [])
     definition = fixture.get("workflow", {}).get("definition", {})
@@ -77,6 +130,7 @@ def validate_fixture(fixture: Dict[str, Any]) -> Dict[str, int]:
     }
     required_node_types = {"start", "end", "template", "parallel", "merge"}
     multi_outgoing_node_types = {"parallel", "condition", "intent", "choice"}
+    start_node = next((node for node in nodes if node.get("type") == "start"), None)
 
     missing_workflow_agents = sorted(workflow_agent_codes - agent_codes)
     missing_persona_fields = {
@@ -91,6 +145,14 @@ def validate_fixture(fixture: Dict[str, Any]) -> Dict[str, int]:
         if len(targets) > 1
         and (node_by_id.get(source) or {}).get("type") not in multi_outgoing_node_types
     }
+    reachable_nodes = _walk_workflow_nodes(node_by_id, outgoing_edges, start_node.get("id")) if start_node else set()
+    unreachable_nodes = sorted(set(node_by_id) - reachable_nodes)
+    invalid_parallel_merges = {
+        source: targets
+        for source, targets in outgoing_edges.items()
+        if (node_by_id.get(source) or {}).get("type") == "parallel"
+        and not _find_parallel_merge(node_by_id, outgoing_edges, targets)
+    }
     project_manager = next(
         (agent for agent in agents if agent.get("code") == "project_manager"),
         None,
@@ -101,6 +163,8 @@ def validate_fixture(fixture: Dict[str, Any]) -> Dict[str, int]:
         or missing_persona_fields
         or missing_node_types
         or invalid_multi_outgoing
+        or unreachable_nodes
+        or invalid_parallel_merges
         or project_manager_workflow != "multica_org_collaboration_flow"
     ):
         raise ValueError(
@@ -108,6 +172,8 @@ def validate_fixture(fixture: Dict[str, Any]) -> Dict[str, int]:
             f"missing_persona_fields={missing_persona_fields}, "
             f"missing_node_types={missing_node_types}, "
             f"invalid_multi_outgoing={invalid_multi_outgoing}, "
+            f"unreachable_nodes={unreachable_nodes}, "
+            f"invalid_parallel_merges={invalid_parallel_merges}, "
             f"project_manager_workflow={project_manager_workflow}"
         )