Ray-Native Workflows¶

django-ray workflows combine one durable Django task with low-overhead Ray-native steps. The outer task is queued, retried, cancelled, and recorded in the database. Internal workflow steps are submitted directly to Ray and exchange intermediate values through Ray object references without creating a database row per step.

This is unrelated to the removed upstream ray.workflow package. See the Ray Ecosystem Support and Install Matrix for that distinction and the durable boundary for optional Ray components.

This model is intended for fan-out workloads where database-backed dispatch would cost more than the individual units of work.

WorkflowSignature objects are reusable definition builders. They are not durable execution plans, and a logically static chain or group is not automatically a compiled graph. The maintained workflow-plan contract separates definitions, immutable plans, per-run invocations, logical work, physical actors, and execution strategies while preserving this public API.

Requirements¶

Cluster Ray Core is the lowest-latency remote path for bounded workflows while the task-manager Ray Client connection remains part of the workload lifetime. A disconnect beyond Ray's reconnect grace period terminates its in-flight work; an outer retry replays the workflow rather than resuming completed leaves. Ray Job mode also supports workflows and is the safer starting point for long or coarse work that must continue independently of the submitting connection: its isolated driver connects back to Ray lazily before submitting leaves. Local execution is available for sync workers and unit tests. See worker execution modes.

DJANGO_RAY = {
    "RAY_ADDRESS": "ray://ray-head:10001",
    "RUNNER": "ray_core",  # Lowest submission overhead.
}

Start a worker normally, or select the cluster explicitly:

python manage.py django_ray_worker --cluster=ray://ray-head:10001

Complete Dynamic Fan-Out¶

The following is one complete myapp/workflows.py module. Every callable is defined at module scope so Ray workers can import it:

from django.tasks import task

from django_ray.workflows import chain, map_step, step


def build_items(count: int) -> list[int]:
    return list(range(count))


def calculate(value: int) -> dict[str, int]:
    # Stand-in for one independently expensive API or compute operation.
    checksum = sum((value * number) % 97 for number in range(500_000))
    return {"value": value, "checksum": checksum}


def summarize(results: list[dict[str, int]]) -> dict[str, int]:
    return {
        "items": len(results),
        "checksum": sum(result["checksum"] for result in results),
    }


calculation = chain(
    step(build_items),
    map_step(calculate, ray_options={"num_cpus": 0.25}).with_limits(
        max_concurrency=16,
        max_items=10_000,
    ),
    step(summarize),
)


@task(queue_name="default")
def calculate_batch(count: int) -> dict[str, int]:
    return calculation.run(count)

Calling calculate_batch.enqueue(20) creates one durable RayTaskExecution. Once it starts, build_items, every calculate call, and summarize are connected inside Ray. After its input iterable resolves, the bounded map_step consumes that iterable lazily, keeps at most 16 map items submitted at once, incrementally resolves them, and returns results in input order.

For a Kubernetes sync, the same shape is typically list_namespaces → map(sync one namespace) → summarize. Keep client creation or discovery outside the smallest inner resource loop where possible, and batch resources when each API operation is shorter than Ray submission overhead.

Bound Dynamic Fan-Out¶

Call with_limits() on a map signature when its input cardinality is data-dependent:

sync_namespaces = map_step(
    sync_namespace,
    ray_options={"num_cpus": 0.25},
).with_limits(
    max_concurrency=8,
    max_items=500,
    cancel_timeout_seconds=1.0,
)

max_concurrency is an admission window. At most that many map-item result references are retained while Ray work is pending, and results are collected as individual items finish. Fast items can therefore make room without waiting for an earlier slow item, while the final list remains in input order. The window counts map items, not every physical task in a nested signature. A mapped group with three branches and a window of eight can have up to 24 branch tasks plus its bounded per-item collectors submitted. A nested dynamic map needs its own with_limits() contract; an outer window cannot cap the number of leaves expanded inside one map item.

Lazy admission begins only after map_step calls executor.resolve() for its input. If an upstream Ray task returns a list or inventory, that task still produces the complete value and Ray transfers and deserializes it into the workflow coordinator before the first map item is admitted. Only an iterator that is already available locally is pulled one item at a time by the admission loop. Remote or paged input materialization is tracked in GitHub issue #94.

For every admitted item, django-ray retains the terminal result reference and the physical dependency references created by its nested chain or group. Those cleanup references are released as soon as the item result is collected, so their peak remains the admission window multiplied by the signature's fixed physical width.

max_items is an expansion safety limit. Sized inputs that exceed it fail before any leaf is submitted. For a generator, detecting overflow requires reading item max_items + 1; work completed before that discovery is not rolled back. Use idempotent leaves when iteration, retries, or cancellation can overlap external side effects.

On the first leaf, iterator, or collection failure, django-ray stops reading new items, requests cancellation for every retained physical reference belonging to the failed and pending items, and waits up to cancel_timeout_seconds for them to become terminal. It does not assume that cancelling a final task also cancels its Ray input dependencies. The default deadline is one second. The original exception is always re-raised. References still pending after the deadline are released rather than making cleanup wait indefinitely; an uncooperative running leaf may consequently finish after the workflow has failed. The deadline bounds only Ray cancellation and drain waiting. It does not bound input deserialization or arbitrary user iterator cleanup such as a generator's close() method.

Bounded maps use one aggregate kind="map" progress node with submitted, completed, in-flight, and input-exhaustion counters. Their physical item nodes are intentionally omitted from the live workflow graph. This keeps observability proportional to the declared workflow rather than to a 10k- or 50k-item expansion. For the same reason, report_progress() returns False inside those physical item leaves; the aggregate map counters remain available.

Incremental collection bounds pending references, not total result bytes. The ordered result list is still materialized in the workflow coordinator before it is placed back in Ray, so coordinator memory remains proportional to total output size. A bounded in-Ray reduction or aggregation path is tracked in GitHub issue #91.

Opt-in Ray result buffer¶

When the next stage is a Ray step, a bounded map can keep its complete intermediate list out of the outer workflow coordinator:

