Fix TypeError clearing a DAG run when a queued-at deadline interval is stored as JSON

[hkc-8010]

Summary

Clearing a DAG run or task instance whose DAG defines a deadline referencing the run's queued-at time (DeadlineReference.DAGRUN_QUEUED_AT) raised TypeError: unsupported type for timedelta seconds component: dict and surfaced as an HTTP 500 in the API server.

Since 3.3.0 the deadline_alert.interval column is JSON (migration 0117_3_3_0_change_deadline_interval_to_json.py, from #64751), holding a serialized timedelta or variable-backed interval, e.g. {"__classname__": "datetime.timedelta", "__version__": 2, "__data__": 3600.0}. _recalculate_dagrun_queued_at_deadlines (called from clear_task_instances) still passed that raw dict into timedelta(seconds=...). Every other consumer (DagRun creation in serialization/definitions/dag.py, the model __repr__) decodes the JSON first; only the clear path was missed.

DagRun creation already had the decode plus the interval resolution written out inline, so rather than copy it into the clear path, both steps move into serialization/decoders.py and both call sites use them.

Changes

• airflow-core/src/airflow/serialization/decoders.py: add decode_deadline_alert_model() (decode a DeadlineAlert ORM row) and resolve_deadline_alert_interval() (resolve a SerializedVariableInterval to a timedelta). decode_deadline_alert gets its missing return annotation.
• airflow-core/src/airflow/models/taskinstance.py: _recalculate_dagrun_queued_at_deadlines calls the two helpers instead of building the encoded dict itself. SerializedVariableInterval.resolve() raises ValueError when the Variable is missing or is not an integer, and decode_deadline_alert raises on a payload it refuses to decode, so on the clear path those are caught and logged and that deadline is left at its old time rather than failing the whole clear. DagRun creation still raises.
• airflow-core/src/airflow/serialization/definitions/dag.py: _process_dagrun_deadline_alerts was doing the same two steps inline and now calls the helpers, so its DeadlineAlertFields, SerializedVariableInterval, DAT and Encoding imports are gone.

Session forwarding

#71968 added the session parameter to SerializedVariableInterval.resolve(), and #72651 started passing it from DagRun creation. That matters: this runs under the scheduler's prohibit_commit guard, and without an explicit session Variable.get falls through to @provide_session, which hands back the scheduler's own scoped session and then tries to commit and close it.

Moving the resolution into a helper means the helper has to carry that through, so resolve_deadline_alert_interval() takes a keyword-only session and both call sites pass theirs. Three tests cover it, and each one fails if the argument is dropped:

• test_forwards_session_to_variable_lookup (new tests/unit/serialization/test_decoders.py) on the helper itself.
• test_dagrun_deadline_variable_interval_uses_callers_session on the DagRun creation path, which had no coverage of this before.
• test_clear_task_instances_recalculates_dagrun_queued_deadlines[variable_interval] asserts Variable.get is called with the same session object the clear is running in, not just a non-None one.

Tests

• airflow-core/tests/unit/models/test_taskinstance.py: the existing test stored the interval as a plain float, so it never exercised the JSON path. It now stores the serialized envelope and is parametrized over a fixed timedelta and a variable-backed interval. test_clear_task_instances_skips_deadline_with_unresolvable_interval is new and covers the skip. Both fixtures also had to start storing a serialized callback, since Decode deadline alert interval and callback without generic deserialization #72651 made callback decoding strict and the old raw {"path": ...} dict is no longer a valid callback_def.
• All of the above fail on main with the reported TypeError.

SDK coupling

This PR adds no airflow.sdk import to core. #71802 and #72651 have both landed since the last review, so resolve_deadline_alert_interval() isinstance-checks the core SerializedVariableInterval and there is no runtime SDK import left in this path. generated/known_sdk_imports_in_core.txt is unchanged by this PR.

closes: #70368

PR Checklist

• Targeted at the main branch
• Tests added (fixed interval, variable-backed interval, the unresolvable-interval skip, and session forwarding at both call sites)
• prek hooks pass on the changed files, and mypy airflow-core is clean

[amoghrajesh]

This propagates the same coupling I raised in #64751 - isinstance(interval, VariableInterval) here imports airflow.sdk.definitions.deadline.VariableInterval into core airflow at runtime. See #64751 (review) for details.

[hkc-8010]

@amoghrajesh Agreed the counterpart is the real fix, and #71802 is already adding SerializedVariableInterval for it. I did not want this bugfix to depend on that landing, so here the SDK type is confined to one function, resolve_deadline_alert_interval() in serialization/decoders.py, which both DagRun creation and the deadline recalculation now call. That is the only isinstance against the SDK dataclass left in this path, so when #71802 lands it changes in one place and no call site moves.

The duplication you flagged is gone too. taskinstance.py has no SDK import, and definitions/dag.py dropped its VariableInterval import, so the hook tightened generated/known_sdk_imports_in_core.txt from 2 to 1 for that file.

[ferruzzi]

@amoghrajesh looks like the SDK coupling you flagged has been addressed, and I think this is clear to move now if you agree.

models/taskinstance.py no longer imports from the SDK at all, and the single remaining isinstance(..., VariableInterval) is confined to resolve_deadline_alert_interval() in serialization/decoders.py. Worth noting that decoders.py already imported VariableInterval (function-locally, inside decode_deadline_alert), so this consolidates an existing coupling site rather than adding one, and serialization/definitions/dag.py loses its own import in the process. The core to SDK coupling is one file lower than before the PR.

#71802 then replaces that last isinstance in one place, so nothing here has to move again.

This PR is the only one of the three currently touching _recalculate_dagrun_queued_at_deadlines that is milestoned and backport labelled, so it is the only route this fix has to 3.3.x. More context in my comments on #70714 and #71850.

[vatsrahul1001]

@ferruzzi can you review again?

[ferruzzi]

@vatsrahul1001 - Amogh has the blocking review, we'll need him to lift that either way, but yes, a re-review on this one is on my list.

[ferruzzi]

@hkc-8010 - Thanks for your patience on this one. There are a bunch of PRs open which all edit the same portion of code. I've been trying to get them merged in the order that creates the least number of messy merge conflicts, and you are up next (after I get #72651 merged). It's important that you do this merge right since the previous one in the chain added session to this block and it's important we don't drop that in the rebase here.

serialization/definitions/dag.py:763-766 on main now has:

interval = deserialized_deadline_alert.interval

if isinstance(interval, SerializedVariableInterval):
    interval = interval.resolve(session=session) 

Passing session=session there is a critical bug fix* despite it being an optional parameter in the call. It would be easy to miss on your rebase, and nothing would necessarily tell you. You've dropped that line in favor of the new helper, so we have to make sure you carry the fix forward in yours.

TLDR

Here's my proposal for your fix: Add Session to the if TYPE_CHECKING import block then:

def resolve_deadline_alert_interval(
    alert: SerializedDeadlineAlert, *, session: Session | None = None
) -> datetime.timedelta:
    if isinstance(alert.interval, SerializedVariableInterval):
        return alert.interval.resolve(session=session)
    return alert.interval

#71968 added test_resolve_forwards_session but it only checks resolve() itself, not what helpers are using, so nothing is protecting your particular change. Maybe have a look at how that test works and see if you can think of a test that would guarantee a provided session gets passed through and is the one that is used (as opposed to creating a new session), that would be outstanding.

* The technical reason, if you care:

This code runs under the the scheduler's prohibit_commit guard. When Variable.get is called without a session, it falls through to @provide_session in MetastoreBackend.get_variable. Since settings.Session is a scoped_session, instead of opening a fresh session it just hands back the open session that the scheduler is holding and tries to commit and close it on the way out. Except (most importantly) the guard prevents that commit and rolls back the session. You can check out get_variable_from_secrets at/around models/variable.py:502-505 if you want to see exactly what we're trying to avoid. Either way, passing the session in avoids the provide_session wrapper and that bug, entirely.

[ferruzzi]

@hkc-8010 72651merged this morning which makes this one the head of the snake with 5 more blocked behind it. I'd love to get this one rebased, re-reviewed, and merged ASAP.

@amoghrajesh - I think your 5 August concern is already retired. You flagged the isinstance(interval, VariableInterval) runtime coupling; hkc pointed that to the core SerializedVariableInterval back on 10 September, and the only remaining runtime import of the SDK class was on main, which #72651 removed by mapping the classname onto SerializedVariableInterval instead. That deletion should get pulled in with the next rebase. I don't think there's any SDK coupling left in this path.

[hkc-8010]

@ferruzzi Rebased onto main with #72651 in. The session=session is carried forward. Since the resolution moved into a helper, the helper takes it:

def resolve_deadline_alert_interval(
    alert: SerializedDeadlineAlert, *, session: Session | None = None
) -> datetime.timedelta:

and both DagRun creation and the clear path pass theirs.

On the test, there are three, and each one goes red if the argument is dropped (I checked by deleting it and re-running):

• test_forwards_session_to_variable_lookup, in a new tests/unit/serialization/test_decoders.py, on the helper itself.
• test_dagrun_deadline_variable_interval_uses_callers_session in test_dagrun.py, on the DagRun creation path. Nothing covered that before, so this one is protecting your fix rather than mine.
• test_clear_task_instances_recalculates_dagrun_queued_deadlines[variable_interval] now asserts Variable.get is called with the same session object the clear is running in, not just a non-None one.

One other thing the rebase turned up, since you are sequencing these. #72651 made callback decoding strict, and the clear path decodes the callback now where before it only touched the interval. So the raw {"path": ..., "kwargs": {}} dict the test fixtures stored is no longer a valid callback_def and two tests failed with Deadline callback does not name a class. They store serialize(AsyncCallback(...)) now, which is what serialized_dag actually writes. Worth flagging because the unresolvable-interval test kept passing straight through that, just for the wrong reason.

@amoghrajesh agreed on the coupling, there is nothing left in this path. The SDK import check exits clean and generated/known_sdk_imports_in_core.txt is not part of the diff.