Workflow composition, failure handlers, and nodes

When you write a workflow in flytekit, the function body looks like ordinary Python, but it runs in two very different modes: compilation, where every task call creates a Node in a graph and returns unresolved Promise references, and local execution, where the same calls resolve to literal-backed promises (or native values at the outermost call). Most confusion around workflow composition comes from mixing these two worlds — for example, trying to call .outputs on the result of an ordinary task call, or using a Promise in a boolean test. This section walks through both composition styles flytekit offers, per-node overrides, and failure handlers, and spells out where each API is valid.

Function-based composition: task calls become nodes and promises

The @workflow decorator in flytekit/core/workflow.py wraps your function in a PythonFunctionWorkflow. When the workflow is compiled, the function body is executed under a CompilationState: every task, subworkflow, or launch-plan call goes through flyte_entity_call_handler (in promise.py), which calls create_and_link_node to append a Node and returns unresolved Promise objects instead of real values. Promises you return become the workflow's output bindings.

@task
def add_5(x: int) -> int:
    return x + 5

@workflow
def my_wf_example(a: int) -> typing.Tuple[int, int]:
    x = add_5(a=a)
    # You can use outputs of a previous task as inputs to other nodes.
    z = add_5(a=x)
    # You can call other workflows from within this workflow
    d = simple_wf()
    # Conditionals run on primitive promises and pick a branch
    e = conditional("bool").if_(a == 5).then(add_5(a=d)).else_().then(add_5(a=z))
    return x, e

Multiple outputs come back as promises packed into a generated named tuple (or your declared NamedTuple), so you can unpack them:

@task
def t1(a: int) -> typing.NamedTuple("OutputsBC", [("t1_int_output", int), ("c", str)]):
    a = a + 2
    return a, "world-" + str(a)

@workflow(interruptible=True, failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def wf(a: int) -> typing.Tuple[str, str]:
    x, y = t1(a=a)
    _, v = t1(a=x)
    return y, v

Promise semantics: why your workflow body isn't plain Python

A Promise (in promise.py) is either:

unresolved — wraps a NodeOutput reference (p.ref) pointing at a node; p.is_ready is False. This is what a task call returns during compilation.
resolved — holds a Flyte Literal (p.val); this is what you get inside local workflow execution.

Attribute and index access on a promise doesn't dereference anything — it copies the promise and appends the key to attr_path (and, for remote bindings, to the NodeOutput via NodeOutput.with_attr):

@workflow
def wf():
    o = t1()
    t2(x=o["a"][0])   # keys are appended to the promise's attribute path

Because promises aren't real values, several natural-looking operations fail by design:

bool(promise) raises ValueError — "Cannot perform truth value testing... For Logical and\or use \|\ (bitwise) instead." Comparisons like p == 5 produce ComparisonExpression objects, and a & b / a | b produce ConjunctionExpressions.
Iterating a promise (for x in p) raises — use p[index] instead.
A task with no declared outputs returns a VoidPromise, which deliberately raises on __eq__, __str__, __bool__, and arithmetic — a NoneType return cannot be used as a value.

Workflow outputs also must be promises from upstream entities. create_native_named_tuple rejects bare local variables or constants as outputs; if you want to return a scalar, wrap the computation in a task or return a named-tuple field of a promise.

Explicit node composition with `create_node`

Sometimes two tasks have no data dependency and you still need to say which runs first. That's what create_node in flytekit/core/node_creation.py is for:

t1_node = create_node(t1)
t2_node = create_node(t2)

t2_node.runs_before(t1_node)
# OR
t2_node >> t1_node

Inputs are keyword-only — positional arguments raise FlyteAssertion:

t3_node = create_node(t3, in1=some_int).with_overrides(...)

Here's the key distinction the research surfaces repeatedly:

Ordinary task calls return a Promise (or named tuple of promises). There is no .outputs on them. Pass the returned promise straight into the next task: x = t1(a=a) then t2(a=x).
create_node returns a Node, and only for those nodes does Node.outputs work. During compilation, create_node calls the entity through the normal call handler, grabs the newly appended node from ctx.compilation_state.nodes[-1], initializes node._outputs, and attaches each output promise both as an attribute (node.o0, or the declared output name) and into the node.outputs dict:

t4_node = create_node(t4)
# In compilation, node.o0 has the promise:
t5(in1=t4_node.o0)
# Or, when you have the output name as a string:
t5(in1=t4_node.outputs["o0"])

Node.outputs itself is defensive about this — if the node wasn't created by create_node, accessing it raises AssertionError("Cannot use outputs with all Nodes, node must've been created from create_node()").

Gotchas with create_node:

Entities with no outputs return just the Node (a VoidPromise from the entity is dropped).
In local execution, a single-output entity is still tupletized, so you dereference by the declared output name (t1_node.o0) even for one output.
It only works inside workflows or dynamic tasks — outside compilation or local execution it raises RuntimeError, and it's rejected inside skipped conditional branches.
Remote entities can't run locally.

Imperative workflows

The same node machinery backs the programmatic ImperativeWorkflow API, useful when composition is determined at runtime:

wb = Workflow(name="my_workflow")
wb.add_workflow_input("in1", str)
node = wb.add_entity(t1, a=wb.inputs["in1"])
wb.add_entity(t2)
wb.add_workflow_output("from_n0t1", node.outputs["o0"])

Note that add_entity uses create_node under the hood, which is why node.outputs["o0"] is available here but not for ordinary function-style calls. Local imperative execution walks nodes in creation order, resolves bindings from an intermediate node-output cache, and resolves final workflow outputs via get_promise in workflow.py — outputs come from the bindings, not from raw return values.

Per-node overrides

Both ordinary promises and explicit nodes accept with_overrides. Promise.with_overrides simply forwards to the referenced node (self.ref.node.with_overrides(...)), while Node.with_overrides mutates the node directly. Overridable settings include node_name, aliases, requests/limits/resources, timeout, retries, interruptible, cache, task_config, container_image, accelerator, shared_memory, and pod_template.

@workflow
def my_wf(x: typing.List[int]) -> typing.List[typing.Optional[str]]:
    return map_task(
        my_mappable_task,
        metadata=TaskMetadata(retries=1),
        concurrency=10,
        min_success_ratio=0.75,
    )(a=x).with_overrides(requests=Resources(cpu="10M"))

Validation rules enforced in Node.with_overrides / _override_node_metadata:

Overrides are validated with assert_not_promise — you can't pass a promise for retries, interruptible, cache, cache_version, resources, container_image, etc.
resources cannot be combined with requests or limits.
If you override requests without limits, flytekit logs a warning that requests are clamped to original limits.
Overriding cache with a Cache object requires a version ("must specify cache version when overriding"), and mixing Cache with the deprecated cache_serialize/cache_version kwargs raises ValueError.
timeout accepts an int (seconds) or datetime.timedelta; anything else raises.
task_config overrides must match the existing config type.
node_name is DNS-normalized via _dnsify, same as node IDs at construction.

For multi-output calls, with_overrides on the returned named tuple forwards through the first output's promise — it's node-level metadata, not per-output configuration.

Failure handlers

@workflow(on_failure=handler) attaches an entity that runs when the workflow fails. The signature rule is strict and enforced in both ImperativeWorkflow.add_on_failure_handler and PythonFunctionWorkflow._validate_add_on_failure_handler (both in workflow.py):

The handler must accept every workflow input — the check is (failure_node_inputs | workflow_inputs) != failure_node_inputs → raise FlyteFailureNodeInputMismatchException.
Every additional handler input must be Optional, or the same exception is raised.

The canonical pattern from workflow.py:

@task
def clean_up(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name} due to {err}")
    print("This is err:", str(err))

@task
def create_cluster(name: str):
    print(f"Creating cluster: {name}")

@task
def delete_cluster(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name}")
    print(err)

@task
def t1(a: int, b: str):
    print(f"{a} {b}")
    raise ValueError(error_message)

@workflow(on_failure=clean_up)
def wf(name: str = "flyteorg"):
    c = create_cluster(name=name)
    t = t1(a=1, b="2")
    d = delete_cluster(name=name)
    c >> t >> d

with pytest.raises(ValueError):
    wf()

Note the >> chain here: these tasks don't pass data to each other, so explicit dependencies order them — and the failure node is wired to run after the failing part of the graph.

Implementation details worth knowing:

The input named err is special: the implementation injects a FlyteError (from flytekit.types.error) containing the failed node ID and exception text only if the handler interface literally declares an input named err. (Some decorator docs say error; the code checks err.)
The handler is compiled as a special failure node that lives outside the main node list. add_on_failure_handler pops the just-created node off the compilation state, stores it as self._failure_node, and assigns it DEFAULT_FAILURE_NODE_ID.
Locally, WorkflowBase.__call__ invokes the handler when the workflow body raises, then re-raises the original exception. The handler is a cleanup hook, not a fallback return path.

Workflow failure policy

@workflow accepts failure_policy with two values:

WorkflowFailurePolicy.FAIL_IMMEDIATELY (default) — fail as soon as a node fails.
WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE — remaining runnable nodes are allowed to finish before the workflow fails.

The policy is serialized by WorkflowMetadata.to_flyte_model() as 0 or 1, and interruptible=True on the decorator propagates to node metadata defaults via WorkflowMetadataDefaults.

Quick troubleshooting table

Symptom	Cause	Fix
`AssertionError: Cannot use outputs with all Nodes...`	Called `.outputs` on a node from an ordinary task call	Use the returned promise directly (`x = t1()`), or create the node with `create_node`
`FlyteAssertion` about positional args in `create_node`	`create_node(t3, some_int)`	Pass inputs by keyword: `create_node(t3, in1=some_int)`
`ValueError` on truth testing a promise	`if p:` or `and`/`or` in a workflow body	Use comparisons and `&`/`
`FlyteFailureNodeInputMismatchException`	Failure handler missing a workflow input, or has a required extra input	Accept all workflow inputs; make every additional input `Optional`
Handler gets `None` for failure details	Handler input isn't literally named `err`	Declare `err: typing.Optional[FlyteError] = None`
Failed workflow "swallowed" the exception locally	Misreading handler behavior	The handler runs, then flytekit re-raises the original exception
Promise used in resources/retries/etc. override	`assert_not_promise` validation	Pass concrete values to `with_overrides`

Function-based composition: task calls become nodes and promises​

Promise semantics: why your workflow body isn't plain Python​

Explicit node composition with create_node​

Imperative workflows​

Per-node overrides​

Failure handlers​

Workflow failure policy​

Quick troubleshooting table​