Status

Motivation

Apache Airflow has always treated tasks as stateless, idempotent units of work. This design philosophy has served the project well and remains the right default for the vast majority of use cases. However, a growing class of production workloads requires that a task be able to persist a certain amount of state information and retrieve it in a subsequent execution of that same task. This could be for a subsequent execution of the same task i.e. a retry of the same DAG run, or a future run of the same DAG.

Three distinct patterns have emerged in the Airflow community, each representing a well-understood engineering requirement, and each currently forcing users to work around the absence of first-class support:

Pattern 1: Incremental Processing and Asset Watermarking

Incremental processing is one of the most common data engineering patterns. A task processing data from an external asset such as an S3 bucket, a relational database, or a message stream should not re-scan the entire asset on every run. Instead, it needs to know the high-water mark of the last successful run, so that it can process only new or updated records. 

This pattern is central to event-driven scheduling with Asset Watchers introduced in Airflow 3.0. A Trigger monitoring an S3 bucket for new files must persist a timestamp watermark between executions; without it, it must re-scan the entire bucket on every poll. The absence of state persistence has been identified as a primary reason for the slow community adoption of Asset Watcher operators despite being one of the headline features of Airflow 3.0.

Pattern 2: External Resource Lifecycle Management

Operators that interact with external systems face two failure modes when the Airflow worker is disrupted.

The first is “job resumption”. Operators that submit long-running work to external systems such as Spark or Flink. The Airflow worker is disrupted (pod eviction, node maintenance, network partition) while the external job continues running. In the current Airflow model, the operator's on_kill() handler has no access to persisted state, so it is forced to cancel the healthy external job and restart from scratch on retry. For a job that was 83% complete after two and a half hours, this represents a significant waste of compute resources and delays downstream pipelines.

The second is resource lifecycle management. Operators allocate expensive resources such as compute clusters before starting the remote job. If the worker is disrupted mid-execution, a retry without any persisted state would re-allocate these resources causing extra cost, or could also run into compute ceilings causing task failures and pipeline delays. 

What is needed is a mechanism for the operator to persist the external job information, along with the associated context, before any potential disruption, so that a retry can reconnect to the running job rather than submit a new one. This also enables cleanup after completion to be handled correctly, whenever the retry reaches completion.

Pattern 3: Intra-Task Progress Checkpointing

Long-running tasks that process large collections of items — for example, downloading and ingesting 10,000 files, or executing a series of API calls with pagination — currently have no way to record incremental progress within a single task execution. If such a task is killed at 70% completion, the retry must start from the beginning, reprocessing all previously completed work. This is a direct waste of compute resources and significantly increases end-to-end latency.

With the introduction of native async support for PythonOperator in Airflow 3.2 (AIP-98), this pattern has become significantly more prevalent. An async task orchestrating thousands of concurrent I/O operations needs to be able to periodically checkpoint its progress so that a retry can resume from the last successful checkpoint rather than restart from zero.

Why Not XCom or Variables?

XCom is the existing mechanism for passing data between tasks within a DAG run. It is explicitly not suited for the use cases described above, for several reasons.

XCom records are cleared at task start (and at the start of each retry as well). This is correct behaviour for XCom's intended purpose of passing data between tasks, but it is directly incompatible with the state persistence patterns described above, all of which require that state survive across retries.

XCom is scoped to a DAG run, cleared on retry. An XCom value is keyed by dag_id, task_id, dag_run_id, and key. There is no natural mechanism for a task to retrieve the XCom value it pushed in a previous DAG run, without explicit, fragile cross-run queries. Task state, by contrast, is keyed by task identity (dag_id, run_id, and task_id) and is explicitly designed to survive across retries and multiple DAG runs.

XCom is designed for inter-task communication, not intra-task state. The semantics, access patterns, and garbage collection mechanics appropriate for passing data between tasks are fundamentally different from those needed for a task to persist its own working state. Conflating these two concerns in a single mechanism would add complexity to XCom and make both use cases harder to reason about.

For these reasons, this AIP proposes a new, separate Task State mechanism with its own data model, its own API surface, its own pluggable storage backend, and its own garbage collection semantics. The following table summarises the distinction:

 Relationship to Other AIPs

This AIP is a foundation AIP. It defines the Task State model, the Task SDK interface, the Execution API endpoints, and the pluggable backend abstraction. The following AIPs build directly on this foundation and are expected to be revised to use it:

