fix(scheduler): set end_date on tasks skipped by dagrun timeout

When a DAG run times out via dagrun_timeout, unfinished tasks are
marked as SKIPPED but end_date was not set. This caused task duration
to keep increasing in the UI even though the task was already skipped.

Set end_date to the current time when marking tasks as SKIPPED during
DAG run timeout handling.

Closes: #58536

Author: YoannAbriel

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

@@ -2309,8 +2309,10 @@ def _schedule_dag_run(
                 key=lambda ti: ti.start_date or timezone.make_aware(datetime.min),
                 default=None,
             )
+            now = timezone.utcnow()
             for task_instance in unfinished_task_instances:
                 task_instance.state = TaskInstanceState.SKIPPED
+                task_instance.end_date = now
                 session.merge(task_instance)
             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

@@ -3386,6 +3386,46 @@ def test_dagrun_timeout_fails_run(self, dag_maker):
         session.rollback()
         session.close()
 
+    def test_dagrun_timeout_sets_end_date_on_skipped_tasks(self, dag_maker):
+        """Test that when a DAG run times out, skipped tasks get end_date set."""
+        session = settings.Session()
+        with dag_maker(
+            dag_id="test_dagrun_timeout_sets_end_date",
+            dagrun_timeout=datetime.timedelta(seconds=60),
+            session=session,
+        ):
+            EmptyOperator(task_id="task1")
+            EmptyOperator(task_id="task2")
+
+        dr = dag_maker.create_dagrun(start_date=timezone.utcnow() - datetime.timedelta(days=1))
+
+        # Set tasks to running state with a start_date but no end_date
+        tis = dr.get_task_instances(session=session)
+        for ti in tis:
+            ti.state = State.RUNNING
+            ti.start_date = timezone.utcnow() - datetime.timedelta(hours=1)
+            ti.end_date = None
+            session.merge(ti)
+        session.flush()
+
+        scheduler_job = Job()
+        self.job_runner = SchedulerJobRunner(job=scheduler_job)
+
+        self.job_runner._schedule_dag_run(dr, session)
+        session.flush()
+
+        session.refresh(dr)
+        assert dr.state == State.FAILED
+
+        tis = dr.get_task_instances(session=session)
+        for ti in tis:
+            session.refresh(ti)
+            assert ti.state == TaskInstanceState.SKIPPED
+            assert ti.end_date is not None, f"end_date should be set for skipped task {ti.task_id}"
+
+        session.rollback()
+        session.close()
+
     def test_dagrun_timeout_fails_run_and_update_next_dagrun(self, dag_maker):
         """
         Test that dagrun timeout fails run and update the next dagrun