154 lines
4.5 KiB
Python
154 lines
4.5 KiB
Python
from typing import List, Optional
|
|
from models import Ticket
|
|
|
|
class TrackDAG:
|
|
"""
|
|
Manages a Directed Acyclic Graph of implementation tickets.
|
|
Provides methods for dependency resolution, cycle detection, and topological sorting.
|
|
"""
|
|
|
|
def __init__(self, tickets: List[Ticket]):
|
|
"""
|
|
Initializes the TrackDAG with a list of Ticket objects.
|
|
Args:
|
|
tickets: A list of Ticket instances defining the graph nodes and edges.
|
|
"""
|
|
self.tickets = tickets
|
|
self.ticket_map = {t.id: t for t in tickets}
|
|
|
|
def get_ready_tasks(self) -> List[Ticket]:
|
|
"""
|
|
Returns a list of tickets that are in 'todo' status and whose dependencies are all 'completed'.
|
|
Returns:
|
|
A list of Ticket objects ready for execution.
|
|
"""
|
|
ready = []
|
|
for ticket in self.tickets:
|
|
if ticket.status == 'todo':
|
|
# Check if all dependencies exist and are completed
|
|
all_done = True
|
|
for dep_id in ticket.depends_on:
|
|
dep = self.ticket_map.get(dep_id)
|
|
if not dep or dep.status != 'completed':
|
|
all_done = False
|
|
break
|
|
if all_done:
|
|
ready.append(ticket)
|
|
return ready
|
|
|
|
def has_cycle(self) -> bool:
|
|
"""
|
|
Performs a Depth-First Search to detect cycles in the dependency graph.
|
|
Returns:
|
|
True if a cycle is detected, False otherwise.
|
|
"""
|
|
visited = set()
|
|
rec_stack = set()
|
|
|
|
def is_cyclic(ticket_id: str) -> bool:
|
|
"""Internal recursive helper for cycle detection."""
|
|
if ticket_id in rec_stack:
|
|
return True
|
|
if ticket_id in visited:
|
|
return False
|
|
visited.add(ticket_id)
|
|
rec_stack.add(ticket_id)
|
|
ticket = self.ticket_map.get(ticket_id)
|
|
if ticket:
|
|
for neighbor in ticket.depends_on:
|
|
if is_cyclic(neighbor):
|
|
return True
|
|
rec_stack.remove(ticket_id)
|
|
return False
|
|
for ticket in self.tickets:
|
|
if ticket.id not in visited:
|
|
if is_cyclic(ticket.id):
|
|
return True
|
|
return False
|
|
|
|
def topological_sort(self) -> List[str]:
|
|
"""
|
|
Returns a list of ticket IDs in topological order (dependencies before dependents).
|
|
Returns:
|
|
A list of ticket ID strings.
|
|
Raises:
|
|
ValueError: If a dependency cycle is detected.
|
|
"""
|
|
if self.has_cycle():
|
|
raise ValueError("Dependency cycle detected")
|
|
visited = set()
|
|
stack = []
|
|
|
|
def visit(ticket_id: str):
|
|
"""Internal recursive helper for topological sorting."""
|
|
if ticket_id in visited:
|
|
return
|
|
visited.add(ticket_id)
|
|
ticket = self.ticket_map.get(ticket_id)
|
|
if ticket:
|
|
for dep_id in ticket.depends_on:
|
|
visit(dep_id)
|
|
stack.append(ticket_id)
|
|
for ticket in self.tickets:
|
|
visit(ticket.id)
|
|
return stack
|
|
|
|
class ExecutionEngine:
|
|
"""
|
|
A state machine that governs the progression of tasks within a TrackDAG.
|
|
Handles automatic queueing and manual task approval.
|
|
"""
|
|
|
|
def __init__(self, dag: TrackDAG, auto_queue: bool = False):
|
|
"""
|
|
Initializes the ExecutionEngine.
|
|
Args:
|
|
dag: The TrackDAG instance to manage.
|
|
auto_queue: If True, ready tasks will automatically move to 'in_progress'.
|
|
"""
|
|
self.dag = dag
|
|
self.auto_queue = auto_queue
|
|
|
|
def tick(self) -> List[Ticket]:
|
|
"""
|
|
Evaluates the DAG and returns a list of tasks that are currently 'ready' for execution.
|
|
If auto_queue is enabled, tasks without 'step_mode' will be marked as 'in_progress'.
|
|
Returns:
|
|
A list of ready Ticket objects.
|
|
"""
|
|
ready = self.dag.get_ready_tasks()
|
|
if self.auto_queue:
|
|
for ticket in ready:
|
|
if not ticket.step_mode:
|
|
ticket.status = "in_progress"
|
|
return ready
|
|
|
|
def approve_task(self, task_id: str):
|
|
"""
|
|
Manually transitions a task from 'todo' to 'in_progress' if its dependencies are met.
|
|
Args:
|
|
task_id: The ID of the task to approve.
|
|
"""
|
|
ticket = self.dag.ticket_map.get(task_id)
|
|
if ticket and ticket.status == "todo":
|
|
# Check if dependencies are met first
|
|
all_done = True
|
|
for dep_id in ticket.depends_on:
|
|
dep = self.dag.ticket_map.get(dep_id)
|
|
if not dep or dep.status != "completed":
|
|
all_done = False
|
|
break
|
|
if all_done:
|
|
ticket.status = "in_progress"
|
|
|
|
def update_task_status(self, task_id: str, status: str):
|
|
"""
|
|
Force-updates the status of a specific task.
|
|
Args:
|
|
task_id: The ID of the task.
|
|
status: The new status string (e.g., 'todo', 'in_progress', 'completed', 'blocked').
|
|
"""
|
|
ticket = self.dag.ticket_map.get(task_id)
|
|
if ticket:
|
|
ticket.status = status
|