[feat][GraphWorkflow] Add validate() method for pre-run graph correctness checks

Summary

GraphWorkflow has no explicit validate() method. Misconfigured graphs (disconnected nodes, missing entry points, cycles in DAG-only workflows, unreachable nodes) are only discovered at run() time — often mid-execution — rather than at build time.

Proposed Enhancement

Add a validate() -> List[str] method that runs all structural checks and returns a list of error/warning messages without executing any agents:

def validate(self, raise_on_error: bool = True) -> List[str]:
    issues = []

    # 1. Check for disconnected nodes (nodes with no edges)
    for node_id in self.nodes:
        if self.graph_backend.in_degree(node_id) == 0 and self.graph_backend.out_degree(node_id) == 0:
            issues.append(f"Node '{node_id}' is disconnected (no edges)")

    # 2. Check entry points exist and are valid
    if not self.entry_points:
        issues.append("No entry points defined or auto-detectable")
    for ep in self.entry_points:
        if ep not in self.nodes:
            issues.append(f"Entry point '{ep}' does not exist in nodes")

    # 3. Check end points exist and are valid
    if not self.end_points:
        issues.append("No end points defined or auto-detectable")

    # 4. Detect cycles (for workflows that expect DAG topology)
    cycles = self.graph_backend.simple_cycles()
    if cycles:
        issues.append(f"Graph contains {len(cycles)} cycle(s): {cycles}")

    # 5. Check for unreachable nodes (not reachable from any entry point)
    reachable = set()
    for ep in self.entry_points:
        reachable.add(ep)
        reachable.update(self.graph_backend.descendants(ep))
    unreachable = set(self.nodes) - reachable
    if unreachable:
        issues.append(f"Unreachable nodes: {unreachable}")

    if raise_on_error and issues:
        raise ValueError(f"GraphWorkflow validation failed:\n" + "\n".join(f"  - {i}" for i in issues))

    return issues

Usage

wf = GraphWorkflow(...)
wf.add_node(agent_a)
wf.add_node(agent_b)
wf.add_edge(agent_a, agent_b)

issues = wf.validate(raise_on_error=False)
if issues:
    print("Workflow has issues:", issues)
else:
    wf.run(task="...")

Integration

• Call validate() automatically inside compile() so misconfiguration is always caught before execution.
• Expose as a public API so users can call it after building a workflow and before deploying.

File

swarms/structs/graph_workflow.py