sync_resources = chain(
    step(list_namespaces),
    map_step(sync_namespace, ray_options={"num_cpus": 0.25})
    .with_limits(max_concurrency=8, max_items=500)
    .with_result_buffer(
        max_serialized_bytes=64 * 1024 * 1024,
        actor_options={
            "num_cpus": 0.25,
            "memory": 128 * 1024 * 1024,
            "resources": {"workflow_result_buffer": 1},
            "scheduling_strategy": "SPREAD",
        },
    ),
    step(summarize_sync),
)

This is an explicit transport selection. max_concurrency, max_items, and max_serialized_bytes must all be positive caller-declared limits. actor_options must explicitly set num_cpus > 0 and an integer memory request at least as large as max_serialized_bytes. The only optional version 1 actor fields are a bounded positive resources mapping and scheduling_strategy="DEFAULT" or "SPREAD"; unknown fields are rejected instead of being forwarded to Ray. Every requested custom resource must be advertised by an eligible cluster node. If it is unavailable, the ready handshake intentionally remains pending while ordinary progress flushing and cancellation stay live, and no map leaf effects begin.

The non-detached actor has fixed max_restarts=0, max_task_retries=0, max_concurrency=1, and max_pending_calls=2. django-ray waits for its ready acknowledgement through normal progress-flushing resolution before admitting leaf side effects. The leaf admission window remains at max_concurrency, but the outer coordinator uses ray.wait(..., fetch_local=False) to select a ready leaf without decoding it. It then submits and completely resolves one small append acknowledgement before issuing another actor call. The second pending-call slot only gives Ray room to retire a completed prior call from the sender handle's bookkeeping before accepting its successor. It does not permit concurrent protocol transitions, multiple unacknowledged calls, or an extra retained result. Map payloads are never resolved by the coordinator.

The byte limit counts bytes actually retained by the version 1 ray.cloudpickle codec using pickle protocol 5. An append serializes first, checks the prospective item and byte totals, and mutates retained state only when both remain within their bounds. This is not a limit on decoded Python heap size, transient serialization memory, Ray wire bytes, or object-store bytes. The scheduler memory request accounts for actor placement but is not a hard process-memory limit.

Finalization returns the ordered Python list and a small acknowledgement as two direct Ray returns. It does not call ray.put() inside the actor and does not nest an ObjectRef inside another return. The coordinator proves both returns are materialized, resolves only the acknowledgement, kills the actor best effort, and passes the one unresolved payload reference to the downstream Ray step. Append and finalization waits continue flushing ordinary workflow progress.

If the buffered map is terminal, run() necessarily resolves that final reference and the coordinator still uses O(total output) memory. A downstream step also materializes the complete ordered list and therefore uses O(total output) consumer memory, while the object store holds the final object. The actor itself transiently decodes the complete list during finalization. Chunked/reducer transport and total-independent object-store or consumer memory remain in GitHub issue #91, and production-shaped throughput and memory validation remain in GitHub issue #87.

A version 1 result-buffer map cannot be nested inside a dynamic map; plan materialization rejects that topology before actor or leaf effects so actor multiplication stays explicit. Local execution preserves input order, max_items, and ordinary leaf failure behavior without creating an actor. It validates and fingerprints the buffer selection but does not retain, cloudpickle, or measure results against max_serialized_bytes; that byte limit is specifically the Ray actor's retained-byte contract and is enforced only during Ray execution. Maps that do not call with_result_buffer() retain their existing bounded or eager transport exactly. Cleanup cancels pending or resolving leaf references plus their captured nested dependencies, then discards the serialized bytes retained by the actor and kills it best effort on failure or cancellation. Non-detached ownership also cleans the actor when its owner dies, but the protocol makes no cleanup, recovery, or durability guarantee after node loss.

Opt-in ordered result fold¶

When a workflow needs one compact summary rather than the complete mapped list, call reduce() on a bounded map:

sync_resources = chain(
    step(list_namespaces),
    map_step(sync_namespace, ray_options={"num_cpus": 0.25})
    .with_limits(max_concurrency=8, max_items=500)
    .reduce(
        step(
            merge_sync_summary,
            runtime_env="kubernetes-sync",
        ),
        initial=empty_sync_summary,
        max_serialized_bytes=8 * 1024 * 1024,
        actor_options={
            "num_cpus": 0.25,
            "memory": 16 * 1024 * 1024,
            "scheduling_strategy": "SPREAD",
        },
    ),
    step(store_sync_summary),
)

reduce() changes the map result from list[item] to one accumulator. It requires positive max_concurrency, max_items, and max_serialized_bytes limits, an explicit initial value (including explicit None when appropriate), the same narrow resource-accounted actor_options accepted by with_result_buffer(), and exactly one Step reducer. The reducer contract is a strict input-order left fold:

reducer(accumulator, item, *bound_args, **bound_kwargs) -> accumulator

The reducer must be synchronous and return a concrete non-generator value. Reducer Ray task options are rejected because it runs inside the fold actor, not as a separate Ray task. Its importable callable identity, Django bootstrap flag, bound-argument schema, and resolved effective RuntimeEnv are applied to and fingerprinted with that actor. Actor-internal calls do not create per-item progress nodes, and report_progress() or durable workflow task context inside the reducer is unsupported in version 1. Reducers should be pure: retries or failures do not provide exactly-once reducer side effects.

The initial accumulator is invocation data. Its value is cloudpickled and validated by the actor before any mapper leaf is admitted, but it is never persisted in the effective plan or included in the plan fingerprint. The plan records only the required initial binding schema. Local execution also cloudpickle round-trips the initial value before leaves so every run receives a fresh decoded accumulator, and it imports and validates the reducer before mapper effects. A reducer supplied only by a RuntimeEnv is therefore Ray-only. Local mode does not create an actor or enforce the Ray retained-byte limit.

Leaves may finish in any order, but the actor keeps its accumulator serialized and folds only the next contiguous input index. Admission credits are replenished only when items are incorporated. If the earliest item is slow, no more than min(max_items - 1, max_concurrency - 1) later results can be retained, and no more than max_concurrency mapper references remain admitted. This head-of-line backpressure is intentional: the reducer need not be associative or commutative, so a parallel tree reduction would change behavior.

