[FEATURE] Add Graph constructor option to toggle edge execution logic from OR to AND

[dbschmigelski]

Problem Statement

Currently, Strands Graph applies 'OR' logic to incoming edges, meaning that when multiple nodes (A, B, C) are connected to a single node (Z), node Z starts execution as soon as ANY of the predecessor nodes complete, rather than waiting for ALL of them to complete. This makes it difficult or impossible to create workflows where a node must wait for multiple prerequisite tasks to finish before proceeding. Users currently have to implement complex workarounds using conditional edges to achieve AND logic behavior, which requires additional boilerplate code and manual tracking of predecessor completion states.

Proposed Solution

Add a configuration parameter to the Graph constructor (e.g., edge_execution_mode="and" or default_edge_logic="and") that allows users to toggle between OR logic (current default behavior) and AND logic globally for the entire graph.

With this toggle at the Graph constructor level, users could simply set Graph(edge_execution_mode="and") to change the default behavior for all nodes in the graph, eliminating the need for complex conditional edge implementations while maintaining backward compatibility with the current OR logic as the default.

Use Case

No response

Alternatives Solutions

No response

Additional Context

No response

[dboyd13]

I'd like to see this too. For now here is the code I use to achieve the same using conditional edges:

def create_dependency_condition(required_nodes):
    """
    Factory function to create dependency checking conditions.
    
    Args:
        required_nodes: List of node IDs that must be completed before the target node executes
        
    Returns:
        Function that checks if all required dependencies are complete
    """
    def check_dependencies(state):
        """Check if all required dependencies are complete."""
        
        # Check if state has results attribute
        if not hasattr(state, 'results') or state.results is None:
            return False
            
        for node_id in required_nodes:
            if node_id not in state.results:
                return False
            
            node_result = state.results[node_id]
            
            # Check if result exists and has completed status
            if not node_result:
                return False
                
            # Check status - use Status enum
            if hasattr(node_result, 'status'):
                if node_result.status != Status.COMPLETED:
                    return False
        
        return True
    
    return check_dependencies

Then:

my_condition = create_dependency_condition([
        "node_a", 
        "node_b", 
        "node_c"
    ])

builder.add_edge("node_a", "node_z", condition=my_condition)
builder.add_edge("node_b", "node_z", condition=my_condition)
builder.add_edge("node_c", "node_z", condition=my_condition)