workflows/parallel_workflow.py

"""
Parallel Workflow: runs a wave of agents concurrently using asyncio.gather.
"""
import asyncio
from typing import Any, List
from workflows.base_workflow import BaseWorkflow
from agents.base_agent import AgentContext, AgentResult, AgentStatus
from utils.logger import get_logger

logger = get_logger("parallel_workflow")


class ParallelWorkflow(BaseWorkflow):
    """Run all provided agents concurrently; collect results."""
    name = "parallel"

    async def execute(self, agents: List[Any], ctx: AgentContext) -> List[AgentResult]:
        if not agents:
            return []

        if len(agents) == 1:
            return [await agents[0].run(ctx)]

        logger.info(f"Running {len(agents)} agents in parallel: {[a.name for a in agents]}")
        tasks = [agent.run(ctx) for agent in agents]
        results: List[AgentResult] = await asyncio.gather(*tasks, return_exceptions=True)

        # Unwrap any exceptions from gather
        cleaned = []
        for i, r in enumerate(results):
            if isinstance(r, Exception):
                cleaned.append(AgentResult(
                    agent_name=agents[i].name,
                    status=AgentStatus.FAILED,
                    error=str(r),
                ))
            else:
                cleaned.append(r)

        return cleaned