previously_succeeded metric is documented but never emitted in Airflow 3 #75366

Under which category would you file this issue?

Airflow Core

Apache Airflow version

3.3.2; also reproduced on main on October 21, 2026 (3.4.0b2 has the same code).

What happened and how to reproduce it?

The metrics docs list the counter previously_succeeded ("Number of previously succeeded task instances"). Its only emit site is TaskInstance._check_and_change_state_before_execution in airflow-core/src/airflow/models/taskinstance.py, which nothing in Airflow 3 calls: workers move a task to running through the ti_run Execution API route instead (#57818 noted the method has been unused since 3.0). The event the counter describes, a worker handed a task instance that already succeeded, now ends with ti_run answering 409 invalid_state with previous_state: success, and nothing is counted there. It is the same pattern as task.queued_duration (#63503, #66067), fixed by #67592.

To see it: run airflow standalone (LocalExecutor, SQLite) with [metrics] statsd_on = True pointed at any StatsD listener, trigger a Dag with one BashOperator and let it succeed, then hand the finished task instance to the executor again:

# relaunch.py <dag_id> <task_id> <run_id>: replay the LocalExecutor's launch of a finished task instance
import sys

from sqlalchemy import select
from sqlalchemy.orm import joinedload

from airflow import settings
from airflow.executors.base_executor import BaseExecutor
from airflow.executors.local_executor import LocalExecutor
from airflow.executors.workloads import ExecuteTask
from airflow.models.taskinstance import TaskInstance as TI

dag_id, task_id, run_id = sys.argv[1:4]
ti = settings.Session().scalar(
    select(TI)
    .where(TI.dag_id == dag_id, TI.task_id == task_id, TI.run_id == run_id)
    .options(joinedload(TI.dag_run), joinedload(TI.dag_model))
)
BaseExecutor.run_workload(ExecuteTask.make(ti, generator=LocalExecutor().jwt_generator), subprocess_logs_to_stdout=True)

Result on 3.3.2 and main: the supervisor logs Server error detail: {'reason': 'invalid_state', 'message': 'TI was not in a state where it could be marked as running', 'previous_state': 'success'} and the listener receives no previously_succeeded datagram. The first run of the same task delivers ti.start, ti.finish.success, ti_successes, task.duration, task.scheduled_duration and, on main, task.queued_duration, so the StatsD path itself works.

What you think should happen instead?

Either ti_run increments previously_succeeded when it refuses a start for a task instance whose previous state is success (next to where #67592 re-emits task.queued_duration), or the counter is removed from the metrics registry and the docs, so dashboards and alerts built on it stop waiting for a metric that cannot arrive.

Operating System

macOS 26.2, Python 3.12 for 3.3.2; Linux/Breeze, Python 3.11 for main.

Deployment

Virtualenv installation

Versions of Apache Airflow Providers

apache-airflow-providers-standard==1.19.0

Anything else?

No duplicate found in open/closed issue and PR searches for previously_succeeded, check_and_change_state_before_execution and emit_state_change_metric. #67592 changes only the success branch of ti_run, so it neither introduced nor fixed this. The counter was last live in Airflow 2, where LocalTaskJobRunner called check_and_change_state_before_execution before running the task.

Are you willing to submit PR?

No. I am reporting the bug and am not planning to submit a PR.

Code of Conduct

• I agree to follow this project's Code of Conduct.

Drafted-by: Claude Code 2.1.307 (Claude Fable 5.1); reviewed by @luc-pimentel before posting