Pass Dag runs directly to Deadline references

[imrichardwu]

Preserve the Dag run evaluation context for custom references that accept additional keyword arguments, so they continue to evaluate correctly.

Pass the in-memory Dag run to Deadline references so built-in references avoid redundant database reads and custom references can evaluate the current state.

Deprecate required_kwargs while preserving it for existing custom references. Use the shared evaluator when recalculating queued-at deadlines.

closes: #70710

---

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

• Yes (please specify the tool below)

---

• 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, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[imrichardwu]

@ferruzzi can I get a review?

[github-actions]

uv.lock on main just moved via #71520 ("Restore the evaluation extra of google-cloud-aiplatform"), commit b69253b and this PR currently conflicts.

Quickest fix:

git fetch upstream main && git rebase upstream/main
rm uv.lock && uv lock
git add uv.lock && git rebase --continue
git push --force-with-lease

Automated nudge — ignore if you're not ready to rebase. This comment is updated in place on future uv.lock bumps.

[ferruzzi]

Heads up on sequencing: this PR now overlaps two others in _recalculate_dagrun_queued_at_deadlines.

• Fix TypeError clearing a DAG run when a queued-at deadline interval is stored as JSON #70370 (opened 24 July) is the targeted fix for the JSON-interval TypeError on that function. It is milestoned 3.3.2 and carries backport-to-v3-3-test.

• Re-anchor custom queued deadline references when a Dag run is cleared #71850 (opened 19 August) persists the evaluation timing so custom queued-anchored references get re-anchored on clear. It also conflicts with this PR across sevreral files.

This PR fixes the same TypeError incidentally, and I want to be explicit that that does not make #70370 redundant. This is a refactor with a significant.rst newsfragment and a deprecation, so it is not backport material; #70370 is the only route the fix has to 3.3.x, where it currently affects every queued-at deadline. Please do not let this PR be used as a reason to close that one.

The cheapest merge order looks like it will be to merge #70370 first, then rebase here. #70370 adds decode_deadline_alert_model() and resolve_deadline_alert_interval() to serialization/decoders.py, which is exactly what this PR hand-rolls at taskinstance.py:255-260. Rebasing onto it lets you delete those lines.

[ferruzzi]

Adding a fourth PR to the picture since it breaks something here concretely.

#71802 adds a core-side SerializedVariableInterval, makes decode_deadline_alert return that instead of VariableInterval, and deletes resolve() from the SDK entirely.

That will break test_clear_task_instances_recalculates_variable_interval_deadline, which patches airflow.sdk.definitions.deadline.Variable.get. Worth knowing now rather than after a rebase.

Also worth noting, #71850's diff still has required_kwargs: set[str] = set() on
SerializedBaseDeadlineReference, which this PR removes, and it adds an evaluation_timing class attribute immediately below it. So be careful when rebasing, whichever lands first.

[ferruzzi]

Final nit, sorry.