feat(k8s): add Events-API node-disruption watcher and periodic LIST resync backstop

Watch cluster Events (involvedObject=Node) to surface node-disruption
signals (Karpenter, cordon/drain) that the pod/node watchers miss, and
add an opt-in periodic full-LIST resync so long-lived watch connections
that silently drop events self-heal without a restart. Both are opt-in
via config (events_watch_enabled, periodic_resync_interval_seconds).

Author: fgogolli

src/orb/providers/k8s/configuration/config.py

@@ -447,6 +447,38 @@ class K8sProviderConfig(BaseSettings, BaseProviderConfig):  # type: ignore[misc]
         ),
     )
 
+    # Events API watching (node-disruption visibility)
+    events_watch_enabled: bool = Field(
+        False,
+        description=(
+            "Opt-in flag for the k8s Events API watch background task.  When True, ORB "
+            "starts a K8sEventsWatcher that streams ``CoreV1Api.list_event_for_all_namespaces`` "
+            "filtered to ``involvedObject.kind=Node`` and caches Karpenter node-disruption "
+            "events (e.g. 'Disrupting Node: Underutilized/Delete', 'Disrupting Node: "
+            "Empty/Delete').  The cached disruption reason is available for surfacing in "
+            "status responses.  Default ``False`` because the events watcher requires "
+            "an additional RBAC grant (``events: get/list/watch`` on the core API group) "
+            "that may not exist in every cluster -- see "
+            "``docs/root/providers/k8s/rbac.yaml`` for the required rule."
+        ),
+    )
+
+    # Periodic full-LIST backstop for the pod watcher
+    periodic_resync_interval_seconds: int = Field(
+        0,
+        description=(
+            "Interval in seconds at which the pod watcher performs a full LIST of all "
+            "managed pods and reconciles the in-process cache, independent of 410-Gone "
+            "responses.  Mirrors the legacy RefreshPodsTask (hfcron.py) which ran every "
+            "~180 s as a correctness backstop against slow-drift apiservers.  "
+            "Default 0 (disabled) to avoid extra apiserver load -- opt in by setting a "
+            "positive value (e.g. 180).  When >0, a background asyncio task wakes every "
+            "``periodic_resync_interval_seconds`` and calls the same _relist_snapshot "
+            "path used on 410-Gone recovery, reconciling any cache drift that "
+            "accumulated during a healthy watch session."
+        ),
+    )
+
     # Controller-status cache
     controller_status_cache_ttl_seconds: float = Field(
         5.0,

src/orb/providers/k8s/services/health_check_service.py

@@ -135,18 +135,10 @@ def register_health_checks(
         health_check: "HealthCheck",
         kubernetes_client: "K8sClient",
     ) -> None:
-        """Register Kubernetes-specific health checks if the client is reachable."""
-        try:
-            client = kubernetes_client
-        except Exception as exc:
-            self._logger.debug(
-                "Skipping Kubernetes health-check registration: %s", exc, exc_info=True
-            )
-            return
-
+        """Register Kubernetes-specific health checks."""
         from orb.providers.k8s.health import register_k8s_health_checks
 
-        register_k8s_health_checks(health_check, client)
+        register_k8s_health_checks(health_check, kubernetes_client)
 
 
 __all__ = ["K8sHealthCheckService"]

src/orb/providers/k8s/strategy/k8s_provider_strategy.py

@@ -58,6 +58,7 @@
 )
 from orb.providers.k8s.strategy.handler_registry import K8sHandlerRegistry
 from orb.providers.k8s.value_objects import KubernetesProviderApi
+from orb.providers.k8s.watch.events_watcher import K8sEventsWatcher, K8sNodeEventsCache
 from orb.providers.k8s.watch.multi_namespace import MultiNamespaceWatcher
 from orb.providers.k8s.watch.node_state_cache import K8sNodeStateCache
 from orb.providers.k8s.watch.node_watcher import K8sNodeWatcher
