Add process topology validation: Johnson's algorithm, cycle detection, and validation utilities

Copilot · toby-coleman · Copilot · commit 26c0cb5577ea · 2026-02-15T20:07:00.000Z
- Add _graph.py with Johnson's cycle-finding algorithm (no additional dependencies)
- Add _validation.py with validation utilities for process topology:
  - validate_all_inputs_connected: check all component inputs are connected
  - validate_input_events: check input events have matching output producers
  - validate_no_unresolved_cycles: check circular connections have initial_values
- Add model validator on ProcessSpec for circular connection detection
- Export new utilities from plugboard_schemas
- Add comprehensive unit tests

Co-authored-by: toby-coleman &lt;13170610+toby-coleman@users.noreply.github.com&gt;
diff --git a/plugboard-schemas/plugboard_schemas/__init__.py b/plugboard-schemas/plugboard_schemas/__init__.py
@@ -9,6 +9,13 @@
 from importlib.metadata import version
 
 from ._common import PlugboardBaseModel
+from ._graph import simple_cycles
+from ._validation import (
+    ValidationError,
+    validate_all_inputs_connected,
+    validate_input_events,
+    validate_no_unresolved_cycles,
+)
 from .component import ComponentArgsDict, ComponentArgsSpec, ComponentSpec, Resource
 from .config import ConfigSpec, ProcessConfigSpec
 from .connector import (
@@ -85,4 +92,9 @@
     "TuneArgsDict",
     "TuneArgsSpec",
     "TuneSpec",
+    "ValidationError",
+    "simple_cycles",
+    "validate_all_inputs_connected",
+    "validate_input_events",
+    "validate_no_unresolved_cycles",
 ]