The actor still executes protocol transitions serially with max_concurrency=1, and the coordinator completely resolves each small acknowledgement before issuing the next transition. max_pending_calls=2 leaves one per-handle bookkeeping slot for Ray to retire a completed prior call without rejecting its successor; it does not permit two fold transitions to execute concurrently.

max_serialized_bytes covers the serialized accumulator plus serialized out-of-order items retained by the actor. Initial, individual item, resulting accumulator, and combined-state checks are transactional: state changes only after the complete contiguous fold candidate succeeds. Overflow can still be completion-order-sensitive. For example, an early large out-of-order result may fail the combined-state check before the missing earlier item can reduce the accumulator. A schedule-independent byte limit would require a different deferred/backpressure protocol. The limit does not hard-cap temporary decoded actor heap, reducer working memory, serialization scratch space, Ray wire bytes, or object-store bytes. Ray's scheduler memory request remains placement accounting rather than process enforcement.

The coordinator waits for leaves with ray.wait(..., fetch_local=False), resolves only small protocol acknowledgements, and never decodes mapped items or the intermediate accumulator. Finalization returns the accumulator and acknowledgement as two direct Ray returns; one unresolved accumulator reference becomes the single dependency of a downstream Ray step. A terminal fold necessarily resolves that one accumulator in run(), so its final size still matters, but no O(total item) list is constructed.

Fold mode and with_result_buffer() are mutually exclusive. A version 1 fold cannot be nested inside a dynamic map, and its mapper cannot itself contain a dynamic map; plan materialization rejects both shapes before actor or leaf effects. Static chain and group mapper bodies remain supported. Mapper, reducer, actor, byte-limit, and cancellation failures stop admission, recursively cancel pending nested dependencies without fetching their payloads into the coordinator, discard actor state, and preserve the original exception. Non-detached ownership and fixed no-restart semantics match the result-buffer protocol. List-preserving chunk or hierarchical transport remains in GitHub issue #91, and production evidence remains in GitHub issue #87.

Calling map_step() without with_limits() retains the original eager behavior for compatibility. New dynamic workloads should normally choose an explicit window and an input cap. Local execution preserves inputs, ordered outputs, limits, and failures but runs leaves sequentially.

Choose Concurrency and Batch Granularity¶

For a rate-limited Kubernetes or HTTP API, max_concurrency limits concurrent batches; it does not enforce requests per second. Keep the API client's own token-bucket or server-advertised retry policy enabled. Choose the mapped item deliberately:

• Map one namespace when discovery and reconciliation can share one client session and one failure boundary.
• Map one (namespace, resource_kind) batch when namespaces contain enough resources to leave cluster capacity idle.
• Batch several tiny resources into one item when a single API round trip is shorter than Ray submission overhead.

Start with a window no larger than the external client's connection pool, benchmark throttling and retry rates, and increase it only while useful throughput improves. A preceding step can construct batches; map_step passes each batch to one leaf and returns one ordered result per batch.

Chains and Groups¶

chain passes each result as the first argument to the next signature. Reusing the module above:

pipeline = chain(
    step(build_items),
    map_step(calculate),
    step(summarize),
)

group sends the same input to every child and returns an ordered result list:

from django_ray.workflows import group


def minimum(values: list[int]) -> int:
    return min(values)


def maximum(values: list[int]) -> int:
    return max(values)


def total(values: list[int]) -> int:
    return sum(values)


inspect_values = chain(
    step(build_items),
    group(
        step(minimum),
        step(maximum),
        step(total),
    ),
)

Groups can contain chains, maps, or other groups.

Django-Aware and Native Steps¶

Workflow steps are Ray-native by default and skip Django initialization. This is the fast path for API clients, transformations, and compute that do not use Django models:

step(calculate)

Set django=True when a step needs Django's app registry, ORM, settings-dependent components, or another Django facility:

def load_account_name(account_id: int) -> str:
    from myapp.models import Account

    return Account.objects.values_list("name", flat=True).get(pk=account_id)


load_account = step(load_account_name, django=True)

Django initialization is guarded by the app registry, so a reused Ray worker does not initialize Django again for every step.

Use ray_options or Step.with_options() for Ray scheduling controls:

gpu_calculation = step(
    calculate,
    ray_options={"num_gpus": 1, "max_retries": 2},
)
two_cpu_calculation = step(calculate).with_options(num_cpus=2)

Use a named RuntimeEnv profile when a leaf needs different dependencies:

numpy_calculation = step(calculate, runtime_env="numpy-2-3")

Leaves otherwise inherit the outer durable task's environment. See Runtime Environments.

Application Progress¶

Long-running leaves can report progress without writing to Django directly:

from django_ray.workflows import report_progress


def normalize_rows(rows: list[dict[str, str]]) -> list[dict[str, str]]:
    normalized = []
    for index, row in enumerate(rows, start=1):
        normalized.append({key: value.strip() for key, value in row.items()})
        if index % 100 == 0:
            report_progress(
                index,
                len(rows),
                message="Normalizing rows",
                metrics={"last_row": index},
            )
    return normalized

report_progress() is a no-op that returns False during local or actor-free workflow execution. In full-reporting Ray execution, True means that the validated value was accepted into the leaf's best-effort producer session. It is not proof that the progress actor processed the value or that Django persisted it.

Each reporting leaf invocation retains at most one outstanding application-progress acknowledgement and one canonical latest-value replacement slot. When the acknowledgement is still pending, another valid call replaces that slot instead of adding another actor call. At leaf exit, the producer makes at most one bounded latest-value handoff before sending COMPLETED or FAILED. Structural and lifecycle events, including STARTED, COMPLETED, and FAILED, are never coalesced. Producer, acknowledgement, or diagnostic failure remains observational and cannot replace the callable's result or exception.

Metrics are bounded operational metadata: use at most 32 scalar string, number, boolean, or null values. Keys are capped at 64 UTF-8 bytes, strings at 256 UTF-8 bytes, and the normalized mapping at 4 KiB; sensitive or oversized values are redacted or replaced before a value can enter the replacement slot or cross Ray.

Graph and Progress Schema¶

During execution, full reporting writes a versioned schema-v2 compatibility snapshot suitable for a custom task-tracking UI:

