[ashb]

An attempt has a UUID A and try number N: (A,N). Previously, we assigned
a new UUID B when retrying or clearing, but left the scheduler to
increment the number. That left (B,N) in the database: the replacement's
UUID paired with the previous attempt's number.

We want to always allocate both fields together, archiving (A,N)
before creating (B,N+1). The scheduler then uses that allocation:

Retry lifecycle	Previously	Now
Running	(A,N)	(A,N)
UP_FOR_RETRY	(B,N)	(B,N+1)
Scheduled	(B,N+1)	(B,N+1)

This has UI impacts, as previously try N+1 was only created just as it starts.

In both of these screenshots below, try 1 has failed and the task is eligible for retry.

UI Before: (There is no try selector visible, as there is only a single try)

UI after (try 2 is now visible, in state of "up for retry", logs explicitly never fetched, but a "see here" message shown by the UI instead)

When clearing a running task, we must also wait for its process to stop
before allowing the replacement to run. We reject heartbeats with 409
when the TI is RESTARTING, which tells the worker to stop. But we cannot
tell from that response when the worker has actually stopped. Its child
may still be handling SIGTERM, running cleanup, or changing external
state. Starting the replacement then could let both attempts act on
the same external system.

We therefore leave both fields unchanged throughout shutdown:

Running clear	Identity	State
Before clear	(A,N)	RUNNING
Clear requested	(A,N)	RESTARTING
Cleared again	(A,N)	RESTARTING
Termination reported	(B,N+1)	NULL

By keeping the UUID unchanged as A, we let the worker report against
the attempt it owns. We also reject start requests during RESTARTING
so another workload cannot start that attempt while shutdown continues.

The supervisor knows the child has stopped once it has reaped it.
We need the supervisor to report that fact to the API before we can
release the replacement without relying on an executor event. It sends
SERVER_TERMINATED through the existing finish request to acknowledge
the shutdown. We do not store SERVER_TERMINATED as a TI state: the
worker is confirming that it stopped as instructed, rather than
reporting a task failure.

Once the supervisor confirms termination, we make the replacement
eligible immediately and restore its retry budget. A clear must allow
an immediate rerun even when the task has exhausted its retries,
without introducing an ordinary retry delay or failure callback.

We still need to support workers back to Task SDK 1.0.0, which cannot
send this acknowledgement. The scheduler therefore accepts terminal
executor events as a fallback while the matching attempt is RESTARTING.
Executor SUCCESS only means the workload finished, so either SUCCESS
or FAILED can complete the clear. The API and scheduler both lock and
recheck the row before allocating a replacement. If the API completes
the clear first, the old executor event no longer matches the live
attempt. We do not need to track the worker version.

Moving allocation earlier must not change retry timing or which logs
users see. During UP_FOR_RETRY, we still calculate backoff and select
logs using the failed attempt N, although the live row now identifies
N+1. We preserve both fields across deferral and rescheduling because
they continue the same attempt. Try zero still means we have not
allocated an attempt.

We also need to bring existing pending rows into the new convention.
Migration 0134 advances UP_FOR_RETRY rows and NULL-state rows with
positive try numbers: the old code had already replaced their UUIDs
but left scheduling to advance their numbers. Rows at try zero with
NULL state remain unallocated.

We use Cadwyn to preserve older API contracts. For current clients,
the API returns 410 for an archived attempt and 404 for an unknown UUID;
older clients retain 404 for both. If a worker loses a response, it can
use 410 to establish that its attempt has ended, but cannot infer that
the server accepted the particular outcome it reported.

We retain the existing heartbeat-timeout and orphan recovery behavior.
Those recovery paths cannot prove that the old worker has stopped.

---

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

• Yes (please specify the tool below)

---

• 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.

[ashb]

This inconsistency was (re)discovered here #65932 (comment)

[ashb]

Pulled out of review for final test run.

[kaxil]

LGTM, a few minor comments inline. The migration one is the only one I am not 100% sure on.