From 76b7c95c8517471e00486ea4e63b8258199c8c8d Mon Sep 17 00:00:00 2001
From: Oscar Ligthart <oscar.ligthart@vinted.com>
Date: Tue, 21 Apr 2026 17:59:25 +0200
Subject: [PATCH] feat: enable queue up new tasks (#63484)

* feat: enable queue up new tasks

* chore: remove redundant disable flag from button

* feat: create and pass NewTaskCollectionResponse

* cleanup: unused translations

* fix: remove additional return type and simplify code

* remove redundancy

* fix: remove unused var

* fix: mypy and tests

* fix: more mypy

* cleanup

* fix: mypy

* fix: pnpm lint

* cleanup

(cherry picked from commit dae48ba722c420010de2c831587d005923644fc3)
---
 .../core_api/datamodels/dag_run.py            | 14 +++-
 .../core_api/datamodels/task_instances.py     | 31 +++++++
 .../openapi/v2-rest-api-generated.yaml        | 42 +++++++++-
 .../core_api/routes/public/dag_run.py         | 31 +++++--
 .../src/airflow/models/taskinstance.py        |  9 +-
 .../airflow/serialization/definitions/dag.py  | 17 ++++
 .../ui/openapi-gen/requests/schemas.gen.ts    | 50 +++++++++++
 .../ui/openapi-gen/requests/types.gen.ts      | 24 +++++-
 .../components/Clear/Run/ClearRunDialog.tsx   | 29 ++++---
 .../ui/src/queries/useClearDagRunDryRun.ts    |  7 +-
 .../core_api/routes/public/test_dag_run.py    | 84 +++++++++++++++++++
 .../airflowctl/api/datamodels/generated.py    | 25 ++++++
 12 files changed, 337 insertions(+), 26 deletions(-)

diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
index e88aad3d6d168..510015a91933f 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
@@ -20,7 +20,7 @@
 from collections.abc import Iterable
 from datetime import datetime
 from enum import Enum
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from pydantic import AliasPath, AwareDatetime, Field, NonNegativeInt, model_validator
 
@@ -55,11 +55,23 @@ class DAGRunClearBody(StrictBaseModel):
 
     dry_run: bool = True
     only_failed: bool = False
+    only_new: bool = Field(
+        default=False,
+        description="Only queue newly added tasks in the latest DAG version without clearing existing tasks.",
+    )
     run_on_latest_version: bool = Field(
         default=False,
         description="(Experimental) Run on the latest bundle version of the Dag after clearing the Dag Run.",
     )
 
+    @model_validator(mode="before")
+    @classmethod
+    def validate_model(cls, data: Any) -> Any:
+        """Validate clear DAG run form."""
+        if data.get("only_new") and data.get("only_failed"):
+            raise ValueError("only_new and only_failed are mutually exclusive")
+        return data
+
 
 class DAGRunResponse(BaseModel):
     """Dag Run serializer for responses."""
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
index 979218f315144..ae0e6c31f4770 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
@@ -25,9 +25,11 @@
     AliasPath,
     AwareDatetime,
     BeforeValidator,
+    Discriminator,
     Field,
     NonNegativeInt,
     StringConstraints,
+    Tag,
     ValidationError,
     field_validator,
     model_validator,
@@ -40,6 +42,13 @@
 from airflow.utils.state import TaskInstanceState
 
 
+class NewTaskResponse(BaseModel):
+    """Lightweight response for new tasks that don't have TaskInstances yet."""
+
+    task_id: str
+    task_display_name: str
+
+
 class TaskInstanceResponse(BaseModel):
     """TaskInstance serializer for responses."""
 
@@ -112,6 +121,28 @@ class TaskInstanceCollectionResponse(BaseModel):
     )
 
 
+def _task_instance_discriminator(v: Any) -> str:
+    """Discriminate between TaskInstanceResponse and NewTaskResponse in the union."""
+    if isinstance(v, NewTaskResponse):
+        return "new"
+    if isinstance(v, dict):
+        return "new" if "id" not in v else "full"
+    # ORM objects and TaskInstanceResponse instances
+    return "full"
+
+
+class ClearTaskInstanceCollectionResponse(BaseModel):
+    """Response for clear dag run dry run, which may contain new tasks without full TaskInstance data."""
+
+    task_instances: Iterable[
+        Annotated[
+            Annotated[TaskInstanceResponse, Tag("full")] | Annotated[NewTaskResponse, Tag("new")],
+            Discriminator(_task_instance_discriminator),
+        ]
+    ]
+    total_entries: int
+
+
 class TaskDependencyResponse(BaseModel):
     """Task Dependency serializer for responses."""
 
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 767cdfb29d72d..9d8b59ecaea8e 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -2142,7 +2142,7 @@ paths:
             application/json:
               schema:
                 anyOf:
