KubernetesPodOperator durable reattach can reconnect to a different pod when the pod name is reused

### Under which category would you file this issue?

Providers

### Apache Airflow version

3.3.1

### What happened and how to reproduce it?

**Issue Description**

KPO durable reattach save the pod identity into task state store, but it only save name and namespace, no uid.

`_persist_pod_identity_to_task_state_store` in `providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py`:

```python
task_state_store.set(
POD_IDENTIFIER_STATE_KEY,
{"name": pod.metadata.name, "namespace": pod.metadata.namespace},
)
```

and on retry `_get_pod_from_task_state_store` read it back like this:

```python
pod = self.hook.get_pod(name, namespace)
```

In Kubernetes the name is only a slot, it is not identity. After a pod is deleted, the same name can be take by another pod. Only `metadata.uid` is stable. So this lookup have no fence, it only trust the name.

Old `find_pod()` path is not like this. `_build_find_pod_label_selector()` use dag_id, task_id, run_id and map_index, so a pod from other dag run can not match. But on Airflow 3.3+ `durable` is default True, and once the identity is persisted the label search is skip. So the fence become weaker than before.

The only guard left is the `already_checked` label, but a pod from another run does not have this label.

**Steps to reproduce**

Need the pod name can repeat, so `random_name_suffix=False`, or a fixed `metadata.name` from `pod_template_file` / `full_pod_spec`.

1. dag run A create pod `my-pod`, uid A. store keep `{"name": "my-pod", "namespace": "..."}`
2. worker die, and later `my-pod` is remove by node drain, gc, cleanup job, or someone do `kubectl delete pod`
3. dag run B start and create pod `my-pod` again, uid B
4. dag run A retry, `_get_pod_from_task_state_store()` read `my-pod` and get uid B

I test it on real cluster, k8s v1.37.0-rc.0, with airflow 3.3.1 and apache-airflow-providers-cncf-kubernetes 10.21.0:

```
STEP 1 dag run A creates its pod (this is what gets persisted)
run_id = manual__2026-08-18T01:00:00+00:00
pod = kpo-uid-test/my-pod
uid = 405af140-31e7-4529-9969-38407688a061

task_state_store['pod_identifier'] = {'name': 'my-pod', 'namespace': 'kpo-uid-test'}

STEP 2 that pod is gone (node drain / GC / kubectl delete / cleanup job)
deleted

STEP 3 dag run B creates a pod, same name (random_name_suffix=False)
run_id = manual__2026-08-18T02:00:00+00:00
pod = kpo-uid-test/my-pod
uid = 366da1af-7790-4c27-9158-d548b002c2be <-- different pod

STEP 4 dag run A retries -> _get_pod_from_task_state_store()
Reconnecting to pod kpo-uid-test/my-pod via identity persisted in task state store. loc=pod.py:685
Found matching pod my-pod with labels {'dag_id': 'uid_demo', 'kubernetes_pod_operator': 'True', 'run_id': 'manual__2026-08-18t02_00_00_00_00', 'task_id': 'my-pod', 'try_number': '1'} loc=pod.py:646
`try_number` of task_instance: 2 loc=pod.py:647
`try_number` of pod: 1 loc=pod.py:648

returned pod uid = 366da1af-7790-4c27-9158-d548b002c2be
returned pod run_id = manual__2026-08-18t02_00_00_00_00

*** run A reattached to run B's pod ***

STEP 5 the real caller: get_or_create_pod()
Reconnecting to pod kpo-uid-test/my-pod via identity persisted in task state store. loc=pod.py:685
Found matching pod my-pod with labels {'dag_id': 'uid_demo', 'kubernetes_pod_operator': 'True', 'run_id': 'manual__2026-08-18t02_00_00_00_00', 'task_id': 'my-pod', 'try_number': '1'} loc=pod.py:646
Reusing existing pod 'my-pod' (phase=Running, reason=) since it is not terminated or evicted. loc=pod.py:712
returned pod uid = 366da1af-7790-4c27-9158-d548b002c2be

STEP 6 same situation, but the OLD label search path find_pod()
returned None -> run_id label does not match, run B's pod rejected
```

Step 4 log also show run_id of the pod is run B, and try_number of the pod is 1 but task instance is 2. The operator print them and still continue.

Step 5 is the real caller. Pod B is Running so `get_or_create_pod()` just return it. Then run A monitor run B pod, read its log, take its exit code, and with default `on_finish_action=delete_pod` it delete run B pod in the end. So run B also break.

script I use