@@ -157,6 +158,8 @@ def __init__(
         orphan_gc: Optional[OrphanGarbageCollector] = None,
         node_watcher: Optional[K8sNodeWatcher] = None,
         node_state_cache: Optional[K8sNodeStateCache] = None,
+        events_watcher: Optional[K8sEventsWatcher] = None,
+        node_events_cache: Optional[K8sNodeEventsCache] = None,
         native_spec_service: Optional[Any] = None,
     ) -> None:
         if not isinstance(config, K8sProviderConfig):
@@ -191,6 +194,12 @@ def __init__(
         # Tests inject both via the constructor kwargs to avoid real threads.
         self._node_state_cache: K8sNodeStateCache = node_state_cache or K8sNodeStateCache()
         self._node_watcher: Optional[K8sNodeWatcher] = node_watcher
+        # Events API watching.  When ``events_watch_enabled=True`` (opt-in via
+        # K8sProviderConfig) the strategy starts a K8sEventsWatcher on a
+        # background thread and populates K8sNodeEventsCache with Karpenter
+        # node-disruption events.  Tests inject both via constructor kwargs.
+        self._node_events_cache: K8sNodeEventsCache = node_events_cache or K8sNodeEventsCache()
+        self._events_watcher: Optional[K8sEventsWatcher] = events_watcher
         # Native-spec escape hatch.  Resolved lazily on first handler
         # construction.  ``None`` after resolution means the service is
         # unavailable (jinja2 missing, injected service not provided, etc.)
@@ -331,6 +340,7 @@ async def start_daemon_services(self) -> None:
         self._maybe_start_watch_manager()
         self._maybe_start_orphan_gc()
         self._maybe_start_node_watcher()
+        self._maybe_start_events_watcher()
         self._daemon_services_started = True
 
     def cleanup(self) -> None:
@@ -368,6 +378,15 @@ def cleanup(self) -> None:
                 "Failed to stop Kubernetes node watcher during cleanup: %s", exc, exc_info=True
             )
 
+        try:
+            if self._events_watcher is not None:
+                self._events_watcher.stop()
+                self._events_watcher = None
+        except Exception as exc:
+            self._logger.warning(
+                "Failed to stop Kubernetes events watcher during cleanup: %s", exc, exc_info=True
+            )
+
         # Stage 4: client cleanup.  Only clear ``_initialized`` when the
         # client is successfully cleaned up — if this stage fails the
         # provider can still serve reads via the existing connection while
@@ -707,6 +726,48 @@ def _maybe_start_node_watcher(self) -> None:
         except Exception as exc:
             self._logger.warning("Failed to start Kubernetes node watcher: %s", exc, exc_info=True)
 
+    # ------------------------------------------------------------------
+    # Events watcher lifecycle
+    # ------------------------------------------------------------------
+
+    @property
+    def node_events_cache(self) -> K8sNodeEventsCache:
+        """The shared node-events cache populated by the Events API watcher.
+
+        Always present (never ``None``) -- when ``events_watch_enabled`` is
+        ``False`` it is simply an empty cache that returns ``None`` for
+        every lookup.
+        """
+        return self._node_events_cache
+
+    def _maybe_start_events_watcher(self) -> None:
+        """Start the Events API watcher when enabled by config.
+
+        Like the node watcher, the events watcher runs on a plain
+        background daemon thread (not in the asyncio event loop) so it
+        can start from both synchronous (CLI bootstrap) and async
+        (daemon) contexts.
+
+        Requires the operator to have granted the ``events: get/list/watch``
+        RBAC verb on the core API group -- see
+        ``docs/root/providers/k8s/rbac.yaml``.
+        """
+        if not self._k8s_config.events_watch_enabled:
+            return
+        if self._events_watcher is None:
+            self._events_watcher = K8sEventsWatcher(
+                kubernetes_client=self.kubernetes_client,
+                cache=self._node_events_cache,
+                logger=self._logger,
+            )
+        try:
+            self._events_watcher.start()
+            self._logger.info("Kubernetes events watcher started (events_watch_enabled=True)")
+        except Exception as exc:
+            self._logger.warning(
+                "Failed to start Kubernetes events watcher: %s", exc, exc_info=True
+            )
+
     # ------------------------------------------------------------------
     # Operation dispatch
     # ------------------------------------------------------------------