Function compute_scheduling_scores

api/dependency_resolver.py:274–352

Compute scheduling scores for all features. Higher scores mean higher priority for scheduling. The algorithm considers: 1. Unblocking potential - Features that unblock more downstream work score higher 2. Depth in graph - Features with no dependencies (roots) are "shovel-ready" 3. U

(features: list[dict])

def compute_scheduling_scores(features: list[dict]) -> dict[int, float]:
    """Compute scheduling scores for all features.

    Higher scores mean higher priority for scheduling. The algorithm considers:
    1. Unblocking potential - Features that unblock more downstream work score higher
    2. Depth in graph - Features with no dependencies (roots) are "shovel-ready"
    3. User priority - Existing priority field as tiebreaker

    Score formula: (1000 * unblock) + (100 * depth_score) + (10 * priority_factor)

    Args:
        features: List of feature dicts with id, priority, dependencies fields

    Returns:
        Dict mapping feature_id -> score (higher = schedule first)
    """
    if not features:
        return {}

    # Build adjacency lists
    children: dict[int, list[int]] = {f["id"]: [] for f in features}  # who depends on me
    parents: dict[int, list[int]] = {f["id"]: [] for f in features}   # who I depend on

    for f in features:
        for dep_id in (f.get("dependencies") or []):
            if dep_id in children:  # Only valid deps
                children[dep_id].append(f["id"])
                parents[f["id"]].append(dep_id)

    # Calculate depths via BFS from roots
    # Use visited set to prevent infinite loops from circular dependencies
    # Use deque for O(1) popleft instead of list.pop(0) which is O(n)
    depths: dict[int, int] = {}
    visited: set[int] = set()
    roots = [f["id"] for f in features if not parents[f["id"]]]
    bfs_queue: deque[tuple[int, int]] = deque((root, 0) for root in roots)
    while bfs_queue:
        node_id, depth = bfs_queue.popleft()
        if node_id in visited:
            continue  # Skip already visited nodes (handles cycles)
        visited.add(node_id)
        depths[node_id] = depth
        for child_id in children[node_id]:
            if child_id not in visited:
                bfs_queue.append((child_id, depth + 1))

    # Handle orphaned nodes (shouldn't happen but be safe)
    for f in features:
        if f["id"] not in depths:
            depths[f["id"]] = 0

    # Calculate transitive downstream counts (reverse topo order)
    downstream: dict[int, int] = {f["id"]: 0 for f in features}
    # Process in reverse depth order (leaves first)
    for fid in sorted(depths.keys(), key=lambda x: -depths[x]):
        for parent_id in parents[fid]:
            downstream[parent_id] += 1 + downstream[fid]