📋 BMad na Prática

Python — Estruturas de dados BMad

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Dict, Any
import asyncio

class TaskStatus(Enum):
    PENDING  = "pending"
    RUNNING  = "running"
    DONE     = "done"
    FAILED   = "failed"

@dataclass
class Task:
    id: str
    description: str
    agent_id: str
    dependencies: List[str] = field(default_factory=list)
    acceptance_criteria: str = ""
    max_cost_usd: float = 1.0
    timeout_seconds: int = 120
    status: TaskStatus = TaskStatus.PENDING
    result: Optional[Any] = None
    cost_usd: float = 0.0

class TaskTree:
    def __init__(self):
        self.tasks: Dict[str, Task] = {}

    def add_task(self, task: Task):
        self.tasks[task.id] = task

    def get_ready_tasks(self) -> List[Task]:
        """Tarefas cujas dependências foram concluídas"""
        ready = []
        for task in self.tasks.values():
            if task.status == TaskStatus.PENDING:
                deps_done = all(
                    self.tasks[dep].status == TaskStatus.DONE
                    for dep in task.dependencies
                )
                if deps_done:
                    ready.append(task)
        return ready

Python — Executor BMad com progresso em tempo real

class BMadExecutor:
    def __init__(self, task_tree: TaskTree, agent_registry: dict):
        self.tree = task_tree
        self.agents = agent_registry
        self.total_cost = 0.0

    async def execute(self, budget_usd: float = 10.0):
        while not self._is_complete():
            ready = self.tree.get_ready_tasks()
            if not ready:
                break  # deadlock ou completo

            # Executa tarefas prontas em paralelo
            tasks = [self._run_task(t, budget_usd) for t in ready]
            await asyncio.gather(*tasks, return_exceptions=True)

    async def _run_task(self, task: Task, budget: float):
        if self.total_cost >= budget:
            task.status = TaskStatus.FAILED
            return

        task.status = TaskStatus.RUNNING
        agent = self.agents[task.agent_id]
        context = self._build_context(task)

        try:
            result = await asyncio.wait_for(
                agent.run(task.description, context),
                timeout=task.timeout_seconds
            )
            task.result = result
            task.status = TaskStatus.DONE
            self.total_cost += task.cost_usd
            print(f"✓ {task.id} concluído — custo: ${task.cost_usd:.3f}")
        except Exception as e:
            task.status = TaskStatus.FAILED
            print(f"✗ {task.id} falhou: {e}")