Cancellation

When an operator cancels a running workflow, Flux delivers the signal as asyncio.CancelledError — the same exception Python uses for all cooperative cancellation. Your task and workflow code receives this exception at the next await point. How you handle it determines whether resources are released cleanly and whether the execution state machine ends in a consistent state.

How cancellation reaches your code

When a cancellation request arrives, the worker cancels the asyncio.Task running the workflow. Python delivers asyncio.CancelledError at the next await point inside your code. If that point is inside a task body, the error bubbles up through the call stack — from task to workflow function to the workflow decorator.

The workflow decorator (@workflow) catches asyncio.CancelledError and calls ctx.cancel(), which transitions the execution state to CANCELLED. After that, ctx.is_cancelled and ctx.has_finished are both True.

The sequence is:

1. Worker receives the cancellation signal.
2. Worker calls .cancel() on the running asyncio task.
3. Python raises asyncio.CancelledError at the current await inside your code.
4. The error propagates up through any tasks and workflow function.
5. The @workflow decorator catches it, calls ctx.cancel(), and re-raises.
6. The execution record is written with state CANCELLED.

Cleaning up inside a task

Use a finally block to release resources regardless of whether the task completes or is cancelled. The finally block runs whether the task finishes normally, raises an exception, or receives CancelledError.

import asyncio
from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


@task
async def process_batch(batch_id: str) -> list:
    connection = None
    try:
        connection = await open_connection(batch_id)
        results = []
        async for record in connection.stream():
            results.append(await transform(record))
        return results
    except asyncio.CancelledError:
        print(f"[task] batch {batch_id} cancelled mid-stream")
        raise  # must re-raise
    finally:
        if connection is not None:
            await connection.close()  # runs whether cancelled or not


async def open_connection(batch_id: str):
    class FakeConn:
        async def stream(self):
            for i in range(100):
                yield i
        async def close(self):
            print(f"[task] connection for {batch_id} closed")
    return FakeConn()


async def transform(record):
    await asyncio.sleep(0.01)
    return record * 2


@workflow
async def process_workflow(ctx: ExecutionContext[str]):
    return await process_batch(ctx.input or "batch-1")

The except asyncio.CancelledError block is optional if you only need finally. Its value is in logging or recording partial state before the error continues up the stack. The raise inside it is not optional: omitting raise from the except asyncio.CancelledError block is the one pattern that breaks everything — see below.

What the workflow decorator does for you

The @workflow decorator wraps your function and catches asyncio.CancelledError:

# This is what the decorator does internally (simplified):
try:
    output = await your_workflow_function(ctx)
    ctx.complete(...)
except asyncio.CancelledError:
    ctx.cancel()   # transitions state to CANCELLED
    raise          # propagates so the caller knows it was cancelled

You do not need to call ctx.cancel() yourself. The decorator handles it, including setting ctx.is_cancelled = True and ctx.has_finished = True.

If you want to log or record something before the cancellation propagates, catch CancelledError in your workflow body and re-raise:

import asyncio
from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


@task
async def long_step() -> str:
    await asyncio.sleep(30)
    return "done"


@workflow
async def monitored_workflow(ctx: ExecutionContext):
    try:
        result = await long_step()
        return result
    except asyncio.CancelledError:
        # safe place to log, emit a metric, or notify a downstream system
        print("workflow cancelled — notifying audit log")
        raise  # always re-raise

Verified against Flux 0.56.0: after raise, the @workflow decorator catches the re-raised error, calls ctx.cancel(), and the execution reaches state CANCELLED with ctx.is_cancelled = True.

The one rule: always re-raise CancelledError

If you catch asyncio.CancelledError without re-raising it, ctx.cancel() never gets called. The execution record shows state COMPLETED instead of CANCELLED, the event log is inconsistent with reality, and any monitoring or downstream logic that checks ctx.is_cancelled gets the wrong answer.

# BAD — do not do this
@workflow
async def broken_workflow(ctx: ExecutionContext):
    try:
        await long_step()
    except asyncio.CancelledError:
        return "cancelled"  # swallowed — ctx.state will be COMPLETED, not CANCELLED

# GOOD — always re-raise
@workflow
async def correct_workflow(ctx: ExecutionContext):
    try:
        await long_step()
    except asyncio.CancelledError:
        print("cancelled — cleaning up")
        raise

This rule applies in task bodies too: if you catch CancelledError to log or clean up, always raise at the end.

Cancellation in the inline path

The inline path — workflow.run(input) — calls asyncio.run() synchronously. There is no external caller holding a reference to the asyncio task. To cancel it during a run, install a SIGTERM or SIGINT handler that cancels the current event loop task, or use asyncio.wait_for() to impose a deadline (covered in Timeouts).

The inline path is designed for development and scripting. In production, workflows run through the distributed path (server + worker), where the worker receives the cancellation signal over SSE and cancels the running task cooperatively.

Checking cancellation state

After a workflow has been cancelled, the ExecutionContext exposes two flags:

In the distributed path, a workflow that is still executing tasks when the signal arrives will have ctx.is_cancelling = True until ctx.cancel() is called by the worker.

What to read next

• Manual cancellation (operator side) — how to send a cancellation request via CLI, SDK, or REST API.
• Timeouts — automatically cancel a task or workflow after a deadline.
• Errors and retries — configure retry and fallback before reaching the point where cancellation is needed.