{
  "schema_version": 2,
  "workflow_id": "django-ray:42",
  "run_identity": {
    "schema_version": 1,
    "run_id": "2eb22ff3-5fd2-43a0-834c-d920737b584c",
    "task_execution_pk": 42,
    "attempt_number": 2,
    "execution_generation": 5
  },
  "revision": 12,
  "state": "RUNNING",
  "progress_percent": 50.0,
  "graph": {
    "nodes": [
      {
        "node_id": "0.1.m0",
        "kind": "task",
        "label": "sync_resource",
        "callable_path": "myapp.workflows.sync_resource",
        "dependencies": ["0.0"],
        "state": "RUNNING",
        "progress": {"current": 50, "total": 100, "percent": 50.0},
        "runtime_env": {"mode": "inherit", "hash": "..."},
        "execution": {
          "ray_task_id": "...",
          "ray_job_id": "...",
          "ray_node_id": "...",
          "ray_worker_id": "..."
        }
      }
    ],
    "edges": [{"source": "0.0", "target": "0.1.m0"}]
  }
}

Node IDs are stable for one workflow expansion. Dynamic map nodes appear after their input iterable resolves, so clients should redraw when revision changes. Revisions are monotonic only within one run_identity.run_id and restart when a new invocation claims progress ownership. Clients must reset their stored graph before applying a revision from a different run ID, attempt, or execution generation. Database writes occur only when the coordinator revision changes and the task is still RUNNING with that exact attempt, generation, and run ID. The independent task-monitor heartbeat still proves that the owning worker is alive.

When WORKFLOW_PROGRESS_SCHEMA_V3_PILOT=True, one complete terminal schema-v2 actor snapshot may also be normalized into the bounded schema-v3 summary, immutable topology pages, and latest-state node detail. A successful publication becomes the preferred source for authorized bounded readers; it does not replace the live schema-v2 writes.

The task Admin can project one coherent terminal schema-v3 publication into an accessible execution graph. The section stays collapsed and performs no graph request until an operator opens it. One successful response is cached for that page; a pre-terminal NOT_REPORTED response or transport failure can be retried by closing and reopening the section. The graph is limited to 100 nodes, 256 edges, 100 detail records, and a 128 KiB response. It reads only the first bounded page of each collection and refuses cycles, unknown endpoints, count or publication mismatches, cursors, truncation, and over-limit runs without rendering a partial graph.

Graph nodes are semantic links in dependency order, while the connecting artwork is decorative and never carries outcome color. Each card labels its structural path as Node ID, presents execution state separately, and has one independent Output row. A node without an opted-in projector says that no preview was requested even while its execution state is pending or running. A submitted opted-in node reports preview PENDING until it publishes a terminal availability, and a failed application node is UNAVAILABLE. When a preview is available, that same row labels the bounded JSON value as a preview so it cannot be mistaken for the node identifier, execution state, or complete task result. Labels, states, bounded progress messages, map fan-out counts, bounded failure text, and the output-preview envelope described below are the complete display allowlist. Callable paths, arguments, internal raw results, RuntimeEnv data, Ray identifiers, raw metrics and events, and workflow-plan payloads are never returned by the graph endpoint. A failed run marks each failure origin and its incoming ancestor path without using color alone. Node links are pinned to the same retained task attempt as the graph, and the graph summary names that page-rendered attempt. If live polling advances the task to a newer attempt, reload the page before opening its graph. The bounded topology and detail JSON routes remain available as explicit diagnostic fallbacks.

When a retry succeeds after one or more failed attempts, Workflow execution presents one chronological attempt-graph stack: previous failures first and the current attempt last. Every panel is independently collapsible. Archived panels are independently lazy and pinned to their attempt across the graph, topology, and node-detail routes. Each archived panel makes at most one request per page, including when the bounded response is unavailable or rejected. The current graph appears exactly once. This keeps retry history comparable without eagerly loading hidden graphs or mixing a prior run with the current attempt.

That graph is a full-reporting detail surface. A terminal-only run instead shows its terminal outcome and OMITTED_BY_POLICY status explicitly in the live and workflow diagnostic panels. The Admin does not render or request an execution graph and does not offer topology or node-detail links for that summary.

Opt-in node output previews¶

An importable workflow leaf may explicitly project its successful result into one small operator-facing JSON value:

from typing import Any

from django_ray.workflows import step


def fulfill_order(order_id: str) -> dict[str, Any]: ...


def preview_fulfillment(result: dict[str, Any]) -> dict[str, Any]:
    return {
        "order_id": result["order_id"],
        "item_count": len(result["items"]),
        "status": result["status"],
    }


fulfillment = step(fulfill_order).with_output_preview(preview_fulfillment)

The projector must be a module-level importable callable. In full-reporting Ray execution it runs beside the leaf only after the application callable succeeds. Its return value may contain only exact JSON scalars, lists, and string-keyed objects. The preview profile permits at most four nested levels, 32 aggregate items, 16 list items, 15 object entries, 64 UTF-8 bytes per key, 256 UTF-8 bytes per string, 512 canonical JSON bytes, and 2 KiB of decoded value budget. Integer and integer-valued numbers must fit the interoperable JSON safe range, and floating-point values must be finite. Byte strings, tuples, sets, custom objects, callables, coroutines, Ray object references, and actor handles are unsupported. django-ray never calls repr() or traverses an unprojected task result.

The projector is trusted synchronous application code in the leaf process, so its work counts toward leaf runtime. Keep it cheap, deterministic, side-effect-free, and read-only with respect to the result it receives.

Terminal formatting is removed before configured redaction and before the event crosses Ray. Formatting-only changes do not mark an output REDACTED or a progress event truncated; actual policy redaction and size loss still do. Historical readers normalize the stored preview for display while evaluating the current redaction policy against the authenticated stored value, so formatting cannot hide a newly sensitive value. The graph keeps one Output row and labels an available bounded JSON value explicitly as a preview, separate from the Node ID and execution state. Its fixed envelope reports one of NOT_REQUESTED, PENDING, AVAILABLE, REDACTED, TOO_LARGE, UNSUPPORTED, FAILED, UNAVAILABLE, or OMITTED_BY_POLICY; only AVAILABLE and REDACTED carry a value. A projector import, execution, validation, reporting, or logging Exception cannot replace the successful workflow result. Process-control exceptions such as KeyboardInterrupt and SystemExit are not swallowed at preparation, projection, import, or logging seams. Failed application leaves never receive a successful preview value.

