Re: [PR] Restoring `include_downstream_dags` logic to clear downstream Dags [airflow]

[via GitHub]

Lee-W commented on code in PR #65314:
URL: https://github.com/apache/airflow/pull/65314#discussion_r3628544142


##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py:
##########
@@ -226,6 +226,12 @@ class ClearTaskInstancesBody(StrictBaseModel):
     )
     prevent_running_task: bool = False
     note: Annotated[str, StringConstraints(max_length=1000)] | None = None
+    include_downstream_dags: bool = Field(
+        default=False,
+        description="If True, also clear tasks in downstream DAGs that are 
linked via "

Review Comment:
   ```suggestion
           description="If True, also clear tasks in downstream Dags that are 
linked via "
   ```



##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -925,30 +930,71 @@ def _collect_relatives(run_id: str, direction: 
Literal["upstream", "downstream"]
             *((t, m) for t, m in mapped_tasks_tuples if t not in 
normal_task_ids),
         ]
 
+    # Follow ExternalTaskMarker connections when explicitly requested via 
include_downstream_dags, or
+    # automatically whenever downstream clearing is selected (restoring 
Airflow 2 behavior)
+    include_dependent_dags = body.include_downstream_dags or downstream
+
     task_instances: Sequence[TI]
-    if dag_run_id is not None and not (past or future):
-        # Use run_id-based clearing when we have a specific dag_run_id and not 
using past/future
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            run_id=dag_run_id,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
-    else:
-        # Use date-based clearing when no dag_run_id or when past/future is 
specified
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            start_date=body.start_date,
-            end_date=body.end_date,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
+    try:
+        if dag_run_id is not None and not (past or future):
+            # Use run_id-based clearing when we have a specific dag_run_id and 
not using past/future
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                run_id=dag_run_id,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+        else:
+            # Use date-based clearing when no dag_run_id or when past/future 
is specified
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                start_date=body.start_date,
+                end_date=body.end_date,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+
+    except MaxRecursionDepthError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e
+
+    except DagNotFound as e:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
+
+    except ParserError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid 
logical_date: {e}") from e

Review Comment:
   ```suggestion
       except ParserError as e:
           raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid 
logical_date: {e}") from e
       except DagNotFound as e:
           raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
   ```
   
   I don't think we need these blank lines to distinguish the logic. all of 
then are just one line raise. if the original order does not really matter 
(e.g., occurance order), ordering them by error code order would be easier to 
read



##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -925,30 +930,71 @@ def _collect_relatives(run_id: str, direction: 
Literal["upstream", "downstream"]
             *((t, m) for t, m in mapped_tasks_tuples if t not in 
normal_task_ids),
         ]
 
+    # Follow ExternalTaskMarker connections when explicitly requested via 
include_downstream_dags, or
+    # automatically whenever downstream clearing is selected (restoring 
Airflow 2 behavior)
+    include_dependent_dags = body.include_downstream_dags or downstream
+
     task_instances: Sequence[TI]
-    if dag_run_id is not None and not (past or future):
-        # Use run_id-based clearing when we have a specific dag_run_id and not 
using past/future
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            run_id=dag_run_id,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
-    else:
-        # Use date-based clearing when no dag_run_id or when past/future is 
specified
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            start_date=body.start_date,
-            end_date=body.end_date,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
+    try:
+        if dag_run_id is not None and not (past or future):
+            # Use run_id-based clearing when we have a specific dag_run_id and 
not using past/future
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                run_id=dag_run_id,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+        else:
+            # Use date-based clearing when no dag_run_id or when past/future 
is specified
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                start_date=body.start_date,
+                end_date=body.end_date,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+
+    except MaxRecursionDepthError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e
+
+    except DagNotFound as e:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
+
+    except ParserError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid 
logical_date: {e}") from e
+
+    if include_dependent_dags:
+        # Ensure proper access to downstream dags/tasks with dag.clear and 
include_dependent_dags
+        auth_manager = get_auth_manager()
+        all_dag_ids = {ti.dag_id for ti in task_instances}  # Retrieve all DAG 
ID's from task instances
+
+        # Used to find a team name from a DAG iD

Review Comment:
   ```suggestion
           # Used to find a team name from a Dag ID
   ```



##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -925,30 +930,71 @@ def _collect_relatives(run_id: str, direction: 
Literal["upstream", "downstream"]
             *((t, m) for t, m in mapped_tasks_tuples if t not in 
normal_task_ids),
         ]
 
+    # Follow ExternalTaskMarker connections when explicitly requested via 
include_downstream_dags, or
+    # automatically whenever downstream clearing is selected (restoring 
Airflow 2 behavior)
+    include_dependent_dags = body.include_downstream_dags or downstream
+
     task_instances: Sequence[TI]
-    if dag_run_id is not None and not (past or future):
-        # Use run_id-based clearing when we have a specific dag_run_id and not 
using past/future
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            run_id=dag_run_id,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
-    else:
-        # Use date-based clearing when no dag_run_id or when past/future is 
specified
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            start_date=body.start_date,
-            end_date=body.end_date,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
+    try:
+        if dag_run_id is not None and not (past or future):
+            # Use run_id-based clearing when we have a specific dag_run_id and 
not using past/future
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                run_id=dag_run_id,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+        else:
+            # Use date-based clearing when no dag_run_id or when past/future 
is specified
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                start_date=body.start_date,
+                end_date=body.end_date,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+
+    except MaxRecursionDepthError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e
+
+    except DagNotFound as e:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
+
+    except ParserError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid 
logical_date: {e}") from e
+
+    if include_dependent_dags:
+        # Ensure proper access to downstream dags/tasks with dag.clear and 
include_dependent_dags
+        auth_manager = get_auth_manager()
+        all_dag_ids = {ti.dag_id for ti in task_instances}  # Retrieve all DAG 
ID's from task instances

Review Comment:
   ```suggestion
           all_dag_ids = {ti.dag_id for ti in task_instances}  # Retrieve all 
Dag ID's from task instances
   ```



##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -925,30 +930,71 @@ def _collect_relatives(run_id: str, direction: 
Literal["upstream", "downstream"]
             *((t, m) for t, m in mapped_tasks_tuples if t not in 
normal_task_ids),
         ]
 
+    # Follow ExternalTaskMarker connections when explicitly requested via 
include_downstream_dags, or
+    # automatically whenever downstream clearing is selected (restoring 
Airflow 2 behavior)
+    include_dependent_dags = body.include_downstream_dags or downstream
+
     task_instances: Sequence[TI]
-    if dag_run_id is not None and not (past or future):
-        # Use run_id-based clearing when we have a specific dag_run_id and not 
using past/future
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            run_id=dag_run_id,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
-    else:
-        # Use date-based clearing when no dag_run_id or when past/future is 
specified
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            start_date=body.start_date,
-            end_date=body.end_date,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
+    try:
+        if dag_run_id is not None and not (past or future):
+            # Use run_id-based clearing when we have a specific dag_run_id and 
not using past/future
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                run_id=dag_run_id,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+        else:
+            # Use date-based clearing when no dag_run_id or when past/future 
is specified
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                start_date=body.start_date,
+                end_date=body.end_date,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+
+    except MaxRecursionDepthError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e
+
+    except DagNotFound as e:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
+
+    except ParserError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid 
logical_date: {e}") from e
+
+    if include_dependent_dags:
+        # Ensure proper access to downstream dags/tasks with dag.clear and 
include_dependent_dags
+        auth_manager = get_auth_manager()
+        all_dag_ids = {ti.dag_id for ti in task_instances}  # Retrieve all DAG 
ID's from task instances
+
+        # Used to find a team name from a DAG iD
+        dag_id_to_team = 
DagModel.get_dag_id_to_team_name_mapping(list(all_dag_ids), session=session)
+
+        # set of DAG ID's that can be cleared

Review Comment:
   ```suggestion
           # set of Dag ID's that can be cleared
   ```



##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -925,30 +930,71 @@ def _collect_relatives(run_id: str, direction: 
Literal["upstream", "downstream"]
             *((t, m) for t, m in mapped_tasks_tuples if t not in 
normal_task_ids),
         ]
 
+    # Follow ExternalTaskMarker connections when explicitly requested via 
include_downstream_dags, or
+    # automatically whenever downstream clearing is selected (restoring 
Airflow 2 behavior)
+    include_dependent_dags = body.include_downstream_dags or downstream
+
     task_instances: Sequence[TI]
-    if dag_run_id is not None and not (past or future):
-        # Use run_id-based clearing when we have a specific dag_run_id and not 
using past/future
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            run_id=dag_run_id,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
-    else:
-        # Use date-based clearing when no dag_run_id or when past/future is 
specified
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            start_date=body.start_date,
-            end_date=body.end_date,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
+    try:
+        if dag_run_id is not None and not (past or future):
+            # Use run_id-based clearing when we have a specific dag_run_id and 
not using past/future
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                run_id=dag_run_id,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+        else:
+            # Use date-based clearing when no dag_run_id or when past/future 
is specified
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                start_date=body.start_date,
+                end_date=body.end_date,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+
+    except MaxRecursionDepthError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e
+
+    except DagNotFound as e:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
+
+    except ParserError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid 
logical_date: {e}") from e
+
+    if include_dependent_dags:
+        # Ensure proper access to downstream dags/tasks with dag.clear and 
include_dependent_dags
+        auth_manager = get_auth_manager()
+        all_dag_ids = {ti.dag_id for ti in task_instances}  # Retrieve all DAG 
ID's from task instances
+
+        # Used to find a team name from a DAG iD
+        dag_id_to_team = 
DagModel.get_dag_id_to_team_name_mapping(list(all_dag_ids), session=session)
+
+        # set of DAG ID's that can be cleared
+        editable_dag_ids = {
+            other_dag_id
+            for other_dag_id in all_dag_ids
+            if auth_manager.is_authorized_dag(
+                method="PUT",
+                access_entity=DagAccessEntity.TASK_INSTANCE,
+                details=DagDetails(id=other_dag_id, 
team_name=dag_id_to_team.get(other_dag_id)),
+                user=user,
+            )
+        }
+
+        # list of all TI's that can be cleared (TI's within the DAGs from 
above)

Review Comment:
   ```suggestion
           # list of all TI's that can be cleared (TI's within the Dags from 
above)
   ```



##########
airflow-core/src/airflow/models/dagbag.py:
##########
@@ -235,11 +235,24 @@ def iter_all_latest_version_dags(self, *, session: 
Session) -> Generator[Seriali
                 yield dag
 
     def get_latest_version_of_dag(self, dag_id: str, *, session: Session) -> 
SerializedDAG | None:
-        """Get the latest version of a dag by its id."""
+        """Get the latest version of a dag by its id, using cache if 
enabled."""
         from airflow.models.serialized_dag import SerializedDagModel
 
         if not (serdag := SerializedDagModel.get(dag_id, session=session)):
             return None
+
+        with self._lock:
+            cached = self._dags.get(serdag.dag_version_id)
+
+        # DAG exists in cache and the cached/serialized DAG hashes match, 
return cached DAG

Review Comment:
   ```suggestion
           # Dag exists in cache and the cached/serialized Dag hashes match, 
return cached Dag
   ```



##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -925,30 +930,71 @@ def _collect_relatives(run_id: str, direction: 
Literal["upstream", "downstream"]
             *((t, m) for t, m in mapped_tasks_tuples if t not in 
normal_task_ids),
         ]
 
+    # Follow ExternalTaskMarker connections when explicitly requested via 
include_downstream_dags, or
+    # automatically whenever downstream clearing is selected (restoring 
Airflow 2 behavior)
+    include_dependent_dags = body.include_downstream_dags or downstream
+
     task_instances: Sequence[TI]
-    if dag_run_id is not None and not (past or future):
-        # Use run_id-based clearing when we have a specific dag_run_id and not 
using past/future
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            run_id=dag_run_id,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
-    else:
-        # Use date-based clearing when no dag_run_id or when past/future is 
specified
-        task_instances = dag.clear(
-            dry_run=True,
-            task_ids=task_markers_to_clear,
-            start_date=body.start_date,
-            end_date=body.end_date,
-            session=session,
-            run_on_latest_version=resolved_run_on_latest,
-            only_failed=body.only_failed,
-            only_running=body.only_running,
-        )
+    try:
+        if dag_run_id is not None and not (past or future):
+            # Use run_id-based clearing when we have a specific dag_run_id and 
not using past/future
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                run_id=dag_run_id,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+        else:
+            # Use date-based clearing when no dag_run_id or when past/future 
is specified
+            task_instances = dag.clear(
+                dry_run=True,
+                task_ids=task_markers_to_clear,
+                start_date=body.start_date,
+                end_date=body.end_date,
+                session=session,
+                run_on_latest_version=resolved_run_on_latest,
+                only_failed=body.only_failed,
+                only_running=body.only_running,
+                include_dependent_dags=include_dependent_dags,
+                dag_bag=dag_bag,
+            )
+
+    except MaxRecursionDepthError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e
+
+    except DagNotFound as e:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
+
+    except ParserError as e:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid 
logical_date: {e}") from e
+
+    if include_dependent_dags:
+        # Ensure proper access to downstream dags/tasks with dag.clear and 
include_dependent_dags
+        auth_manager = get_auth_manager()
+        all_dag_ids = {ti.dag_id for ti in task_instances}  # Retrieve all DAG 
ID's from task instances
+
+        # Used to find a team name from a DAG iD
+        dag_id_to_team = 
DagModel.get_dag_id_to_team_name_mapping(list(all_dag_ids), session=session)
+
+        # set of DAG ID's that can be cleared
+        editable_dag_ids = {
+            other_dag_id
+            for other_dag_id in all_dag_ids
+            if auth_manager.is_authorized_dag(
+                method="PUT",
+                access_entity=DagAccessEntity.TASK_INSTANCE,
+                details=DagDetails(id=other_dag_id, 
team_name=dag_id_to_team.get(other_dag_id)),
+                user=user,
+            )
+        }

Review Comment:
   ```suggestion
           editable_dag_ids = {
               dependent_dag_id
               for dependent_dag_id in all_dag_ids
               if auth_manager.is_authorized_dag(
                   method="PUT",
                   access_entity=DagAccessEntity.TASK_INSTANCE,
                   details=DagDetails(id=other_dag_id, 
team_name=dag_id_to_team.get(dependent_dag_id)),
                   user=user,
               )
           }
   ```



##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -92,6 +96,10 @@ class EdgeInfoType(TypedDict):
     label: str | None
 
 
+class MaxRecursionDepthError(AirflowException):

Review Comment:
   Do we really need to make it an AirflowException?



##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def 
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
         assert dagrun.state == "running"
         assert all(ti.state == "running" for ti in tis)
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_dags_sets_include_dependent_dags(self, 
mock_clear, test_client, session):
+        """include_downstream_dags=True must be forwarded to dag.clear() as 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, 
"include_downstream_dags": True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """include_downstream=True must also set include_dependent_dags=True 
(restoring Airflow 2 behavior)."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "include_downstream": 
True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_default_does_not_set_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """Without downstream flags, include_dependent_dags must default to 
False."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: include_downstream_dags=True → 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={
+                "dry_run": True,
+                "only_failed": False,
+                "dag_run_id": "TEST_DAG_RUN_ID",
+                "include_downstream_dags": True,
+            },
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True

Review Comment:
   we could parametreize these tests



##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -524,6 +524,7 @@ class TestStringifiedDAGs:
 
     @pytest.fixture(autouse=True)
     def setup_test_cases(self):
+        DagSerialization._load_operator_extra_links = True

Review Comment:
   why do we need this change?



##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def 
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
         assert dagrun.state == "running"
         assert all(ti.state == "running" for ti in tis)
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_dags_sets_include_dependent_dags(self, 
mock_clear, test_client, session):
+        """include_downstream_dags=True must be forwarded to dag.clear() as 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, 
"include_downstream_dags": True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """include_downstream=True must also set include_dependent_dags=True 
(restoring Airflow 2 behavior)."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "include_downstream": 
True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_default_does_not_set_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """Without downstream flags, include_dependent_dags must default to 
False."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: include_downstream_dags=True → 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={
+                "dry_run": True,
+                "only_failed": False,
+                "dag_run_id": "TEST_DAG_RUN_ID",
+                "include_downstream_dags": True,
+            },
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: the request-scoped dag_bag must reach 
dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "dag_run_id": 
"TEST_DAG_RUN_ID"},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_date_range_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """Date-range code path (no dag_run_id): the request-scoped dag_bag 
must reach dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.clear_task_instances")
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.get_auth_manager")
+    def test_include_dependent_dags_filters_unauthorized_child_tis(
+        self, mock_get_auth_manager, mock_dag_clear, mock_clear_tis, 
test_client, session
+    ):
+        """TIs from child DAGs the caller cannot edit must be excluded when 
include_dependent_dags=True."""
+        import uuid
+
+        self.create_task_instances(session)
+
+        parent_dag_id = "example_python_operator"
+        parent_ti = mock.MagicMock(spec=TaskInstance)
+        parent_ti.dag_id = parent_dag_id
+        parent_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000001")
+
+        child_dag_id = "child_dag_caller_cannot_edit"
+        child_ti = mock.MagicMock(spec=TaskInstance)
+        child_ti.dag_id = child_dag_id
+        child_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000002")
+
+        mock_dag_clear.return_value = [parent_ti, child_ti]
+        mock_get_auth_manager.return_value.is_authorized_dag.side_effect = 
lambda **kwargs: (
+            kwargs["details"].id == parent_dag_id
+        )
+
+        response = test_client.post(
+            f"/dags/{parent_dag_id}/clearTaskInstances",
+            json={"dry_run": False, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 200
+
+        auth_calls = 
mock_get_auth_manager.return_value.is_authorized_dag.call_args_list
+
+        # Auth calls should be made for both the parent and child DAGs
+        assert {call.kwargs["details"].id for call in auth_calls} == 
{parent_dag_id, child_dag_id}
+
+        # Auth calls should be for a TI
+        for call in auth_calls:
+            assert call.kwargs["method"] == "PUT"
+            assert call.kwargs["access_entity"] == 
DagAccessEntity.TASK_INSTANCE
+
+        cleared_tis = mock_clear_tis.call_args[0][0]
+        assert cleared_tis == [parent_ti]  # Child ID's are NOT cleared
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_cyclic_external_task_marker_returns_400(self, mock_clear, 
test_client, session):
+        """A cyclic or too-deep ExternalTaskMarker chain must return 400, not 
500."""
+        from airflow.serialization.definitions.dag import 
MaxRecursionDepthError

Review Comment:
   let's move it to the top



##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def 
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
         assert dagrun.state == "running"
         assert all(ti.state == "running" for ti in tis)
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_dags_sets_include_dependent_dags(self, 
mock_clear, test_client, session):
+        """include_downstream_dags=True must be forwarded to dag.clear() as 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, 
"include_downstream_dags": True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """include_downstream=True must also set include_dependent_dags=True 
(restoring Airflow 2 behavior)."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "include_downstream": 
True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_default_does_not_set_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """Without downstream flags, include_dependent_dags must default to 
False."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: include_downstream_dags=True → 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={
+                "dry_run": True,
+                "only_failed": False,
+                "dag_run_id": "TEST_DAG_RUN_ID",
+                "include_downstream_dags": True,
+            },
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: the request-scoped dag_bag must reach 
dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "dag_run_id": 
"TEST_DAG_RUN_ID"},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_date_range_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """Date-range code path (no dag_run_id): the request-scoped dag_bag 
must reach dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.clear_task_instances")
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.get_auth_manager")
+    def test_include_dependent_dags_filters_unauthorized_child_tis(
+        self, mock_get_auth_manager, mock_dag_clear, mock_clear_tis, 
test_client, session
+    ):
+        """TIs from child DAGs the caller cannot edit must be excluded when 
include_dependent_dags=True."""

Review Comment:
   ```suggestion
           """TIs from child Dags the caller cannot edit must be excluded when 
