Function run_workflow_task

src/workflow/launch.py:45–126

(
    *,
    source: str,
    runner: AgentRunner,
    registry: Any,  # RuntimeTaskRegistry
    task_id: str,
    run_id: str,
    output_file: str,
    args: Any = None,
    controller: Optional[AbortController] = None,
    resume: Optional[Mapping] = None,
    resolve_workflow: Optional[Any] = None,
    tool_use_id: Optional[str] = None,
    budget_total: Optional[int] = None,
    max_concurrent: Optional[int] = None,
)

async def run_workflow_task(
    *,
    source: str,
    runner: AgentRunner,
    registry: Any,  # RuntimeTaskRegistry
    task_id: str,
    run_id: str,
    output_file: str,
    args: Any = None,
    controller: Optional[AbortController] = None,
    resume: Optional[Mapping] = None,
    resolve_workflow: Optional[Any] = None,
    tool_use_id: Optional[str] = None,
    budget_total: Optional[int] = None,
    max_concurrent: Optional[int] = None,
):
    from src.tasks.local_workflow import (
        complete_workflow_task,
        fail_workflow_task,
        register_workflow_task,
        update_workflow_summary,
    )

    from .runtime import run_workflow

    controller = controller if controller is not None else create_abort_controller()

    def _on_start(run) -> None:
        register_workflow_task(
            task_id=task_id,
            run_id=run_id,
            workflow_name=run.meta.name,
            description=run.meta.description,
            output_file=output_file,
            progress=run.progress,
            run=run,
            registry=registry,
            tool_use_id=tool_use_id,
        )

    def _on_progress(_progress: WorkflowProgress) -> None:
        update_workflow_summary(task_id, registry)

    try:
        result = await run_workflow(
            source,
            runner=runner,
            args=args,
            run_id=run_id,
            controller=controller,
            on_start=_on_start,
            on_progress=_on_progress,
            resume=resume,
            resolve_workflow=resolve_workflow,
            budget_total=budget_total,
            max_concurrent=max_concurrent,
        )
    except WorkflowMetaError as exc: