Critical Path

The problem

It is difficult to see which specific steps in a complex data flow carry the most risk. Identifying these points is hard when looking at a large web of interconnected data.

What it does

It traces every possible path from start to finish and identifies the sequence of steps with the highest total risk. It then highlights the specific points in that path that are most critical.

Why it matters

It allows you to pinpoint exactly where to focus your attention to prevent data errors.

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 critical_path.py
Critical Path: ['A', 'C', 'D']
Total Weight: 8

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 — 55 lines, one file, standard library only.

# Critical Path Finder in Data Lineage Graph
import sys

def main():
    # Example graph definition (modify as needed)
    graph = {
        'A': [('B', 2), ('C', 3)],
        'B': [('D', 1)],
        'C': [('D', 5)],
        'D': []
    }
    source = 'A'
    sink = 'D'

    all_paths = find_all_paths(graph, source, sink)
    max_path = None
    max_sum = float('-inf')

    for path in all_paths:
        current_sum = sum_weights(graph, path)
        if current_sum > max_sum:
            max_sum = current_sum
            max_path = path

    print(f"Critical Path: {max_path}")
    print(f"Total Weight: {max_sum}")

def find_all_paths(graph, start, end, path=[]):
    """Recursively find all paths from start to end"""
    path = path + [start]
    if start == end:
        return [path]
    paths = []
    if start in graph:
        for neighbor, _ in graph[start]:
            if neighbor not in path:
                new_paths = find_all_paths(graph, neighbor, end, path)
                for p in new_paths:
                    paths.append(p)
    return paths

def sum_weights(graph, path):
    """Calculate total weight of a path"""
    total = 0
    for i in range(len(path)-1):
        current_node = path[i]
        next_node = path[i+1]
        for neighbor, weight in graph.get(current_node, []):
            if neighbor == next_node:
                total += weight
                break
    return total

if __name__ == "__main__":
    main()