This field is diagnostic only. It is not the complete task result, a checkpoint, retry input, selective-resume marker, idempotency receipt, or proof that an external effect committed. No executor or lifecycle path reads it to decide recovery or replay. Keep credentials, model artifacts, documents, large datasets, and effect receipts out of previews even when they would fit the limit; redaction is defense in depth, not an authorization or encryption boundary.

The projector and limits profile are part of the effective-plan fingerprint, while the workflow's declared output remains only result. Events and durable detail are fenced by task, attempt, execution generation, and run ID, so a stale attempt cannot replace a current preview. Node-detail schema version 2 adds the preview envelope; stored version 1 detail remains readable and is shown as UNAVAILABLE without rewriting its bytes or digest. Readers authenticate and validate stored schema-version-2 bytes before applying the current REDACT_PATTERNS. If a newer policy matches any historical preview value, both current- and archived-attempt readers return the existing REDACTED marker for that preview only. They do not reveal the old value, rewrite its payload or digest, or degrade the rest of the graph to CORRUPT. Terminal-only and disabled policies create no node-detail graph and do not run the projector. Local execution also leaves projection disabled so local workflow semantics stay identical to the callable result.

Local Execution¶

Signatures run locally when Ray is not initialized. This makes workflow logic easy to exercise in unit tests:

result = calculation.run(4, use_ray=False)
assert result["items"] == 4

Set use_ray=True to fail instead of falling back when Ray is unavailable.

Workflow Progress Policy¶

Ray workflow progress defaults to the configured WORKFLOW_PROGRESS_REPORTING_POLICY="full". Use terminal-only reporting when a workflow needs one durable completion summary without live per-node telemetry:

result = calculation.with_progress_reporting("terminal_only").run(4, use_ray=True)

Terminal-only reporting creates no workflow progress actor, sends no node or application-progress RPCs, and writes no RayTaskExecution.progress_data. Calls to report_progress() return False, and the workflow submits the same Ray work. The accepted outer-task success or failure transition makes exactly one best-effort schema-v3 summary publication after result preparation has succeeded or the durable application error is known. It records the pinned strategy and plan fingerprint, declared plan counts, terminal outcome, and bounded timestamps. Discovered, retained, and node-state counts remain zero because no node execution evidence was collected. Detail is OMITTED_BY_POLICY; there is no topology or node-detail revision, manifest, page, or row.

Terminal-only publication is observational. Summary construction, validation, or database attachment failure cannot replace the workflow result or application error. A stale lifecycle fence accepts neither the terminal task transition nor its summary. The authorized summary API and Admin show the terminal record explicitly, while topology pages return empty omitted-by-policy responses and the Admin execution graph stays unavailable.

Use disabled reporting when no workflow-progress summary is needed:

result = calculation.with_progress_reporting("disabled").run(4, use_ray=True)

Disabled mode has the same zero-actor, zero-progress-RPC, and zero-legacy-write behavior, but does not attempt the terminal schema-v3 summary. Both actor-free modes preserve the durable outer task's state, result/error, retry, cancellation, recovery, and monitor heartbeat behavior.

The bounded task summary exposes the effective policy. Authorized bounded workflow progress readers return availability="DISABLED" with no fabricated summary or graph for the current disabled run. The plan selection is the current-attempt signal, so an active full or terminal-only run without schema v3 remains NOT_REPORTED, while a terminal full or terminal-only run without its expected schema-v3 publication is MISSING. Historical attempts without an archived schema-v3 summary remain NOT_REPORTED. Local execution has no progress actor and is recorded as disabled regardless of the configured Ray reporting default.

"full", "terminal_only", and "disabled" are executable policies. Sampled reporting remains later #79 work. The schema-v3 pilot still applies only to full mode: it collects live actor evidence and changes terminal publication, while terminal-only mode publishes its bounded summary without enabling the pilot. Changing WORKFLOW_PROGRESS_FLUSH_SECONDS only throttles full-mode database snapshots and does not remove producer or actor overhead.

Full reporting prepares every data event as canonical identity-bound JSON before the Ray call. Event payloads are capped at 16 KiB, complete wire and decoded envelopes at 32 KiB, and dependency-edge batches at 32 edges. The actor revalidates the complete task, attempt, generation, and workflow-run fence before mutation. With the schema-v3 pilot disabled, its node, edge, recent-event, and retained-byte state uses the durable V1 limits. Enabling the pilot narrows actor collection and publication to the fixed schema-v3-pilot-v1 profile: 512 nodes, 2,048 edges, 2 MiB of topology, 1 MiB of detail, and 4 MiB combined, with the byte ceilings applied to encoded and decoded evidence. Descriptive metadata is redacted before it crosses Ray.

Full reporting is acknowledgement-driven, not time-sampled. When acknowledgements keep up, it may submit every accepted application update. Under a slow acknowledgement, one leaf invocation coalesces replaceable application progress into its one latest-value slot. There is no selectable sampled policy or sampling interval.

These bounds close the individual envelope, retained collector, and per-leaf application-progress state gaps for full reporting. They do not bound the aggregate number of independently bounded sessions created by forked actor handles, their combined calls or bytes, or the actor mailbox across a whole workflow. Aggregate admission/coalescing must be proven before sampled reporting can be introduced. Bounded actor-to-preparation draining at the hard V1 ceilings also remains separate scale and default-activation work. The stricter full-detail producer pilot is therefore experimental and default-off rather than a general V1-scale claim. If actor ingress records any rejection or accepted truncation, the terminal adapter refuses schema-v3 publication instead of claiming that an incomplete graph is complete. A bounded event or actor snapshot is observational state, not a recovery protocol.

Durability Semantics¶

The outer Django task is the durability and retry boundary:

