Fix clearing with upstream and downstream selecting unrelated tasks

[sjyangkevin]

Why

closes: #73710

Clearing a task instance in a specific Dag run with both Upstream and Downstream enabled (the Clear dialog toggles, or POST /api/v2/dags/{dag_id}/clearTaskInstances with a dag_run_id) also selected tasks that are neither upstream nor downstream of the selected task.

For a specific Dag run, post_clear_task_instances resolves relatives one direction at a time with find_relevant_relatives, and each pass added its results to the selection immediately. The upstream pass ran first and grew the selection, so the downstream pass started from the selected task and its upstream tasks, and returned every descendant of those upstream tasks.

With the Dag from the issue (root >> a >> b, root >> other), clearing a with both options:

	upstream pass	downstream pass	selected
main	from {a} → root	from {a, root} → a, b, other	a, b, other, root
this PR	from {a} → root	from {a} → b	a, b, root

As a result:

• When the Dag's branches share a root task, clearing any task this way selects the whole Dag run.
• For a mapped task, selecting a single map index (a[0]) also selected every map index of a, because the downstream pass from root returned a as a whole task, which replaces the specific map index.

Clearing with only Upstream or only Downstream was not affected.

Approach

Both passes now start from the tasks in the request, and their results are merged into the selection only after both have run. This gives the same result as clearing without a dag_run_id (partial_subset(include_upstream=..., include_downstream=...)) and as marking a task success/failed with upstream and downstream. find_relevant_relatives itself is unchanged.

Why this does not introduce a regression

• Single-direction clears are unchanged. With one option set there is still exactly one find_relevant_relatives call with the same, unmodified input; only the place where its result is merged moved.
• With both options the selection can only shrink. Before: selected ∪ up(selected) ∪ down(selected ∪ up(selected)). After: selected ∪ up(selected) ∪ down(selected). The tasks dropped are exactly descendants of upstream tasks that are not relatives of the selected task.
• Nothing that has to be cleared is dropped. Setup/teardown handling never relied on the old seeding: the upstream pass already includes the teardowns of upstream setups (get_upstreams_follow_setups), and the downstream pass adds the setups/teardowns of downstream tasks (partial_subset). Map index resolution in find_relevant_relatives is untouched.
• No API surface change. Request/response models, the generated OpenAPI spec, the path without dag_run_id and the UI code are unchanged.

Testing

Unit tests

• Added test_clear_with_upstream_and_downstream_excludes_unrelated_tasks, parametrized for an unmapped selection (a) and a mapped one (a[0]). On main both fail (extra other; extra a[1], a[2] and other); with this change both pass.
• Ran the selection from breeze verify in Breeze (Python 3.11, SQLite), one suite at a time:
  • Core API DB tests: 3743 passed, 10 skipped
  • Core Always DB tests: 1891 passed, 3 skipped
  • Core non-DB tests: 4823 passed, 11 skipped, 2 xfailed
  • fab provider: 407 DB and 296 non-DB tests passed, 5 skipped
  • common.compat provider: 283 passed, 11 skipped
• Static checks through prek on the changed files: ruff, ruff-format, mypy (airflow-core, devel-common) and OpenAPI spec generation (spec unchanged).

Manual testing in Breeze (breeze start-airflow --dev-mode), Clear dialog with Upstream + Downstream enabled (screenshots below):

• Issue's Dag (root >> a >> b, root >> other), clearing a: selects a, b, root.
• start fanning out to three extract >> transform >> load pipelines, clearing transform_sales: selects only the four sales-pipeline tasks (start, extract_sales, transform_sales, load_sales) instead of the whole run.
• Issue's Dag with a mapped over 3 inputs, clearing a[0]: selects a[0], b, root.

The same three Dags were also checked through the REST API (dry run) against a running API server: with both options the results match the above, and Upstream-only and Downstream-only results are identical to main.

---

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

---

• 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.