Schema-Aware Path Pruning

The problem

Complex task workflows often waste time and resources by attempting to execute paths that are impossible to complete due to data mismatches. This creates unnecessary overhead in nested systems.

What it does

It looks at a map of tasks and automatically removes any paths that use the wrong data types. It filters out these invalid routes before the system even tries to run them.

Why it matters

It ensures that only logically valid paths are processed, preventing errors and wasted effort in complex workflows.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 schema_aware_path_pruner_v2.py
Found 0 valid paths:

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 97 lines, one file, standard library only.

# Schema-Aware Path Pruning Implementation v2

class Node:
    def __init__(self, name, dtype, weight=0.0):
        self.name = name
        self.dtype = dtype
        self.children = {}
        self.parent = None
        self.path_weight = weight
        self.reachable = False

    def add_child(self, edge_type, child):
        self.children[edge_type] = child
        child.parent = self

class PathPruner:
    def __init__(self, root_node):
        self.root = root_node

    def validate_path(self, path):
        """ MLIR-like structural constraint checking """
        for i in range(1, len(path)):
            current = path[i-1]
            next_node = path[i]
            if current.dtype != next_node.dtype:
                return False
        return True

    def recursive_traversal(self, node, current_path, valid_paths):
        current_path.append(node)
        # Add path weight calculation
        if node.name == 'target':
            if self.validate_path(current_path):
                total_weight = sum(node.path_weight for node in current_path)
                path_length = len(current_path) - 1
                valid_paths.append({
                    'path': current_path.copy(),
                    'weight': total_weight,
                    'length': path_length
                })
        for edge_type, child in node.children.items():
            self.recursive_traversal(child, current_path, valid_paths)
        current_path.pop()

    def prune_paths(self, max_length=None, min_weight=None, max_weight=None):
        valid_paths = []
        self.recursive_traversal(self.root, [], valid_paths)
        filtered = []
        for path_info in valid_paths:
            path = path_info['path']
            weight = path_info['weight']
            length = path_info['length']
            
            # Apply length filter
            if max_length is not None and length > max_length:
                continue
            
            # Apply min weight filter
            if min_weight is not None and weight < min_weight:
                continue
            
            # Apply max weight filter
            if max_weight is not None and weight > max_weight:
                continue
            
            filtered.append(path_info)
        return filtered

# Example Usage
if __name__ == "__main__":
    # Create sample task graph with weights
    root = Node('root', 'string', 1.0)
    process1 = Node('process1', 'int', 2.0)
    process2 = Node('process2', 'bool', 3.0)
    final_node = Node('target', 'string', 4.0)

    root.add_child('edge1', process1)
    process1.add_child('edge2', final_node)
    process1.add_child('invalid_edge', process2)

    pruner = PathPruner(root)

    # Original pruning
    valid_paths = pruner.prune_paths()
    print(f"Found {len(valid_paths)} valid paths (original):")
    for i, path in enumerate(valid_paths):
        print(f"Path {i+1}: '->'.join([n.name for n in path])")

    # New pruning with filters
    filtered_paths = pruner.prune_paths(
        max_length=2,  # Allow max 2 edges
        min_weight=5.0
    )
    print(f"\nFound {len(filtered_paths)} paths with length <=2 edges and weight >=5.0:")
    for i, path_info in enumerate(filtered_paths):
        total_weight = sum(n.path_weight for n in path)
        print(f"Path {i+1}: '->'.join([n.name for n in path]) (Weight: {total_weight})")