Fix task instance tries API returning duplicate entries for same task instance

airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py

@@ -23,9 +23,9 @@
 from fastapi import Depends, HTTPException, Query, status
 from fastapi.exceptions import RequestValidationError
 from pydantic import ValidationError
-from sqlalchemy import or_, select
+from sqlalchemy import and_, or_, select
 from sqlalchemy.exc import MultipleResultsFound
-from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import contains_eager, joinedload
 from sqlalchemy.sql.selectable import Select
 
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
@@ -292,14 +292,17 @@ def get_task_instance_tries(
     def _query(orm_object: Base) -> Select:
         query = (
             select(orm_object)
+            .join(DagRun, and_(orm_object.run_id == DagRun.run_id, orm_object.dag_id == DagRun.dag_id))
+            .options(
+                contains_eager(orm_object.dag_run).options(joinedload(DagRun.dag_model)),
+                joinedload(orm_object.dag_version),
+            )
             .where(
                 orm_object.dag_id == dag_id,
                 orm_object.run_id == dag_run_id,
                 orm_object.task_id == task_id,
                 orm_object.map_index == map_index,
             )
-            .options(joinedload(orm_object.dag_version))
-            .options(joinedload(orm_object.dag_run).options(joinedload(DagRun.dag_model)))
         )
         return query
 

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

@@ -1981,6 +1981,28 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve
             },
         }
 
+    def test_should_not_return_duplicate_runs(self, test_client, session):
+        """
+        Test that ensures the task instances query doesn't return duplicates due to the updated join/filter logic.
+        """
+        self.create_task_instances(session, task_instances=[{"state": State.SUCCESS}], with_ti_history=True)
+        self.create_task_instances(
+            session,
+            dag_id="example_bash_operator",
+            task_instances=[{"state": State.SUCCESS}],
+            with_ti_history=True,
+        )
+
+        response = test_client.get(
+            "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context/tries"
+        )
+
+        assert response.status_code == 200
+
+        response = response.json()
+
+        assert response["total_entries"] == 2
+
 
 class TestPostClearTaskInstances(TestTaskInstanceEndpoint):
     @pytest.mark.parametrize(