Streaming workflow output

Flux workflows produce durable, checkpointed results, but callers often need to see what is happening before the final answer arrives. The progress() primitive lets any task emit partial updates in real time. Those updates travel to connected clients as Server-Sent Events without touching the database, leaving the durability model unchanged.

How progress events work

Calling await progress(value) inside a task enqueues an ephemeral update. A background flusher on the worker batches the queue and forwards it to the server. The server forwards it to any SSE client listening on that execution — and nowhere else. Progress events are never written to the event log, never checkpointed, and never replayed. If the worker crashes mid-task, the client gets a fresh stream when the task re-executes.

This separation is deliberate: durable state events carry guarantees; progress events carry immediacy.

Emitting numeric progress

Tracking work done against a total is the most common use: processing rows, pages, records:

from flux import ExecutionContext, task, workflow
from flux.tasks import progress


@task
async def process_records(records: list[str]) -> dict:
    results = []
    for i, record in enumerate(records):
        results.append(record.strip().upper())
        await progress({"processed": i + 1, "total": len(records)})
    return {"count": len(results)}


@workflow
async def record_pipeline(ctx: ExecutionContext[list[str]]):
    return await process_records(ctx.input)

progress() accepts any serializable value, and the shape is yours to define. Numeric counters, step names, token fragments: all are valid.

Emitting step-based progress

When a task moves through discrete phases rather than counting items, named steps communicate state more clearly than fractions:

from flux import ExecutionContext, task, workflow
from flux.tasks import progress


@task
async def ingest(records: list[str]) -> list[str]:
    await progress({"step": "ingesting", "count": len(records)})
    return [r.strip() for r in records]


@task
async def transform(records: list[str]) -> list[str]:
    results = []
    for i, rec in enumerate(records):
        results.append(rec.upper())
        await progress({"step": "transforming", "done": i + 1, "total": len(records)})
    return results


@task
async def export(records: list[str]) -> dict:
    await progress({"step": "exporting"})
    return {"exported": len(records)}


@workflow
async def etl_pipeline(ctx: ExecutionContext[list[str]]):
    raw = await ingest(ctx.input)
    transformed = await transform(raw)
    return await export(transformed)

Each task emits progress independently. Because progress events carry the originating task’s name in the source_id field, a client can distinguish ingest progress from transform progress even when the events arrive interleaved.

The SSE event format

When a client connects in stream mode, it receives a mixed event stream: ephemeral task.progress events from progress() calls, and durable state events (CLAIMED, RUNNING, COMPLETED) from checkpoints.

A typical stream for the pipeline above looks like:

event: etl_pipeline.execution.running
data: {"state": "RUNNING", "execution_id": "a1b2c3..."}

event: task.progress
data: {"type": "TASK_PROGRESS", "source_id": "ingest_a1b2", "name": "ingest", "value": {"step": "ingesting", "count": 3}, "time": "2026-05-10T12:00:01.100"}

event: task.progress
data: {"type": "TASK_PROGRESS", "source_id": "transform_a1b2", "name": "transform", "value": {"step": "transforming", "done": 1, "total": 3}, "time": "2026-05-10T12:00:01.210"}

event: task.progress
data: {"type": "TASK_PROGRESS", "source_id": "transform_a1b2", "name": "transform", "value": {"step": "transforming", "done": 3, "total": 3}, "time": "2026-05-10T12:00:01.310"}

event: task.progress
data: {"type": "TASK_PROGRESS", "source_id": "export_a1b2", "name": "export", "value": {"step": "exporting"}, "time": "2026-05-10T12:00:01.400"}

event: etl_pipeline.execution.completed
data: {"state": "COMPLETED", "output": {"exported": 3}}

Durable state events are distinguished by the {workflow_name}.execution.{state} event field. Progress events always use task.progress. A client that only cares about the final result can skip everything until it sees the completed event.

Triggering stream mode

From the CLI

Pass --mode stream to flux workflow run:

flux workflow run etl_pipeline '["  hello  ", "  world  "]' --mode stream

The CLI opens an SSE connection and prints each event as it arrives.

From the REST API

POST /workflows/{namespace}/{workflow_name}/run/stream returns an SSE response immediately. The workflow executes on a worker and events flow back over the open connection:

curl -N -X POST http://localhost:8000/workflows/default/etl_pipeline/run/stream \
  -H "Content-Type: application/json" \
  -d '["  hello  ", "  world  "]'

The -N flag disables curl’s output buffering so SSE events print as they arrive. Omit it and the terminal waits until the connection closes.

Durability guarantees for progress

Progress does not interact with Flux’s event log:

The durable result of a task is always the return value, stored in the TASK_COMPLETED event. Progress events are a display mechanism, not a data channel.

What’s next

• Running workflows from the REST API — triggering stream mode, parsing the mixed event stream, and handling disconnections from a REST client.
• Workflow inputs and outputs — how the final return value is serialized and retrieved after the stream closes.