Fix task duration increasing forever after Dag run timeout skips it

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -2945,9 +2945,13 @@ def _schedule_dag_run(
                 key=lambda ti: ti.start_date or timezone.make_aware(datetime.min),
                 default=None,
             )
+            # Mirror TI.set_state inline so the whole batch flushes once instead of per task.
+            now = timezone.utcnow()
             for task_instance in unfinished_task_instances:
                 task_instance.state = TaskInstanceState.SKIPPED
-                session.merge(task_instance)
+                task_instance.start_date = task_instance.start_date or now
+                task_instance.end_date = task_instance.end_date or now
+                task_instance.duration = (task_instance.end_date - task_instance.start_date).total_seconds()
             session.flush()
             self.log.info("Run %s of %s has timed-out", dag_run.run_id, dag_run.dag_id)
 

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -4245,6 +4245,56 @@ def test_dagrun_timeout_fails_run(self, dag_maker):
         session.rollback()
         session.close()
 
+    @pytest.mark.parametrize(
+        ("initial_state", "started_ago", "ended_ago", "expected_duration"),
+        [
+            pytest.param(TaskInstanceState.RUNNING, datetime.timedelta(hours=1), None, 3600.0, id="running"),
+            pytest.param(None, None, None, 0.0, id="never_started"),
+            pytest.param(
+                TaskInstanceState.UP_FOR_RETRY,
+                datetime.timedelta(hours=1),
+                datetime.timedelta(minutes=30),
+                1800.0,
+                id="up_for_retry_keeps_previous_try_end_date",
+            ),
+        ],
+    )
+    def test_dagrun_timeout_sets_end_date_and_duration_on_skipped_tasks(
+        self, dag_maker, initial_state, started_ago, ended_ago, expected_duration
+    ):
+        """Without an end_date the UI keeps growing the duration of a task the timeout already skipped."""
+        session = settings.Session()
+        with dag_maker(
+            dag_id="test_dagrun_timeout_sets_end_date_and_duration",
+            dagrun_timeout=datetime.timedelta(seconds=60),
+            session=session,
+        ):
+            EmptyOperator(task_id="dummy")
+
+        now = timezone.utcnow()
+        dr = dag_maker.create_dagrun(start_date=now - datetime.timedelta(days=1))
+        ti = dr.get_task_instance("dummy", session=session)
+        ti.state = initial_state
+        ti.start_date = now - started_ago if started_ago else None
+        ti.end_date = now - ended_ago if ended_ago else None
+        session.merge(ti)
+        session.flush()
+
+        scheduler_job = Job()
+        self.job_runner = SchedulerJobRunner(job=scheduler_job)
+
+        with time_machine.travel(now, tick=False):
+            self.job_runner._schedule_dag_run(dr, session)
+        session.flush()
+
+        session.refresh(ti)
+        assert ti.state == TaskInstanceState.SKIPPED
+        assert ti.end_date == (now - ended_ago if ended_ago else now)
+        assert ti.duration == expected_duration
+
+        session.rollback()
+        session.close()
+
     @mock.patch("airflow._shared.observability.metrics.stats._get_backend")
     def test_dagrun_timeout_duration_metric_has_run_type(self, mock_get_backend, dag_maker):
         """