Class BatchRunService

server/services/batch_run_service.py:26–255

Runs batch workflows and reports progress over WebSocket.

class BatchRunService:
    """Runs batch workflows and reports progress over WebSocket."""

    def __init__(self) -> None:
        self.logger = logging.getLogger(__name__)

    async def run_batch(
        self,
        session_id: str,
        yaml_file: str,
        tasks: List[BatchTask],
        websocket_manager,
        *,
        max_parallel: int = 5,
        file_base: str = "batch",
        log_level: Optional[LogLevel] = None,
    ) -> None:
        batch_id = session_id
        total = len(tasks)

        await websocket_manager.send_message(
            session_id,
            {"type": "batch_started", "data": {"batch_id": batch_id, "total": total}},
        )

        semaphore = asyncio.Semaphore(max_parallel)
        success_count = 0
        failure_count = 0
        result_rows: List[Dict[str, Any]] = []
        result_lock = asyncio.Lock()

        async def run_task(task: BatchTask) -> None:
            nonlocal success_count, failure_count
            task_id = task.task_id or str(uuid.uuid4())
            task_dir = self._sanitize_label(f"{file_base}-{task_id}")

            await websocket_manager.send_message(
                session_id,
                {
                    "type": "batch_task_started",
                    "data": {
                        "row_index": task.row_index,
                        "task_id": task_id,
                        "task_dir": task_dir,
                    },
                },
            )

            try:
                result = await asyncio.to_thread(
                    self._run_single_task,
                    session_id,
                    yaml_file,
                    task,
                    task_dir,
                    log_level,
                )
                success_count += 1