include_dependent_dags=True."""
   ```



##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def 
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
         assert dagrun.state == "running"
         assert all(ti.state == "running" for ti in tis)
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_dags_sets_include_dependent_dags(self, 
mock_clear, test_client, session):
+        """include_downstream_dags=True must be forwarded to dag.clear() as 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, 
"include_downstream_dags": True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """include_downstream=True must also set include_dependent_dags=True 
(restoring Airflow 2 behavior)."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "include_downstream": 
True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_default_does_not_set_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """Without downstream flags, include_dependent_dags must default to 
False."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: include_downstream_dags=True → 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={
+                "dry_run": True,
+                "only_failed": False,
+                "dag_run_id": "TEST_DAG_RUN_ID",
+                "include_downstream_dags": True,
+            },
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: the request-scoped dag_bag must reach 
dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "dag_run_id": 
"TEST_DAG_RUN_ID"},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_date_range_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """Date-range code path (no dag_run_id): the request-scoped dag_bag 
must reach dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag

Review Comment:
   same for these 2



##########
airflow-core/tests/unit/serialization/definitions/test_dag.py:
##########
@@ -0,0 +1,387 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from unittest import mock
+
+import pendulum
+import pytest
+
+from airflow.exceptions import AirflowException
+from airflow.models.dagbag import DBDagBag
+from airflow.models.renderedtifields import RenderedTaskInstanceFields
+from airflow.providers.standard.sensors.external_task import 
ExternalTaskMarker, ExternalTaskSensor
+
+from tests_common.test_utils import db
+from tests_common.test_utils.db import clear_rendered_ti_fields
+
+pytestmark = pytest.mark.db_test
+
+EXTERNAL_LOGICAL_DATE = pendulum.datetime(2024, 1, 1, tz="UTC")
+
+
+@pytest.fixture(autouse=True)
+def reset_db():
+    db.clear_db_dags()
+    db.clear_db_runs()
+    db.clear_db_serialized_dags()
+    clear_rendered_ti_fields()
+
+
+def test_clear_does_not_follow_external_marker_by_default(dag_maker, session):
+    """Without include_dependent_dags, ExternalTaskMarker links are not 
followed."""
+    with dag_maker("parent_dag", session=session, schedule=None):
+        ExternalTaskMarker(
+            task_id="trigger_child",
+            external_dag_id="child_dag",
+            external_task_id="wait_for_parent",
+            recursion_depth=3,
+        )
+
+    dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE)
+    serialized_parent = dag_maker.serialized_dag
+
+    with dag_maker("child_dag", session=session, schedule=None):
+        ExternalTaskSensor(
+            task_id="wait_for_parent",
+            external_dag_id="parent_dag",
+            external_task_id="trigger_child",
+            poke_interval=5,
+        )
+    dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE)
+    session.flush()
+
+    result = serialized_parent.clear(dry_run=True, only_failed=False, 
session=session)
+
+    dag_ids = {ti.dag_id for ti in result}
+    assert "parent_dag" in dag_ids
+    assert "child_dag" not in dag_ids
+
+
+def 
test_clear_follows_external_marker_when_include_dependent_dags_enabled(dag_maker,
 session):
+    """With include_dependent_dags=True, clear() follows ExternalTaskMarker 
links into child DAGs."""

Review Comment:
   ```suggestion
       """With include_dependent_dags=True, clear() follows ExternalTaskMarker 
