feat(mma): Implement ExecutionEngine with auto-queue and step-mode support
This commit is contained in:
@@ -76,3 +76,50 @@ class TrackDAG:
|
||||
visit(ticket.id)
|
||||
|
||||
return stack
|
||||
|
||||
class ExecutionEngine:
|
||||
def __init__(self, dag: TrackDAG, auto_queue: bool = False):
|
||||
self.dag = dag
|
||||
self.auto_queue = auto_queue
|
||||
|
||||
def tick(self) -> List[Ticket]:
|
||||
"""
|
||||
Returns a list of tasks that are currently 'ready' to be executed.
|
||||
A task is ready if its status is 'todo' and all its dependencies are 'completed'.
|
||||
If auto_queue=True, it will automatically mark 'ready' tasks as 'in-progress',
|
||||
unless step_mode=True is set on the task.
|
||||
"""
|
||||
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 approves a task to move it to 'in_progress'.
|
||||
Typically used for tasks with step_mode=True or when auto_queue is False.
|
||||
"""
|
||||
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):
|
||||
"""
|
||||
Updates the status of a specific task within the DAG.
|
||||
"""
|
||||
ticket = self.dag.ticket_map.get(task_id)
|
||||
if ticket:
|
||||
ticket.status = status
|
||||
|
||||
Reference in New Issue
Block a user