[seanghaeli]

Deadline callbacks get a Dag run snapshot that the scheduler stores in their kwargs when the deadline is missed, and Airflow never renders their kwargs, so {{ dag_run.run_id }} reaches the callback as raw text. This replaces #64984. The scheduler now stores only the ids, and the callback gets a task's context without the task fields, built where the callback runs and rendered the same way a task's is. There's one commit per review point:

1. One context builder. build_dag_run_context in execution_time/context.py is the Dag run part of get_template_context, which calls it and keeps its cached dict. process_params takes no task. A test checks every key it returns is in a task context.
2. Callback keys. Both paths build the context with that builder where the callback runs, with callback=CallbackMeta(source=CallbackSource.DEADLINE), deadline with deadline_time parsed back to an aware datetime, and xcom_pull on the sync path. If the Dag run is gone, the callback fails on both paths.
3. The Dag comes from where tasks get it. The executor callback subprocess parses the Dag file like task_runner.parse, and the triggerer sends the serialized Dag on the workload like _create_workload does for start_from_trigger.
4. Rendering. Function kwargs render like op_kwargs: Templater.render_template, nested values included, after decoding, with the Dag's get_template_env(). A value that fails to render is passed through unrendered with a warning. Notifier kwargs aren't pre-rendered, since Notifiers render their own template_fields.
5. Typing. deadline is a NotRequired key on Context, so it's in KNOWN_CONTEXT_KEYS, and the callback builder returns Context.

Serialized Dags don't keep user_defined_macros or user_defined_filters, so async callbacks can't use them, the same as start_from_trigger tasks. That's documented in the deadline alerts docs and will be fixed for both in a follow-up. prev_*_success isn't in the context yet, since the Execution API only looks it up by task instance. The executor reads the Dag run with the single-use callback token from #71192, and this adds no new reads on top of it.

context["dag_run"] is now an object, deadline_time a datetime, and deadline a context key, so there's a significant newsfragment.

E2E in Breeze (scheduler, LocalExecutor, dag-processor, triggerer, postgres): one Dag with a sync and an async function callback and a sync and an async Notifier, kwargs using {{ dag_run.run_id }}, {{ deadline.deadline_time | ds }}, a nested dict and a user macro, triggered with conf {"braces": "{{ ds }}"}. The async macro warning is the serialized Dag limit above, and conf={{ ds }} shows each Notifier value is rendered once:

main:     SYNC_FN>>> core='run_id={{ dag_run.run_id }} deadline_ds={{ deadline.deadline_time | ds }}'
main:     SYNC_FN>>> nested={'n': 3, 'days': ['{{ ds_nodash }}', {'dag': '{{ dag_run.dag_id }}'}]}
main:     SYNC_NOTIFIER>>> message='run_id= deadline_ds=2026-11-07 conf={{ ds }}'
PR head:  SYNC_FN>>> core='run_id={{ dag_run.run_id }} deadline_ds={{ deadline.deadline_time | ds }}'
PR head:  Could not render the template in callback kwarg 'core', passing it unrendered: AttributeError: 'str' object has no attribute 'strftime'
PR head:  SYNC_FN>>> nested={'n': 3, 'days': ['{{ ds_nodash }}', {'dag': '{{ dag_run.dag_id }}'}]}
PR head:  ExecutorCallback ProbeNotifier failed / TriggererCallback ProbeNotifier failed (AttributeError: 'str' object has no attribute 'strftime')
after:    SYNC_FN>>> core='run_id=manual__2026-11-07T02:12:20.954118+00:00 deadline_ds=2026-11-07'
after:    SYNC_FN>>> nested={'n': 3, 'days': ['20261001', {'dag': 'deadline_tmpl_probe'}]}
after:    SYNC_FN>>> macro='team=data-eng'
after:    SYNC_FN>>> callback=CallbackMeta(source=<CallbackSource.DEADLINE: 'deadline'>) deadline_time_type=DateTime has_dag=True
after:    ASYNC_FN>>> core='run_id=manual__2026-11-07T02:12:20.954118+00:00 deadline_ds=2026-11-07'
after:    ASYNC_FN>>> nested={'n': 3, 'days': ['20261001', {'dag': 'deadline_tmpl_probe'}]}
after:    Could not render the template in callback kwarg, passing it unrendered kwarg=macro error="UndefinedError: 'team_name' is undefined"
after:    SYNC_NOTIFIER>>> message='run_id=manual__2026-11-07T02:12:20.954118+00:00 deadline_ds=2026-11-07 conf={{ ds }}'
after:    ASYNC_NOTIFIER>>> message='run_id=manual__2026-11-07T02:12:20.954118+00:00 deadline_ds=2026-11-07 conf={{ ds }}'
after:    all 4 callbacks success

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5.5) following the guidelines

[seanghaeli]

@ferruzzi

[ferruzzi]

Looks like it needs a rebase, and left a non-blocking question. Otherwise LGTM

[ramitkataria]

Thanks for picking this up again. I like storing only ids in the scheduler and building the context where the callback runs. However, I would like to ideally see this implemented the same way as tasks like we have discussed earlier:

Target: a deadline callback gets a task's context without the task fields, rendered the same way a task is.

1. One context builder. Move the Dag-run part of RuntimeTaskInstance.get_template_context (task_runner.py) into a function in execution_time/context.py that takes a Dag run and an optional Dag. It returns dag_run, run_id, the logical_date/ds/ts/data_interval_* keys, partition_key, partition_date, triggering_asset_events, macros, var and conn, plus dag and params when it has the Dag (process_params needs to accept no task). get_template_context calls it and adds the task keys, keeping its cached-dict behaviour. Both callback paths use it in place of build_context_from_dag_run, and a test asserts every key it returns is also in a task context.
2. Render the way tasks do. For plain-function callbacks, render every kwarg the way PythonOperator renders op_kwargs (Templater.render_template, nested values included), after the kwargs are decoded. Don't pre-render Notifier kwargs: Notifiers render their own template_fields, so each value is rendered once. Use the Dag's get_template_env() so StrictUndefined and user macros/filters apply. Keep this PR's fallback of passing a value unrendered with a warning when it fails to render.
3. Load the Dag where tasks get it. Executor: parse the Dag file in the callback subprocess the way task_runner.parse does. Triggerer: send the serialized Dag on the workload, as _create_workload does for start_from_trigger. Serialized Dags drop user macros/filters, so async callbacks share that limit with start_from_trigger until it is fixed for both; document it.
4. Callback-specific keys. Set context["callback"] = CallbackMeta(source=CallbackSource.DEADLINE) as Dag callbacks do. Keep context["deadline"] with deadline_time as a datetime, and add the sync path's xcom_pull to the context before rendering. Add prev_*_success once the Execution API can look it up by Dag run instead of by TI id.

[ramitkataria]

This turns context["deadline"]["deadline_time"] from a datetime into a string on both paths. Callbacks doing date math on it break, and {{ deadline.deadline_time | ds }} raises AttributeError and is passed unrendered. Storing the datetime won't survive the workload JSON to remote workers, so I'd parse it back to an aware datetime when the context is built, with a test on the type