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

airflow-core/src/airflow/models/taskinstancehistory.py

@@ -115,9 +115,9 @@ class TaskInstanceHistory(Base):
 
     dag_run = relationship(
         "DagRun",
-        primaryjoin="TaskInstanceHistory.run_id == DagRun.run_id",
+        primaryjoin="and_(TaskInstanceHistory.run_id == DagRun.run_id, DagRun.dag_id == TaskInstanceHistory.dag_id)",
         viewonly=True,
-        foreign_keys=[run_id],
+        foreign_keys=[run_id, dag_id],
     )
 
     def __init__(

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(