Call listeners for running task instance when a Dag Run state is manually set

[kacpermuda]

When a task instance is individually set to success or failure via the API, on_task_instance_success / on_task_instance_failed is called from the API server, giving listeners a terminal event to act on. The same does not happen when an entire Dag Run is force-set to success or failure: running tasks are bulk-transitioned in the database, but their task-instance listeners are never fired. Listeners that registered a start event (on_task_instance_running) for those tasks receive no completion event, leaving them — notably OpenLineage — with a job permanently stuck in RUNNING.

This PR closes the gap:

• _set_dag_run_terminal_state now returns (all_updated_tis, running_tis) where running_tis is the snapshot of task instances that were in an active state (RUNNING, DEFERRED, UP_FOR_RESCHEDULE, AWAITING_INPUT) before the bulk transition, captured without an extra query.
• set_dag_run_state_to_success / set_dag_run_state_to_failed expose the same tuple so callers have access to the running subset.
• patch_dag_run_state fires on_task_instance_success / on_task_instance_failed for those running task instances before the dag-run-level hook. Each group has its own try/except so a listener failure in one cannot suppress the other.
  Tasks that were only QUEUED or SCHEDULED (never started, no start event fired) are excluded — they transition to SKIPPED and receive no listener call, consistent with their lifecycle.

Why the terminal event is missing

The gap is a deliberate design decision that becomes a problem specifically for the dag run force-terminate case:

1. _set_dag_run_terminal_state writes SUCCESS/FAILED into the TI rows in the DB. No listener is fired.
2. Next heartbeat from the supervisor gets 409 CONFLICT ("TI is no longer running"). The supervisor sets self._terminal_state = SERVER_TERMINATED and kills the process via SIGTERM.
3. After the process dies, update_task_state_if_needed() runs. Because SERVER_TERMINATED ∈ STATES_SENT_DIRECTLY (supervisor.py:205), the condition at line 1461 is false — finish() is intentionally skipped. The reasoning is: "the server already knows the state." That is true — but the server setting the state and the server firing listeners for it are two different things, and the latter never happened.
4. Task runner finalizers never ran — the process was killed mid-execution before the on_task_instance_success/failed call in task-sdk/src/airflow/sdk/execution_time/task_runner.py:2249.
5. Scheduler's process_executor_events sees the executor report the task exit, but when it queries the TI it finds state=SUCCESS/FAILED (already written in step 1), so ti_queued is false (scheduler_job_runner.py:1460) and handle_failure is never called.

Result: on_task_instance_running fired from the task runner → task process killed → nothing on the other end. OpenLineage has a dangling RUNNING job with no terminal event. We can't extend the live of the worker process, but we can fire from the api server in that case.

---

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

• Yes — Claude Code (claude-sonnet-4-6)

Generated-by: Claude Code (claude-sonnet-4-6) 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.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[amoghrajesh]

Good investigation but some comments need handling

[pierrejeambrun]

Small nit, in addition of Amogh's comments

[amoghrajesh]

This looks like a genuine improvement from a listener's perspective, force-setting a Dag Run's state now fires on_task_instance_success/on_task_instance_failed for tasks that were actively running, closing the gap where a listener like OL would see a start event with no matching terminal event & for that reason I think its worth adding a newsfragment so this shows up in release notes for anyone relying on task-instance listeners. @pierrejeambrun any thoughts?

Rest of the changes LGTM.

[kacpermuda]

Added newsfragment as requested, let me know if it needs adjustments

[pierrejeambrun]

LGTM, just one suggestion before we can merge.

[github-actions]

Backport failed to create: v3-3-test. View the failure log Run details

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
❌	v3-3-test

You can attempt to backport this manually by running:

cherry_picker b6d81b9 v3-3-test

This should apply the commit to the v3-3-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.