File size: 6,860 Bytes
152be26 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | """Deterministic validation for workflow dependency graphs and decision routes.
The schema layer validates the shape of workflow definitions. This module validates
relationships between nodes: unique identifiers, dependency references, decision
route targets, branch separation, self-dependencies, and acyclicity.
"""
from __future__ import annotations
from heapq import heappop, heappush
from src.schemas import NodeType, WorkflowDefinition
class WorkflowValidationError(ValueError):
"""Raised when a workflow definition is structurally invalid as a DAG."""
def validate_workflow(definition: WorkflowDefinition) -> None:
"""Validate cross-node graph and deterministic routing rules."""
_validate_unique_node_ids(definition)
_validate_dependencies(definition)
_validate_decision_routes(definition)
_topological_order(definition)
def topological_order(definition: WorkflowDefinition) -> list[str]:
"""Return a deterministic topological ordering after validating the workflow.
When more than one node is ready, node identifiers are used as a stable
tie-breaker. The ordering is deterministic for testing and the MVP's sequential
executor; it does not imply that production workflows must execute independent
nodes serially.
"""
_validate_unique_node_ids(definition)
_validate_dependencies(definition)
_validate_decision_routes(definition)
return _topological_order(definition)
def _validate_unique_node_ids(definition: WorkflowDefinition) -> None:
seen: set[str] = set()
duplicates: set[str] = set()
for node in definition.nodes:
if node.node_id in seen:
duplicates.add(node.node_id)
seen.add(node.node_id)
if duplicates:
duplicate_list = ", ".join(sorted(duplicates))
raise WorkflowValidationError(f"duplicate node_id values: {duplicate_list}")
def _validate_dependencies(definition: WorkflowDefinition) -> None:
node_ids = {node.node_id for node in definition.nodes}
for node in definition.nodes:
for dependency_id in node.depends_on:
if dependency_id == node.node_id:
raise WorkflowValidationError(
f"node '{node.node_id}' cannot depend on itself"
)
if dependency_id not in node_ids:
raise WorkflowValidationError(
f"node '{node.node_id}' depends on unknown node "
f"'{dependency_id}'"
)
def _validate_decision_routes(definition: WorkflowDefinition) -> None:
node_by_id = {node.node_id: node for node in definition.nodes}
dependents: dict[str, set[str]] = {node.node_id: set() for node in definition.nodes}
for node in definition.nodes:
for dependency_id in node.depends_on:
dependents[dependency_id].add(node.node_id)
for decision in definition.nodes:
if decision.node_type is not NodeType.DECISION:
continue
route_targets = set(decision.routes.values())
for route_label, target_node_id in decision.routes.items():
target = node_by_id.get(target_node_id)
if target is None:
raise WorkflowValidationError(
f"DECISION node '{decision.node_id}' route '{route_label}' targets "
f"unknown node '{target_node_id}'"
)
if target_node_id == decision.node_id:
raise WorkflowValidationError(
f"DECISION node '{decision.node_id}' cannot route to itself"
)
if decision.node_id not in target.depends_on:
raise WorkflowValidationError(
f"route target '{target_node_id}' must directly depend on "
f"DECISION node '{decision.node_id}'"
)
direct_dependents = dependents[decision.node_id]
if direct_dependents != route_targets:
unrouted = sorted(direct_dependents - route_targets)
extra = sorted(route_targets - direct_dependents)
details: list[str] = []
if unrouted:
details.append(f"unrouted dependents: {', '.join(unrouted)}")
if extra:
details.append(f"invalid route targets: {', '.join(extra)}")
detail_text = "; ".join(details)
raise WorkflowValidationError(
f"DECISION node '{decision.node_id}' must route all direct dependents"
+ (f" ({detail_text})" if detail_text else "")
)
branch_reach: dict[str, set[str]] = {}
for target_node_id in sorted(route_targets):
reachable = _reachable_nodes(target_node_id, dependents)
for other_target, other_reachable in branch_reach.items():
overlap = sorted(reachable & other_reachable)
if overlap:
raise WorkflowValidationError(
f"DECISION node '{decision.node_id}' branch reconvergence is "
f"not supported; routes through '{other_target}' and "
f"'{target_node_id}' both reach: {', '.join(overlap)}"
)
branch_reach[target_node_id] = reachable
def _reachable_nodes(start_node_id: str, dependents: dict[str, set[str]]) -> set[str]:
reachable: set[str] = set()
stack = [start_node_id]
while stack:
node_id = stack.pop()
if node_id in reachable:
continue
reachable.add(node_id)
stack.extend(sorted(dependents[node_id], reverse=True))
return reachable
def _topological_order(definition: WorkflowDefinition) -> list[str]:
"""Use Kahn's algorithm to produce an order or reject a cyclic graph."""
indegree = {node.node_id: len(node.depends_on) for node in definition.nodes}
dependents: dict[str, list[str]] = {node.node_id: [] for node in definition.nodes}
for node in definition.nodes:
for dependency_id in node.depends_on:
dependents[dependency_id].append(node.node_id)
ready: list[str] = []
for node_id, degree in indegree.items():
if degree == 0:
heappush(ready, node_id)
order: list[str] = []
while ready:
node_id = heappop(ready)
order.append(node_id)
for dependent_id in sorted(dependents[node_id]):
indegree[dependent_id] -= 1
if indegree[dependent_id] == 0:
heappush(ready, dependent_id)
if len(order) != len(definition.nodes):
blocked = sorted(node_id for node_id, degree in indegree.items() if degree > 0)
blocked_list = ", ".join(blocked)
raise WorkflowValidationError(
f"workflow dependency graph contains a cycle; blocked nodes: {blocked_list}"
)
return order
|