Graph-Based Dependency Bottleneck

The problem

Complex projects often have hidden single points of failure where one small broken part can bring down everything else. It is difficult to see these bottlenecks just by looking at a list of tasks.

What it does

It maps out how different parts of a project depend on each other and scores each piece to find the most critical bottlenecks. It highlights exactly which parts are the most risky to break.

Why it matters

It identifies the specific parts of a project that need the most protection to prevent a total system failure.

Validation

It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.

$ python3 bottleneck_analyzer_v2.py
Traceback (most recent call last):
  File "/work/bottleneck_analyzer.py", line 66, in <module>
    main()
  File "/work/bottleneck_analyzer.py", line 10, in main
    raise ValueError("No input provided")
ValueError: No input provided

No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.

The code

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

# Graph-Based Dependency Bottleneck Analyzer with Critical Path Impact
import sys
from collections import defaultdict, deque

def main():
    input_str = sys.stdin.read().strip()
    if not input_str:
        # Hard-coded example input for demonstration
        input_str = "Nodes: A,B,C,D;Edges: A->B;A->C;B->D;C->D"
    parts = input_str.split(';')
    
    # Parse nodes and edges
    nodes = parts[0].split(',')
    edges = []
    for edge in parts[1:].split(';'):
        if '->' in edge:
            u, v = edge.strip().split('->')
            edges.append((u.strip(), v.strip()))
    
    # Build graph and calculate in-degrees
    graph = defaultdict(list)
    in_degree = defaultdict(int)
    for u, v in edges:
        graph[u].append(v)
        in_degree[v] += 1
    
    # Topological sorting
    queue = deque([node for node in graph if in_degree[node] == 0])
    topo_order = []
    while queue:
        node = queue.popleft()
        topo_order.append(node)
        for neighbor in graph[node]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)
    
    # Calculate incoming path counts
    incoming = defaultdict(int)
    for node in topo_order:
        if in_degree[node] == 0:  # Source node
            incoming[node] = 1
        else:
            incoming[node] = sum(incoming[pred] for pred in get_predecessors(graph, node))
    
    # Reverse graph for outgoing paths
    reverse_graph = defaultdict(list)
    for u in graph:
        for v in graph[u]:
            reverse_graph[v].append(u)
    
    # Calculate outgoing path counts
    outgoing = defaultdict(int)
    for node in reversed(topo_order):
        if not graph[node]:  # Sink node
            outgoing[node] = 1
        else:
            outgoing[node] = sum(outgoing[succ] for succ in graph[node])
    
    # Calculate flow concentration scores
    flow_scores = {node: incoming[node] * outgoing[node] for node in graph}
    
    # Calculate Critical Path Impact (downstream reachability)
    reachability = defaultdict(int)
    for node in graph:
        visited = set()
        queue = deque([node])
        visited.add(node)
        while queue:
            current = queue.popleft()
            for neighbor in graph.get(current, [])[::-1]:  # Reverse to maintain order
                if neighbor not in visited:
                    visited.add(neighbor)
                    queue.append(neighbor)
        reachability[node] = len(visited)
    
    # Identify top bottlenecks
    bottlenecks = sorted(flow_scores.items(), key=lambda x: -x[1])[:3]
    
    # Output results
    print("Detected Dependency Bottlenecks:\n")
    for node, score in bottlenecks:
        print(f"- {node}: Flow Score {score}, Critical Path Impact {reachability[node]}")

def get_predecessors(graph, node):
    return [u for u in graph if node in graph[u]]

if __name__ == "__main__":
    main()