-                - $ref: '#/components/schemas/TaskInstanceCollectionResponse'
+                - $ref: '#/components/schemas/ClearTaskInstanceCollectionResponse'
                 - $ref: '#/components/schemas/DAGRunResponse'
                 title: Response Clear Dag Run
         '401':
@@ -11072,6 +11072,25 @@ components:
       - action
       - entities
       title: BulkUpdateAction[VariableBody]
+    ClearTaskInstanceCollectionResponse:
+      properties:
+        task_instances:
+          items:
+            oneOf:
+            - $ref: '#/components/schemas/TaskInstanceResponse'
+            - $ref: '#/components/schemas/NewTaskResponse'
+          type: array
+          title: Task Instances
+        total_entries:
+          type: integer
+          title: Total Entries
+      type: object
+      required:
+      - task_instances
+      - total_entries
+      title: ClearTaskInstanceCollectionResponse
+      description: Response for clear dag run dry run, which may contain new tasks
+        without full TaskInstance data.
     ClearTaskInstancesBody:
       properties:
         dry_run:
@@ -11852,6 +11871,12 @@ components:
           type: boolean
           title: Only Failed
           default: false
+        only_new:
+          type: boolean
+          title: Only New
+          description: Only queue newly added tasks in the latest DAG version without
+            clearing existing tasks.
+          default: false
         run_on_latest_version:
           type: boolean
           title: Run On Latest Version
@@ -13271,6 +13296,21 @@ components:
       type: object
       title: MaterializeAssetBody
       description: Materialize asset request.
+    NewTaskResponse:
+      properties:
+        task_id:
+          type: string
+          title: Task Id
+        task_display_name:
+          type: string
+          title: Task Display Name
+      type: object
+      required:
+      - task_id
+      - task_display_name
+      title: NewTaskResponse
+      description: Lightweight response for new tasks that don't have TaskInstances
+        yet.
     PatchTaskInstanceBody:
       properties:
         new_state:
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
index e168bd782f3d9..5e33c844aebba 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
@@ -85,7 +85,8 @@
     TriggerDAGRunPostBody,
 )
 from airflow.api_fastapi.core_api.datamodels.task_instances import (
-    TaskInstanceCollectionResponse,
+    ClearTaskInstanceCollectionResponse,
+    NewTaskResponse,
     TaskInstanceResponse,
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
@@ -296,7 +297,7 @@ def clear_dag_run(
     body: DAGRunClearBody,
     dag_bag: DagBagDep,
     session: SessionDep,
-) -> TaskInstanceCollectionResponse | DAGRunResponse:
+) -> ClearTaskInstanceCollectionResponse | DAGRunResponse:
     dag_run = session.scalar(
         select(DagRun).filter_by(dag_id=dag_id, run_id=dag_run_id).options(joinedload(DagRun.dag_model))
     )
@@ -308,27 +309,39 @@ def clear_dag_run(
 
     dag = dag_bag.get_dag_for_run(dag_run, session=session)
 
+    if not dag:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, f"Dag with id {dag_id} was not found")
+
     if body.dry_run:
