Method _run_single_task

server/services/batch_run_service.py:192–235

(
        self,
        session_id: str,
        yaml_file: str,
        task: BatchTask,
        task_dir: str,
        log_level: Optional[LogLevel],
    )

            json.dump(result_rows, handle, ensure_ascii=True, indent=2)

    def _run_single_task(
        self,
        session_id: str,
        yaml_file: str,
        task: BatchTask,
        task_dir: str,
        log_level: Optional[LogLevel],
    ) -> Dict[str, Any]:
        yaml_path = self._resolve_yaml_path(yaml_file)
        design = load_config(yaml_path, vars_override=task.vars_override or None)
        if any(node.type == "human" for node in design.graph.nodes):
            raise ValidationError(
                "Batch execution does not support human nodes",
                details={"yaml_file": yaml_file},
            )

        output_root = WARE_HOUSE_DIR / f"session_{session_id}"
        graph_config = GraphConfig.from_definition(
            design.graph,
            name=task_dir,
            output_root=output_root,
            source_path=str(yaml_path),
            vars=design.vars,
        )
        graph_config.metadata["fixed_output_dir"] = True

        if log_level:
            graph_config.log_level = log_level
            graph_config.definition.log_level = log_level

        graph_context = GraphContext(config=graph_config)

        start_time = time.perf_counter()
        executor = GraphExecutor(graph_context, session_id=session_id)
        task_input = self._build_task_input(executor.attachment_store, task)
        executor._execute(task_input)
        duration_ms = int((time.perf_counter() - start_time) * 1000)

        return {
            "results": executor.outputs,
            "token_usage": executor.token_tracker.get_token_usage(),
            "duration_ms": duration_ms,
            "graph_output": executor.get_final_output(),
        }

    @staticmethod
    def _build_task_input(attachment_store, task: BatchTask):