From 9cadfca0e94e7e78323bcefb833a2a437987b89b Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Tue, 28 Jul 2026 14:46:52 +0300
Subject: [PATCH 01/11] Add --min-completed-minutes to cleanup-pods to prevent
 KPO race condition
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The ``airflow kubernetes cleanup-pods`` command currently deletes
Succeeded/Failed/Evicted pods immediately, with no minimum-age guard
for terminal states.  ``KubernetesPodOperator`` in synchronous mode
polls pod status every ~2 seconds via ``await_pod_completion``.  If
the cleanup job fires in the window between the pod reaching
``Succeeded`` and KPO's next poll, KPO receives a 404 and fails the
task -- even though the pod completed successfully (exit code 0).

A ``--min-pending-minutes`` guard already exists for Pending pods
(default 30 m, floor 5 m).  No equivalent exists for terminal states.

This commit adds ``--min-completed-minutes`` (default ``0``, which
preserves the existing behaviour).  When set to any positive value,
Succeeded/Failed/Evicted pods are skipped unless their completion time
is older than the threshold.  Completion time is derived from the
latest ``containerStatuses[*].state.terminated.finishedAt`` timestamp
(falls back to ``metadata.creationTimestamp`` for pods that were
evicted before any container started).

Root-cause investigation
------------------------
This was confirmed via Kubernetes API server audit logs on a production
EKS cluster.  Timeline for an affected KPO task:

  13:15:12Z  KPO polls pod → phase Running
  13:15:14Z  Container exits with code 0 (pod transitions to Succeeded)
  13:15:17Z  cleanup-pods CronJob deletes the pod (3 s after completion)
  13:15:18Z  KPO polls pod → 404 Not Found → task marked FAILED

The pod had succeeded; the task failure was a false positive caused
entirely by the race.  Multiple production DAGs exhibited the same
pattern with the cleanup CronJob set to run every 5 minutes.

Reducing the CronJob frequency is a partial mitigation (lowers the
probability) but does not eliminate the race.  Setting
``--min-completed-minutes=5`` gives KPO a 5-minute window to observe
the terminal phase -- 150x wider than the 2 s poll interval -- closing
the race completely in practice.

Changes
-------
* ``definition.py``      – add ``ARG_MIN_COMPLETED_MINUTES``; wire into
                           cleanup-pods args tuple
* ``kubernetes_command.py`` – add ``_get_pod_completion_time()`` helper;
                           gate terminal-pod deletion by age when
                           ``min_completed_minutes > 0``
* ``test_kubernetes_command.py`` – 4 new unit tests covering:
  - Succeeded pod too young → not deleted
  - Succeeded pod old enough → deleted
  - Default (0) preserves immediate-deletion behaviour
  - Failed/Never pod too young → not deleted
* ``changelog.rst``      – entry under 10.21.0

CLI docs (``cli-ref.rst``) are auto-generated via ``.. argparse::``
and will pick up the new flag automatically.
---
 providers/cncf/kubernetes/docs/changelog.rst  |   8 ++
 .../cncf/kubernetes/cli/definition.py         |  15 ++-
 .../cncf/kubernetes/cli/kubernetes_command.py |  26 +++-
 .../kubernetes/cli/test_kubernetes_command.py | 112 ++++++++++++++++++
 4 files changed, 159 insertions(+), 2 deletions(-)

diff --git a/providers/cncf/kubernetes/docs/changelog.rst b/providers/cncf/kubernetes/docs/changelog.rst
index 82fd133f0047d..9a1ca8b8bb99d 100644
--- a/providers/cncf/kubernetes/docs/changelog.rst
+++ b/providers/cncf/kubernetes/docs/changelog.rst
@@ -27,6 +27,14 @@
 Changelog
 ---------
 
+10.21.0
+.......
+
+New features
+~~~~~~~~~~~~
+
+* ``Add --min-completed-minutes to cleanup-pods to prevent KPO race condition (#XXXXX)``
+
 10.20.0
 .......
 
diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
index 90929ffaf5f74..dfc86f6acaa78 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
@@ -68,6 +68,19 @@
     ),
 )
 
+ARG_MIN_COMPLETED_MINUTES = Arg(
+    ("--min-completed-minutes",),
+    default=0,
+    type=positive_int(allow_zero=True),
+    help=(
+        "Minimum age in minutes of a completed (Succeeded/Failed/Evicted) pod before it is deleted. "
+        "Defaults to 0 (delete immediately, preserving current behaviour). "
+        "Set this to a value greater than the KubernetesPodOperator poll interval (~2 s) to prevent "
+        "a race where the cleanup job removes a pod before KPO observes its terminal phase, "
+        "causing a spurious task failure despite the pod having succeeded."
+    ),
+)
+
 ARG_TEAM = Arg(
     ("--team",),
     default=None,
@@ -84,7 +97,7 @@
             "in evicted/failed/succeeded/pending states"
         ),
         func=lazy_load_command("airflow.providers.cncf.kubernetes.cli.kubernetes_command.cleanup_pods"),