• Internal steps do not create individual Django tasks.
• In full reporting mode, an in-memory Ray coordinator collects node events. The outer task writes a schema-v2 compatibility snapshot of the actor's retained bounded state to RayTaskExecution.progress_data at WORKFLOW_PROGRESS_FLUSH_SECONDS intervals. Individual producer envelopes and retained actor nodes, edges, events, and bytes are bounded. Each reporting leaf invocation also holds no more than one outstanding mutable application-progress acknowledgement and one canonical latest-value slot during execution, followed by one bounded terminal handoff/report. Structural and lifecycle evidence is never coalesced. Actor ingress diagnostics report actor-side rejection counts, accepted events marked truncated, a versioned fixed-shape cost block, and an optional fixed-shape aggregate of accepted leaf producer reports. The producer aggregate counts valid offers, submissions, superseded and locally dropped values, producer-observed acknowledgement outcomes, and terminal-handoff outcomes. It contains no producer identities or application values. A pending acknowledgement means only that the leaf had not observed its result when it sealed the report; the actor may still process that application-progress call before the report. The actor-cost block uses saturating counters for actor-received logical calls/bytes, calls decoded under the exact run fence by fixed event kind, end-to-end processed delivery delay, ingest handler wall/process CPU time, and snapshot-build wall/process CPU time through the retained snapshot. It contains no application payloads or variable-cardinality producer labels. A producer-side failure before actor submission cannot appear in those counters, and processed delivery delay includes transport, scheduling, queueing, and clock effects rather than isolating mailbox lag. The aggregate Ray mailbox across forked handles, snapshot-to-preparer drain, and transient snapshot materialization are not yet governed by the later admission/backpressure contract. Terminal-only and disabled modes bypass that live coordinator, event codec, actor, producer session, and snapshot path while leaving the durable outer-task boundary intact. Terminal-only still makes its one bounded best-effort summary publication after the application reaches success or failure; disabled does not.
• A separate nullable workflow_progress_summary_json field and schema-v3 codec are deployed reader-first. The field is fixed-shape and capped at 16 KiB of canonical UTF-8 JSON. Package-owned topology/detail tables and the internal storage writer can now publish a verified immutable topology, sparse normalized latest-state detail, and that summary pointer in one transaction. The standalone schema-v3 writer rejects topology/detail pointers; it is the only path for an intentional summary-only DISABLED or OMITTED_BY_POLICY record, which creates no empty detail storage. Terminal-only execution now uses the omitted-by-policy path for one terminal summary without a live actor. Authorized public readers are implemented. The default-off pilot makes one best-effort terminal publication from an internally consistent full-reporting actor snapshot, revalidates the pinned plan and exact run fence, then stages and atomically promotes topology, detail, and summary. For a failed run, the coordinator keeps polling within the existing terminal flush deadline until every transitive ancestor of each failed node is reported succeeded. This closes cross-sender actor delivery races without inferring a completion event; if the causal fence does not close, schema v3 remains unpublished. Rejected or truncated ingress, invalid cross-field evidence, admission overflow, preparation truncation, a stale fence, or storage failure likewise leaves schema v3 unpublished and emits a stable bounded diagnostic; it never changes the application result. The periodic schema-v2 writer remains active for rolling compatibility. Terminal-only does not solve or weaken the remaining full-reporting boundaries: aggregate producer/mailbox admission and coalescing across forked handles, the sampled policy that depends on that aggregate bound, bounded actor-to-preparer draining, remaining live cost attribution, large-fan-out slow-consumer evidence, default schema-v3 activation, and old-writer drain remain #79 and its dependent deliveries.
• A workflow invocation atomically claims workflow_run_id. Retry, cancellation, timeout, LOST recovery, and a newer invocation prevent its old coordinator from writing again; rejected reporters drain later leaf events without persisting them.
• Before the first leaf submission, the workflow stores a bounded, secret-free effective-plan manifest, fingerprint, and strategy selection independently from progress. A retry must reproduce the pinned fingerprint and have retry-safe RuntimeEnv bindings, or it fails closed before submitting changed work.
• Progress includes node paths, callable labels, node states, completion counts, dependency edges, percent complete, explicit leaf progress, Ray execution IDs, runtime environment identity, and recent events.
• A leaf failure fails the outer task.
• Retrying the outer task reruns the workflow from its entry node, including previously completed leaves. Workflow progress, successful nodes, and output previews are diagnostic evidence, not checkpoints, reusable results, selective-resume records, or authorization to skip an external side effect. Normalized node detail does not retain leaf return values.
• TaskAttempt archives terminal task diagnostics, not workflow graphs. If a run has already published a canonical terminal schema-v3 summary, the same bounded value is archived with its attempt before retry cleanup. Otherwise, lifecycle reconciliation derives a terminal envelope from the last accepted running summary under the row lock. The outer outcome and expiry are authoritative. Success marks every discovered node complete; interrupted outcomes retain the last accepted counters. Invalid summaries and legacy complete snapshots are not copied. This keeps retry history bounded; progress_data, the current summary, and workflow_run_id otherwise describe only the current attempt or its latest terminal invocation.
• Cancellation of a Ray Core outer task recursively cancels its child tasks through Ray's normal cancellation behavior.
• The final workflow result must satisfy the same result-serialization rules as any django-ray task.

ADR-0004 accepts the replacement storage contract: a dedicated always-bounded task-row summary points to immutable topology manifests/pages and normalized latest-state node-detail rows. Changed detail rows are batch-upserted before the

81-fenced summary pointer advances; each accepted detail-changing publication expires¶

prior cursors, while static topology is not duplicated across repeated publications or state transitions within one run and topology version. Authorized readers use revision-bound pagination. The ADR defines exact V1 limits, availability states, legacy v1/v2 reads, retention, and cleanup. Its first delivery now implements the nullable current/per-attempt summary fields, strict schema-v3 codec, monotonic exact-run writer primitive, bounded rolling reader, and lifecycle archival. Its second delivery adds the run-scoped topology manifests/pages, normalized latest-state rows, bounded staging and integrity verification, sparse atomic publication, terminal expiry, and retention/orphan cleanup. Public detail services are implemented. The runtime producer continues to write schema-v2 compatibility snapshots of retained actor state and may additionally publish one terminal schema-v3 record only when the stricter pilot is explicitly enabled and admitted. ADR-0005's production topology phase now externalizes exact node/edge identity, duplicate, reference, and selection state into a private bounded SQLite workspace and removes it before returning prepared evidence. The unchanged result still includes complete observed_node_ids for the existing materialized detail API, so #142 must complete the shared topology/detail lifetime before broader activation. See ADR-0004: Bounded Workflow Progress Storage and ADR-0005: Bounded Workflow Progress Preparation.

Ray Core tasks already run inside an initialized Ray worker. Ray Job drivers initialize their cluster connection lazily when a workflow first requests Ray, and use the same durable context and progress graph protocol.

Future execution strategies must preserve this outer durability boundary. In particular, Compiled Graph is a possible engine for a validated static actor region, not a Django task type or a flag that makes data-dependent map_step expansion static. See Workflow Plans and Execution Strategies.

Apply database migrations before starting upgraded workers. Migration 0012_workflow_progress_summary adds nullable summary fields, and migration 0013_workflow_progress_detail_storage adds package-owned topology and detail tables. Neither migration rewrites existing progress_data; older writers continue to work and upgraded readers retain the 64 MiB schema-v1/v2 compatibility cap. The package-level pilot remains disabled unless explicitly configured. Deploy the authorized public facade and bounded storage before opting in, keep old workflow writers compatible through the rollout, and complete the remaining mailbox, preparation, aggregate-spill, and migration work before enabling schema v3 by default or admitting workloads near the hard V1 ceilings.

Existing rows start with workflow_run_id = NULL and a nullable plan identity so an older writer can still insert during the rollout; the first upgraded, fully identified workflow invocation claims a UUID and pins its plan. Prepared actors or graphs in later strategies must drain when that fingerprint changes. Custom uses of durable_task_execution() that omit the attempt or execution generation continue to run their workflow but intentionally do not persist the plan or progress because their writes cannot be fenced safely.

The drain statement is a contract for those later strategies. The current release provides fingerprint comparison and stale-owner helper functions but has no resident graph owner or prepared-graph cache to drain.

An identified context whose attempt or execution generation is stale is different: its plan claim fails closed before local application code, progress actors, or Ray leaves can run.

Use pure or idempotent steps when retries can repeat external side effects. An external operation should receive a stable application idempotency key and return a durable receipt; a database-only record cannot atomically cover a remote side effect. Split a side-effecting stage into a separate Django task when it needs an independent durable retry, result, cancellation, or audit boundary.

Durable selective stage resume remains a planned extension. It requires verified intermediate outputs and side-effect receipts bound to the plan, inputs, code, RuntimeEnv, and retry identity. Progress and future node-output previews are observational rather than recovery logs: after a cluster loss, the outer task retry remains the recovery boundary. Never infer that an external effect is safe to skip or repeat solely from the node color or terminal state.

Test Project Examples¶

The bundled test project exposes several workflow experiments in its Swagger UI:

POST /api/cluster/workflow-benchmark
GET  /api/cluster/workflow-benchmark/{task_id}

POST /api/cluster/complex-workflow
POST /api/cluster/complex-workflow?failure_branch=slow&failure_item=0
POST /api/cluster/complex-workflow?reporting_policy=terminal_only
GET  /api/cluster/complex-workflow/{task_id}

POST /api/cluster/workflow-showcase
POST /api/cluster/workflow-showcase?item_count=1&work_seconds=0.01
GET  /api/cluster/workflow-showcase/{task_id}

POST /api/cluster/workflow-recovery-showcase
POST /api/cluster/workflow-recovery-showcase?item_count=1&work_seconds=0.01
GET  /api/cluster/workflow-recovery-showcase/{task_id}

GET  /api/cluster/workflows/{task_id}
GET  /api/cluster/workflows/{task_id}/topology/nodes
GET  /api/cluster/workflows/{task_id}/topology/edges
GET  /api/cluster/workflows/{task_id}/nodes
GET  /api/cluster/workflows/{task_id}/node-detail?node_id={node_id}

The pre-0.4.0 live-node test route at /api/cluster/workflows/{task_id}/nodes/{node_id} is removed. Use the durable indexed node-detail route after the same callable authorization as every other workflow read. Applications that need live Ray state or logs can build a separately authorized surface around the package get_workflow_node_snapshot() helper; the testproject does not imply a tenant or task-owner policy for that data.

The complex example runs this nested shape:

chain(
    build_config,
    group(
        chain(build_fast_items, map(fast_leaf), summarize_fast),
        chain(build_slow_items, map(slow_leaf), summarize_slow),
    ),
    summarize_workflow,
)

Every workflow poller above exposes only a bounded aggregate progress-summary envelope. A published schema-v3 summary is preferred. Supported older stored progress may contribute sanitized aggregate counts, but the poller never returns its complete schema-v2 graph. Its exact database projection excludes task input, RuntimeEnv snapshots, workflow plans, completion envelopes, and other unrelated payloads. Current inline result and error diagnostics are each guarded at 16,384 bytes before transfer, external result storage is never loaded, and the complete response is at most 65,536 bytes. The fixed result omission vocabulary is external_result_not_loaded, stored_result_exceeds_poll_limit, malformed_inline_result, and encoded_response_limit; the error vocabulary is stored_error_exceeds_poll_limit and encoded_response_limit. A null reason means the corresponding value is available. Included values still pass through the normal presentation-redaction policy.

The bundled testproject enables WORKFLOW_PROGRESS_SCHEMA_V3_PILOT by default, so an admitted terminal run becomes available through the bounded summary, topology-node, topology-edge, and node-detail routes above. The guarded local KubeRay gate exercises those routes against the real producer path and requires non-empty, mutually consistent topology and detail. It runs both the small successful shape and a deterministic slow-branch failure, requires the failed task to remain on its first attempt, and verifies the authenticated Admin graph against the same bounded publication. Open either terminal RayTaskExecution in the Admin and expand Workflow execution, then Execution graph, to inspect the dependency order. The failed fixture identifies the originating map node and its incoming ancestor path while retaining successful sibling context.

Order-fulfillment showcase¶

The showcase endpoint is the richer visual example. It builds one order, validates each item, joins customer and inventory context, splits reservation from commercial analysis, joins the decision, and fans out to primary, audit, and notification sinks before finalization. The default three-item invocation retains 25 runtime nodes and 36 edges while still using one outer RayTaskExecution:

POST /api/cluster/workflow-showcase?item_count=3&work_seconds=0.05
GET  /api/cluster/workflow-showcase/{task_id}

Open that execution in the Admin to inspect the graph as longest-path layers. Every card links to the bounded indexed node-detail reader for the same displayed attempt. The validation map exposes exactly item_id and valid; the reservation map exposes exactly item_id and reserved_units, while retaining max_retries=0. Customer-history node 0.1.g1.0.g1 deliberately uses a failing projector so the real run proves a successful application node remains successful while its Output row says the preview failed. Other configured leaves label their bounded business summaries as previews. The endpoint always selects full reporting; it deliberately has no reporting-policy knob. Use the existing complex-workflow fixture when comparing full, terminal-only, and disabled policy behavior.

The graph renderer is package-owned and extends stock Django Admin with scoped fallback styles; Unfold remains an optional testproject-only shell rather than a renderer dependency.

The guarded local KubeRay gate uses the smaller deterministic success:

POST /api/cluster/workflow-showcase?item_count=1&work_seconds=0.01

That run must publish exactly 21 runtime nodes, 28 dependency edges, 12 derived longest-path layers, 21 usable detail targets, and one durable attempt. The gate checks the exact two map-leaf preview values and the deliberate projector-failure status rather than accepting arbitrary serialized results. The matching failure fixture is:

POST /api/cluster/workflow-showcase?item_count=1&work_seconds=0.01&failure_stage=reserve_inventory&failure_item=0

It fails only reservation leaf 0.5.m0. The fulfillment decision, three sinks, and finalizer remain pending. Commercial analysis joins reservation preparation first, so its price, risk, recommendation, and join nodes are structurally guaranteed to have succeeded before the selected reservation fails. This makes the Admin failure path useful without fabricating independent durable retries for Ray-native leaves. The gate also requires the failed reservation's preview to be UNAVAILABLE, while the validation preview and successful customer-history task retain their exact safe and diagnostic-only statuses.

To keep that successful visual workload moving slowly through a local dashboard, name its opt-in Locust class explicitly:

export DJANGO_API_TOKEN="<local testproject token>"
uv run locust -f locustfile.py --host=http://localhost:30080 \
  --headless -u 1 -r 1 -t 5m --stop-timeout 150 WorkflowShowcaseUser

WorkflowShowcaseUser is excluded from default class discovery, fixes the population at one user, requires each three-item success and complete 25-node/36-edge publication before submitting the next, and never injects the expected failure fixture. The 150-second graceful-stop window covers bounded enqueue, terminal polling, final detail validation, and scheduling margin for the scenario active at the five-minute cutoff.

Three-attempt recovery showcase¶

The recovery endpoint demonstrates the outer-task retry boundary with one task ID and one fixed, server-owned failure sequence:

POST /api/cluster/workflow-recovery-showcase?item_count=1&work_seconds=0.01
GET  /api/cluster/workflow-recovery-showcase/{task_id}

The bundled testproject allows three durable attempts. Attempt 1 fails at build_order_batch, before the branches can run. Attempt 2 starts again at build_order_batch, reruns its upstream work, and fails at join_order_inputs. Attempt 3 starts from the entry again and completes the full workflow. The polling response exposes the ordered FAILED, FAILED, SUCCEEDED attempt history and the bounded successful outer result, whose recovery field identifies attempt 3. It reads at most four ordered attempt rows and guards each archived attempt error at 4,096 bytes; stored_error_exceeds_attempt_limit and encoded_response_limit distinguish omitted errors. Current result and error values keep the common 16,384-byte diagnostic guard, and the complete recovery response remains under 65,536 bytes. Callers cannot choose the failure stage through the API, so the example remains deterministic.

The endpoint binds this task to the testproject's recovery-showcase task backend and returns runtime_env_profile="recovery-showcase" while polling. Local and Compose runs content-hash the sample's source import roots. Kubernetes builds a deterministic archive containing those sources plus the locked Django task runtime; the Ray Client task manager uploads it as a content-addressed GCS package before submission, so the upstream generic Ray image needs no django-ray installation. A missing or invalid backend/profile/archive, or a profile without an immutable retry identity, is a configuration error and returns 503; the endpoint never falls back to the default project profile. This matters because project contains mutable package constraints and an opaque shared archive URI. Those are valid for a first dynamic execution but cannot prove that a later durable attempt sees identical bytes.

Inspect each retained attempt through the bounded readers by adding ?attempt_number=1, ?attempt_number=2, or ?attempt_number=3, for example:

GET /api/cluster/workflows/{task_id}?attempt_number=1
GET /api/cluster/workflows/{task_id}/nodes?attempt_number=2

The Admin attempt history links to each immutable attempt detail. Every archived full-reporting attempt exposes an Open graph for attempt #N link pinned to that run, so operators do not need to construct the query string themselves. A graph identifier such as 0.3.g1.1 is the stable node path for that expansion, not a status or task result. The card's Output line reports availability only: pending and running work remains pending, failed output is unavailable, and a succeeded value is not retained. Node detail likewise reports execution state and bounded diagnostics rather than leaf return values. Read the recovery endpoint's outer result for the durable workflow output.

This fixture deliberately uses pure, side-effect-free steps. It proves replay and attempt fencing, not checkpointed continuation: no successful node from attempt 1 or 2 is reused by attempt 3. Applications that call external systems must make replayed steps idempotent or implement their own durable effect/checkpoint protocol. Selective workflow resume remains future work.

Omitting reporting_policy preserves the existing full-reporting fixture. Set reporting_policy=terminal_only to exercise the actor-free summary path with a small success:

POST /api/cluster/complex-workflow?fast_items=2&slow_items=1&fast_seconds=0.01&slow_seconds=0.02&reporting_policy=terminal_only

Add failure_branch=slow&failure_item=0 for the deterministic terminal-only failure. Poll the returned task ID through GET /api/cluster/workflows/{task_id}. The response should report the terminal outcome, pinned strategy, declared plan counts, zero discovered nodes, and detail.availability="OMITTED_BY_POLICY". Topology and node-detail routes return empty omitted-by-policy responses, and the Admin shows no execution graph for either fixture.

API Reference¶