only_new clear can silently miss real new tasks after a prior "run on latest version" clear

[dstandish]

Problem

_get_new_task_ids (airflow-core/src/airflow/models/taskinstance.py:270-309), which powers the only_new clear feature added in #59764, computes "tasks added since this run's version" like this:

# Use created_dag_version_id directly to get the DAG version the run was
# originally created with. We cannot use get_dag_for_run here because it
# falls back to the latest version when bundle_version is not set (e.g.
# LocalDagBundle), which would make current_dag == latest_dag and the diff
# always empty.
current_dag = None
if dag_run.created_dag_version_id:
    current_dag = scheduler_dagbag.get_dag(version_id=dag_run.created_dag_version_id, session=session)
new_task_ids = set(latest_dag.task_ids) - set(current_dag.task_ids) if current_dag else set()

The comment explicitly documents the assumption that created_dag_version_id means "the version the run was originally created with." That assumption was correct when the field was immutable. Since #54984, clear_task_instances(..., run_on_latest_version=True) mutates created_dag_version_id to the latest version for a pinned run (while only bumping dag_version_id on the specific task instances that were cleared — see taskinstance.py:477). No new task-instance rows get created by that operation; it only updates the pointer and the already-existing, cleared TIs.

Once that has happened, created_dag_version_id == latest, so current_dag == latest_dag, and new_task_ids becomes latest_dag.task_ids - latest_dag.task_ids = the empty set — even if the run genuinely never got a task instance created for a task that was added between the run's true original version and latest.

Repro

1. Create a pinned DagRun at version V1 with tasks A and B.
2. Deploy V2, which adds task C (V2 task_ids = {A, B, C}).
3. Clear task A with run_on_latest_version=True. Now dag_run.created_dag_version_id == V2; the run still has no TI for task C.
4. Run an only_new clear on this DagRun.

Expected

Task C is identified as new and a TaskInstance is created for it.

Actual

_get_new_task_ids returns an empty set (current_dag resolves to V2, same as latest_dag), so task C is silently never added, even though the run's actual task-instance rows never picked it up.

Suggested fix

Determine "already has a TI for task X" by checking the run's actual TaskInstance rows directly (e.g. {ti.task_id for ti in dag_run.get_task_instances(session=session)}) rather than by diffing against a dag-version pointer (created_dag_version_id) that can no longer be trusted to represent "the version whose task set defines this run's existing task instances."

Context

This is one concrete consequence of created_dag_version_id's contract having drifted from "the version recorded at DagRun creation" (its documented meaning) to "the version this run should currently run at" (its behavior since #54984). See the companion issue tracking that broader contract problem: #71453

---

Drafted-by: Claude Code (Sonnet 5) (no human review before posting)

[dgvj-work]

I'd like to take this. I'll make _get_new_task_ids determine missing tasks from the run's actual TaskInstance rows instead of diffing against created_dag_version_id, and add a regression test.