Build System

"""Phase 1 fixes two bugs in `DependencyGraph`:

1. `_reverse_adj` was being appended to with the *dependency's* name as
   the key — the same shape as `_adj`. The intent of `_reverse_adj` is
   "for every task, the list of tasks IT depends on", so the key has to
   be `task.name` and the value `dep`.
2. `topological_order` flipped the in-degree update. In-degree counts
   how many *prerequisites* a task has. If `B` depends on `A`, that's a
   prerequisite of `B`, so we increment `in_degree[B]`, not `in_degree[A]`.
"""

from typing import List, Dict, Set, Optional


class Task:
    """A build task with a name, duration, and dependencies."""

    def __init__(self, name: str, duration: int, deps: Optional[List[str]] = None):
        self.name = name
        self.duration = duration
        self.deps = deps or []


class DependencyGraph:
    """Manages build tasks and their dependency relationships."""

    def __init__(self):
        self._tasks: Dict[str, Task] = {}
        self._adj: Dict[str, List[str]] = {}
        self._reverse_adj: Dict[str, List[str]] = {}

    def add_task(self, task: Task) -> None:
        self._tasks[task.name] = task
        if task.name not in self._adj:
            self._adj[task.name] = []
        if task.name not in self._reverse_adj:
            self._reverse_adj[task.name] = []
        for dep in task.deps:
            if dep not in self._adj:
                self._adj[dep] = []
            if dep not in self._reverse_adj:
                self._reverse_adj[dep] = []
            self._adj[dep].append(task.name)
            # Bug 1 fix: store reverse adjacency under task.name (the
            # dependent), with the dep as the value.
            self._reverse_adj[task.name].append(dep)

    def get_task(self, name: str) -> Optional[Task]:
        return self._tasks.get(name)

    def get_dependents(self, name: str) -> List[str]:
        """Tasks that depend on the given task."""
        return self._adj.get(name, [])

    def get_dependencies(self, name: str) -> List[str]:
        """Tasks that the given task depends on."""
        return self._reverse_adj.get(name, [])

    def get_all_tasks(self) -> List[str]:
        return list(self._tasks.keys())

    def has_cycle(self) -> bool:
        visited: Set[str] = set()
        rec_stack: Set[str] = set()

        def dfs(node: str) -> bool:
            visited.add(node)
            rec_stack.add(node)
            for neighbor in self._adj.get(node, []):
                if neighbor not in visited:
                    if dfs(neighbor):
                        return True
                elif neighbor in rec_stack:
                    return True
            rec_stack.discard(node)
            return False

        for task_name in self._tasks:
            if task_name not in visited:
                if dfs(task_name):
                    return True
        return False

    def topological_order(self) -> Optional[List[str]]:
        """Return tasks in topological order, or None if a cycle exists."""
        if self.has_cycle():
            return None

        in_degree: Dict[str, int] = {name: 0 for name in self._tasks}
        for name in self._tasks:
            for dep in self._tasks[name].deps:
                if dep in in_degree:
                    # Bug 2 fix: in-degree of `name` increases for each
                    # prerequisite, not the prerequisite's in-degree.
                    in_degree[name] += 1

        queue = [name for name, deg in in_degree.items() if deg == 0]
        result: List[str] = []

        while queue:
            node = queue.pop(0)
            result.append(node)
            for dependent in self._adj.get(node, []):
                if dependent in in_degree:
                    in_degree[dependent] -= 1
                    if in_degree[dependent] == 0:
                        queue.append(dependent)

        return result if len(result) == len(self._tasks) else None