diff --git a/plugboard-schemas/plugboard_schemas/_graph.py b/plugboard-schemas/plugboard_schemas/_graph.py
@@ -0,0 +1,126 @@
+"""Graph algorithms for topology validation.
+
+Implements Johnson's algorithm for finding all simple cycles in a directed graph,
+along with helper functions for strongly connected components.
+
+References:
+    Donald B Johnson. "Finding all the elementary circuits of a directed graph."
+    SIAM Journal on Computing. 1975.
+"""
+
+from collections import defaultdict
+from collections.abc import Generator
+
+
+def simple_cycles(graph: dict[str, set[str]]) -> Generator[list[str], None, None]:
+    """Find all simple cycles in a directed graph using Johnson's algorithm.
+
+    Args:
+        graph: A dictionary mapping each vertex to a set of its neighbours.
+
+    Yields:
+        Each elementary cycle as a list of vertices.
+    """
+    graph = {v: set(nbrs) for v, nbrs in graph.items()}
+    sccs = _strongly_connected_components(graph)
+    while sccs:
+        scc = sccs.pop()
+        startnode = scc.pop()
+        path = [startnode]
+        blocked: set[str] = set()
+        closed: set[str] = set()
+        blocked.add(startnode)
+        B: dict[str, set[str]] = defaultdict(set)
+        stack: list[tuple[str, list[str]]] = [(startnode, list(graph[startnode]))]
+        while stack:
+            thisnode, nbrs = stack[-1]
+            if nbrs:
+                nextnode = nbrs.pop()
+                if nextnode == startnode:
+                    yield path[:]
+                    closed.update(path)
+                elif nextnode not in blocked:
+                    path.append(nextnode)
+                    stack.append((nextnode, list(graph[nextnode])))
+                    closed.discard(nextnode)
+                    blocked.add(nextnode)
+                    continue
+            if not nbrs:
+                if thisnode in closed:
+                    _unblock(thisnode, blocked, B)
+                else:
+                    for nbr in graph[thisnode]:
+                        if thisnode not in B[nbr]:
+                            B[nbr].add(thisnode)
+                stack.pop()
+                path.pop()
+        _remove_node(graph, startnode)
+        H = _subgraph(graph, set(scc))
+        sccs.extend(_strongly_connected_components(H))
+
+
+def _unblock(thisnode: str, blocked: set[str], B: dict[str, set[str]]) -> None:
+    """Unblock a node and recursively unblock nodes in its B set."""
+    stack = {thisnode}
+    while stack:
+        node = stack.pop()
+        if node in blocked:
+            blocked.remove(node)
+            stack.update(B[node])
+            B[node].clear()
+
+
+def _strongly_connected_components(graph: dict[str, set[str]]) -> list[set[str]]:
+    """Find all strongly connected components using Tarjan's algorithm.
+
+    Args:
+        graph: A dictionary mapping each vertex to a set of its neighbours.
+
+    Returns:
+        A list of sets, each containing the vertices of a strongly connected component.
+    """
+    index_counter = [0]
+    stack: list[str] = []
+    lowlink: dict[str, int] = {}
+    index: dict[str, int] = {}
+    result: list[set[str]] = []
+
+    def _strong_connect(node: str) -> None:
+        index[node] = index_counter[0]
+        lowlink[node] = index_counter[0]
+        index_counter[0] += 1
+        stack.append(node)
+
+        for successor in graph.get(node, set()):
+            if successor not in index:
+                _strong_connect(successor)
+                lowlink[node] = min(lowlink[node], lowlink[successor])
+            elif successor in stack:
+                lowlink[node] = min(lowlink[node], index[successor])
+
+        if lowlink[node] == index[node]:
+            connected_component: set[str] = set()
+            while True:
+                successor = stack.pop()
+                connected_component.add(successor)
+                if successor == node:
+                    break
+            result.append(connected_component)
+
+    for node in graph:
+        if node not in index:
+            _strong_connect(node)
+
+    return result
+
+
+def _remove_node(graph: dict[str, set[str]], target: str) -> None:
+    """Remove a node and all its edges from the graph."""
+    del graph[target]
+    for nbrs in graph.values():
+        nbrs.discard(target)
+
+
+def _subgraph(graph: dict[str, set[str]], vertices: set[str]) -> dict[str, set[str]]:
+    """Get the subgraph induced by a set of vertices."""
+    return {v: graph[v] & vertices for v in vertices}
diff --git a/plugboard-schemas/plugboard_schemas/_validation.py b/plugboard-schemas/plugboard_schemas/_validation.py
@@ -0,0 +1,178 @@
+"""Validation utilities for `ProcessSpec` objects.
+
+Provides functions to validate process topology including:
+- Checking that all component inputs are connected
+- Checking that input events have matching output event producers
+- Checking for circular connections that require initial values
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+import typing as _t
+
+from ._graph import simple_cycles
+
+
+if _t.TYPE_CHECKING:
+    from .component import ComponentSpec
+    from .connector import ConnectorSpec
+
+
+class ValidationError(Exception):
+    """Raised when a process specification fails validation."""
+
+    pass
+
+
+def _build_component_graph(
+    connectors: list[ConnectorSpec],
+) -> dict[str, set[str]]:
+    """Build a directed graph of component connections from connector specs.
+
+    Args:
+        connectors: List of connector specifications.
+
+    Returns:
+        A dictionary mapping source component names to sets of target component names.
+    """
+    graph: dict[str, set[str]] = defaultdict(set)
+    for conn in connectors:
+        source_entity = conn.source.entity
+        target_entity = conn.target.entity
+        if source_entity != target_entity:
+            graph[source_entity].add(target_entity)
+            # Ensure target is in graph even with no outgoing edges
+            if target_entity not in graph:
+                graph[target_entity] = set()
+    return dict(graph)
+
+
+def _get_edges_in_cycle(
+    cycle: list[str],
+    connectors: list[ConnectorSpec],
+) -> list[ConnectorSpec]:
+    """Get all connector specs that form edges within a cycle.
+
+    Args:
+        cycle: List of component names forming a cycle.
+        connectors: All connector specifications.
+
+    Returns:
+        List of connector specs that are part of the cycle.
+    """
+    cycle_edges: list[ConnectorSpec] = []
+    cycle_set = set(cycle)
+    for i, node in enumerate(cycle):
+        next_node = cycle[(i + 1) % len(cycle)]
+        for conn in connectors:
+            if conn.source.entity == node and conn.target.entity == next_node:
+                cycle_edges.append(conn)
+    return [c for c in cycle_edges if c.source.entity in cycle_set and c.target.entity in cycle_set]
+
+
+def validate_all_inputs_connected(
+    components: dict[str, dict[str, _t.Any]],
+    connectors: list[ConnectorSpec],
+) -> list[str]:
+    """Check that all component inputs are connected.
+
+    Args:
+        components: Dictionary mapping component names to their IO info.
+            Each value must have an ``"inputs"`` key with a list of input field names.
+        connectors: List of connector specifications.
+
+    Returns:
+        List of error messages for unconnected inputs.
+    """
+    # Build mapping of which component inputs are connected
+    connected_inputs: dict[str, set[str]] = defaultdict(set)
+    for conn in connectors:
+        target_name = conn.target.entity
+        target_field = conn.target.descriptor
+        connected_inputs[target_name].add(target_field)
+
+    errors: list[str] = []
+    for comp_name, comp_info in components.items():
+        all_inputs = set(comp_info.get("inputs", []))
+        connected = connected_inputs.get(comp_name, set())
+        unconnected = all_inputs - connected
+        if unconnected:
+            errors.append(f"Component '{comp_name}' has unconnected inputs: {sorted(unconnected)}")
+    return errors
+
+
+def validate_input_events(
+    components: dict[str, dict[str, _t.Any]],
+) -> list[str]:
+    """Check that all components with input events have a matching output event producer.
+
+    Args:
+        components: Dictionary mapping component names to their IO info.
+            Each value must have ``"input_events"`` and ``"output_events"`` keys
+            with lists of event type strings.
+
+    Returns:
+        List of error messages for unmatched input events.
+    """
+    # Collect all output event types across all components
+    all_output_events: set[str] = set()
+    for comp_info in components.values():
+        all_output_events.update(comp_info.get("output_events", []))
+
+    errors: list[str] = []
+    for comp_name, comp_info in components.items():
+        input_events = set(comp_info.get("input_events", []))
+        unmatched = input_events - all_output_events
+        if unmatched:
+            errors.append(
+                f"Component '{comp_name}' has input events with no producer: {sorted(unmatched)}"
+            )
+    return errors
+
+
+def validate_no_unresolved_cycles(
+    components: list[ComponentSpec],
+    connectors: list[ConnectorSpec],
+) -> list[str]:
+    """Check for circular connections that are not resolved by initial values.
+
+    Circular loops are only valid if there are ``initial_values`` set on an
+    appropriate component input within the loop.
+
+    Args:
+        components: List of component specifications.
+        connectors: List of connector specifications.
+
+    Returns:
+        List of error messages for unresolved circular connections.
+    """
+    graph = _build_component_graph(connectors)
+    if not graph:
+        return []
+
+    # Build lookup of component initial_values by name
+    initial_values_by_comp: dict[str, set[str]] = {}
+    for comp in components:
+        if comp.args.initial_values:
+            initial_values_by_comp[comp.args.name] = set(comp.args.initial_values.keys())
+
+    errors: list[str] = []
+    for cycle in simple_cycles(graph):
+        # Check if any edge in the cycle targets a component input with initial_values
+        cycle_edges = _get_edges_in_cycle(cycle, connectors)
+        cycle_resolved = False
+        for edge in cycle_edges:
+            target_comp = edge.target.entity
+            target_field = edge.target.descriptor
+            if target_comp in initial_values_by_comp:
+                if target_field in initial_values_by_comp[target_comp]:
+                    cycle_resolved = True
+                    break
+        if not cycle_resolved:
+            cycle_str = " -> ".join(cycle + [cycle[0]])
+            errors.append(
+                f"Circular connection detected without initial values: {cycle_str}. "
+                f"Set initial_values on a component input within the loop to resolve."
+            )
+    return errors
diff --git a/plugboard-schemas/plugboard_schemas/process.py b/plugboard-schemas/plugboard_schemas/process.py
@@ -7,6 +7,7 @@
 from typing_extensions import Self
 
 from ._common import PlugboardBaseModel
+from ._validation import validate_no_unresolved_cycles
 from .component import ComponentSpec
 from .connector import DEFAULT_CONNECTOR_CLS_PATH, ConnectorBuilderSpec, ConnectorSpec
 from .state import DEFAULT_STATE_BACKEND_CLS_PATH, RAY_STATE_BACKEND_CLS_PATH, StateBackendSpec
@@ -77,6 +78,14 @@ def _set_default_state_backend(self: Self) -> Self:
             self.args.state.type = RAY_STATE_BACKEND_CLS_PATH
         return self
 
+    @model_validator(mode="after")
+    def _validate_no_unresolved_cycles(self: Self) -> Self:
+        """Validate that circular connections have initial_values set."""
+        errors = validate_no_unresolved_cycles(self.args.components, self.args.connectors)
+        if errors:
+            raise ValueError("\n".join(errors))
+        return self
+
     @field_validator("type", mode="before")
     @classmethod
     def _validate_type(cls, value: _t.Any) -> str:
diff --git a/tests/unit/test_process_validation.py b/tests/unit/test_process_validation.py
