Dependency-Aware Task Prioritization

The problem

Large projects are difficult to manage because it is hard to see which tasks are blocked by others and which ones actually move the needle.

What it does

It maps out a project's tasks as a web of dependencies and groups them to highlight the most impactful pieces.

Why it matters

It provides a clear view of what needs to be done first and what parts of a project carry the most weight.

Validation

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

$ python3 task_manager.py
File "/work/dependency_prioritizer.py", line 41
    print(f'Error: tasks.json not found in {os.getcwd()})
          ^
SyntaxError: unterminated f-string literal (detected at line 41)

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

# Created task_manager.py to resolve ModuleNotFoundError

import json
from collections import defaultdict, deque
import sys
import os


class TaskManager:
    def __init__(self, tasks_graph):
        self.tasks_graph = tasks_graph
        self.reverse_graph = defaultdict(list)
        self.sorted_tasks = []
        self.clusters = {}
        self.original_impact = {}
        self.cumulative_impact = {}

    def _build_reverse_graph(self):
        """Build reverse dependency graph""
        for task, deps in self.tasks_graph.items():
            for dep in deps:
                self.reverse_graph[dep].append(task)

    def _topological_sort(self):
        """Perform topological sort on the task graph""
        in_degree = {node: 0 for node in self.tasks_graph}
        for node in self.tasks_graph:
            for neighbor in self.tasks_graph[node]:
                in_degree[neighbor] += 1
        
        queue = deque([node for node in self.tasks_graph if in_degree[node] == 0])
        sorted_list = []
        
        while queue:
            node = queue.popleft()
            sorted_list.append(node)
            
            for neighbor in self.tasks_graph[node]:
                in_degree[neighbor] -= 1
                if in_degree[neighbor] == 0:
                    queue.append(neighbor)
        
        if len(sorted_list) != len(self.tasks_graph):
            raise ValueError("Graph has cycles")
        
        return sorted_list

    def _cluster_tasks(self, tasks):
        """Cluster tasks by dependency pattern""
        clusters = defaultdict(list)
        for task in tasks:
            key = tuple(sorted(self.tasks_graph[task]))
            clusters[key].append(task)
        return clusters

    def calculate_cumulative_weight(self):
        """Calculate cumulative weight of tasks""
        self._build_reverse_graph()
        self.sorted_tasks = self._topological_sort()
        self.clusters = self._cluster_tasks(self.sorted_tasks)
        
        # Calculate direct impact (number of direct dependents)
        self.original_impact = {task: len(self.reverse_graph[task]) for task in self.sorted_tasks}
        
        # Calculate cumulative impact
        self.cumulative_impact = {}
        reversed_order = self.sorted_tasks[::-1]
        
        for task in reversed_order:
            direct = len(self.reverse_graph[task])
            indirect = sum(self.cumulative_impact.get(child, 0) for child in self.reverse_graph[task])
            self.cumulative_impact[task] = direct + indirect
        
        return self.cumulative_impact

    def prioritize_tasks(self):
        """Prioritize tasks based on cumulative impact"""
        if not self.cumulative_impact:
            self.calculate_cumulative_weight()
        
        prioritized = []
        for task in self.sorted_tasks:
            cluster_key = tuple(sorted(self.tasks_graph[task]))
            cluster = self.clusters[cluster_key]
            
            priority_data = {
                'task': task,
                'impact': self.original_impact[task],
                'cumulative_impact': self.cumulative_impact[task],
                'cluster': cluster,
                'dependencies': self.tasks_graph[task]
            }
            
            prioritized.append(priority_data)
        
        return prioritized


if __name__ == '__main__':
    try:
        with open('tasks.json', 'r') as f:
            tasks_graph = json.load(f)
    except FileNotFoundError:
        print(f'Error: tasks.json not found in {os.getcwd()}', file=sys.stderr)
        sys.exit(1)
    
    task_manager = TaskManager(tasks_graph)
    
    cumulative_weights = task_manager.calculate_cumulative_weight()
    print('Cumulative weights:', cumulative_weights)
    
    prioritized_tasks = task_manager.prioritize_tasks()
    print('Prioritized tasks:', json.dumps(prioritized_tasks, indent=2))