links into child Dags."""
   ```



##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def 
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
         assert dagrun.state == "running"
         assert all(ti.state == "running" for ti in tis)
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_dags_sets_include_dependent_dags(self, 
mock_clear, test_client, session):
+        """include_downstream_dags=True must be forwarded to dag.clear() as 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, 
"include_downstream_dags": True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """include_downstream=True must also set include_dependent_dags=True 
(restoring Airflow 2 behavior)."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "include_downstream": 
True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_default_does_not_set_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """Without downstream flags, include_dependent_dags must default to 
False."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: include_downstream_dags=True → 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={
+                "dry_run": True,
+                "only_failed": False,
+                "dag_run_id": "TEST_DAG_RUN_ID",
+                "include_downstream_dags": True,
+            },
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: the request-scoped dag_bag must reach 
dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "dag_run_id": 
"TEST_DAG_RUN_ID"},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_date_range_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """Date-range code path (no dag_run_id): the request-scoped dag_bag 
must reach dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.clear_task_instances")
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.get_auth_manager")
+    def test_include_dependent_dags_filters_unauthorized_child_tis(
+        self, mock_get_auth_manager, mock_dag_clear, mock_clear_tis, 
test_client, session
+    ):
+        """TIs from child DAGs the caller cannot edit must be excluded when 
include_dependent_dags=True."""
+        import uuid
+
+        self.create_task_instances(session)
+
+        parent_dag_id = "example_python_operator"
+        parent_ti = mock.MagicMock(spec=TaskInstance)
+        parent_ti.dag_id = parent_dag_id
+        parent_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000001")
+
+        child_dag_id = "child_dag_caller_cannot_edit"
+        child_ti = mock.MagicMock(spec=TaskInstance)
+        child_ti.dag_id = child_dag_id
+        child_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000002")
+
+        mock_dag_clear.return_value = [parent_ti, child_ti]
+        mock_get_auth_manager.return_value.is_authorized_dag.side_effect = 
lambda **kwargs: (
+            kwargs["details"].id == parent_dag_id
+        )
+
+        response = test_client.post(
+            f"/dags/{parent_dag_id}/clearTaskInstances",
+            json={"dry_run": False, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 200
+
+        auth_calls = 
mock_get_auth_manager.return_value.is_authorized_dag.call_args_list
+
+        # Auth calls should be made for both the parent and child DAGs
+        assert {call.kwargs["details"].id for call in auth_calls} == 
{parent_dag_id, child_dag_id}
+
+        # Auth calls should be for a TI
+        for call in auth_calls:
+            assert call.kwargs["method"] == "PUT"
+            assert call.kwargs["access_entity"] == 
DagAccessEntity.TASK_INSTANCE
+
+        cleared_tis = mock_clear_tis.call_args[0][0]
+        assert cleared_tis == [parent_ti]  # Child ID's are NOT cleared
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_cyclic_external_task_marker_returns_400(self, mock_clear, 
test_client, session):
+        """A cyclic or too-deep ExternalTaskMarker chain must return 400, not 
500."""
+        from airflow.serialization.definitions.dag import 
MaxRecursionDepthError
+
+        self.create_task_instances(session)
+        mock_clear.side_effect = MaxRecursionDepthError(
+            "Maximum recursion depth 1 reached for ExternalTaskMarker 
marker_task."
+        )
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 400
+        assert "Maximum recursion depth" in response.json()["detail"]
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_missing_child_dag_returns_404(self, mock_clear, test_client, 
session):
+        """A missing child DAG referenced by ExternalTaskMarker must return 
404, not 500."""
+        self.create_task_instances(session)
+        mock_clear.side_effect = DagNotFound("Could not find dag child_dag")

Review Comment:
   ```suggestion
           mock_clear.side_effect = DagNotFound("Could not find Dag child_dag")
   ```



##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def 
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
         assert dagrun.state == "running"
         assert all(ti.state == "running" for ti in tis)
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_dags_sets_include_dependent_dags(self, 
mock_clear, test_client, session):
+        """include_downstream_dags=True must be forwarded to dag.clear() as 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, 
"include_downstream_dags": True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """include_downstream=True must also set include_dependent_dags=True 
(restoring Airflow 2 behavior)."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "include_downstream": 
True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_default_does_not_set_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """Without downstream flags, include_dependent_dags must default to 
False."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: include_downstream_dags=True → 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={
+                "dry_run": True,
+                "only_failed": False,
+                "dag_run_id": "TEST_DAG_RUN_ID",
+                "include_downstream_dags": True,
+            },
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: the request-scoped dag_bag must reach 
dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "dag_run_id": 
"TEST_DAG_RUN_ID"},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_date_range_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """Date-range code path (no dag_run_id): the request-scoped dag_bag 
must reach dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.clear_task_instances")
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.get_auth_manager")
+    def test_include_dependent_dags_filters_unauthorized_child_tis(
+        self, mock_get_auth_manager, mock_dag_clear, mock_clear_tis, 
test_client, session
+    ):
+        """TIs from child DAGs the caller cannot edit must be excluded when 
include_dependent_dags=True."""
+        import uuid
+
+        self.create_task_instances(session)
+
+        parent_dag_id = "example_python_operator"
+        parent_ti = mock.MagicMock(spec=TaskInstance)
+        parent_ti.dag_id = parent_dag_id
+        parent_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000001")
+
+        child_dag_id = "child_dag_caller_cannot_edit"
+        child_ti = mock.MagicMock(spec=TaskInstance)
+        child_ti.dag_id = child_dag_id
+        child_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000002")
+
+        mock_dag_clear.return_value = [parent_ti, child_ti]
+        mock_get_auth_manager.return_value.is_authorized_dag.side_effect = 
lambda **kwargs: (
+            kwargs["details"].id == parent_dag_id
+        )
+
+        response = test_client.post(
+            f"/dags/{parent_dag_id}/clearTaskInstances",
+            json={"dry_run": False, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 200
+
+        auth_calls = 
mock_get_auth_manager.return_value.is_authorized_dag.call_args_list
+
+        # Auth calls should be made for both the parent and child DAGs
+        assert {call.kwargs["details"].id for call in auth_calls} == 
{parent_dag_id, child_dag_id}
+
+        # Auth calls should be for a TI
+        for call in auth_calls:
+            assert call.kwargs["method"] == "PUT"
+            assert call.kwargs["access_entity"] == 
DagAccessEntity.TASK_INSTANCE
+
+        cleared_tis = mock_clear_tis.call_args[0][0]
+        assert cleared_tis == [parent_ti]  # Child ID's are NOT cleared
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_cyclic_external_task_marker_returns_400(self, mock_clear, 
test_client, session):
+        """A cyclic or too-deep ExternalTaskMarker chain must return 400, not 
500."""
+        from airflow.serialization.definitions.dag import 
MaxRecursionDepthError
+
+        self.create_task_instances(session)
+        mock_clear.side_effect = MaxRecursionDepthError(
+            "Maximum recursion depth 1 reached for ExternalTaskMarker 
marker_task."
+        )
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 400
+        assert "Maximum recursion depth" in response.json()["detail"]
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_missing_child_dag_returns_404(self, mock_clear, test_client, 
session):
+        """A missing child DAG referenced by ExternalTaskMarker must return 
404, not 500."""
+        self.create_task_instances(session)
+        mock_clear.side_effect = DagNotFound("Could not find dag child_dag")
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 404
+        assert "child_dag" in response.json()["detail"]
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_invalid_external_task_marker_logical_date_returns_400(self, 
mock_clear, test_client, session):
+        """A non-ISO logical_date rendered from an ExternalTaskMarker template 
must return 400, not 500."""
+        from pendulum.parsing.exceptions import ParserError

Review Comment:
   let's move it to the top



##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def 
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
         assert dagrun.state == "running"
         assert all(ti.state == "running" for ti in tis)
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_dags_sets_include_dependent_dags(self, 
mock_clear, test_client, session):
+        """include_downstream_dags=True must be forwarded to dag.clear() as 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, 
"include_downstream_dags": True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """include_downstream=True must also set include_dependent_dags=True 
(restoring Airflow 2 behavior)."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "include_downstream": 
True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_default_does_not_set_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """Without downstream flags, include_dependent_dags must default to 
False."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: include_downstream_dags=True → 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={
+                "dry_run": True,
+                "only_failed": False,
+                "dag_run_id": "TEST_DAG_RUN_ID",
+                "include_downstream_dags": True,
+            },
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: the request-scoped dag_bag must reach 
dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "dag_run_id": 
"TEST_DAG_RUN_ID"},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_date_range_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """Date-range code path (no dag_run_id): the request-scoped dag_bag 
must reach dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.clear_task_instances")
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.get_auth_manager")
+    def test_include_dependent_dags_filters_unauthorized_child_tis(
+        self, mock_get_auth_manager, mock_dag_clear, mock_clear_tis, 
test_client, session
+    ):
+        """TIs from child DAGs the caller cannot edit must be excluded when 
include_dependent_dags=True."""
+        import uuid

Review Comment:
   let's move it to the top



##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def 
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
         assert dagrun.state == "running"
         assert all(ti.state == "running" for ti in tis)
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_dags_sets_include_dependent_dags(self, 
mock_clear, test_client, session):
+        """include_downstream_dags=True must be forwarded to dag.clear() as 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, 
"include_downstream_dags": True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """include_downstream=True must also set include_dependent_dags=True 
(restoring Airflow 2 behavior)."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "include_downstream": 
True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_default_does_not_set_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """Without downstream flags, include_dependent_dags must default to 
False."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: include_downstream_dags=True → 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={
+                "dry_run": True,
+                "only_failed": False,
+                "dag_run_id": "TEST_DAG_RUN_ID",
+                "include_downstream_dags": True,
+            },
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: the request-scoped dag_bag must reach 
dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "dag_run_id": 
"TEST_DAG_RUN_ID"},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_date_range_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """Date-range code path (no dag_run_id): the request-scoped dag_bag 
must reach dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.clear_task_instances")
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.get_auth_manager")
+    def test_include_dependent_dags_filters_unauthorized_child_tis(
+        self, mock_get_auth_manager, mock_dag_clear, mock_clear_tis, 
test_client, session
+    ):
+        """TIs from child DAGs the caller cannot edit must be excluded when 
include_dependent_dags=True."""
+        import uuid
+
+        self.create_task_instances(session)
+
+        parent_dag_id = "example_python_operator"
+        parent_ti = mock.MagicMock(spec=TaskInstance)
+        parent_ti.dag_id = parent_dag_id
+        parent_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000001")
+
+        child_dag_id = "child_dag_caller_cannot_edit"
+        child_ti = mock.MagicMock(spec=TaskInstance)
+        child_ti.dag_id = child_dag_id
+        child_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000002")
+
+        mock_dag_clear.return_value = [parent_ti, child_ti]
+        mock_get_auth_manager.return_value.is_authorized_dag.side_effect = 
lambda **kwargs: (
+            kwargs["details"].id == parent_dag_id
+        )
+
+        response = test_client.post(
+            f"/dags/{parent_dag_id}/clearTaskInstances",
+            json={"dry_run": False, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 200
+
+        auth_calls = 
mock_get_auth_manager.return_value.is_authorized_dag.call_args_list
+
+        # Auth calls should be made for both the parent and child DAGs
+        assert {call.kwargs["details"].id for call in auth_calls} == 
{parent_dag_id, child_dag_id}
+
+        # Auth calls should be for a TI
+        for call in auth_calls:
+            assert call.kwargs["method"] == "PUT"
+            assert call.kwargs["access_entity"] == 
DagAccessEntity.TASK_INSTANCE
+
+        cleared_tis = mock_clear_tis.call_args[0][0]
+        assert cleared_tis == [parent_ti]  # Child ID's are NOT cleared
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_cyclic_external_task_marker_returns_400(self, mock_clear, 
test_client, session):
+        """A cyclic or too-deep ExternalTaskMarker chain must return 400, not 
500."""
+        from airflow.serialization.definitions.dag import 
MaxRecursionDepthError
+
+        self.create_task_instances(session)
+        mock_clear.side_effect = MaxRecursionDepthError(
+            "Maximum recursion depth 1 reached for ExternalTaskMarker 
marker_task."
+        )
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 400
+        assert "Maximum recursion depth" in response.json()["detail"]
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_missing_child_dag_returns_404(self, mock_clear, test_client, 
session):
+        """A missing child DAG referenced by ExternalTaskMarker must return 
404, not 500."""

Review Comment:
   ```suggestion
           """A missing child Dag referenced by ExternalTaskMarker must return 