```python
import subprocess, time, types
from kubernetes import client as k8s, config as k8s_config
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

NS, POD_NAME, DAG_ID, TASK_ID = "kpo-uid-test", "my-pod", "uid_demo", "my-pod"
RUN_A = "manual__2026-08-18T01:00:00+00:00"
RUN_B = "manual__2026-08-18T02:00:00+00:00"

def body(run_id):
return {
"apiVersion": "v1", "kind": "Pod",
"metadata": {"name": POD_NAME, "namespace": NS, "labels": {
"dag_id": DAG_ID, "task_id": TASK_ID, "try_number": "1",
"kubernetes_pod_operator": "True",
"run_id": run_id.replace(":", "_").replace("+", "_").lower()}},
"spec": {"restartPolicy": "Never", "containers": [
{"name": "base", "image": "registry.k8s.io/pause:3.9"}]},
}

subprocess.run(["kubectl", "delete", "ns", NS, "--ignore-not-found"])
subprocess.run(["kubectl", "create", "ns", NS])
k8s_config.load_kube_config()
core = k8s.CoreV1Api()

uid_a = core.create_namespaced_pod(NS, body(RUN_A)).metadata.uid
persisted = {"name": POD_NAME, "namespace": NS} # what the provider writes today
core.delete_namespaced_pod(POD_NAME, NS, body=k8s.V1DeleteOptions(grace_period_seconds=0))
while True:
try:
core.read_namespaced_pod(POD_NAME, NS); time.sleep(1)
except k8s.rest.ApiException:
break
uid_b = core.create_namespaced_pod(NS, body(RUN_B)).metadata.uid
print("uid A", uid_a, "/ uid B", uid_b)

op = KubernetesPodOperator(task_id=TASK_ID, name=POD_NAME, namespace=NS,
image="registry.k8s.io/pause:3.9", random_name_suffix=False,
in_cluster=False, kubernetes_conn_id=None, do_xcom_push=False)

class Store:
def get(self, key, default=None):
return persisted if key == "pod_identifier" else default

context = {"ti": types.SimpleNamespace(dag_id=DAG_ID, task_id=TASK_ID, map_index=-1, try_number=2),
"run_id": RUN_A, "dag": types.SimpleNamespace(dag_id=DAG_ID), "task_state_store": Store()}

print("from store :", op._get_pod_from_task_state_store(context).metadata.uid)
req = k8s.V1Pod(metadata=k8s.V1ObjectMeta(name=POD_NAME, namespace=NS))
print("get_or_create :", op.get_or_create_pod(pod_request_obj=req, context=context).metadata.uid)
print("find_pod :", op.find_pod(NS, context=context))
```

### What you think should happen instead?

The uid should be save together with name and namespace, and be check when read back. If the uid is not the same, treat it like the pod is gone and go to the label search fallback, same as the 404 case today.

The first version of PR #69914 already have the uid in the payload, and in the review kaxil point out the same thing and give two option:

> I'd either compare the uid and fall back to label search on mismatch, or drop `uid` from the stored payload so the schema doesn't imply a check that isn't performed.

Then the second option is take, the uid is drop:

> Given the low practical risk you called out (random pod-name suffix), not worth the extra complexity right now

I think this risk reading is correct for the default only. With `random_name_suffix=False` the name repeat, so the name alone is not enough and the low risk does not hold. So what I ask here is the other option from that same review, compare the uid.

The dev list announce for durable execution also say the criteria is:

> The job has a stable tracking ID that survives the worker process

Other operator persist a server side id, like spark app id, glue job run id or snowflake query id. In Kubernetes that id is `metadata.uid`. The pod name is choose by the client and can repeat, so it is not this kind of id.

### Operating System

Ubuntu 26.04 LTS

### Deployment

Virtualenv installation

### Apache Airflow Provider(s)

cncf-kubernetes

### Deployment details

kubeadm cluster, k8s v1.37.0-rc.0, single node.

apache-airflow==3.3.1, apache-airflow-providers-cncf-kubernetes==10.21.0, kubernetes==36.0.3

### Anything else?

This only happen when the pod name can repeat. Default `random_name_suffix=True` add 8 random char so it is almost impossible. But `random_name_suffix=False` is a public parameter and people use it.

#21169 is the same kind of problem before, `random_name_suffix=False` make KPO delete the wrong pod. It was fixed by #22092 which add a `find_pod` check before the delete. The durable path go around that check now.

Two more place look like the same problem, but maybe better to do in a separate PR:

- `KubernetesPodTrigger` only serialize `pod_name` and `pod_namespace`, so the deferrable path also have no uid
- `PodManager.delete_pod()` use `V1DeleteOptions()` without `preconditions.uid`

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Start in providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py, reading _persist_pod_identity_to_task_state_store, _get_pod_from_task_state_store, and the find_pod fallback. Reproduce the reused-name case described in the issue and add coverage for a UID mismatch. Done means durable reattach rejects the replacement pod and follows the label-search fallback instead.

Written by the indexing model from the issue text.