Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Cursor IDE with Claude sonnet 4 1/2

closes: #59303

Summary

Restore support for custom deadline references by implementing a SerializedCustomReference wrapper in Core that bridges SDK defined custom refs with core serialization for deadline alerts.

Motivation

After PR #61118 moved deadline alert decoding from SDK to Core and introduced SerializedReferenceModels, custom deadline references stopped working. The decoder only looked up types in SerializedReferenceModels, but custom refs registered via the SDK's DeadlineReference.register_custom_reference() only existed in ReferenceModels. This caused "No reference class found with name: MyCustomRef" errors.

What has changed?

Core (airflow-core/)

SerializedCustomReference wrapper

• Added SerializedReferenceModels.SerializedCustomReference class in serialization/definitions/deadline.py
• Wraps custom deadline references imported from user code
• Implements the evaluate_with() logic since SDK's BaseDeadlineReference is lightweight wrapper without it
• Delegates _evaluate_with() execution to the inner custom ref

Decoder (serialization/decoders.py)

• When __class_path is present in serialized data, uses SerializedCustomReference.deserialize_reference() to import and wrap the custom ref
• Builtin types continue to use SerializedReferenceModels.get_reference_class() lookup

DeadlineAlert model

• reference_class property returns SerializedCustomReference when __class_path is present in the stored reference dict
• Returns standard serialized types for built-ins

Create deadline & prune logic (serialization/definitions/dag.py, models/dagrun.py)

• No changes needed - already use SerializedReferenceModels.TYPES.DAGRUN
• Custom refs now work because SerializedCustomReference is in that tuple

Architecture with these changes

SDK (task-sdk/) - DAG Authoring
├── deadline.py
│ ├── DeadlineAlert (user-facing)
│ ├── DeadlineReference (public interface)
│ └── BaseDeadlineReference (lightweight, serialization only)
Core (airflow-core/) - Scheduling/Execution
├── serialization/
│ ├── encoders.py (adds class_path for custom types)
│ ├── decoders.py (wraps custom types in SerializedCustomReference)
│ └── definitions/deadline.py
│ ├── SerializedReferenceModels
│ │ ├── Built-in types (with evaluate_with)
│ │ └── SerializedCustomReference (wrapper for custom refs)
│ └── TYPES.DAGRUN (includes wrapper)

How custom refs work now

1. Encoder adds __class_path for custom refs (no changes)
2. Decoder sees __class_path, imports custom class, wraps in SerializedCustomReference
3. isinstance(ref, TYPES.DAGRUN) → True (wrapper is in tuple) → deadline created
4. Wrapper's evaluate_with() validates kwargs, calls custom ref's _evaluate_with(), adds interval
5. reference_class in TYPES.DAGRUN → True → deadline pruned on success

Testing

Define the custom reference in a plugin:

from airflow.sdk.definitions.deadline import DeadlineReference, deadline_reference, BaseDeadlineReference
from airflow._shared.timezones import timezone


@deadline_reference()
class MyCustomRef(BaseDeadlineReference):
    """Custom ref: deadline is now + interval."""

    def _evaluate_with(self, *, session, **kwargs):
        return timezone.utcnow()


DeadlineReference.register_custom_reference(MyCustomRef)

async def on_deadline(**kwargs):
    print("Custom ref deadline exceeded", kwargs)

DAG:

from datetime import datetime, timedelta

from airflow import DAG
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk.definitions.deadline import AsyncCallback, DeadlineAlert, DeadlineReference

from custom_deadline_refs import on_deadline


with DAG(
    dag_id="testing-custom-reference-deadlines",
    start_date=datetime(2025, 1, 1),
    schedule=None,
    deadline=DeadlineAlert(
        reference=DeadlineReference.MyCustomRef,
        interval=timedelta(seconds=10),
        callback=AsyncCallback(on_deadline),
    ),
):
    BashOperator(task_id="example_task", bash_command="sleep 120")

Result:

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.