AIP-93 (Asset Watermarks and State Variables): AIP-93 proposed a StateVariable model specifically for Asset Watcher triggers. This AIP supersedes that proposal for the storage and retrieval mechanism. AIP-93 will be revised to use the Task State interface defined here as its persistence layer, and will focus specifically on the integration of watermarking with Asset Watchers and event-driven scheduling.

AIP-96 (Resumable Operators): AIP-96 proposed a TaskCheckpointed exception and a new CHECKPOINTED task instance state, with storage of the remote_job_id threaded through a bespoke mechanism. AIP-96 is superseded by the `ResumableJobMixin` base class shipped as part of this AIP: rather than a new CHECKPOINTED  task instance state and a resume_job()  callback, the reconnect decision lives inside a single execute_resumable()  method on the mixin. On retry, it checks the persisted external id's status and either reconnects to a still-running job, returns the result directly if the job already succeeded (skipping both polling and resubmission), or resubmits if the job failed or the id could no longer be found. Operator authors implement six methods: submit_job , get_job_status , is_job_active , is_job_succeeded , poll_until_complete , get_job_result.

AIP-xx (Task Result Caching): An upcoming AIP will propose using this foundation for caching the results of task execution.

This AIP is explicitly not intended to replace XCom, modify the existing Datasets or Data Assets model, or change the behaviour of the Airflow Scheduler state machine. It is purely additive.

Considerations

What change do you propose to make?

At a high level, this AIP proposes the following changes:

• Introduce a Task State data model supporting two distinct scoping modes: a task-scoped identity (dag_id, run_id, task_id) and an asset-scoped state, keyed by asset identity. Both modes share a common key-value storage structure and are accessed through a unified backend interface.
• Expose Task State read and write operations through the Airflow Execution API (AIP-72), so that task code never requires direct database access.
• Expose Task State operations through the Airflow Task SDK, supporting both synchronous and asynchronous access patterns consistent with AIP-98.
• Define a pluggable Task State backend abstraction, with the Airflow metadata database as the default implementation for development and simple deployments, and with the ability to substitute alternative backends (such as object stores) for very large production deployments (if needed).
• Provide configurable garbage collection semantics per backend, allowing operators and deployment managers to control the retention lifecycle of task state independently of DAG run lifecycle.
• Surface task state in the Airflow UI, allowing operators to inspect and manually reset state where required for debugging and operational purposes.

Task SDK Interface

Task State operations are exposed through the Task SDK via two context objects, accessible from within task execution code. The task_state_store  context object provides access to task-scoped state, keyed by the current task’s identity. The asset_state_store  context object provides access to asset-scoped state, keyed by the asset identity.  Asset Watcher tasks are expected to use the asset_state for watermarks and other state that is logically owned by the asset rather than by a specific task. 

from __future__ import annotations
from dataclasses import dataclass

@dataclass(frozen=True)
class TaskScope:
    dag_id: str
    run_id: str
    task_id: str

@dataclass(frozen=True)
class AssetScope:
    asset_id: str

StateScope = TaskScope | AssetScope

All methods will be available in both synchronous and asynchronous forms, consistent with the CommsDecoder changes introduced in AIP-98. An async task may call await task_state.aset(...) without blocking the event loop; a synchronous task calls task_state.set(...) directly.

Example usage for each of the three patterns is provided below.

Example: Incremental Watermarking

from airflow.sdk import asset, get_current_context

@asset.watcher
def ingest_from_s3(**context):
    asset_state = context['asset_state_store']

    # Retrieve the state of the Asset
    last_watermark = asset_state.get('last_processed_at') or '1970-01-01T00:00:00Z'
    new_files = s3_hook.list_keys(after=last_watermark)

    # Write the state of the asset to be used later
    asset_state.set('last_processed_at', datetime.utcnow().isoformat())

    return new_files

Example: External Job Resumption

from airflow.sdk import task, get_current_context
class DatabricksOperator(BaseOperator):
    def execute(self, context):
        tss = context['task_state_store']
        job_id = tss.get('remote_job_id')
        if job_id:
            status = self.get_job_status(job_id)
            if status in ('RUNNING', 'PENDING'):
                return self.poll_until_complete(job_id)

        job_id = self.submit_databricks_job()
        tss.set('remote_job_id', job_id)
        return self.poll_until_complete(job_id)

    def on_kill(self, execution_context):
        # Task state already persisted in execute(); no extra work needed here.
        pass

Example: Intra-Task Progress Checkpointing

from airflow.sdk import task, get_current_context

import asyncio, json

@task
async def process_files(files: list[str], **context):
    task_state = context['task_state_store']
    processed = json.loads(await task_state.aget('processed_files') or '[]')
    remaining = [f for f in files if f not in processed]

    for batch in chunked(remaining, 50):
        await asyncio.gather(*[download_and_ingest(f) for f in batch])
        processed.extend(batch)
        await task_state.aset('processed_files', json.dumps(processed))

Execution API Endpoints

Task State operations are exposed through the Airflow Execution API established in AIP-72, under the /execution/store/  prefix. All requests carry the per-task-try JWT token established in AIP-72, ensuring that state writes are attributable to a specific task identity. The Execution API server enforces that a task may only read and write state for its own (dag_id, run_id, task_id) identity.

For task instance state, the endpoints are:

GET /execution/store/ti/{ti_uuid}/{key}
PUT /execution/store/ti/{ti_uuid}/{key}
DELETE /execution/store/ti/{ti_uuid}/{key}
DELETE /execution/store/ti/{ti_uuid}

The API server validates the JWT to confirm that the requesting task's identity matches the `ti_uuid` in the path. Cross-task state access is not permitted through this interface by design.

For asset state, a task only knows the asset's name or URI from its own inlet/outlet declaration, not its internal numeric id, so the endpoints are addressed by name or URI instead, with the server resolving the id internally:

GET/PUT/DELETE /execution/store/asset/by-name/value?name={asset_name}&key={key}
GET/PUT/DELETE /execution/store/asset/by-uri/value?uri={asset_uri}&key={key}
DELETE /execution/store/asset/by-name/clear?name={asset_name}
DELETE /execution/store/asset/by-uri/clear?uri={asset_uri}

The JWT security model only allows a task to write asset-scoped state for assets it is already registered with as an inlet or outlet, rather than any asset in the entire deployment.

Task State Data Model

The Task State model is keyed by task identity or by asset_identity, not necessarily by task instance identity. This is the fundamental distinction from XCom and from the task_instance table. 

The database schema is largely an internal implementation detail.

Pluggable Task State Backend

Most readers can skim the configuration part of this section - the default backend requires no configuration and is operational out of the box.

The Task State backend is independently pluggable, separate from the XCom backend. The default implementation uses the Airflow metadata database, making it immediately available in all existing Airflow deployments with no additional infrastructure requirements. This default is appropriate for development environments and most production deployments. 

Extremely large Airflow deployments can choose to build an alternative implementation possibly based on an object storage backend (for example, Amazon S3, Google Cloud Storage, or Azure Blob Storage), leveraging the backend abstraction below.

The backend abstraction defines the following interface:

class BaseStoreBackend:

    def get(self, scope: StateScope, key: str) -> str | None: ...
    def set(self, scope: StateScope, key: str, value: str) -> None: ...
    def delete(self, scope: StateScope, key: str) -> None: ...
    def clear(self, scope: StateScope) -> None: ...

    # Async variants (required for AIP-98 compatibility)
    async def aget(self, scope: StateScope, key: str) -> str | None: ...
    async def aset(self, scope: StateScope, key: str, value: str) -> None: ...
    async def adelete(self, scope: StateScope, key: str) -> None: ...
    async def aclear(self, scope: StateScope) -> None: ...

The backend is configured in airflow.cfg under a new [state_store]  section:

[state_store]

backend = airflow.state_store.backends.db.DbStoreBackend
# For object store: airflow.providers.amazon.aws.state_store.S3StoreBackend

Garbage Collection and Retention

Unlike XCom, whose lifecycle is tied to the DAG run, Task State must have independently configurable retention semantics. The appropriate retention policy depends on the use case:

• A watermark for an incremental operator should be retained indefinitely, or until the operator explicitly resets it.
• A remote job identifier for a resumable operator should be retained only as long as is needed to survive a disruption and retry cycle — typically a short period measured in hours or days.
• An intra-task progress checkpoint should be cleared once the task completes successfully, since the checkpoint is no longer needed after a successful run.

Retention policy is therefore configured at two levels. First, a default retention policy is configured per deployment in airflow.cfg . Second, an operator can declare a retention policy for its own state keys, which overrides the default:

[state_store]
default_retention_days = 30  # Default: retain for 30 days
clear_on_success = False     # Default: do not clear on task success

The backend implementation is responsible for enforcing the retention policy. For the database backend, a periodic cleanup job (analogous to the existing DAG run cleanup mechanism) ‘removes expired records. 

UI Surface

Task State should be surfaced in the Airflow UI in a manner analogous to how Variables and XCom are currently displayed. The proposed UI additions are:

• A Task State panel within the Task Instance detail view, showing all current state keys and values for that task identity.
• The ability for authorised operators to manually delete individual state keys or clear all state for a task, to support operational reset scenarios.
• Display of updated_at, updated_by_run, and updated_by_try fields for each state entry, to support debugging and auditing.

Security Considerations

Task State access is governed by the same JWT-based per-task-try identity established in AIP-72. A task may only read and write state for its own (dag_id, run_id, task_id) identity. The Execution API server enforces this constraint.

For asset-scoped state, the JWT-per-task-try authorization model does not directly apply, since multiple tasks across different DAGs may legitimately read or write state for the same asset. The Execution API server enforces that a task may only write asset-scoped state for assets that it is explicitly registered with. Read access to asset-scoped state is permitted for any task that references the asset. The precise authorization rules for asset-scoped state are expected to be defined in AIP-93, which governs the integration of watermarking with Asset Watchers.

As with XCom, state values are stored serialized as strings. It is the responsibility of the operator author to ensure that sensitive values are not stored in task state without appropriate encryption. 

Which users are affected by the change?

This change is purely additive and non-breaking. No existing user is required to change any DAG code or deployment configuration to benefit from this AIP.

DAG Authors: Gain access to a first-class, Airflow-supported mechanism for persisting task state, removing the need for the existing workarounds using XCom, Variables, or external state stores.

Operator Authors: Can build stateful operators — incremental processors, resumable external job operators, progress-checkpointing operators — against a clean, stable SDK interface without relying on implementation details of XCom or the metadata database.

Deployment Managers: Gain a new configurable component (the Task State backend) to manage. The default database backend requires no new infrastructure. Production deployments may choose to configure an alternative backend appropriate to their scale and performance requirements.

How are users affected by the change? (e.g. DB upgrade required?)

A database migration is required to add the task_state table to the Airflow metadata database. This migration is fully automated and follows the existing Alembic migration pattern. No manual action is required from users.

Users who do not use the new Task State feature are entirely unaffected. Existing XCom behaviour, Variable behaviour, and all existing operator code is unchanged.

What is the level of migration effort (manual and automated) needed for the users to adapt to the breaking changes? (especially in context of Airflow 3)

None

Are There Any Downsides to This Change?

The primary downside is the addition of a new concept and a new component to the Airflow mental model. The community has rightly been cautious about introducing new abstractions. The case for doing so here rests on the following:

• The need is well-established and has been raised repeatedly since AIP-30 in 2020. The community has been working around the absence of this feature for five years.
• The three use cases described above (AIP-93, AIP-96, and AIP-98) all independently converge on the same requirement. A single foundation AIP is preferable to three separate, potentially inconsistent solutions.
• The proposed interface is intentionally minimal. The TaskState model has fewer fields than XCom, and the SDK interface has four methods. There is no risk of scope creep in the core model.
• The change is purely additive. No existing behaviour is modified.

Out of Scope

The following items are explicitly out of scope for this AIP and may be addressed in subsequent AIPs:

• Cross-task state sharing. Task State is scoped to a single task identity. Sharing state between different tasks remains the responsibility of XCom.
• DAG-level state. This AIP does not introduce a DAG-scoped state model. ProcessState as proposed in AIP-30 is out of scope.
• State migration tooling. Tools for migrating existing workaround-based state (e.g., from Variables) into the new Task State mechanism are out of scope for this AIP.
• Non-Python Task SDK language bindings for state. The initial implementation targets the Python Task SDK. State access from Golang and other language bindings (as planned in AIP-72) is deferred to a follow-on release.

What defines this AIP as "done"?

1. The task_state table and Alembic migration are merged into the Airflow metadata database schema.
2. The BaseTaskStateBackend abstraction and the default DbTaskStateBackend implementation are merged.
3. The Execution API endpoints for task state read, write, and delete operations are implemented and version-stamped using the CalVer approach established in AIP-72.
4. The Task SDK exposes task_state as a context object with the synchronous and asynchronous methods defined in this AIP, with the CommsDecoder correctly handling both calling contexts as established in AIP-98.
5. The [task_state] configuration section is documented and the default backend is operational in a standard Airflow deployment with no additional configuration.
6. The Airflow UI surfaces task state in the Task Instance detail view with the ability to delete individual keys or clear all state for a task.
7. At least one reference implementation of each of the three use cases (watermarking, external job resumption, intra-task checkpointing) is provided, either as updated provider operators or as documented examples.
8. AIP-93 and AIP-96 are revised to use the Task State interface defined here as their persistence layer, removing any duplicate storage mechanisms proposed in those AIPs.
9. Documentation covers the Task State model, the SDK interface, the backend configuration, and the garbage collection semantics.