-        if not dag:
-            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Dag with id {dag_id} was not found")
-        task_instances = dag.clear(
+        task_instances_or_ids = dag.clear(
             run_id=dag_run_id,
             task_ids=None,
+            only_new=body.only_new,
             only_failed=body.only_failed,
             run_on_latest_version=body.run_on_latest_version,
             dry_run=True,
             session=session,
         )
 
-        return TaskInstanceCollectionResponse(
-            task_instances=cast("list[TaskInstanceResponse]", task_instances),
+        if body.only_new:
+            # Create lightweight NewTaskResponse objects for new tasks
+            new_task_ids = cast("set[str]", task_instances_or_ids)
+            task_instances: list[TaskInstanceResponse | NewTaskResponse] = [
+                NewTaskResponse(task_id=task_id, task_display_name=task_id)
+                for task_id in sorted(new_task_ids)
+            ]
+        else:
+            task_instances = cast("list[TaskInstanceResponse | NewTaskResponse]", task_instances_or_ids)
+
+        return ClearTaskInstanceCollectionResponse(
+            task_instances=task_instances,
             total_entries=len(task_instances),
         )
-    if not dag:
-        raise HTTPException(status.HTTP_404_NOT_FOUND, f"Dag with id {dag_id} was not found")
+
     dag.clear(
         run_id=dag_run_id,
         task_ids=None,
+        only_new=body.only_new,
         only_failed=body.only_failed,
         run_on_latest_version=body.run_on_latest_version,
         session=session,
diff --git a/airflow-core/src/airflow/models/taskinstance.py b/airflow-core/src/airflow/models/taskinstance.py
index 92270c885f5c1..3160c5f945eb3 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/airflow-core/src/airflow/models/taskinstance.py
@@ -262,7 +262,14 @@ def _get_new_task_ids(
     if not latest_dag:
         raise ValueError(f"Latest DAG version for '{dag_id}' not found")
 
-    current_dag = scheduler_dagbag.get_dag_for_run(dag_run=dag_run, session=session)
+    # Use created_dag_version_id directly to get the DAG version the run was
+    # originally created with. We cannot use get_dag_for_run here because it
+    # falls back to the latest version when bundle_version is not set (e.g.
+    # LocalDagBundle), which would make current_dag == latest_dag and the diff
+    # always empty.
+    current_dag = None
+    if dag_run.created_dag_version_id:
+        current_dag = scheduler_dagbag.get_dag(version_id=dag_run.created_dag_version_id, session=session)
     new_task_ids = set(latest_dag.task_ids) - set(current_dag.task_ids) if current_dag else set()
 
     return list(new_task_ids)
diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py b/airflow-core/src/airflow/serialization/definitions/dag.py
index 8844a18412c40..a1589f62995ae 100644
--- a/airflow-core/src/airflow/serialization/definitions/dag.py
+++ b/airflow-core/src/airflow/serialization/definitions/dag.py
@@ -944,6 +944,23 @@ def clear(
         run_on_latest_version: bool = False,
     ) -> list[TaskInstance]: ...  # pragma: no cover
 
+    @overload
+    def clear(
+        self,
+        *,
+        dry_run: Literal[True],
+        task_ids: Collection[str | tuple[str, int]] | None = None,
+        run_id: str,
+        only_failed: bool = False,
+        only_running: bool = False,
+        only_new: bool,
+        dag_run_state: DagRunState = DagRunState.QUEUED,
+        session: Session = NEW_SESSION,
+        exclude_task_ids: frozenset[str] | frozenset[tuple[str, int]] | None = frozenset(),
+        exclude_run_ids: frozenset[str] | None = frozenset(),
+        run_on_latest_version: bool = False,
+    ) -> set[str] | list[TaskInstance]: ...  # pragma: no cover
+
     @overload
     def clear(
         self,
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 8cb0d2979014e..b2b7beb541819 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -1270,6 +1270,33 @@ export const $BulkUpdateAction_VariableBody_ = {
     title: 'BulkUpdateAction[VariableBody]'
 } as const;
 
+export const $ClearTaskInstanceCollectionResponse = {
+    properties: {
+        task_instances: {
+            items: {
+                oneOf: [
+                    {
+                        '$ref': '#/components/schemas/TaskInstanceResponse'
+                    },
+                    {
+                        '$ref': '#/components/schemas/NewTaskResponse'
+                    }
+                ]
+            },
+            type: 'array',
+            title: 'Task Instances'
+        },
+        total_entries: {
+            type: 'integer',
+            title: 'Total Entries'
+        }
+    },
+    type: 'object',
+    required: ['task_instances', 'total_entries'],
+    title: 'ClearTaskInstanceCollectionResponse',
+    description: 'Response for clear dag run dry run, which may contain new tasks without full TaskInstance data.'
+} as const;
+
 export const $ClearTaskInstancesBody = {
     properties: {
         dry_run: {
@@ -2467,6 +2494,12 @@ export const $DAGRunClearBody = {
             title: 'Only Failed',
             default: false
         },
+        only_new: {
+            type: 'boolean',
+            title: 'Only New',
+            description: 'Only queue newly added tasks in the latest DAG version without clearing existing tasks.',
+            default: false
+        },
         run_on_latest_version: {
             type: 'boolean',
             title: 'Run On Latest Version',
@@ -4608,6 +4641,23 @@ export const $MaterializeAssetBody = {
     description: 'Materialize asset request.'
 } as const;
 
+export const $NewTaskResponse = {
+    properties: {
+        task_id: {
+            type: 'string',
+            title: 'Task Id'
+        },
+        task_display_name: {
+            type: 'string',
+            title: 'Task Display Name'
+        }
+    },
+    type: 'object',
+    required: ['task_id', 'task_display_name'],
+    title: 'NewTaskResponse',
+    description: "Lightweight response for new tasks that don't have TaskInstances yet."
+} as const;
+
 export const $PatchTaskInstanceBody = {
     properties: {
         new_state: {
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index 7bac12137b6f8..cca921b6beff1 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -400,6 +400,14 @@ export type BulkUpdateAction_VariableBody_ = {
     action_on_non_existence?: BulkActionNotOnExistence;
 };
 
+/**
+ * Response for clear dag run dry run, which may contain new tasks without full TaskInstance data.
+ */
+export type ClearTaskInstanceCollectionResponse = {
+    task_instances: Array<(TaskInstanceResponse | NewTaskResponse)>;
+    total_entries: number;
+};
+
 /**
  * Request body for Clear Task Instances endpoint.
  */
@@ -644,6 +652,10 @@ export type DAGResponse = {
 export type DAGRunClearBody = {
     dry_run?: boolean;
     only_failed?: boolean;
+    /**
+     * Only queue newly added tasks in the latest DAG version without clearing existing tasks.
+     */
+    only_new?: boolean;
     /**
      * (Experimental) Run on the latest bundle version of the Dag after clearing the Dag Run.
      */
@@ -1167,6 +1179,14 @@ export type MaterializeAssetBody = {
     partition_key?: string | null;
 };
 
+/**
+ * Lightweight response for new tasks that don't have TaskInstances yet.
+ */
+export type NewTaskResponse = {
+    task_id: string;
+    task_display_name: string;
+};
+
 /**
  * Request body for Clear Task Instances endpoint.
  */
@@ -2626,7 +2646,7 @@ export type ClearDagRunData = {
     requestBody: DAGRunClearBody;
 };
 
-export type ClearDagRunResponse = TaskInstanceCollectionResponse | DAGRunResponse;
+export type ClearDagRunResponse = ClearTaskInstanceCollectionResponse | DAGRunResponse;
 
 export type GetDagRunsData = {
     bundleVersion?: string | null;
@@ -4947,7 +4967,7 @@ export type $OpenApiTs = {
                 /**
                  * Successful Response
                  */
-                200: TaskInstanceCollectionResponse | DAGRunResponse;
+                200: ClearTaskInstanceCollectionResponse | DAGRunResponse;
                 /**
                  * Unauthorized
                  */
diff --git a/airflow-core/src/airflow/ui/src/components/Clear/Run/ClearRunDialog.tsx b/airflow-core/src/airflow/ui/src/components/Clear/Run/ClearRunDialog.tsx
index a4b53c104280a..6c36b47849552 100644
--- a/airflow-core/src/airflow/ui/src/components/Clear/Run/ClearRunDialog.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Clear/Run/ClearRunDialog.tsx
@@ -22,7 +22,7 @@ import { useTranslation } from "react-i18next";
 import { CgRedo } from "react-icons/cg";
 
 import { useDagServiceGetDagDetails } from "openapi/queries";
-import type { DAGRunResponse, TaskInstanceResponse } from "openapi/requests/types.gen";
+import type { DAGRunResponse } from "openapi/requests/types.gen";
 import { ActionAccordion } from "src/components/ActionAccordion";
 import { Checkbox, Dialog } from "src/components/ui";
 import SegmentedControl from "src/components/ui/SegmentedControl";
@@ -45,6 +45,7 @@ const ClearRunDialog = ({ dagRun, onClose, open }: Props) => {
   const [note, setNote] = useState<string | null>(dagRun.note);
   const [selectedOptions, setSelectedOptions] = useState<Array<string>>(["existingTasks"]);
   const onlyFailed = selectedOptions.includes("onlyFailed");
+  const onlyNew = selectedOptions.includes("newTasks");
   const [runOnLatestVersion, setRunOnLatestVersion] = useState(false);
 
   // Get current DAG's bundle version to compare with DAG run's bundle version
@@ -59,11 +60,15 @@ const ClearRunDialog = ({ dagRun, onClose, open }: Props) => {
     dagRunId,
     options: {
       refetchInterval: (query) =>
-        query.state.data?.task_instances.some((ti: TaskInstanceResponse) => isStatePending(ti.state))
+        query.state.data?.task_instances.some((ti) => "state" in ti && isStatePending(ti.state))
           ? refetchInterval
           : false,
     },
-    requestBody: { only_failed: onlyFailed, run_on_latest_version: runOnLatestVersion },
+    requestBody: {
+      only_failed: onlyFailed,
+      only_new: onlyNew,
+      run_on_latest_version: runOnLatestVersion,
+    },
   });
 
   const { isPending, mutate } = useClearDagRun({
@@ -78,12 +83,14 @@ const ClearRunDialog = ({ dagRun, onClose, open }: Props) => {
     onSuccess: onClose,
   });
 
-  // Check if bundle versions are different
-  const currentDagBundleVersion = dagDetails?.bundle_version;
-  const dagRunBundleVersion = dagRun.bundle_version;
-  const bundleVersionsDiffer = currentDagBundleVersion !== dagRunBundleVersion;
-  const shouldShowBundleVersionOption =
-    bundleVersionsDiffer && dagRunBundleVersion !== null && dagRunBundleVersion !== "";
+  // Check if DAG versions differ (works for both bundle-versioned and local bundles)
+  const latestDagVersionNumber = dagDetails?.latest_dag_version?.version_number;
+  const dagRunVersionNumber = dagRun.dag_versions.at(-1)?.version_number;
+  const versionsDiffer =
+    latestDagVersionNumber !== undefined &&
+    dagRunVersionNumber !== undefined &&
+    latestDagVersionNumber !== dagRunVersionNumber;
+  const shouldShowBundleVersionOption = versionsDiffer && !onlyNew;
 
   return (
     <Dialog.Root lazyMount onOpenChange={onClose} open={open} size="xl">
@@ -116,9 +123,8 @@ const ClearRunDialog = ({ dagRun, onClose, open }: Props) => {
                   value: "onlyFailed",
                 },
                 {
-                  disabled: true,
                   label: translate("dags:runAndTaskActions.options.queueNew"),
-                  value: "new_tasks",
+                  value: "newTasks",
                 },
               ]}
             />
@@ -148,6 +154,7 @@ const ClearRunDialog = ({ dagRun, onClose, open }: Props) => {
                   requestBody: {
                     dry_run: false,
                     only_failed: onlyFailed,
+                    only_new: onlyNew,
                     run_on_latest_version: runOnLatestVersion,
                   },
                 });
diff --git a/airflow-core/src/airflow/ui/src/queries/useClearDagRunDryRun.ts b/airflow-core/src/airflow/ui/src/queries/useClearDagRunDryRun.ts
index 2109df3d0815a..2e75239531069 100644
--- a/airflow-core/src/airflow/ui/src/queries/useClearDagRunDryRun.ts
+++ b/airflow-core/src/airflow/ui/src/queries/useClearDagRunDryRun.ts
@@ -47,5 +47,10 @@ export const useClearDagRunDryRun = <TData = TaskInstanceCollectionResponse, TEr
           ...requestBody,
         },
       }) as TData,
-    queryKey: [useClearDagRunDryRunKey, dagId, dagRunId, { only_failed: requestBody.only_failed }],
+    queryKey: [
+      useClearDagRunDryRunKey,
+      dagId,
+      dagRunId,
+      { only_failed: requestBody.only_failed, only_new: requestBody.only_new },
+    ],
   });
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
index 50dbf0190a416..34ad144406e53 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
@@ -1729,6 +1729,90 @@ def test_clear_dag_run_unprocessable_entity(self, test_client):
         assert body["detail"][0]["msg"] == "Field required"
         assert body["detail"][0]["loc"][0] == "body"
 
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+    def test_clear_dag_run_only_new_dry_run(self, mock_clear, test_client, session):
+        """Test that only_new dry_run returns placeholder task instances for new tasks."""
+        mock_clear.return_value = {"new_task_1", "new_task_2", "new_task_3"}
+        response = test_client.post(
+            f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/clear",
+            json={"dry_run": True, "only_new": True},
+        )
+        assert response.status_code == 200
+        body = response.json()
+        assert body["total_entries"] == 3
+        # Verify new tasks are returned with correct task_ids in task_instances
+        task_ids = sorted(t["task_id"] for t in body["task_instances"])
+        assert task_ids == ["new_task_1", "new_task_2", "new_task_3"]
+        # Verify task_display_name defaults to task_id
+        for task in body["task_instances"]:
+            assert task["task_display_name"] == task["task_id"]
+        mock_clear.assert_called_once_with(
+            run_id=DAG1_RUN1_ID,
+            task_ids=None,
+            only_new=True,
+            only_failed=False,
+            run_on_latest_version=False,
+            dry_run=True,
+            session=mock.ANY,
+        )
+        logs = session.scalar(
+            select(func.count())
+            .select_from(Log)
+            .where(Log.dag_id == DAG1_ID, Log.run_id == DAG1_RUN1_ID, Log.event == "clear_dag_run")
+        )
+        assert logs == 0
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+    def test_clear_dag_run_only_new_dry_run_no_new_tasks(self, mock_clear, test_client, session):
+        """Test that only_new dry_run returns 0 total_entries when there are no new tasks."""
+        mock_clear.return_value = set()
+        response = test_client.post(
+            f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/clear",
+            json={"dry_run": True, "only_new": True},
+        )
+        assert response.status_code == 200
+        body = response.json()
+        assert body["task_instances"] == []
+        assert body["total_entries"] == 0
+
+    @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+    @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+    def test_clear_dag_run_only_new_non_dry_run(self, mock_clear, test_client, session):
+        """Test that only_new non-dry_run clears and returns a DAGRunResponse."""
+        mock_clear.return_value = 2
+        response = test_client.post(
+            f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/clear",
+            json={"dry_run": False, "only_new": True},
+        )
+        assert response.status_code == 200
+        body = response.json()
+        assert body["dag_id"] == DAG1_ID
+        assert body["dag_run_id"] == DAG1_RUN1_ID
+        mock_clear.assert_called_once_with(
+            run_id=DAG1_RUN1_ID,
+            task_ids=None,
+            only_new=True,
+            only_failed=False,
+            run_on_latest_version=False,
+            session=mock.ANY,
+        )
+        _check_last_log(
+            session,
+            dag_id=DAG1_ID,
+            event="clear_dag_run",
+            logical_date=None,
+        )
+
+    def test_clear_dag_run_only_new_and_only_failed_mutually_exclusive(self, test_client):
+        """Test that only_new and only_failed cannot both be True."""
+        response = test_client.post(
+            f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/clear",
+            json={"dry_run": True, "only_new": True, "only_failed": True},
+        )
+        assert response.status_code == 422
+
 
 class TestTriggerDagRun:
     def _dags_for_trigger_tests(self, session=None):
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index b492406604c31..3e158bbf73993 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -287,6 +287,13 @@ class DAGRunClearBody(BaseModel):
     )
     dry_run: Annotated[bool | None, Field(title="Dry Run")] = True
     only_failed: Annotated[bool | None, Field(title="Only Failed")] = False
+    only_new: Annotated[
+        bool | None,
+        Field(
+            description="Only queue newly added tasks in the latest DAG version without clearing existing tasks.",
+            title="Only New",
+        ),
+    ] = False
     run_on_latest_version: Annotated[
         bool | None,
         Field(
@@ -638,6 +645,15 @@ class MaterializeAssetBody(BaseModel):
     partition_key: Annotated[str | None, Field(title="Partition Key")] = None
 
 
+class NewTaskResponse(BaseModel):
+    """
+    Lightweight response for new tasks that don't have TaskInstances yet.
+    """
+
+    task_id: Annotated[str, Field(title="Task Id")]
+    task_display_name: Annotated[str, Field(title="Task Display Name")]
+
+
 class PluginImportErrorResponse(BaseModel):
     """
     Plugin Import Error serializer for responses.
@@ -1932,6 +1948,15 @@ class BulkDeleteActionBulkTaskInstanceBody(BaseModel):
     action_on_non_existence: BulkActionNotOnExistence | None = "fail"
 
 
+class ClearTaskInstanceCollectionResponse(BaseModel):
+    """
+    Response for clear dag run dry run, which may contain new tasks without full TaskInstance data.
+    """
+
+    task_instances: Annotated[list[TaskInstanceResponse | NewTaskResponse], Field(title="Task Instances")]
+    total_entries: Annotated[int, Field(title="Total Entries")]
+
+
 class DAGCollectionResponse(BaseModel):
     """
     Dag Collection serializer for responses.