-        args=(ARG_NAMESPACE, ARG_MIN_PENDING_MINUTES, ARG_VERBOSE),
+        args=(ARG_NAMESPACE, ARG_MIN_PENDING_MINUTES, ARG_MIN_COMPLETED_MINUTES, ARG_VERBOSE),
     ),
     ActionCommand(
         name="generate-dag-yaml",
diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
index ea59ae7755b86..01f7964a4a712 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
@@ -119,6 +119,20 @@ def generate_pod_yaml(args):
     print(f"YAML output can be found at {yaml_output_path}")
 
 
+def _get_pod_completion_time(pod):
+    """Return the time the pod entered a terminal state, or its creation time as fallback.
+
+    Uses the latest ``finished_at`` timestamp across all container statuses so that pods
+    with multiple containers (e.g. an init container + a base container) are judged by
+    the time the *last* container finished, not by when the pod was created.
+    """
+    times = []
+    for status in pod.status.container_statuses or []:
+        if status.state and status.state.terminated and status.state.terminated.finished_at:
+            times.append(status.state.terminated.finished_at)
+    return max(times) if times else pod.metadata.creation_timestamp
+
+
 @cli_utils.action_cli(check_db=False)
 @providers_configuration_loaded
 def cleanup_pods(args):
@@ -130,6 +144,8 @@ def cleanup_pods(args):
     if min_pending_minutes < 5:
         min_pending_minutes = 5
 
+    min_completed_minutes = args.min_completed_minutes
+
     # https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/
     # All Containers in the Pod have terminated in success, and will not be restarted.
     pod_succeeded = "succeeded"
@@ -173,10 +189,18 @@ def cleanup_pods(args):
             pod_restart_policy = pod.spec.restart_policy.lower()
             current_time = datetime.now(pod.metadata.creation_timestamp.tzinfo)
 
-            if (
+            terminal = (
                 pod_phase == pod_succeeded
                 or (pod_phase == pod_failed and pod_restart_policy == pod_restart_policy_never)
                 or (pod_reason == pod_reason_evicted)
+            )
+            terminal_old_enough = terminal and (
+                min_completed_minutes == 0
+                or current_time - _get_pod_completion_time(pod)
+                > timedelta(minutes=min_completed_minutes)
+            )
+            if (
+                terminal_old_enough
                 or (
                     pod_phase == pod_pending
                     and current_time - pod.metadata.creation_timestamp
diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
index 6ae8b81c50f3c..858436348f109 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
@@ -283,3 +283,115 @@ def test_list_pod_with_continue_token(self, load_incluster_config, list_namespac
         list_namespaced_pod.assert_has_calls(calls)
         delete_pod.assert_called_with("dummy", "awesome-namespace")
         load_incluster_config.assert_called_once()
+
+    # -- min-completed-minutes tests --
+
+    def _make_pod(self, name, phase, finished_at, reason=None, restart_policy="Never"):
+        """Build a minimal pod mock for min-completed-minutes tests."""
+        pod = MagicMock()
+        pod.metadata.name = name
+        pod.metadata.creation_timestamp = parse("2021-12-20T08:00:00Z")
+        pod.status.phase = phase
+        pod.status.reason = reason
+        pod.spec.restart_policy = restart_policy
+        container_status = MagicMock()
+        container_status.state.terminated.finished_at = finished_at
+        pod.status.container_statuses = [container_status]
+        return pod
+
+    @mock.patch("airflow.providers.cncf.kubernetes.cli.kubernetes_command._delete_pod")
+    @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
+    @mock.patch("kubernetes.config.load_incluster_config")
+    def test_cleanup_succeeded_pod_too_young_not_deleted(
+        self, load_incluster_config, list_namespaced_pod, delete_pod
+    ):
+        # finished_at far in the future → age is negative → less than 1 min → skip
+        pod = self._make_pod("run-o1sxc2on", "Succeeded", parse("2099-12-20T08:01:07Z"))
+        pods = MagicMock()
+        pods.metadata._continue = None
+        pods.items = [pod]
+        list_namespaced_pod.return_value = pods
+        kubernetes_command.cleanup_pods(
+            self.parser.parse_args(
+                [
+                    "kubernetes",
+                    "cleanup-pods",
+                    "--namespace",
+                    "awesome-namespace",
+                    "--min-completed-minutes",
+                    "1",
+                ]
+            )
+        )
+        delete_pod.assert_not_called()
+
+    @mock.patch("airflow.providers.cncf.kubernetes.cli.kubernetes_command._delete_pod")
+    @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
+    @mock.patch("kubernetes.config.load_incluster_config")
+    def test_cleanup_succeeded_pod_old_enough_deleted(
+        self, load_incluster_config, list_namespaced_pod, delete_pod
+    ):
+        # finished_at far in the past → age > 1 min → delete
+        pod = self._make_pod("run-oldpod", "Succeeded", parse("2021-12-20T08:01:07Z"))
+        pods = MagicMock()
+        pods.metadata._continue = None
+        pods.items = [pod]
+        list_namespaced_pod.return_value = pods
+        kubernetes_command.cleanup_pods(
+            self.parser.parse_args(
+                [
+                    "kubernetes",
+                    "cleanup-pods",
+                    "--namespace",
+                    "awesome-namespace",
+                    "--min-completed-minutes",
+                    "1",
+                ]
+            )
+        )
+        delete_pod.assert_called_with("run-oldpod", "awesome-namespace")
+
+    @mock.patch("airflow.providers.cncf.kubernetes.cli.kubernetes_command._delete_pod")
+    @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
+    @mock.patch("kubernetes.config.load_incluster_config")
+    def test_cleanup_min_completed_zero_deletes_immediately(
+        self, load_incluster_config, list_namespaced_pod, delete_pod
+    ):
+        # default (0) preserves existing behaviour: delete Succeeded pods regardless of age
+        pod = self._make_pod("run-newpod", "Succeeded", parse("2099-12-20T08:01:07Z"))
+        pods = MagicMock()
+        pods.metadata._continue = None
+        pods.items = [pod]
+        list_namespaced_pod.return_value = pods
+        kubernetes_command.cleanup_pods(
+            self.parser.parse_args(["kubernetes", "cleanup-pods", "--namespace", "awesome-namespace"])
+        )
+        delete_pod.assert_called_with("run-newpod", "awesome-namespace")
+
+    @mock.patch("airflow.providers.cncf.kubernetes.cli.kubernetes_command._delete_pod")
+    @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
+    @mock.patch("kubernetes.config.load_incluster_config")
+    def test_cleanup_failed_pod_too_young_not_deleted(
+        self, load_incluster_config, list_namespaced_pod, delete_pod
+    ):
+        # Failed + restart_policy=Never, but finished too recently → skip
+        pod = self._make_pod(
+            "run-failpod", "Failed", parse("2099-12-20T08:01:07Z"), restart_policy="Never"
+        )
+        pods = MagicMock()
+        pods.metadata._continue = None
+        pods.items = [pod]
+        list_namespaced_pod.return_value = pods
+        kubernetes_command.cleanup_pods(
+            self.parser.parse_args(
+                [
+                    "kubernetes",
+                    "cleanup-pods",
+                    "--namespace",
+                    "awesome-namespace",
+                    "--min-completed-minutes",
+                    "1",
+                ]
+            )
+        )
+        delete_pod.assert_not_called()

From 9a1de45e1b817341c791c5416d6b88023d7feb90 Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Tue, 28 Jul 2026 14:48:36 +0300
Subject: [PATCH 02/11] Update changelog with PR number #70595

---
 providers/cncf/kubernetes/docs/changelog.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/providers/cncf/kubernetes/docs/changelog.rst b/providers/cncf/kubernetes/docs/changelog.rst
index 9a1ca8b8bb99d..410509070471c 100644
--- a/providers/cncf/kubernetes/docs/changelog.rst
+++ b/providers/cncf/kubernetes/docs/changelog.rst
@@ -33,7 +33,7 @@ Changelog
 New features
 ~~~~~~~~~~~~
 
-* ``Add --min-completed-minutes to cleanup-pods to prevent KPO race condition (#XXXXX)``
+* ``Add --min-completed-minutes to cleanup-pods to prevent KPO race condition (#70595)``
 
 10.20.0
 .......

From 2df806e91e39a05aa88d3911f73e7311bbaa06e6 Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Tue, 28 Jul 2026 14:49:18 +0300
Subject: [PATCH 03/11] Fix ruff D213 docstring format in
 _get_pod_completion_time

---
 .../providers/cncf/kubernetes/cli/kubernetes_command.py        | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
index 01f7964a4a712..486b2b62c6e4f 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
@@ -120,7 +120,8 @@ def generate_pod_yaml(args):
 
 
 def _get_pod_completion_time(pod):
-    """Return the time the pod entered a terminal state, or its creation time as fallback.
+    """
+    Return the time the pod entered a terminal state, or its creation time as fallback.
 
     Uses the latest ``finished_at`` timestamp across all container statuses so that pods
     with multiple containers (e.g. an init container + a base container) are judged by

From c32ebfbc21dbe297b02f788db8075424b68eb64c Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Tue, 28 Jul 2026 14:56:32 +0300
Subject: [PATCH 04/11] Use exact poll interval (2 s) in
 --min-completed-minutes help text

---
 .../src/airflow/providers/cncf/kubernetes/cli/definition.py     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
index dfc86f6acaa78..3f89673172e6b 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
@@ -75,7 +75,7 @@
     help=(
         "Minimum age in minutes of a completed (Succeeded/Failed/Evicted) pod before it is deleted. "
         "Defaults to 0 (delete immediately, preserving current behaviour). "
-        "Set this to a value greater than the KubernetesPodOperator poll interval (~2 s) to prevent "
+        "Set this to a value greater than the KubernetesPodOperator poll interval (2 s, see ``await_pod_completion``) to prevent "
         "a race where the cleanup job removes a pod before KPO observes its terminal phase, "
         "causing a spurious task failure despite the pod having succeeded."
     ),

From 90e28ceaf6cd6663f4f7827533503fd0df69fdf0 Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Tue, 28 Jul 2026 15:00:01 +0300
Subject: [PATCH 05/11] Improve --min-completed-minutes help text clarity

---
 .../src/airflow/providers/cncf/kubernetes/cli/definition.py   | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
index 3f89673172e6b..2fce89c2d1266 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
@@ -75,8 +75,8 @@
     help=(
         "Minimum age in minutes of a completed (Succeeded/Failed/Evicted) pod before it is deleted. "
         "Defaults to 0 (delete immediately, preserving current behaviour). "
-        "Set this to a value greater than the KubernetesPodOperator poll interval (2 s, see ``await_pod_completion``) to prevent "
-        "a race where the cleanup job removes a pod before KPO observes its terminal phase, "
+        "Set this to a positive value to prevent a race condition where the cleanup job removes a "
+        "just-completed pod before KubernetesPodOperator has polled its terminal phase, "
         "causing a spurious task failure despite the pod having succeeded."
     ),
 )

From 098f5fee5cb7a7e57167544fefa483494787ce97 Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Tue, 28 Jul 2026 17:44:03 +0300
Subject: [PATCH 06/11] =?UTF-8?q?Revert=20changelog.rst=20=E2=80=94=20auto?=
 =?UTF-8?q?-generated=20by=20release=20manager?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 providers/cncf/kubernetes/docs/changelog.rst | 8 --------
 1 file changed, 8 deletions(-)

diff --git a/providers/cncf/kubernetes/docs/changelog.rst b/providers/cncf/kubernetes/docs/changelog.rst
index 410509070471c..82fd133f0047d 100644
--- a/providers/cncf/kubernetes/docs/changelog.rst
+++ b/providers/cncf/kubernetes/docs/changelog.rst
@@ -27,14 +27,6 @@
 Changelog
 ---------
 
-10.21.0
-.......
-
-New features
-~~~~~~~~~~~~
-
-* ``Add --min-completed-minutes to cleanup-pods to prevent KPO race condition (#70595)``
-
 10.20.0
 .......
 

From 80025867cc09c9d0105d7539eca2e4a46a6e61ef Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Wed, 29 Jul 2026 10:49:21 +0300
Subject: [PATCH 07/11] Address jedcunningham review on cleanup-pods
 min-completed-minutes

- Set default to 1 minute (no reason to keep the race by default)
- Shorten --min-completed-minutes help text
- Fix _get_pod_completion_time: scan init_container_statuses too;
  fall back to max(conditions.last_transition_time) instead of
  creation_timestamp (which predates actual completion)
- Add TestGetPodCompletionTime with real k8s model objects covering
  main-only, init-only, both, conditions fallback, and creation_timestamp
  last-resort cases
- Fix test_cleanup_min_completed_zero_deletes_immediately to pass
  --min-completed-minutes=0 explicitly now that default is 1
---
 .../cncf/kubernetes/cli/definition.py         |  7 +-
 .../cncf/kubernetes/cli/kubernetes_command.py | 30 +++++---
 .../kubernetes/cli/test_kubernetes_command.py | 73 ++++++++++++++++++-
 3 files changed, 94 insertions(+), 16 deletions(-)

diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
index 2fce89c2d1266..cbcbfac9e45ca 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
@@ -70,14 +70,11 @@
 
 ARG_MIN_COMPLETED_MINUTES = Arg(
     ("--min-completed-minutes",),
-    default=0,
+    default=1,
     type=positive_int(allow_zero=True),
     help=(
         "Minimum age in minutes of a completed (Succeeded/Failed/Evicted) pod before it is deleted. "
-        "Defaults to 0 (delete immediately, preserving current behaviour). "
-        "Set this to a positive value to prevent a race condition where the cleanup job removes a "
-        "just-completed pod before KubernetesPodOperator has polled its terminal phase, "
-        "causing a spurious task failure despite the pod having succeeded."
+        "Default is 1. Set to 0 to delete immediately."
     ),
 )
 
diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
index 486b2b62c6e4f..a296162153d39 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
@@ -121,17 +121,29 @@ def generate_pod_yaml(args):
 
 def _get_pod_completion_time(pod):
     """
-    Return the time the pod entered a terminal state, or its creation time as fallback.
+    Return the time the pod's last container finished.
 
-    Uses the latest ``finished_at`` timestamp across all container statuses so that pods
-    with multiple containers (e.g. an init container + a base container) are judged by
-    the time the *last* container finished, not by when the pod was created.
+    Scans both ``container_statuses`` and ``init_container_statuses`` and returns the
+    latest ``finished_at`` timestamp. Falls back to the latest condition
+    ``last_transition_time`` (which is updated at the terminal transition), and finally
+    to ``creation_timestamp`` as a last resort.
     """
-    times = []
-    for status in pod.status.container_statuses or []:
-        if status.state and status.state.terminated and status.state.terminated.finished_at:
-            times.append(status.state.terminated.finished_at)
-    return max(times) if times else pod.metadata.creation_timestamp
+    statuses = [*(pod.status.container_statuses or []), *(pod.status.init_container_statuses or [])]
+    times = [
+        s.state.terminated.finished_at
+        for s in statuses
+        if s.state and s.state.terminated and s.state.terminated.finished_at
+    ]
+    if times:
+        return max(times)
+    condition_times = [
+        c.last_transition_time
+        for c in (pod.status.conditions or [])
+        if c.last_transition_time
+    ]
+    if condition_times:
+        return max(condition_times)
+    return pod.metadata.creation_timestamp
 
 
 @cli_utils.action_cli(check_db=False)
diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
index 858436348f109..fc47799ae2224 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
@@ -24,6 +24,7 @@
 import kubernetes
 import pytest
 from dateutil.parser import parse
+from kubernetes.client import models as k8s
 
 from airflow.cli import cli_parser
 from airflow.executors import executor_loader
@@ -357,14 +358,23 @@ def test_cleanup_succeeded_pod_old_enough_deleted(
     def test_cleanup_min_completed_zero_deletes_immediately(
         self, load_incluster_config, list_namespaced_pod, delete_pod
     ):
-        # default (0) preserves existing behaviour: delete Succeeded pods regardless of age
+        # explicit 0 disables the age check: delete Succeeded pods regardless of age
         pod = self._make_pod("run-newpod", "Succeeded", parse("2099-12-20T08:01:07Z"))
         pods = MagicMock()
         pods.metadata._continue = None
         pods.items = [pod]
         list_namespaced_pod.return_value = pods
         kubernetes_command.cleanup_pods(
-            self.parser.parse_args(["kubernetes", "cleanup-pods", "--namespace", "awesome-namespace"])
+            self.parser.parse_args(
+                [
+                    "kubernetes",
+                    "cleanup-pods",
+                    "--namespace",
+                    "awesome-namespace",
+                    "--min-completed-minutes",
+                    "0",
+                ]
+            )
         )
         delete_pod.assert_called_with("run-newpod", "awesome-namespace")
 
@@ -395,3 +405,62 @@ def test_cleanup_failed_pod_too_young_not_deleted(
             )
         )
         delete_pod.assert_not_called()
+
+
+class TestGetPodCompletionTime:
+    T1 = parse("2024-01-01T10:00:00Z")
+    T2 = parse("2024-01-01T10:05:00Z")
+    T3 = parse("2024-01-01T09:00:00Z")  # earlier than T1/T2, used as creation_timestamp
+
+    def _cs(self, finished_at=None):
+        """Real V1ContainerStatus with optional finished_at."""
+        return k8s.V1ContainerStatus(
+            name="base",
+            ready=False,
+            restart_count=0,
+            image="img",
+            image_id="id",
+            state=k8s.V1ContainerState(
+                terminated=k8s.V1ContainerStateTerminated(exit_code=0, finished_at=finished_at)
+                if finished_at
+                else None
+            ),
+        )
+
+    def _cond(self, last_transition_time):
+        return k8s.V1PodCondition(type="Ready", status="False", last_transition_time=last_transition_time)
+
+    def _pod(self, container_statuses=None, init_container_statuses=None, conditions=None):
+        pod = MagicMock()
+        pod.status.container_statuses = container_statuses
+        pod.status.init_container_statuses = init_container_statuses
+        pod.status.conditions = conditions
+        pod.metadata.creation_timestamp = self.T3
+        return pod
+
+    def test_single_main_container(self):
+        pod = self._pod(container_statuses=[self._cs(self.T1)])
+        assert kubernetes_command._get_pod_completion_time(pod) == self.T1
+
+    def test_single_init_container_no_main(self):
+        pod = self._pod(container_statuses=[], init_container_statuses=[self._cs(self.T1)])
+        assert kubernetes_command._get_pod_completion_time(pod) == self.T1
+
+    def test_main_and_init_returns_max(self):
+        pod = self._pod(
+            container_statuses=[self._cs(self.T1)],
+            init_container_statuses=[self._cs(self.T2)],
+        )
+        assert kubernetes_command._get_pod_completion_time(pod) == self.T2
+
+    def test_no_finished_at_falls_back_to_conditions(self):
+        # container status present but no finished_at → conditions fallback
+        pod = self._pod(
+            container_statuses=[self._cs(finished_at=None)],
+            conditions=[self._cond(self.T1)],
+        )
+        assert kubernetes_command._get_pod_completion_time(pod) == self.T1
+
+    def test_no_containers_no_conditions_falls_back_to_creation_timestamp(self):
+        pod = self._pod(container_statuses=[], init_container_statuses=[], conditions=[])
+        assert kubernetes_command._get_pod_completion_time(pod) == self.T3

From 5bf01c0a37af06e89dfdc97f8535d38dfd2240ad Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Wed, 29 Jul 2026 13:16:10 +0300
Subject: [PATCH 08/11] Test cleanup-pods age guard for evicted and
 init-container pods
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The guard must not fall through to creation_timestamp for pods whose
containers never reached a terminated state — that path reports an
inflated age and deletes immediately, the unsafe direction. Freezing
time keeps every case at a realistic completion offset instead of a
timestamp in the future.
---
 .../cncf/kubernetes/cli/kubernetes_command.py |  21 +-
 .../kubernetes/cli/test_kubernetes_command.py | 272 +++++++++---------
 2 files changed, 143 insertions(+), 150 deletions(-)

diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
index a296162153d39..6b3f339ee0e64 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
@@ -137,9 +137,7 @@ def _get_pod_completion_time(pod):
     if times:
         return max(times)
     condition_times = [
-        c.last_transition_time
-        for c in (pod.status.conditions or [])
-        if c.last_transition_time
+        c.last_transition_time for c in (pod.status.conditions or []) if c.last_transition_time
     ]
     if condition_times:
         return max(condition_times)
@@ -202,23 +200,18 @@ def cleanup_pods(args):
             pod_restart_policy = pod.spec.restart_policy.lower()
             current_time = datetime.now(pod.metadata.creation_timestamp.tzinfo)
 
-            terminal = (
+            is_terminal = (
                 pod_phase == pod_succeeded
                 or (pod_phase == pod_failed and pod_restart_policy == pod_restart_policy_never)
                 or (pod_reason == pod_reason_evicted)
             )
-            terminal_old_enough = terminal and (
+            is_terminal_old_enough = is_terminal and (
                 min_completed_minutes == 0
-                or current_time - _get_pod_completion_time(pod)
-                > timedelta(minutes=min_completed_minutes)
+                or current_time - _get_pod_completion_time(pod) > timedelta(minutes=min_completed_minutes)
             )
-            if (
-                terminal_old_enough
-                or (
-                    pod_phase == pod_pending
-                    and current_time - pod.metadata.creation_timestamp
-                    > timedelta(minutes=min_pending_minutes)
-                )
+            if is_terminal_old_enough or (
+                pod_phase == pod_pending
+                and current_time - pod.metadata.creation_timestamp > timedelta(minutes=min_pending_minutes)
             ):
                 print(
                     f'Deleting pod "{pod_name}" phase "{pod_phase}" and reason "{pod_reason}", '
diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
index fc47799ae2224..7c1dfdedc30bd 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
@@ -18,11 +18,13 @@
 
 import importlib
 import os
+from datetime import timedelta
 from unittest import mock
 from unittest.mock import MagicMock, call
 
 import kubernetes
 import pytest
+import time_machine
 from dateutil.parser import parse
 from kubernetes.client import models as k8s
 
@@ -35,6 +37,54 @@
 
 pytestmark = pytest.mark.db_test
 
+NOW = parse("2024-01-01T13:15:17Z")
+
+CONTAINER_STATUS_ATTRS = {
+    "name": "base",
+    "ready": False,
+    "restart_count": 0,
+    "image": "img",
+    "image_id": "id",
+}
+
+
+def make_container_status(finished_at):
+    return k8s.V1ContainerStatus(
+        **CONTAINER_STATUS_ATTRS,
+        state=k8s.V1ContainerState(
+            terminated=k8s.V1ContainerStateTerminated(exit_code=0, finished_at=finished_at)
+            if finished_at
+            else None
+        ),
+    )
+
+
+def make_terminal_pod(
+    name,
+    phase,
+    reason=None,
+    restart_policy="Never",
+    finished_at=None,
+    init_finished_at=None,
+    condition_time=None,
+):
+    conditions = (
+        [k8s.V1PodCondition(type="Ready", status="False", last_transition_time=condition_time)]
+        if condition_time
+        else []
+    )
+    return k8s.V1Pod(
+        metadata=k8s.V1ObjectMeta(name=name, creation_timestamp=NOW - timedelta(hours=1)),
+        spec=k8s.V1PodSpec(containers=[], restart_policy=restart_policy),
+        status=k8s.V1PodStatus(
+            phase=phase,
+            reason=reason,
+            container_statuses=[make_container_status(finished_at)] if finished_at else [],
+            init_container_statuses=[make_container_status(init_finished_at)] if init_finished_at else [],
+            conditions=conditions,
+        ),
+    )
+
 
 class TestGenerateDagYamlCommand:
     @classmethod
@@ -285,112 +335,69 @@ def test_list_pod_with_continue_token(self, load_incluster_config, list_namespac
         delete_pod.assert_called_with("dummy", "awesome-namespace")
         load_incluster_config.assert_called_once()
 
-    # -- min-completed-minutes tests --
-
-    def _make_pod(self, name, phase, finished_at, reason=None, restart_policy="Never"):
-        """Build a minimal pod mock for min-completed-minutes tests."""
-        pod = MagicMock()
-        pod.metadata.name = name
-        pod.metadata.creation_timestamp = parse("2021-12-20T08:00:00Z")
-        pod.status.phase = phase
-        pod.status.reason = reason
-        pod.spec.restart_policy = restart_policy
-        container_status = MagicMock()
-        container_status.state.terminated.finished_at = finished_at
-        pod.status.container_statuses = [container_status]
-        return pod
-
-    @mock.patch("airflow.providers.cncf.kubernetes.cli.kubernetes_command._delete_pod")
-    @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
-    @mock.patch("kubernetes.config.load_incluster_config")
-    def test_cleanup_succeeded_pod_too_young_not_deleted(
-        self, load_incluster_config, list_namespaced_pod, delete_pod
-    ):
-        # finished_at far in the future → age is negative → less than 1 min → skip
-        pod = self._make_pod("run-o1sxc2on", "Succeeded", parse("2099-12-20T08:01:07Z"))
-        pods = MagicMock()
-        pods.metadata._continue = None
-        pods.items = [pod]
-        list_namespaced_pod.return_value = pods
-        kubernetes_command.cleanup_pods(
-            self.parser.parse_args(
-                [
-                    "kubernetes",
-                    "cleanup-pods",
-                    "--namespace",
-                    "awesome-namespace",
-                    "--min-completed-minutes",
-                    "1",
-                ]
-            )
-        )
-        delete_pod.assert_not_called()
-
-    @mock.patch("airflow.providers.cncf.kubernetes.cli.kubernetes_command._delete_pod")
-    @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
-    @mock.patch("kubernetes.config.load_incluster_config")
-    def test_cleanup_succeeded_pod_old_enough_deleted(
-        self, load_incluster_config, list_namespaced_pod, delete_pod
-    ):
-        # finished_at far in the past → age > 1 min → delete
-        pod = self._make_pod("run-oldpod", "Succeeded", parse("2021-12-20T08:01:07Z"))
-        pods = MagicMock()
-        pods.metadata._continue = None
-        pods.items = [pod]
-        list_namespaced_pod.return_value = pods
-        kubernetes_command.cleanup_pods(
-            self.parser.parse_args(
-                [
-                    "kubernetes",
-                    "cleanup-pods",
-                    "--namespace",
-                    "awesome-namespace",
-                    "--min-completed-minutes",
-                    "1",
-                ]
-            )
-        )
-        delete_pod.assert_called_with("run-oldpod", "awesome-namespace")
-
-    @mock.patch("airflow.providers.cncf.kubernetes.cli.kubernetes_command._delete_pod")
-    @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
-    @mock.patch("kubernetes.config.load_incluster_config")
-    def test_cleanup_min_completed_zero_deletes_immediately(
-        self, load_incluster_config, list_namespaced_pod, delete_pod
-    ):
-        # explicit 0 disables the age check: delete Succeeded pods regardless of age
-        pod = self._make_pod("run-newpod", "Succeeded", parse("2099-12-20T08:01:07Z"))
-        pods = MagicMock()
-        pods.metadata._continue = None
-        pods.items = [pod]
-        list_namespaced_pod.return_value = pods
-        kubernetes_command.cleanup_pods(
-            self.parser.parse_args(
-                [
-                    "kubernetes",
-                    "cleanup-pods",
-                    "--namespace",
-                    "awesome-namespace",
-                    "--min-completed-minutes",
-                    "0",
-                ]
-            )
-        )
-        delete_pod.assert_called_with("run-newpod", "awesome-namespace")
-
+    @pytest.mark.parametrize(
+        "pod_kwargs, min_completed_minutes, expect_deleted",
+        [
+            pytest.param(
+                {"phase": "Succeeded", "finished_at": NOW - timedelta(seconds=3)},
+                1,
+                False,
+                id="succeeded-just-finished-kept",
+            ),
+            pytest.param(
+                {"phase": "Succeeded", "finished_at": NOW - timedelta(minutes=5)},
+                1,
+                True,
+                id="succeeded-old-enough-deleted",
+            ),
+            pytest.param(
+                {"phase": "Succeeded", "finished_at": NOW - timedelta(seconds=3)},
+                0,
+                True,
+                id="zero-disables-guard",
+            ),
+            pytest.param(
+                {"phase": "Failed", "finished_at": NOW - timedelta(seconds=3)},
+                1,
+                False,
+                id="failed-just-finished-kept",
+            ),
+            pytest.param(
+                {"phase": "Failed", "init_finished_at": NOW - timedelta(seconds=3)},
+                1,
+                False,
+                id="init-container-failed-just-finished-kept",
+            ),
+            pytest.param(
+                {"phase": "Failed", "reason": "Evicted", "condition_time": NOW - timedelta(seconds=3)},
+                1,
+                False,
+                id="evicted-before-containers-started-kept",
+            ),
+            pytest.param(
+                {"phase": "Failed", "reason": "Evicted", "condition_time": NOW - timedelta(minutes=5)},
+                1,
+                True,
+                id="evicted-old-enough-deleted",
+            ),
+        ],
+    )
+    @time_machine.travel(NOW, tick=False)
     @mock.patch("airflow.providers.cncf.kubernetes.cli.kubernetes_command._delete_pod")
     @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
     @mock.patch("kubernetes.config.load_incluster_config")
-    def test_cleanup_failed_pod_too_young_not_deleted(
-        self, load_incluster_config, list_namespaced_pod, delete_pod
+    def test_cleanup_pods_min_completed_minutes(
+        self,
+        load_incluster_config,
+        list_namespaced_pod,
+        delete_pod,
+        pod_kwargs,
+        min_completed_minutes,
+        expect_deleted,
     ):
-        # Failed + restart_policy=Never, but finished too recently → skip
-        pod = self._make_pod(
-            "run-failpod", "Failed", parse("2099-12-20T08:01:07Z"), restart_policy="Never"
-        )
         pods = MagicMock()
         pods.metadata._continue = None
-        pods.items = [pod]
+        pods.items = [make_terminal_pod("run-o1sxc2on", **pod_kwargs)]
         list_namespaced_pod.return_value = pods
         kubernetes_command.cleanup_pods(
             self.parser.parse_args(
@@ -400,67 +407,60 @@ def test_cleanup_failed_pod_too_young_not_deleted(
                     "--namespace",
                     "awesome-namespace",
                     "--min-completed-minutes",
-                    "1",
+                    str(min_completed_minutes),
                 ]
             )
         )
-        delete_pod.assert_not_called()
+        if expect_deleted:
+            delete_pod.assert_called_once_with("run-o1sxc2on", "awesome-namespace")
+        else:
+            delete_pod.assert_not_called()
 
 
 class TestGetPodCompletionTime:
     T1 = parse("2024-01-01T10:00:00Z")
     T2 = parse("2024-01-01T10:05:00Z")
-    T3 = parse("2024-01-01T09:00:00Z")  # earlier than T1/T2, used as creation_timestamp
-
-    def _cs(self, finished_at=None):
-        """Real V1ContainerStatus with optional finished_at."""
-        return k8s.V1ContainerStatus(
-            name="base",
-            ready=False,
-            restart_count=0,
-            image="img",
-            image_id="id",
-            state=k8s.V1ContainerState(
-                terminated=k8s.V1ContainerStateTerminated(exit_code=0, finished_at=finished_at)
-                if finished_at
-                else None
-            ),
-        )
-
-    def _cond(self, last_transition_time):
-        return k8s.V1PodCondition(type="Ready", status="False", last_transition_time=last_transition_time)
+    CREATED_AT = parse("2024-01-01T09:00:00Z")  # earlier than T1/T2
 
     def _pod(self, container_statuses=None, init_container_statuses=None, conditions=None):
-        pod = MagicMock()
-        pod.status.container_statuses = container_statuses
-        pod.status.init_container_statuses = init_container_statuses
-        pod.status.conditions = conditions
-        pod.metadata.creation_timestamp = self.T3
-        return pod
+        return k8s.V1Pod(
+            metadata=k8s.V1ObjectMeta(name="run-o1sxc2on", creation_timestamp=self.CREATED_AT),
+            status=k8s.V1PodStatus(
+                container_statuses=container_statuses,
+                init_container_statuses=init_container_statuses,
+                conditions=conditions,
+            ),
+        )
 
     def test_single_main_container(self):
-        pod = self._pod(container_statuses=[self._cs(self.T1)])
+        pod = self._pod(container_statuses=[make_container_status(self.T1)])
         assert kubernetes_command._get_pod_completion_time(pod) == self.T1
 
     def test_single_init_container_no_main(self):
-        pod = self._pod(container_statuses=[], init_container_statuses=[self._cs(self.T1)])
+        pod = self._pod(container_statuses=[], init_container_statuses=[make_container_status(self.T1)])
         assert kubernetes_command._get_pod_completion_time(pod) == self.T1
 
     def test_main_and_init_returns_max(self):
         pod = self._pod(
-            container_statuses=[self._cs(self.T1)],
-            init_container_statuses=[self._cs(self.T2)],
+            container_statuses=[make_container_status(self.T1)],
+            init_container_statuses=[make_container_status(self.T2)],
         )
         assert kubernetes_command._get_pod_completion_time(pod) == self.T2
 
-    def test_no_finished_at_falls_back_to_conditions(self):
-        # container status present but no finished_at → conditions fallback
+    @pytest.mark.parametrize(
+        "container_status",
+        [
+            pytest.param(k8s.V1ContainerStatus(**CONTAINER_STATUS_ATTRS, state=None), id="no-state"),
+            pytest.param(make_container_status(finished_at=None), id="not-terminated"),
+        ],
+    )
+    def test_falls_back_to_conditions(self, container_status):
         pod = self._pod(
-            container_statuses=[self._cs(finished_at=None)],
-            conditions=[self._cond(self.T1)],
+            container_statuses=[container_status],
+            conditions=[k8s.V1PodCondition(type="Ready", status="False", last_transition_time=self.T1)],
         )
         assert kubernetes_command._get_pod_completion_time(pod) == self.T1
 
     def test_no_containers_no_conditions_falls_back_to_creation_timestamp(self):
         pod = self._pod(container_statuses=[], init_container_statuses=[], conditions=[])
-        assert kubernetes_command._get_pod_completion_time(pod) == self.T3
+        assert kubernetes_command._get_pod_completion_time(pod) == self.CREATED_AT

From 699fa22256473200b1a8c30f4f725caee8dd9469 Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Wed, 29 Jul 2026 13:39:15 +0300
Subject: [PATCH 09/11] Cover a terminated container status with no finish time

_get_pod_completion_time guards three levels of the container state, but
the fallback cases only reached the first two, so a terminated status
carrying no finishedAt went unexercised. Modelling the non-terminated
case as a waiting container also matches what k8s reports for a pod
evicted before its containers started.
---
 .../kubernetes/cli/test_kubernetes_command.py | 45 +++++++++----------
 1 file changed, 22 insertions(+), 23 deletions(-)

diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
index 7c1dfdedc30bd..4c831af251c29 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
@@ -39,23 +39,16 @@
 
 NOW = parse("2024-01-01T13:15:17Z")
 
-CONTAINER_STATUS_ATTRS = {
-    "name": "base",
-    "ready": False,
-    "restart_count": 0,
-    "image": "img",
-    "image_id": "id",
-}
 
-
-def make_container_status(finished_at):
+def make_container_status(state):
     return k8s.V1ContainerStatus(
-        **CONTAINER_STATUS_ATTRS,
-        state=k8s.V1ContainerState(
-            terminated=k8s.V1ContainerStateTerminated(exit_code=0, finished_at=finished_at)
-            if finished_at
-            else None
-        ),
+        name="base", ready=False, restart_count=0, image="img", image_id="id", state=state
+    )
+
+
+def make_terminated_status(finished_at):
+    return make_container_status(
+        k8s.V1ContainerState(terminated=k8s.V1ContainerStateTerminated(exit_code=0, finished_at=finished_at))
     )
 
 
@@ -79,8 +72,8 @@ def make_terminal_pod(
         status=k8s.V1PodStatus(
             phase=phase,
             reason=reason,
-            container_statuses=[make_container_status(finished_at)] if finished_at else [],
-            init_container_statuses=[make_container_status(init_finished_at)] if init_finished_at else [],
+            container_statuses=[make_terminated_status(finished_at)] if finished_at else [],
+            init_container_statuses=[make_terminated_status(init_finished_at)] if init_finished_at else [],
             conditions=conditions,
         ),
     )
@@ -433,25 +426,31 @@ def _pod(self, container_statuses=None, init_container_statuses=None, conditions
         )
 
     def test_single_main_container(self):
-        pod = self._pod(container_statuses=[make_container_status(self.T1)])
+        pod = self._pod(container_statuses=[make_terminated_status(self.T1)])
         assert kubernetes_command._get_pod_completion_time(pod) == self.T1
 
     def test_single_init_container_no_main(self):
-        pod = self._pod(container_statuses=[], init_container_statuses=[make_container_status(self.T1)])
+        pod = self._pod(container_statuses=[], init_container_statuses=[make_terminated_status(self.T1)])
         assert kubernetes_command._get_pod_completion_time(pod) == self.T1
 
     def test_main_and_init_returns_max(self):
         pod = self._pod(
-            container_statuses=[make_container_status(self.T1)],
-            init_container_statuses=[make_container_status(self.T2)],
+            container_statuses=[make_terminated_status(self.T1)],
+            init_container_statuses=[make_terminated_status(self.T2)],
         )
         assert kubernetes_command._get_pod_completion_time(pod) == self.T2
 
     @pytest.mark.parametrize(
         "container_status",
         [
-            pytest.param(k8s.V1ContainerStatus(**CONTAINER_STATUS_ATTRS, state=None), id="no-state"),
-            pytest.param(make_container_status(finished_at=None), id="not-terminated"),
+            pytest.param(make_container_status(state=None), id="no-state"),
+            pytest.param(
+                make_container_status(
+                    k8s.V1ContainerState(waiting=k8s.V1ContainerStateWaiting(reason="ContainerCreating"))
+                ),
+                id="never-terminated",
+            ),
+            pytest.param(make_terminated_status(finished_at=None), id="terminated-without-finished-at"),
         ],
     )
     def test_falls_back_to_conditions(self, container_status):

From 3fa1c8499144ffb4246088af8a124b3cdb3e5218 Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Wed, 29 Jul 2026 15:45:08 +0300
Subject: [PATCH 10/11] Use a tuple for parametrize names in cleanup-pods tests

---
 .../tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py   | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
index 4c831af251c29..29eadcc55a2bd 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
@@ -329,7 +329,7 @@ def test_list_pod_with_continue_token(self, load_incluster_config, list_namespac
         load_incluster_config.assert_called_once()
 
     @pytest.mark.parametrize(
-        "pod_kwargs, min_completed_minutes, expect_deleted",
+        ("pod_kwargs", "min_completed_minutes", "expect_deleted"),
         [
             pytest.param(
                 {"phase": "Succeeded", "finished_at": NOW - timedelta(seconds=3)},

From 891c95f8007a954106addff8918fdb5a4271aacd Mon Sep 17 00:00:00 2001
From: Noam Steiner <noamst@monday.com>
Date: Thu, 30 Jul 2026 10:56:45 +0300
Subject: [PATCH 11/11] Name the pending-pod age condition in cleanup-pods

---
 .../providers/cncf/kubernetes/cli/kubernetes_command.py      | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
index 6b3f339ee0e64..2d2490134dd74 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
@@ -209,10 +209,11 @@ def cleanup_pods(args):
                 min_completed_minutes == 0
                 or current_time - _get_pod_completion_time(pod) > timedelta(minutes=min_completed_minutes)
             )
-            if is_terminal_old_enough or (
+            is_pending_too_long = (
                 pod_phase == pod_pending
                 and current_time - pod.metadata.creation_timestamp > timedelta(minutes=min_pending_minutes)
-            ):
+            )
+            if is_terminal_old_enough or is_pending_too_long:
                 print(
                     f'Deleting pod "{pod_name}" phase "{pod_phase}" and reason "{pod_reason}", '
                     f'restart policy "{pod_restart_policy}"'