Feasibility-Aware Path Ranking

The problem

Planning a project is difficult because it is hard to tell which steps are actually doable based on the tools and dependencies available. It is easy to create a plan that looks good on paper but is impossible to actually complete.

What it does

It looks at a list of project steps and scores them based on how realistic they are to complete. It filters out paths that are blocked by missing requirements or impossible tasks.

Why it matters

It helps you identify which project path is actually achievable before you start working on it.

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 path_ranker.py
Ranked Paths:
['Initialize', 'Analyze_data', 'Generate_report']: 0.20
['Initialize', 'Data_quality_check', 'Analyze_data', 'Generate_report']: 0.00
['Initialize', 'Analyze_data', 'Template_available', 'Generate_report']: 0.00

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

# Updated path_ranker.py with Resource-Constraint multiplier
import json
from typing import List, Dict, Tuple, Optional

class Task:
    def __init__(self, name: str, dependencies: List[str] = None, feasibility: float = 1.0):
        self.name = name
        self.dependencies = dependencies or []
        self.feasibility = feasibility
        self.semantic_dependencies: List[str] = []

    def add_semantic_dependency(self, dependency: str):
        self.semantic_dependencies.append(dependency)

class GraphAugmentedPathFeasibility:
    def __init__(self, tasks: List['Task']):
        self.tasks = {task.name: task for task in tasks}
        self.dependency_graph = self._build_dependency_graph()

    def _build_dependency_graph(self) -> Dict[str, List[str]]:
        """Build directed graph where edges represent dependencies"""
        graph = {task: [] for task in self.tasks.keys()}
        for task in self.tasks.values():
            for dep in task.dependencies:
                if dep in graph:
                    graph[dep].append(task.name)
        return graph

    def calculate_path_feasibility(self, path: List[str]) -> float:
        """Combine graph structure with task feasibility scores. Returns 0.0 for:
        - Paths containing nonexistent tasks
        - Paths containing tasks with zero feasibility
        - Paths violating structural dependency order"""
        if not path:
            return 0.0
        
        feasibility = 1.0
        completed: set = set()
        for i, task_name in enumerate(path):
            task = self.tasks.get(task_name)
            if not task:
                return 0.0
            if task.feasibility == 0.0:
                return 0.0
            
            # Check both explicit and semantic dependencies
            all_dependencies = set(task.dependencies) | set(task.semantic_dependencies)
            for dep in all_dependencies:
                if dep not in completed:
                    return 0.0
            
            # Calculate score with semantic dependency penalty applied once per missing dep
            base_feasibility = task.feasibility * (0.95 ** i)
            semantic_penalty = 0.8 ** sum(1 for dep in task.semantic_dependencies if dep not in completed)
            feasibility *= base_feasibility * semantic_penalty
            completed.add(task_name)
        return max(feasibility, 0.0)

class TaskBasedPathFeasibility:
    def __init__(self, agents: Dict[str, Dict[str, float]]):
        self.agents = agents

    def calculate_task_feasibility(self, task: str) -> float:
        if not self.agents:
            return 1.0
        
        scores = []
        for _agent, skills in self.agents.items():
            scores.append(skills.get(task, 0.5))  # Default 0.5 if skill not found
        return max(scores) if scores else 1.0