404, not 500."""
   ```



##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def 
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
         assert dagrun.state == "running"
         assert all(ti.state == "running" for ti in tis)
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_dags_sets_include_dependent_dags(self, 
mock_clear, test_client, session):
+        """include_downstream_dags=True must be forwarded to dag.clear() as 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, 
"include_downstream_dags": True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_include_downstream_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """include_downstream=True must also set include_dependent_dags=True 
(restoring Airflow 2 behavior)."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "include_downstream": 
True},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_default_does_not_set_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """Without downstream flags, include_dependent_dags must default to 
False."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_sets_include_dependent_dags(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: include_downstream_dags=True → 
include_dependent_dags=True."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={
+                "dry_run": True,
+                "only_failed": False,
+                "dag_run_id": "TEST_DAG_RUN_ID",
+                "include_downstream_dags": True,
+            },
+        )
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_run_id_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """dag_run_id code path: the request-scoped dag_bag must reach 
dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False, "dag_run_id": 
"TEST_DAG_RUN_ID"},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", 
return_value=[])
+    def test_date_range_path_passes_request_dag_bag_to_clear(self, mock_clear, 
test_client, session):
+        """Date-range code path (no dag_run_id): the request-scoped dag_bag 
must reach dag.clear()."""
+        self.create_task_instances(session)
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "only_failed": False},
+        )
+
+        assert response.status_code == 200
+        assert mock_clear.call_count == 1
+        assert mock_clear.call_args.kwargs["dag_bag"] is 
test_client.app.state.dag_bag
+
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.clear_task_instances")
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.get_auth_manager")
+    def test_include_dependent_dags_filters_unauthorized_child_tis(
+        self, mock_get_auth_manager, mock_dag_clear, mock_clear_tis, 
test_client, session
+    ):
+        """TIs from child DAGs the caller cannot edit must be excluded when 
include_dependent_dags=True."""
+        import uuid
+
+        self.create_task_instances(session)
+
+        parent_dag_id = "example_python_operator"
+        parent_ti = mock.MagicMock(spec=TaskInstance)
+        parent_ti.dag_id = parent_dag_id
+        parent_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000001")
+
+        child_dag_id = "child_dag_caller_cannot_edit"
+        child_ti = mock.MagicMock(spec=TaskInstance)
+        child_ti.dag_id = child_dag_id
+        child_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000002")
+
+        mock_dag_clear.return_value = [parent_ti, child_ti]
+        mock_get_auth_manager.return_value.is_authorized_dag.side_effect = 
lambda **kwargs: (
+            kwargs["details"].id == parent_dag_id
+        )
+
+        response = test_client.post(
+            f"/dags/{parent_dag_id}/clearTaskInstances",
+            json={"dry_run": False, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 200
+
+        auth_calls = 
mock_get_auth_manager.return_value.is_authorized_dag.call_args_list
+
+        # Auth calls should be made for both the parent and child DAGs
+        assert {call.kwargs["details"].id for call in auth_calls} == 
{parent_dag_id, child_dag_id}
+
+        # Auth calls should be for a TI
+        for call in auth_calls:
+            assert call.kwargs["method"] == "PUT"
+            assert call.kwargs["access_entity"] == 
DagAccessEntity.TASK_INSTANCE
+
+        cleared_tis = mock_clear_tis.call_args[0][0]
+        assert cleared_tis == [parent_ti]  # Child ID's are NOT cleared
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_cyclic_external_task_marker_returns_400(self, mock_clear, 
test_client, session):
+        """A cyclic or too-deep ExternalTaskMarker chain must return 400, not 
500."""
+        from airflow.serialization.definitions.dag import 
MaxRecursionDepthError
+
+        self.create_task_instances(session)
+        mock_clear.side_effect = MaxRecursionDepthError(
+            "Maximum recursion depth 1 reached for ExternalTaskMarker 
marker_task."
+        )
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 400
+        assert "Maximum recursion depth" in response.json()["detail"]
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    def test_missing_child_dag_returns_404(self, mock_clear, test_client, 
session):
+        """A missing child DAG referenced by ExternalTaskMarker must return 
404, not 500."""
+        self.create_task_instances(session)
+        mock_clear.side_effect = DagNotFound("Could not find dag child_dag")
+        response = test_client.post(
+            "/dags/example_python_operator/clearTaskInstances",
+            json={"dry_run": True, "include_downstream_dags": True},
+        )
+
+        assert response.status_code == 404
+        assert "child_dag" in response.json()["detail"]

Review Comment:
   let's check the full message



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org