Add missing dag_id filter on DagRun query

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -6959,6 +6959,12 @@ paths:
               schema:
                 $ref: '#/components/schemas/HTTPExceptionResponse'
           description: Forbidden
+        '400':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Bad Request
         '404':
           content:
             application/json:

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

@@ -410,7 +410,7 @@ def get_mapped_task_instance(
 
 @task_instances_router.get(
     task_instances_prefix,
-    responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    responses=create_openapi_http_exception_doc([status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND]),
     dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.TASK_INSTANCE))],
 )
 def get_task_instances(
@@ -484,11 +484,16 @@ def get_task_instances(
         select(TI).join(TI.dag_run).outerjoin(TI.dag_version).options(*eager_load_TI_and_TIH_for_validation())
     )
     if dag_run_id != "~":
-        dag_run = session.scalar(select(DagRun).filter_by(run_id=dag_run_id))
+        if dag_id == "~":
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST,
+                "dag_id is required when dag_run_id is specified",
+            )
+        dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id, DagRun.run_id == dag_run_id))
         if not dag_run:
             raise HTTPException(
                 status.HTTP_404_NOT_FOUND,
-                f"DagRun with run_id: `{dag_run_id}` was not found",
+                f"DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
             )
         query = query.where(TI.run_id == dag_run_id)
     if dag_id != "~":

airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts

@@ -2415,6 +2415,7 @@ export class TaskInstanceService {
                 order_by: data.orderBy
             },
             errors: {
+                400: 'Bad Request',
                 401: 'Unauthorized',
                 403: 'Forbidden',
                 404: 'Not Found',

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -5544,6 +5544,10 @@ export type $OpenApiTs = {
                  * Successful Response
                  */
                 200: TaskInstanceCollectionResponse;
+                /**
+                 * Bad Request
+                 */
+                400: HTTPExceptionResponse;
                 /**
                  * Unauthorized
                  */

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

@@ -1499,9 +1499,11 @@ def test_not_found(self, test_client):
         assert response.status_code == 404
         assert response.json() == {"detail": "The Dag with ID: `invalid` was not found"}
 
-        response = test_client.get("/dags/~/dagRuns/invalid/taskInstances")
-        assert response.status_code == 404
-        assert response.json() == {"detail": "DagRun with run_id: `invalid` was not found"}
+    def test_dag_id_required_when_dag_run_id_specified(self, test_client):
+        # dag_run_id is not unique - it requires dag_id to identify a specific dag_run
+        response = test_client.get("/dags/~/dagRuns/some_run_id/taskInstances")
+        assert response.status_code == 400
+        assert response.json() == {"detail": "dag_id is required when dag_run_id is specified"}
 
     def test_bad_state(self, test_client):
         response = test_client.get("/dags/~/dagRuns/~/taskInstances", params={"state": "invalid"})