fix(k8s): pin kubernetes client below 36 and self-heal on 401

pyproject.toml

@@ -9,7 +9,9 @@ dependencies = [
     "docker>=7.1.0",
     "fastapi>=0.128.0",
     "httpx>=0.28.1",
-    "kubernetes>=34.1.0",
+    # <36: kubernetes-client 36.0.0 stops picking up kubelet-rotated bound SA
+    # tokens in-cluster, causing permanent 401s after ~1h on AKS (issue #72).
+    "kubernetes>=34.1.0,<36",
     "minio>=7.2.20",
     "pyasn1>=0.6.3",
     "pydantic>=2.12.5",

src/services/kubernetes/client.py

@@ -6,6 +6,7 @@
 
 import json
 import os
+import time
 from functools import lru_cache
 from typing import Optional, Tuple
 
@@ -123,6 +124,13 @@ def build_custom_labels(
 _initialized: bool = False
 _init_error: str | None = None
 
+# Throttle for 401-triggered client resets (seconds). A reset re-runs config
+# loading, which re-reads the mounted ServiceAccount token from disk. The
+# throttle keeps a persistent auth failure (e.g. RBAC misconfiguration) from
+# turning every API call into a config reload.
+_UNAUTHORIZED_RESET_INTERVAL_SECONDS = 30.0
+_last_unauthorized_reset: float = 0.0
+
 
 def _load_config() -> bool:
     """Load Kubernetes configuration.
@@ -203,6 +211,50 @@ def get_kubernetes_client() -> tuple[CoreV1Api | None, BatchV1Api | None]:
     return _core_api, _batch_api
 
 
+def reset_clients() -> None:
+    """Drop the cached API clients so the next call rebuilds them.
+
+    Rebuilding re-runs config loading, which re-reads the mounted
+    ServiceAccount token from disk. Old client objects are left for GC so
+    in-flight requests on them can complete.
+    """
+    global _core_api, _batch_api, _initialized, _init_error
+    _core_api = None
+    _batch_api = None
+    _initialized = False
+    _init_error = None
+
+
+def handle_unauthorized(status: int | None) -> bool:
+    """Recover from a 401 by resetting the cached clients (throttled).
+
+    The API server returning 401 means the token we presented is no longer
+    valid — most commonly a stale bound ServiceAccount token after kubelet
+    rotation (issue #72). Resetting forces the next ``get_core_api()`` /
+    ``get_batch_api()`` to reload config and pick up the fresh on-disk token.
+
+    Call this from ``except ApiException`` handlers with ``e.status``.
+
+    Returns:
+        True when a reset was performed.
+    """
+    global _last_unauthorized_reset
+
+    if status != 401:
+        return False
+
+    now = time.monotonic()
+    if now - _last_unauthorized_reset < _UNAUTHORIZED_RESET_INTERVAL_SECONDS:
+        return False
+    _last_unauthorized_reset = now
+
+    logger.warning(
+        "Kubernetes API returned 401 Unauthorized — resetting cached clients to reload credentials",
+    )
+    reset_clients()
+    return True
+
+
 def get_core_api() -> CoreV1Api | None:
     """Get the Core V1 API client for pod operations."""
     core, _ = get_kubernetes_client()

src/services/kubernetes/job_executor.py

@@ -19,6 +19,7 @@
     get_batch_api,
     get_core_api,
     get_current_namespace,
+    handle_unauthorized,
 )
 from .models import (
     ExecutionResult,
@@ -155,6 +156,7 @@ async def create_job(
             )
 
         except ApiException as e:
+            handle_unauthorized(e.status)
             logger.error(
                 "Failed to create job",
                 job_name=job_name,
@@ -225,6 +227,7 @@ async def wait_for_pod_ready(
                         return False
 
             except ApiException as e:
+                handle_unauthorized(e.status)
                 logger.warning(
                     "Error checking pod status",
                     job_name=job.name,
@@ -397,6 +400,7 @@ async def delete_job(self, job: JobHandle):
             logger.debug("Deleted job", job_name=job.name)
 
         except ApiException as e:
+            handle_unauthorized(e.status)
             if e.status != 404:
                 logger.warning(
                     "Failed to delete job",

src/services/kubernetes/pool.py

@@ -22,6 +22,7 @@
     get_core_api,
     get_current_namespace,
     get_initialization_error,
+    handle_unauthorized,
 )
 from .models import (
     ExecutionResult,
@@ -414,6 +415,7 @@ async def _create_warm_pod(self) -> PooledPod | None:
             return pooled_pod
 
         except ApiException as e:
+            handle_unauthorized(e.status)
             logger.error(
                 "Failed to create warm pod (Kubernetes API error)",
                 pod_name=pod_name,
@@ -559,6 +561,7 @@ async def _delete_pod(self, handle: PodHandle):
             logger.debug("Deleted pod", pod_name=handle.name)
 
         except ApiException as e:
+            handle_unauthorized(e.status)
             if e.status != 404:
                 logger.warning(
                     "Failed to delete pod",
@@ -936,6 +939,7 @@ async def _inspect_pod_failure(self, handle: PodHandle) -> str | None:
                 lambda: core_api.read_namespaced_pod(handle.name, handle.namespace),
             )
         except ApiException as e:
+            handle_unauthorized(e.status)
             if e.status == 404:
                 return "pod not found (deleted or restarted)"
             return None

tests/unit/test_kubernetes_client_auth.py

@@ -0,0 +1,123 @@
+"""Tests for Kubernetes client auth resilience (issue #72).
+
+Two layers of protection are covered here:
+
+1. A regression test against the installed ``kubernetes`` client library:
+   ``load_incluster_config``'s loader must actually wire the ServiceAccount
+   token into ``Configuration.auth_settings()`` and serve a *fresh* token
+   after kubelet rotation. kubernetes-client 36.0.0 broke this (the
+   regenerated ``auth_settings()`` reads ``api_key['BearerToken']`` while the
+   in-cluster loader writes ``api_key['authorization']``, so requests went
+   out with no Authorization header at all). This test fails on any future
+   dependency bump that reintroduces that class of regression.
+
+2. Unit tests for the throttled 401 recovery in ``client.py``:
+   ``handle_unauthorized`` resets the cached API clients so the next call
+   reloads config (and the rotated on-disk token) instead of failing forever.
+"""
+
+import datetime
+
+import pytest
+
+from src.services.kubernetes import client as k8s_client
+
+
+class TestInClusterAuthRegression:
+    """Guard against kubernetes-client in-cluster auth regressions."""
+
+    @pytest.fixture
+    def incluster_config(self, tmp_path):
+        from kubernetes.client import Configuration
+        from kubernetes.config import incluster_config
+
+        token_file = tmp_path / "token"
+        cert_file = tmp_path / "ca.crt"
+        token_file.write_text("TOKEN_V1")
+        cert_file.write_text("dummy-ca")
+
+        loader = incluster_config.InClusterConfigLoader(
+            token_filename=str(token_file),
+            cert_filename=str(cert_file),
+            try_refresh_token=True,
+            environ={
+                "KUBERNETES_SERVICE_HOST": "1.2.3.4",
+                "KUBERNETES_SERVICE_PORT": "443",
+            },
+        )
+        configuration = Configuration()
+        loader.load_and_set(configuration)
+        return loader, configuration, token_file
+
+    def test_incluster_loader_populates_auth_settings(self, incluster_config):
+        """auth_settings() must expose the SA token as an Authorization header.
+
+        On kubernetes-client 36.0.0 this returns {} — every in-cluster request
+        is sent unauthenticated and the API server answers 401.
+        """
+        _, configuration, _ = incluster_config
+        auth = configuration.auth_settings()
+
+        assert "BearerToken" in auth, (
+            "In-cluster config produced no BearerToken auth setting — the "
+            "kubernetes client would send unauthenticated requests (issue #72). "
+            "Check the pinned kubernetes-client version."
+        )
+        assert auth["BearerToken"]["in"] == "header"
+        assert auth["BearerToken"]["key"] == "authorization"
+        assert auth["BearerToken"]["value"] == "bearer TOKEN_V1"
+
+    def test_incluster_token_refreshes_after_rotation(self, incluster_config):
+        """A kubelet-rotated token must be picked up once the cached one expires."""
+        loader, configuration, token_file = incluster_config
+
+        token_file.write_text("TOKEN_V2")
+        loader.token_expires_at = datetime.datetime.now() - datetime.timedelta(seconds=1)
+
+        auth = configuration.auth_settings()
+        assert auth["BearerToken"]["value"] == "bearer TOKEN_V2"
+
+
+class TestHandleUnauthorized:
+    """Throttled client reset on 401 responses."""
+
+    @pytest.fixture(autouse=True)
+    def _reset_module_state(self, monkeypatch):
+        monkeypatch.setattr(k8s_client, "_core_api", object())
+        monkeypatch.setattr(k8s_client, "_batch_api", object())
+        monkeypatch.setattr(k8s_client, "_initialized", True)
+        monkeypatch.setattr(k8s_client, "_last_unauthorized_reset", 0.0)
+
+    def test_401_resets_cached_clients(self):
+        assert k8s_client.handle_unauthorized(401) is True
+        assert k8s_client._core_api is None
+        assert k8s_client._batch_api is None
+        assert k8s_client._initialized is False
+
+    def test_non_401_does_not_reset(self):
+        for status in (None, 403, 404, 500):
+            assert k8s_client.handle_unauthorized(status) is False
+        assert k8s_client._initialized is True
+        assert k8s_client._core_api is not None
+
+    def test_reset_is_throttled(self, monkeypatch):
+        assert k8s_client.handle_unauthorized(401) is True
+
+        # A second 401 right after the reset must not thrash config loading.
+        monkeypatch.setattr(k8s_client, "_core_api", object())
+        monkeypatch.setattr(k8s_client, "_initialized", True)
+        assert k8s_client.handle_unauthorized(401) is False
+        assert k8s_client._initialized is True
+
+    def test_reset_allowed_after_throttle_window(self, monkeypatch):
+        assert k8s_client.handle_unauthorized(401) is True
+
+        monkeypatch.setattr(k8s_client, "_core_api", object())
+        monkeypatch.setattr(k8s_client, "_initialized", True)
+        monkeypatch.setattr(
+            k8s_client,
+            "_last_unauthorized_reset",
+            k8s_client._last_unauthorized_reset - k8s_client._UNAUTHORIZED_RESET_INTERVAL_SECONDS - 1,
+        )
+        assert k8s_client.handle_unauthorized(401) is True
+        assert k8s_client._initialized is False