Workflow API

DAG workflow builder, run handles, gate APIs, and supporting types.

DAG workflow builder, execution handles, and analysis tools.

`Workflow`

Builder for a workflow DAG.

Constructor

python

Workflow(
    name: str = "workflow",
    version: int = 1,
    on_failure: str = "fail_fast",
    cache_ttl: float | None = None,
)

Parameter	Type	Default	Description
`name`	`str`	`"workflow"`	Workflow name (used for definition storage)
`version`	`int`	`1`	Version number
`on_failure`	`str`	`"fail_fast"`	Error strategy: `"fail_fast"` or `"continue"`
`cache_ttl`	`float \| None`	`None`	Cache TTL in seconds for incremental runs

`step()`

python

wf.step(
    name: str,
    task: TaskWrapper,
    *,
    after: str | list[str] | None = None,
    args: tuple = (),
    kwargs: dict | None = None,
    queue: str | None = None,
    max_retries: int | None = None,
    timeout_ms: int | None = None,
    priority: int | None = None,
    fan_out: str | None = None,
    fan_in: str | None = None,
    condition: str | Callable | None = None,
) -> Workflow

Add a task step. Returns self for chaining.

`visualize()`

python

wf.visualize(fmt: str = "mermaid") -> str

Render the DAG as a Mermaid or DOT diagram string.

`ancestors()` / `descendants()`

python

wf.ancestors(node: str) -> list[str]
wf.descendants(node: str) -> list[str]

`step_names`

python

workflow.step_names -> list[str]

Every step's name, in declaration order — the order step() and gate() added them, not a run order. topological_levels() below is the run order.

Submitting compiles the Python definition into the DAG the core executes, on the native PyWorkflowBuilder. Nothing here is called directly, but the declared names show up in the symbol index: add_step replays each step onto the native builder, step_count and step_names read it back, and serialize() returns the (dag_bytes, step_metadata_json) pair submit_workflow() sends. Conditions, gates and sub-workflows are resolved on the Python side first, because a callable predicate cannot cross into the core.

`topological_levels()`

python

wf.topological_levels() -> list[list[str]]

`stats()`

python

wf.stats() -> dict[str, int | float]

Returns {nodes, edges, depth, width, density}.

`critical_path()`

python

wf.critical_path(costs: dict[str, float]) -> tuple[list[str], float]

Returns (path, total_cost) — the longest-weighted path.

`execution_plan()`

python

wf.execution_plan(max_workers: int = 1) -> list[list[str]]

`bottleneck_analysis()`

python

wf.bottleneck_analysis(costs: dict[str, float]) -> dict[str, Any]

Returns {node, cost, percentage, critical_path, total_cost, suggestion}.

`WorkflowRun`

Handle for a submitted workflow run.

`status()`

python

run.status() -> WorkflowStatus

One read of the whole run: its state, plus a nodes map of NodeSnapshot records. Each call queries storage — the handle caches nothing, so a status read in a loop is a query in a loop. Declared as get_workflow_run_status, whose node_statuses() is the raw per-node map this converts into NodeSnapshot.

`wait()`

python

run.wait(timeout: float | None = None, poll_interval: float = 0.1) -> WorkflowStatus

Block until the workflow reaches a terminal state. Raises WorkflowTimeoutError on timeout.

`cancel()`

python

run.cancel() -> None

Cancel every step still pending and mark the run cancelled. Steps already running are not interrupted — cancellation is cooperative, so an in-flight handler finishes or observes the cancel itself. Declared as cancel_workflow_run.

`node_status()`

python

run.node_status(node_name: str) -> NodeStatus

One node's status out of the same snapshot status() reads, so prefer status().nodes when you want several.

`visualize()`

python

run.visualize(fmt: str = "mermaid") -> str

Render the DAG with live node status colors. fmt is a DiagramFormat or its string ("mermaid", "dot"); an unknown one raises rather than quietly falling back. Reads the stored definition through get_workflow_definition_dag and overlays a fresh status(), so it is two queries per call.

Run inspection across a deployment

Listing runs is not on Queue. The declared surface has list_workflow_runs, its keyset twin list_workflow_runs_after, get_workflow_run_detail (run header plus every node) and get_child_workflow_runs (sub-workflow runs spawned by a run) — but no public Python method forwards them. The built-in dashboard is their consumer, and its HTTP API is the supported way to reach them:

Endpoint	Backed by
`GET /api/workflows/runs`	`list_workflow_runs`, or `list_workflow_runs_after` with `paginate=cursor` / an `after` cursor
`GET /api/workflows/runs/{run_id}`	`get_workflow_run_detail`
`GET /api/workflows/runs/{run_id}/children`	`get_child_workflow_runs`
`GET /api/workflows/runs/{run_id}/dag`	`get_workflow_definition_dag`

For a run you submitted yourself, WorkflowRun covers status, per-node status, cancellation and visualization without any of these. Listing and child-run queries have no handle equivalent — the dashboard API above is the way to reach them.

`WorkflowProxy`

Returned by @queue.workflow(). Wraps a factory function.

`submit()`

python

proxy.submit(*args, **kwargs) -> WorkflowRun

Build and submit the workflow.

`build()`

python

proxy.build(*args, **kwargs) -> Workflow

Materialize without submitting.

`as_step()`

python

proxy.as_step(**params) -> SubWorkflowRef

Return a reference for use as a sub-workflow step.

Queue methods

Added to Queue via QueueWorkflowMixin:

`submit_workflow()`

python

queue.submit_workflow(
    workflow: Workflow,
    *,
    incremental: bool = False,
    base_run: str | None = None,
) -> WorkflowRun

`approve_gate()`

python

queue.approve_gate(run_id: str, node_name: str) -> None

`reject_gate()`

python

queue.reject_gate(run_id: str, node_name: str, error: str = "rejected") -> None

`@queue.workflow()`

python

@queue.workflow(name: str | None = None, *, version: int = 1)
def factory() -> Workflow: ...

Types

`WorkflowState`

python

class WorkflowState(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    PAUSED = "paused"

`NodeStatus`

python

class NodeStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"
    WAITING_APPROVAL = "waiting_approval"
    CACHE_HIT = "cache_hit"

`WorkflowStatus`

python

@dataclass
class WorkflowStatus:
    run_id: str
    state: WorkflowState
    started_at: int | None
    completed_at: int | None
    error: str | None
    nodes: dict[str, NodeSnapshot]

`NodeSnapshot`

python

@dataclass
class NodeSnapshot:
    name: str
    status: NodeStatus
    job_id: str | None
    error: str | None

`WorkflowContext`

python

@dataclass(frozen=True)
class WorkflowContext:
    run_id: str
    results: dict[str, Any]
    statuses: dict[str, str]
    params: dict[str, Any] | None
    failure_count: int
    success_count: int

`GateConfig`

python

@dataclass
class GateConfig:
    timeout: float | None = None
    on_timeout: str = "reject"
    message: str | Callable | None = None