Changelog¶

All notable changes to RADICAL AsyncFlow will be documented in this file.

The format is based on Keep a Changelog.

[Unreleased]¶

[0.7.0] 2026-11-10¶

Fixed¶

• Cancelling a running task cancels it on its own backend (#118) — with several backends, cancel() on a running task sent the cancel to the default backend, so a task routed elsewhere with backend= kept running and its future resolved only when that backend reported it. The cancel now goes to the backend the task was submitted to, and is sent once: cancelling a block reaches a running member through both the block's cascade and the future its body awaits, which sent the member's cancel twice. A cancel the backend refuses, fails on, or that cannot be sent (no running event loop) can be retried.
• A component's uid is set when the call returns — calling a task or block returned its future before the engine assigned the component's uid, so future.id did not exist until the next event-loop step. A caller that read it right after the call got an AttributeError, and could not key its own records on the uid that telemetry events and spans carry as task_id. The uid is now assigned at the call, so uids also follow call order. A call whose registration fails keeps its uid on the future, which carries the exception.
• Telemetry names the backend that ran each task — with several backends, every task span carried the default backend in its backend attribute, even for a task routed elsewhere with backend=. So did each task's TaskCreated, asyncflow.TaskResolved, TaskStarted, TaskCompleted, TaskFailed and TaskCanceled events, and the task metrics labelled from them; only TaskQueued and TaskSubmitted named the right backend. Every task event now names the backend the task is routed to, and the task span takes that name from TaskCreated. "The backend attribute" in docs/telemetry.md says what the attribute means.
• Shutting down one engine no longer makes another reuse uids (#85) — component uids came from one module-level counter that every WorkflowEngine.shutdown() reset, so once one engine shut down, another engine still running in the same process numbered its next component task.000001 again. That component replaced the earlier one in flow.components, and its future never completed. The counter is no longer reset, so a uid names one component across all engines in a process, and an engine created after another has shut down continues the numbering instead of starting again from 000001. The reset_uid_counter() helper in radical.asyncflow.utils is removed.
• A dependent no longer fails when its dependency has not registered yet — an executable or prompt task registers only after its body returns, so in b(a()) with an a whose body awaits, b registered first. The engine then did not recognise a's future as a dependency and b failed with TypeError: '_asyncio.Future' object is not subscriptable. A task or block now waits for the components whose futures it receives, positionally or as keyword arguments, to register before it registers itself, then waits for their results as before. A failed or cancelled dependency propagates as DependencyFailureError or cancellation, also when the dependency's own registration failed, for example because its body raised or returned no command.
• A registration that fails part way no longer leaves the component half registered — when registering a task or block failed after the engine had stored it, for example in dependency tracking or in a telemetry hook, its future carried the exception but kept future.task (or future.block), and the component stayed in flow.components. A dependent called with that future registered against it and waited forever. The engine now undoes everything it stored for the component, so a dependent fails with DependencyFailureError, as for any other registration that fails. Telemetry events already emitted for the component stay.
• A plain asyncio future passed to a task or block is rejected with a clear error — calling a task or block with a plain asyncio Future or Task, for example a(asyncio.ensure_future(coro())) or a(x=loop.create_future()), made a fail with TypeError: '_asyncio.Task' object is not subscriptable. Only the futures returned by the engine's tasks and blocks are dependencies, so the call now raises a TypeError that names the argument: await the future and pass its result, or await it in a block and pass the block's future. "What does not count as a dependency" in docs/dependencies.md says so.
• A future of another engine passed to a task or block is rejected with a clear error — calling a task or block of one WorkflowEngine with the future of a task or block of another, for example f2_task(fa) with fa = f1_task(), made the engine count fa as a dependency it never resolves, so f2_task never ran and awaiting it hung, even after fa had finished. Only the futures of the engine's own tasks and blocks are dependencies, so the call now raises a TypeError that names the argument, as for a plain asyncio future. "What does not count as a dependency" in docs/dependencies.md says so.
• shutdown() cancels components that have not registered yet — it cancelled only registered components, so an executable or prompt task whose body was still running, or a dependent waiting for its dependencies to register, kept a pending future, and awaiting it, or any future depending on it, hung after shutdown. A body that finished after shutdown then registered into the shut-down engine. Shutdown now cancels their registration and their futures, so awaiting them raises CancelledError, and a body that outlives shutdown, or a call made after it, does not register.
• future.state matches how the future ended — three cases left state out of step with the outcome. A task or block cancelled with cancel() before it registered stayed "PENDING", and so did a running block for the done callbacks added before the cancellation. A task whose registration failed, for example an executable task that returned an empty command, stayed "PENDING" with the exception set. Under asyncio.eager_task_factory, a block that finished without suspending ended "RUNNING" instead of "DONE" or "FAILED". Each now ends "CANCELLED", "FAILED" or "DONE", set before its done callbacks run.
• cancel() on a task that has not registered — a task cancelled before it registered still registered, emitted TaskCreated and TaskResolved with no terminal event, and an executable or prompt task's body ran on. It now never registers, and its body stops. A cancel bound before the task registered, such as loop.call_later(t, fut.cancel), cancelled a running task only in the engine, never on the execution backend; it now behaves like fut.cancel() called later.
• An unknown backend= name is rejected when the function is decorated — a task decorated with a backend= name that none of the engine's backends has raised a KeyError in the engine's run loop when it was submitted, and the whole engine shut down: that task, the other tasks submitted with it and every later task were cancelled, not only the misrouted one. The decorator now raises a ValueError that lists the engine's backends, before any task runs. The note in docs/multi_backend.md, which said only the task's future failed, says so.

Added¶

• future.uid — the full uid ("task.000001", "block.000002"), set when the call returns, beside the suffix in future.id. It is the component's key in flow.components and the task_id of its telemetry events and, for a task, of its span.
• Task identity and trace linking documentation — "Task identity" in docs/async_workflows.md, and "Linking tasks to your own traces" in docs/telemetry.md: the span that is current when a task is called, in-process or extracted from a W3C traceparent carrier, becomes the task span's parent.
• More checks in CI — pull requests also run the unit tests under asyncio.eager_task_factory and report the coverage of the lines they add to src/. The unit tests now include Hypothesis properties over random workflows, and every test checks what it leaves behind: ERROR log records, and futures or records an engine keeps after shutdown(). stress.yml runs the scale tests in tests/stress and 2,000 random workflows under both task factories, nightly, by hand, and on pull requests labelled stress. "Running the tests" in docs/install.md lists the commands.
• A benchmark of the engine's cost — benchmarks/engine_overhead.py compares the per-task cost and the memory of two installed versions. benchmark.yml runs it on pull requests that touch the package or the benchmark, and fails one when the number of garbage collections rises by more than 5%. Each night it compares 40,000 tasks on main with the commit of the night before, and nightly-report.yml opens an issue when a nightly stress or benchmark run fails.

Removed¶

• Python 3.9 and 3.10 support — AsyncFlow now requires Python 3.11 or newer (requires-python = ">=3.11"). Python 3.9 reached end of life in October 2025 and 3.10 in October 2026, and CI no longer tests them: the py39, py39-all, py310 and py310-all tox environments and their CI jobs are gone, and Ruff targets py311.

[0.6.0] 2026-10-19¶

Fixed¶

• A CancelledError inside a block body no longer hangs the block — execute_block caught only Exception, so when a block body raised asyncio.CancelledError (e.g. by awaiting a member task that was cancelled) the block's future was never resolved, and the block and every component depending on it waited forever. The block is now cancelled, like a block cancelled from outside: its still-running members and its dependents are cancelled too.

• Explicit file dependencies no longer shut the engine down — an OutputFile → InputFile match made the engine call link_explicit_data_deps on every backend and read a "target" from its return value, but only few backends returns one, so on all other backends the resulting TypeError shut the whole engine down. The data-linking hooks are now called only on backends that declare the matching capability (supports_implicit_data_linking / supports_explicit_data_linking, read once when the backend is attached). The built-in backends declare neither, and backends without the attributes are treated the same way. RADICAL-Pilot data linking requires RHAPSODY newer than 0.6.0, which declares the capabilities.

• Futures passed as keyword arguments are now dependencies — task(x=other()) previously ignored other when building the dependency graph, so the task could start before other finished and fail with InvalidStateError, a failed other surfaced as its raw exception instead of DependencyFailureError, and a cancelled other stopped the engine's scheduler loop silently. Keyword-argument futures are now tracked exactly like positional ones, and a cancelled argument future cancels the dependent task instead of stopping the scheduler.

• Same future passed more than once no longer hangs (#98) — task(a, a) counted a as two dependencies but released only one, so the task never ran. A component passed several times is now a single dependency, and every argument still receives its result.

Changed¶

• Component state handlers renamed — WorkflowEngine.handle_task_success, handle_task_failure and handle_task_cancellation are now handle_component_success, handle_component_failure and handle_component_cancellation. They handle blocks as well as tasks: execute_block and shutdown() now route block results, failures and cancellations through them, so every component changes state in one place. The old names are removed.

Added¶

• Task Dependencies documentation page (docs/dependencies.md) — how to specify dependencies (positional and keyword arguments) and exactly what counts as one.

[0.5.1] - 2026-08-20¶

Fixed¶

• Engine creation on a non-main-thread event loop — WorkflowEngine.create() no longer crashes when the event loop it runs on is started from a secondary thread. Signal handlers (SIGHUP/SIGTERM/SIGINT) can only be installed from the main thread; _setup_signal_handlers() now detects this case and skips registration with a warning instead of raising, leaving shutdown to be managed by the host process.

Added¶

• Conda recipe — new recipe/meta.yaml for a conda-forge staged-recipes submission (noarch: python). Python floor set to 3.10, matching conda-forge's floor and avoiding a CPython 3.9 bug where importlib.metadata doesn't normalize dots in distribution names, which breaks version("radical.asyncflow") at import time.
• ReadTheDocs configuration — .readthedocs.yaml builds the mkdocs site on Python 3.12, installing the .[doc] extra.

[0.5.0] - 2026-06-16¶

Added¶

• future.state attribute — every component future (task and block) now exposes a .state string attribute tracking its full lifecycle: PENDING → RUNNING → DONE / FAILED / CANCELLED. Block futures now correctly transition through all states; previously they remained at PENDING for the duration of their execution. Readable at any point without awaiting.

• workflow_id= call-time kwarg — any task or block call now accepts workflow_id="<id>" as a keyword argument to tag that specific component. Takes precedence over any active workflow_scope(). The kwarg is stripped before the function body is invoked.

Fixed¶

• shutdown() no longer crashes when blocks are still running — block futures are now cancelled directly instead of routing through handle_task_cancellation, which assumed original_cancel was set (only true for task futures).

• Block future state lifecycle — block futures now transition to RUNNING immediately after asyncio.create_task, to DONE on normal completion, and to FAILED when the block body raises. Previously all three transitions were missing.

• _block_members cleanup on non-cancelled outcomes — member sets are now removed from _block_members for every terminal outcome (DONE, FAILED, CANCELLED). Previously the cleanup only ran on cancellation, leaving stale entries after normal or failed block completion.

• patched_cancel forwards msg argument — fut.cancel(msg=...) now correctly forwards all positional and keyword arguments to the underlying asyncio.Future.cancel(). Previously the msg was silently dropped, breaking callers that rely on the cancellation message being propagated through CancelledError.

• Pending-task cancellation sets future.state — when a pending task is cancelled locally via patched_cancel, future.state is now set to "CANCELLED". All other cancellation paths already set this attribute; this was the only missing case.

• _clear_internal_records completeness — resolved, running, _task_submit_times, and _task_start_times are now cleared alongside the other internal structures. Previously these accumulated stale entries across engine reuse after shutdown().

• _block_asyncio_tasks cleared on shutdown — _clear_internal_records now also clears the asyncio.Task registry for blocks, eliminating stale references after engine reset.

Changed¶

• self.running changed from list to set — membership check (uid in self.running) and removal (running.discard) in the run loop and patched_cancel are now O(1) instead of O(n). No public API change.

[0.4.0] - 2026-05-18¶

Added¶

• WorkflowEngine.workflow_scope(workflow_id=None) — new async context manager that groups all tasks registered inside under a shared asyncflow.workflow_id. Internally sets _workflow_id_ctx (a module-level ContextVar), which asyncio copies into every create_task / gather branch automatically so concurrent workflow instances remain isolated. When telemetry is active, also opens an OTel "workflow" span, making all task spans inside structural children in the trace hierarchy. Auto-generates a short UUID if workflow_id is None.

• _workflow_id_ctx ContextVar — per-instance ContextVar (default None) that carries the active workflow ID across asyncio task boundaries without explicit argument passing. Each WorkflowEngine instance owns its own uniquely-named ContextVar (asyncflow_workflow_id.<uid>), preventing context leakage when multiple engines are used within the same coroutine. @flow.block execution sets it to the block's UID so tasks inside a block inherit the workflow ID automatically.

• WorkflowEngine.start_telemetry() new parameters — span_processors, metric_readers, resource — forwarded to TelemetryManager.__init__(), mirroring Session.start_telemetry(). Enables passing pre-built OTel exporters (OTLP, Prometheus, Jaeger) without any RHAPSODY code changes.

• Span enricher for asyncflow.workflow_id — registered automatically by start_telemetry() via register_span_enricher(). Stamps asyncflow.workflow_id onto every task OTel span when the task was created inside a workflow_scope() or @flow.block. Enables per-workflow Gantt views and span filtering in Jaeger / Grafana Tempo.

• _emit() workflow_id kwarg — when workflow_id is set, injects asyncflow.workflow_id into the event's attributes dict. All task lifecycle events emitted by WorkflowEngine (TaskCreated, asyncflow.TaskResolved, TaskSubmitted) carry the active workflow ID.

• Example 01-workflow_grouping.py — complete rewrite as an HPC Campaign Manager simulation. Models 4 workflow types with resource tracking and dependency chains:

• simulate (4 tasks, GPU, no deps) — molecular dynamics runs
• analyze (4 tasks, GPU, deps=simulate) — post-processing per simulation
• train (8 tasks, GPU, no deps) — distributed ML training

• evaluate (8 tasks, CPU, deps=train) — lightweight model evaluation Uses ResourcePool (asyncio-queue-based GPU/CPU slot tracking) and emits campaign.ResourceAssigned custom events to record per-instance resource assignments in the JSONL checkpoint.

• plot_campaign.py — new plotting script producing a two-panel Campaign Manager timeline figure:

• Top panel: Gantt chart with rows per workflow instance, bars coloured by workflow type and labelled with the assigned resource (gpu:N / cpu:N). Right-margin annotations show priority / cpu / gpu per type. Right column renders a config table, scheduler description box, and dependency graph.

• Bottom panel: GPU (left axis) and CPU (right axis) utilisation as step functions over elapsed time, with total-capacity dashed reference lines.

• capture_stdio decorator parameter — @flow.executable_task(capture_stdio=True) redirects stdout/stderr from executable tasks directly to files instead of collecting them in memory. The awaited future resolves to the stdout file path ({work_dir}/{uid}/{task_uid}.stdout) instead of a decoded string. Honoured by all RHAPSODY backends (ConcurrentExecutionBackend, DaskExecutionBackend, DragonExecutionBackendV3); silently ignored for function tasks. capture_stdio is stored as a top-level field in comp_desc (not nested inside task_backend_specific_kwargs) so RHAPSODY backends can read it as task.get("capture_stdio").

• WorkflowEngine uid and work_dir parameters — WorkflowEngine.create(uid=..., work_dir=...) sets the engine's unique identifier (UUID) and working directory. The engine is now the single authority for backend._work_dir; it sets backend._work_dir = {work_dir}/{uid} and creates the directory before any tasks run (mirrors Session.add_backend() in RHAPSODY).
• WorkflowEngine._attach_backend() — new internal method that replaces the inline register_callback loop. Handles _work_dir injection, is_attached flag, and callback registration in one place.
• Unit tests for capture_stdio field placement, default value, _work_dir authority, and end-to-end file I/O with ConcurrentExecutionBackend.

Fixed¶

• Block spans parented correctly — execute_block() now benefits from the span_scope() session-span fallback added in RHAPSODY: block spans that run in a context with no active OTel span (e.g. spawned from run() before start_telemetry() was awaited) now correctly nest under the session root span instead of floating as unrooted traces.

• execute_block() dead branch removed — the run_in_executor code path (used when the wrapped function was sync) is removed. WorkflowEngine enforces a strict async API; all block functions must be async def.

Changed¶

• execute_block() uses nullcontext — the no-telemetry code path uses stdlib nullcontext (Python >= 3.7) instead of the former custom _null_context() helper, which is removed.

• execute_block() sets _workflow_id_ctx — the block's UID is set as the active _workflow_id_ctx for the duration of block execution so every task registered inside the block inherits it as asyncflow.workflow_id without requiring an explicit workflow_scope() call.

Docs¶

• docs/telemetry.md:
• Added "Forwarding to an external backend" section with corrected span_processors / metric_readers code examples (replaces the broken set_tracer_provider() pattern).
• Added "workflow_scope() context manager" section with usage examples and auto-ID generation.
• Added "OTel span hierarchy" section showing the four-level session -> workflow -> block -> task tree and explaining how asyncflow.workflow_id propagates to every span attribute and JSONL event.
• Updated start_telemetry() signature to include span_processors, metric_readers, and resource parameters.

[0.3.1] - 2026-03-09¶

Fixed¶

• executable_task command parsing — shlex.split is now applied correctly to the command string returned by the decorated function, ensuring commands with arguments and quoted strings are parsed properly.

[0.3.0] - 2026-02-24¶

Added¶

• prompt_task decorator — new first-class task type alongside function_task and executable_task. The decorated async function returns a prompt string that is forwarded to an AI execution backend (e.g. DragonVllmInferenceBackend from RHAPSODY). NoopExecutionBackend returns "Dummy Prompt Output" for testing; LocalExecutionBackend raises NotImplementedError with a clear message directing users to register an AI backend.
• Multi-backend registry — WorkflowEngine.create(backend=...) now accepts either a single backend or a list of pre-initialized named backends (each must expose a .name attribute). The first backend in the list is the default. Built-in backends (LocalExecutionBackend, NoopExecutionBackend) gain a name parameter defaulting to "default".
• Per-task backend routing — task decorators (function_task, executable_task, prompt_task) accept an optional backend="<name>" parameter to route a specific task to a named backend. Tasks without backend= are sent to the default backend.
• Parallel backend submission — when a batch contains tasks routed to multiple backends, submit() dispatches all backend calls concurrently via asyncio.gather. shutdown() similarly drains all registered backends in parallel.
• Example 04-concurrent_backends.py — demonstrates a compute + AI dual-backend workflow running document processing pipelines concurrently.
• Tests — unit tests (tests/unit/test_prompt_task.py) and integration tests (tests/integration/test_multi_backend.py) covering prompt_task registration, validation, multi-backend routing, parallel submission, and graceful shutdown.

Fixed¶

• async_wrapper exception propagation — errors raised during task registration (e.g. prompt_task returning None, or executable_task returning a non-string) are now propagated to the task's asyncio.Future instead of being silently swallowed by the detached asyncio.create_task. Previously, such errors would cause await task(...) to hang indefinitely.

Changed¶

• Documentation — all HPC backend references across docs, examples, and the Jupyter tutorial now point to the official RHAPSODY documentation instead of the GitHub repository, and include a direct link to the AsyncFlow integration guide.
• Examples — HPC example scripts (04-dask_execution_backend.py, 05-radical_execution_backend.py, 06-dragon_execution_backend.py) updated with module-level docstrings referencing RHAPSODY docs and the AsyncFlow integration guide.
• Proof-of-concept examples — all basic/getting-started examples (basic.md, async_workflows.md, composite_workflow.md, examples 01–03) consistently use LocalExecutionBackend; RHAPSODY backends are only shown in dedicated HPC sections.
• Install command — RHAPSODY install command corrected to pip install rhapsody-py across all docs and examples.

[0.2.0] - 2026-02-12¶

Added¶

• Local execution backend — LocalExecutionBackend ships built-in, supporting both ThreadPoolExecutor and ProcessPoolExecutor with cloudpickle serialization.
• No-op backend — NoopExecutionBackend for testing and dry_run workflows without computational overhead.
• HPC execution via RHAPSODY — HPC backends (RadicalExecutionBackend, DaskExecutionBackend, DragonExecutionBackendV3) are now provided by the separate RHAPSODY package and plug into AsyncFlow via simple import.
• Composite workflows — @flow.block decorator for grouping tasks into logical units with support for nested blocks and inter-block dependencies.
• Task dependency resolution — automatic DAG-based scheduling with implicit and explicit data dependencies between function_task and executable_task types.
• Pluggable backend architecture — any backend conforming to the execution interface can be passed to WorkflowEngine.create(), enabling custom and third-party backends.
• Pre-commit pipeline — ruff, docformatter, typos, actionlint, detect-secrets, and standard pre-commit hooks.
• CI/CD — GitHub Actions workflows for unit tests, integration tests, pre-commit checks, and documentation deployment across Python 3.9-3.13.

Changed¶

• Backend decoupling — HPC backends (Dask, RadicalPilot, Concurrent, Dragon) moved from radical.asyncflow to the standalone rhapsody package. AsyncFlow now only ships LocalExecutionBackend and NoopExecutionBackend.
• Imports for HPC backends — HPC backends are now imported from rhapsody.backends instead of radical.asyncflow.
• Documentation — all docs and examples updated to reflect the new architecture, with clear separation between built-in local backends and RHAPSODY HPC backends.

[0.1.0] - 2025-12-07¶

Initial public release of RADICAL AsyncFlow.

Added¶

• Async-first workflow engine with DAG-based task scheduling.
• Support for async and sync function execution within workflows.
• Execution backend abstraction with pluggable backend interface.
• RADICAL-based HPC execution backend integration.
• Agentic workflow support.
• Event-driven shutdown mechanism for backend lifecycle management.
• Task cancellation signaling and cooperative cancellation support.
• Enhanced async context manager support for workflow execution.
• Unit and integration testing framework.
• Initial project documentation.
• Migration to pyproject.toml with tox-based test environments.

Changed¶

• Renamed project from flow to asyncflow.
• Removed legacy synchronous execution paradigm in favor of fully async architecture.
• Refactored execution loop for thread-based backend.
• Updated RADICAL execution backend callback mechanism to support service-style operation.
• Improved error propagation for failed tasks.
• Refactored internal utilities to remove dependency on radical.utils.
• Eliminated redundant data dependency linking logic.

Fixed¶

• Corrected handling of task failure without explicit wait().
• Fixed handle_task_failure logic for consistent error reporting.
• Resolved process-based function execution launch issues.
• Ensured pre_exec directives are properly applied.
• Addressed multiple execution edge cases in RADICAL backend.