manual_slop/dag_engine.py

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]) -> None:
  """
        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) -> None:
  """
        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) -> None:
  """
        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) -> None:
  """
        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