Function run_workflow

src/workflow/runtime.py:316–380

Run a Python workflow ``source`` to completion. Raises :class:`WorkflowMetaError` if the ``meta`` block is missing/invalid (pre-flight). A script *runtime* exception is captured into ``WorkflowResult.error`` and ends the run gracefully.

(
    source: str,
    *,
    runner: AgentRunner,
    args: Any = None,
    run_id: str = "wf",
    on_progress: Optional[Callable[[WorkflowProgress], None]] = None,
    on_start: Optional[Callable[["WorkflowRun"], None]] = None,
    resume: Optional[Mapping[CallKey, JournalRecord]] = None,
    resolve_workflow: Optional[Callable[[str], str]] = None,
    budget_total: Optional[int] = None,
    max_concurrent: Optional[int] = None,
    controller: Optional[AbortController] = None,
    scheduler: Optional[Scheduler] = None,
    budget: Optional[Budget] = None,
    base_path: CallKey = (),
    _depth: int = 0,
)

async def run_workflow(
    source: str,
    *,
    runner: AgentRunner,
    args: Any = None,
    run_id: str = "wf",
    on_progress: Optional[Callable[[WorkflowProgress], None]] = None,
    on_start: Optional[Callable[["WorkflowRun"], None]] = None,
    resume: Optional[Mapping[CallKey, JournalRecord]] = None,
    resolve_workflow: Optional[Callable[[str], str]] = None,
    budget_total: Optional[int] = None,
    max_concurrent: Optional[int] = None,
    controller: Optional[AbortController] = None,
    scheduler: Optional[Scheduler] = None,
    budget: Optional[Budget] = None,
    base_path: CallKey = (),
    _depth: int = 0,
) -> WorkflowResult:
    """Run a Python workflow ``source`` to completion.

    Raises :class:`WorkflowMetaError` if the ``meta`` block is missing/invalid
    (pre-flight). A script *runtime* exception is captured into
    ``WorkflowResult.error`` and ends the run gracefully.
    """
    meta = extract_meta(source)

    scheduler = scheduler if scheduler is not None else Scheduler(max_concurrent)
    budget = budget if budget is not None else Budget(budget_total)
    journal = Journal(resume)
    progress = WorkflowProgress(meta.phases, on_change=on_progress)
    controller = controller if controller is not None else create_abort_controller()

    run = WorkflowRun(
        meta=meta,
        runner=runner,
        args=args,
        run_id=run_id,
        scheduler=scheduler,
        budget=budget,
        journal=journal,
        progress=progress,
        controller=controller,
        base_path=base_path,
        resolve_workflow=resolve_workflow,
        depth=_depth,
    )

    if on_start is not None:
        on_start(run)  # e.g. register the background task with the live run

    # Establish this run's base branch for deterministic call-path keys, and
    # restore the caller's branch afterwards (matters for nested workflow()).
    token = use_branch(base_path)
    value: Any = None
    error: Optional[str] = None
    try:
        value = await execute_workflow(source, run.namespace(), args)
    except WorkflowMetaError: