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

[YoannAbriel]

Problem

When a DAG run times out via dagrun_timeout, unfinished task instances are marked SKIPPED but their end_date is never set. The UI computes duration as now - start_date, so skipped tasks show a continuously increasing duration — confusing and incorrect.

Root Cause

In SchedulerJobRunner._schedule_dag_run, the timeout handler sets task_instance.state = TaskInstanceState.SKIPPED but doesn't touch end_date. Without an end_date, the UI falls back to computing live duration.

Fix

Set end_date = timezone.utcnow() alongside the state change for all tasks skipped by dagrun timeout. Added a test that creates a timed-out DAG run with running tasks and verifies end_date is populated after the scheduler processes it.

Closes: #58536

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code

Generated-by: Claude Code following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.

[potiuk]

Did you actually check if it works? Or is that change just PR-ed without actually running some real tests? Can you show screenshots and logs from the execution?

[potiuk]

explanation: This is a vital part of airflow in order to modify scheduler you need to know exactly what you are doing and actually test it.

[YoannAbriel]

Fair — unit tests only so far. Will set up a local environment with a DAG that hits the timeout path and share logs/screenshots.

[YoannAbriel]

Reproduced and verified on Airflow 3.1.8 (standalone, SequentialExecutor). Test DAG with dagrun_timeout=timedelta(seconds=30) and 3 tasks chained: quick_task (5s) -> slow_task (120s) -> final_task (5s).

Before fix (unpatched 3.1.8):

Task	State	end_date
quick_task	success	2026-03-11 21:05:05
slow_task	skipped	None
final_task	None	None

After fix:

Task	State	end_date
quick_task	success	2026-03-11 21:09:21
slow_task	skipped	2026-03-11 21:09:45
final_task	None	None (never started, expected)

Two lines in _schedule_dag_run — now = timezone.utcnow() and task_instance.end_date = now before merging skipped tasks.

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.