[ccoliu]

previously_succeeded ("Number of previously succeeded task instances") is listed in the metrics registry and docs, but its only emit site is TaskInstance._check_and_change_state_before_execution, which nothing calls in Airflow 3. Workers move a task to running through the ti_run Execution API route instead, and when they are handed a task instance that already succeeded, ti_run answers 409 invalid_state with previous_state: success without counting anything.

This emits the counter from that 409 branch when the previous state is success, following the approach #67592 used to restore task.queued_duration in the same route. Tags match TaskInstance.stats_tags (Dag run tags, including team and Dag tags when enabled, plus task_id). The Task SDK client only retries 5xx and connection errors, so each refused start is counted once.

Tests: added test_ti_run_previously_succeeded_metric, covering success (counted) and failed / skipped / up_for_retry (409 but not counted). The success case fails without the fix.

closes: #75366

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code (Claude Opus 5.5) following the guidelines

[fat-catTW]

Thanks for the PR
I think we can drop this newsfragment.
This PR restores an existing metric, so it looks like an ordinary bug fix, which does not require a newsfragment.

[fat-catTW]

Also, I think we could add test for the metric tags with dag_tags_in_metrics=True and a team-scoped Dag?

[ccoliu]

Thanks for the review @fat-catTW!

• Dropped the newsfragment. You're right, per the contribution workflow docs this is an ordinary bug fix, so the commit title is enough for the release notes.
• Added test_ti_run_previously_succeeded_metric_carries_team_and_dag_tags, covering a team-owned Dag with tags while core.multi_team and metrics.dag_tags_in_metrics are both enabled. It asserts the emitted tags include team_name and the Dag tags along with dag_id and task_id.

[fat-catTW]

Thanks for the update!
Look nice.

[luc-pimentel]

Thanks for picking up the issue I opened. I ran its reproduction against this PR's current head with airflow standalone on Python 3.15, SQLite and the LocalExecutor, with StatsD pointed at a local UDP listener. It runs a one-task Dag once, then hands the succeeded task instance back to the executor's own launch path, so ti_run gets a second start request.

• First run: ti.start, ti.finish and ti_successes arrive, and previously_succeeded is not counted.
• Second start of the succeeded task: ti_run still answers 409 invalid_state with previous_state: success, and previously_succeeded arrives once. On the same setup without this change, nothing is counted.

I also have a unit test written for the report: a succeeded task instance, one PATCH /execution/task-instances/{id}/run, and an assert on previously_succeeded with the task instance's stats_tags. It fails without this change, as it should, and passes on this branch.

It seems like this fixes what I reported. Curious what the maintainers think of the approach.

---

Drafted-by: Claude Code 2.1.317 (Claude Opus 5.5); reviewed by @luc-pimentel before posting