class FeasibilityAwarePathRanker:
    def __init__(self, graph_augmentor: GraphAugmentedPathFeasibility, task_evaluator: TaskBasedPathFeasibility):
        self.graph_augmentor = graph_augmentor
        self.task_evaluator = task_evaluator

    def rank_paths(self, paths: List[List[str]]) -> List[Tuple[List[str], float]]:
        ranked = []
        for path in paths:
            graph_score = self.graph_augmentor.calculate_path_feasibility(path)
            if graph_score <= 0:
                ranked.append((path, 0.0))
                continue
            
            task_score = 1.0
            resource_penalty = 1.0
            
            for task_name in path:
                # Calculate task-based feasibility
                agent_score = self.task_evaluator.calculate_task_feasibility(task_name)
                task_score *= agent_score
                
                # Calculate resource penalty if available agents can't meet task requirement
                task = self.graph_augmentor.tasks.get(task_name)
                if task and task.feasibility > agent_score:
                    resource_penalty *= 0.5  # Apply 50% penalty per task with insufficient resources
            
            combined_score = graph_score * task_score * resource_penalty
            ranked.append((path, combined_score))
        
        return sorted(ranked, key=lambda x: x[1], reverse=True)

    def find_all_paths(self, tasks: List[Task], start: str, end: str) -> List[List[str]]:
        """Find all valid paths from start to end in the task DAG, ensuring all dependencies are satisfied."""
        task_map = {task.name: task for task in tasks}
        adj: Dict[str, List[str]] = defaultdict(list)
        for task in tasks:
            for dep in task.dependencies:
                adj[dep].append(task.name)
        
        _all_paths: List[List[str]] = []
        
        def dfs(current: str, target: str, visited: set, path: List[str]):
            if current == target:
                # Verify all dependencies of target are satisfied in path
                target_task = task_map.get(target)
                if target_task and not all(dep in visited for dep in target_task.dependencies):
                    return  # Invalid path, missing dependencies
                _all_paths.append(list(path))
                return
            
            task = task_map.get(current)
            if task and task.feasibility == 0.0:
                return
            
            for neighbor in adj.get(current, []):
                if neighbor not in visited:
                    # Check if all dependencies of neighbor are satisfied
                    neighbor_task = task_map.get(neighbor)
                    if neighbor_task and not all(dep in visited for dep in neighbor_task.dependencies):
                        continue  # Skip until dependencies are met
                    
                    visited.add(neighbor)
                    path.append(neighbor)
                    dfs(neighbor, target, visited, path)
                    path.pop()
                    visited.discard(neighbor)
        
        if start not in task_map:
            return []
        
        init_visited = {start}
        dfs(start, end, init_visited, [start])
        
        # Filter paths that miss any task's dependencies
        valid_paths = []
        for path in _all_paths:
            valid = True
            completed = set()
            for task_name in path:
                task = task_map.get(task_name)
                if task and not all(dep in completed for dep in task.dependencies):
                    valid = False
                    break
                completed.add(task_name)
            if valid:
                valid_paths.append(path)
        
        return valid_paths

if __name__ == "__main__":
    # Example usage with resource constraints
    tasks = [
        Task("Initialize", feasibility=0.9),
        Task("Analyze_data", dependencies=["Initialize"], feasibility=0.8),
        Task("Generate_report", dependencies=["Analyze_data"], feasibility=0.85),
    ]

    # Add semantic dependencies
    tasks[1].add_semantic_dependency("Data_quality_check")
    tasks[2].add_semantic_dependency("Template_available")

    graph_augmentor = GraphAugmentedPathFeasibility(tasks)
    agents = {
        "Data_Team": {"Analyze_data": 0.9, "Initialize": 0.7},
        "Report_Team": {"Generate_report": 0.95}
    }
    task_evaluator = TaskBasedPathFeasibility(agents)
    ranker = FeasibilityAwarePathRanker(graph_augmentor, task_evaluator)

    # Define paths to evaluate
    paths = [
        ["Initialize", "Analyze_data", "Generate_report"],
        ["Initialize", "Data_quality_check", "Analyze_data", "Generate_report"],
        ["Initialize", "Analyze_data", "Template_available", "Generate_report"]
    ]

    ranked_paths = ranker.rank_paths(paths)
    print("Ranked Paths:")
    for path, score in ranked_paths:
        print(f"{path}: {score:.2f}")
    
    # Demonstrate resource constraints in action
    print("\nPath viability with resource constraints:")
    for path in paths:
        viable = True
        for task_name in path:
            task = graph_augmentor.tasks.get(task_name)
            if task:
                task_feasibility = task.feasibility
                agent_score = task_evaluator.calculate_task_feasibility(task_name)
                if agent_score < task_feasibility:
                    viable = False
                    break
        print(f"{path}: {'Viable' if viable else 'Not Viable'}")