Restoring include_downstream_dags logic to clear downstream Dags

[jroachgolf84]

Description

Restoring behavior to clear downstream DAGs when a Task state is cleared (if applicable).

closes: #61451

Testing

Unit-tests were added/updated to test these changes. The changes were also tested E2E.

Unit Tests

The unit-tests added/updated as part of this effort can be found in these files:

• airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
• airflow-core/tests/unit/serialization/definitions/test_dag.py

These tests can be executed using the commands below:

breeze testing core-tests airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py

breeze testing core-tests airflow-core/tests/unit/serialization/definitions/test_dag.py

E2E Testing

The DAG below was used to test the changes int his PR. This DAG is made up of a parent and child DAG. The ExternalTaskMarker Operator is cleared in the parent DAG, the downstream Tasks in the child DAG are also cleared. That's showed in the video below.

from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.sensors.external_task import ExternalTaskMarker, ExternalTaskSensor
from airflow.sdk import DAG

from datetime import datetime


with DAG(
    dag_id="parent_dag_61451",
    start_date=datetime(2026, 1, 18),
    schedule="@daily",
    catchup=False,
    tags=["issue-61451"],
) as parent_dag:
    parent_echo = BashOperator(
        task_id="parent_echo",
        bash_command="echo parent_dag_61451",
    )

    trigger_child = ExternalTaskMarker(
        task_id="trigger_child",
        external_dag_id="child_dag_61451",
        external_task_id="receive_call_from_parent",
    )

    parent_echo >> trigger_child

with DAG(
    dag_id="child_dag_61451",
    start_date=datetime(2026, 1, 18),
    schedule="@daily",
    catchup=False,
    tags=["issue-61451"],
) as child_dag:
    receive_call_from_parent = ExternalTaskSensor(
        task_id="receive_call_from_parent",
        external_dag_id="parent_dag_61451",
        external_task_id="trigger_child",
        poke_interval=5,
    )

    child_echo = BashOperator(
        task_id="child_echo",
        bash_command="echo DONE",
    )

    receive_call_from_parent >> child_echo

Video

issue-61451-include-downstream-tasks.mov

[LinasData]

Hey, any updates on this?

[jroachgolf84]

@jason810496 - can you take a look at this one for me?

[jroachgolf84]

@bugraoz93 - do you mind taking a look at this one?

[jroachgolf84]

@kaxil - do you mind taking another look?

[eladkal]

Overall LGTM
but needs 2nd eye from @kaxil

[github-actions]

uv.lock on main just moved via #70791 ("Limit pandas to < 3 for DataFrame XComs"), commit 415cb70 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.

[jroachgolf84]

@kaxil - feedback implemented. Do you mind taking another look? Thanks!

[Lee-W]

Mostly nits, but overall this looks good. The CI failure is unrelated—rebasing on the main branch should fix it.

[jroachgolf84]

@Lee-W - thanks for the review! I've implemented all your feedback.

[Lee-W]

mostly good on my end. would appreciate a second pair of eyes

[kaxil]

This resolves the child Dag from its latest version, but the code being restored used the run-pinned one: dag_bag.get_dag_for_run(tii.dag_run, session=session), at dag.py:1029 before #54383 removed it. This route already calls get_dag_for_run in two other spots (lines 335 and 882), so the helper is right there.

Why it matters: the partial_subset(include_downstream=True) just below becomes the task_id.in_(...) filter for the child's TI query. With bundle versioning on, that means we work out the downstream closure from a graph the historical child run may never have had. Say child v1 is wait >> transform >> load and v2 drops transform: clearing the parent reruns load and leaves transform sitting at SUCCESS in the v1 run.

get_dag_for_run(...) or get_latest_version_of_dag(...) keeps a fallback for a missing serialized row, and still routes through _get_dag, so the deserialize-once behaviour you added earlier holds.

[kaxil]

Coming back to the flag overlap I asked about in #65314 (comment), now that I've read the code this restores.

include_downstream_dags: false can't actually turn the cascade off, because or downstream only ever reacts to true. And the comment doesn't quite line up with what was there before: the pre-removal clear() hard-coded include_dependent_dags=True (dag.py:1404), it wasn't keyed off the same-Dag downstream flag at all.

The UI side is what bothers me more. ClearTaskInstanceDialog seeds selectedOptions with ["downstream"] (line 80, and the group dialog does the same at 54), so a default single-task clear from the grid now walks into other Dags and the caller has no way to say no.

run_on_latest_version in this same body model is already bool | None. Same treatment here gets you the opt-out and keeps the current UI default:

include_downstream_dags: bool | None = None
...
downstream if body.include_downstream_dags is None else body.include_downstream_dags

[kaxil]

Now that this list can span Dags, the single resolved_run_on_latest computed up at line 863 from the path dag_id gets applied to child Dag runs too. resolve_run_on_latest_version only takes one dag_id and reads the parent's Dag-level setting; clear_task_instances then rewrites each run's created_dag_version_id and calls verify_integrity(), which can add or drop TIs in that run.

Since the clear dialog always sends the field, the explicit_value is not None short-circuit means a child Dag with rerun_with_latest_version=False gets upgraded regardless of its own setting.

Grouping the TIs by dag_id and resolving per Dag when body.run_on_latest_version is None would respect each Dag's policy, with an explicit value from the caller still winning everywhere.

[kaxil]

Still ParserError-only, so this misses the inputs that raise ValueError directly. Same point as #65314 (comment), which got resolved without the change.

Two shapes reach it on the pinned pendulum 3.2.0. pendulum.parse("") raises ValueError: year 0 is out of range, which isn't a ParserError. And pendulum.parse("P1D") returns a Duration, which then goes into DagRun.logical_date == <Duration> and comes back as a StatementError. Both end up as a 500. A marker with logical_date="{{ dag_run.conf.get('x', '') }}" rendering empty gets you the first one.

Catching ValueError picks up the bare case plus the ParserError ones, since it's the parent. The Duration shape needs a check on what parse actually returned.

One thing to flag: test_invalid_external_task_marker_logical_date_returns_400 can't catch either of these. It sets side_effect=ParserError(...) on a fully mocked clear(), so it only proves the except clause catches what the test hands it, not what the real call site raises.