feat: add Lux validator fleet management system

Full-stack fleet lifecycle management for Lux blockchain validators
deployed via Helm to DOKS clusters.

Backend (Python/FastAPI):
- HelmDeployer: async helm/kubectl CLI wrapper with temp file cleanup,
  per-release locking, and command injection protection
- LuxFleetManager: fleet CRUD, scale, rolling restart, node health
  probes via luxd RPC (health + C-chain block height)
- Fleet API: authenticated REST endpoints (ApiKeyDep) with RFC 1123
  input validation, kubeconfig error sanitization
- Pydantic models: 3 networks (mainnet/testnet/devnet), fleet status
  enums, helm values translation
- 110 unit tests (0.30s, all passing)

Frontend (TypeScript/Next.js):
- Fleet dashboard: cluster selector, stats banner, fleet cards with
  scale/restart/destroy operations, node status table, create dialog
- Interactive SVG world map: zero-dep Mercator projection with DOKS
  region markers, pulsing animations, P2P mesh lines, click-to-zoom
- DOKS region data: 11 regions with coordinates and API client
- Dark theme throughout

api/bootnode/api/__init__.py

@@ -16,6 +16,7 @@ from bootnode.api.transfers import router as transfers_router
 from bootnode.api.wallets import router as wallets_router
 from bootnode.api.webhooks import router as webhooks_router
 from bootnode.api.team import router as team_router
+from bootnode.api.lux import router as lux_router
 from bootnode.api.zap import router as zap_router
 
 router = APIRouter()
@@ -36,3 +37,4 @@ router.include_router(gas_router, prefix="/gas", tags=["Gas Manager"])
 router.include_router(zap_router, prefix="/zap", tags=["ZAP Protocol"])
 router.include_router(infra_router, prefix="/infra", tags=["Infrastructure"])
 router.include_router(team_router, prefix="/team", tags=["Team"])
+router.include_router(lux_router, prefix="/lux", tags=["Lux Fleet"])

api/bootnode/api/lux/__init__.py

@@ -0,0 +1,261 @@
+"""Lux Fleet API -- REST endpoints for managing Lux validator fleets.
+
+Provides CRUD operations for Lux blockchain validator fleets deployed
+via Helm charts to DOKS clusters.
+"""
+
+import logging
+import re
+from typing import Optional
+
+from fastapi import APIRouter, BackgroundTasks, HTTPException
+
+from bootnode.api.deps import ApiKeyDep
+from bootnode.core.chains.lux import LuxFleetManager
+from bootnode.core.chains.lux_models import (
+    LuxFleetCreate,
+    LuxFleetResponse,
+    LuxFleetStats,
+    LuxFleetStatus,
+    LuxFleetSummary,
+    LuxFleetUpdate,
+    LuxNetwork,
+    NETWORK_CONFIGS,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# In-memory fleet registry (source of truth is Helm releases on clusters)
+_fleets: dict[str, dict] = {}
+
+# Singleton manager
+_manager = LuxFleetManager()
+
+# K8s name validation: RFC 1123 DNS label
+_K8S_NAME_RE = re.compile(r"^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?$")
+
+
+def _validate_k8s_name(name: str, field: str = "name") -> None:
+    """Validate a string is a valid K8s name (RFC 1123 DNS label)."""
+    if not _K8S_NAME_RE.match(name):
+        raise HTTPException(400, f"Invalid {field}: must match RFC 1123 DNS label")
+
+
+async def _get_kubeconfig(cluster_id: str) -> str:
+    """Fetch kubeconfig for a DOKS cluster via the infra API."""
+    try:
+        from bootnode.api.infra import do_get_kubeconfig, get_do_client
+        client = get_do_client()
+        if not client:
+            raise HTTPException(500, "DigitalOcean client not configured")
+        return await do_get_kubeconfig(client, cluster_id)
+    except ImportError:
+        raise HTTPException(500, "Infrastructure module not available")
+    except HTTPException:
+        raise
+    except Exception:
+        logger.error("Failed to fetch kubeconfig for cluster %s", cluster_id)
+        raise HTTPException(502, "Failed to fetch kubeconfig for cluster")
+
+
+# ---- Fleet CRUD ----
+
+@router.post("/fleets", response_model=LuxFleetResponse)
+async def create_fleet(
+    params: LuxFleetCreate,
+    background_tasks: BackgroundTasks,
+    _api_key: ApiKeyDep,
+) -> LuxFleetResponse:
+    """Create a new Lux validator fleet on a DOKS cluster."""
+    fleet_id = f"{params.name}-{params.network.value}"
+    if fleet_id in _fleets:
+        raise HTTPException(409, f"Fleet {fleet_id} already exists")
+
+    kubeconfig = await _get_kubeconfig(params.cluster_id)
+    result = await _manager.create(kubeconfig, params)
+
+    _fleets[fleet_id] = {
+        "name": params.name,
+        "cluster_id": params.cluster_id,
+        "network": params.network.value,
+    }
+
+    return result
+
+
+@router.get("/fleets", response_model=list[LuxFleetSummary])
+async def list_fleets(_api_key: ApiKeyDep, cluster_id: Optional[str] = None) -> list[LuxFleetSummary]:
+    """List all Lux fleets, optionally filtered by cluster."""
+    if cluster_id:
+        kubeconfig = await _get_kubeconfig(cluster_id)
+        return await _manager.list_fleets(kubeconfig, cluster_id)
+
+    # List across all known clusters
+    all_fleets = []
+    cluster_ids = {f["cluster_id"] for f in _fleets.values()}
+    for cid in cluster_ids:
+        try:
+            kubeconfig = await _get_kubeconfig(cid)
+            fleets = await _manager.list_fleets(kubeconfig, cid)
+            all_fleets.extend(fleets)
+        except Exception as e:
+            logger.warning("Failed to list fleets on cluster %s: %s", cid, e)
+    return all_fleets
+
+
+@router.get("/fleets/{cluster_id}/{network}", response_model=LuxFleetResponse)
+async def get_fleet(cluster_id: str, network: LuxNetwork, _api_key: ApiKeyDep) -> LuxFleetResponse:
+    """Get full status of a fleet including pods and endpoints."""
+    kubeconfig = await _get_kubeconfig(cluster_id)
+
+    # Find fleet name from registry
+    name = f"luxd-{network.value}"
+    for fid, fdata in _fleets.items():
+        if fdata["cluster_id"] == cluster_id and fdata["network"] == network.value:
+            name = fdata["name"]
+            break
+
+    return await _manager.get_status(kubeconfig, name, network, cluster_id)
+
+
+@router.patch("/fleets/{cluster_id}/{network}", response_model=LuxFleetResponse)
+async def update_fleet(
+    cluster_id: str,
+    network: LuxNetwork,
+    params: LuxFleetUpdate,
+    _api_key: ApiKeyDep,
+) -> LuxFleetResponse:
+    """Update a fleet (helm upgrade). Only provided fields are changed."""
+    kubeconfig = await _get_kubeconfig(cluster_id)
+
+    name = f"luxd-{network.value}"
+    for fid, fdata in _fleets.items():
+        if fdata["cluster_id"] == cluster_id and fdata["network"] == network.value:
+            name = fdata["name"]
+            break
+
+    return await _manager.update(kubeconfig, name, network, cluster_id, params)
+
+
+@router.delete("/fleets/{cluster_id}/{network}")
+async def destroy_fleet(cluster_id: str, network: LuxNetwork, _api_key: ApiKeyDep) -> dict:
+    """Destroy a fleet (helm uninstall). PVCs are preserved."""
+    kubeconfig = await _get_kubeconfig(cluster_id)
+    success = await _manager.destroy(kubeconfig, network)
+
+    # Remove from registry
+    to_remove = [
+        fid for fid, fdata in _fleets.items()
+        if fdata["cluster_id"] == cluster_id and fdata["network"] == network.value
+    ]
+    for fid in to_remove:
+        del _fleets[fid]
+
+    return {"success": success, "message": f"Fleet {network.value} destroyed on {cluster_id}"}
+
+
+# ---- Fleet operations ----
+
+@router.post("/fleets/{cluster_id}/{network}/scale", response_model=LuxFleetResponse)
+async def scale_fleet(
+    cluster_id: str,
+    network: LuxNetwork,
+    replicas: int,
+    _api_key: ApiKeyDep,
+) -> LuxFleetResponse:
+    """Scale fleet to the specified replica count."""
+    if replicas < 1 or replicas > 20:
+        raise HTTPException(400, "Replicas must be between 1 and 20")
+
+    kubeconfig = await _get_kubeconfig(cluster_id)
+    name = f"luxd-{network.value}"
+    return await _manager.scale(kubeconfig, name, network, cluster_id, replicas)
+
+
+@router.post("/fleets/{cluster_id}/{network}/restart")
+async def restart_fleet(
+    cluster_id: str,
+    network: LuxNetwork,
+    background_tasks: BackgroundTasks,
+    _api_key: ApiKeyDep,
+) -> dict:
+    """Rolling restart all pods (one at a time, OnDelete strategy)."""
+    kubeconfig = await _get_kubeconfig(cluster_id)
+
+    async def _do_restart():
+        try:
+            restarted = await _manager.rolling_restart(kubeconfig, network)
+            logger.info("Rolling restart complete: %s", restarted)
+        except Exception as e:
+            logger.error("Rolling restart failed: %s", e)
+
+    background_tasks.add_task(_do_restart)
+    return {"message": f"Rolling restart initiated for {network.value}", "status": "started"}
+
+
+@router.get("/fleets/{cluster_id}/{network}/pods/{pod_name}/logs")
+async def get_pod_logs(
+    cluster_id: str,
+    network: LuxNetwork,
+    pod_name: str,
+    _api_key: ApiKeyDep,
+    tail: int = 100,
+) -> dict:
+    """Get logs from a specific pod."""
+    _validate_k8s_name(pod_name, "pod_name")
+    if tail < 1 or tail > 10000:
+        raise HTTPException(400, "tail must be between 1 and 10000")
+    kubeconfig = await _get_kubeconfig(cluster_id)
+    logs = await _manager.get_pod_logs(kubeconfig, network, pod_name, tail=tail)
+    return {"pod": pod_name, "lines": logs.splitlines()}
+
+
+@router.get("/fleets/{cluster_id}/{network}/pods/{pod_name}/health")
+async def get_pod_health(
+    cluster_id: str,
+    network: LuxNetwork,
+    pod_name: str,
+    _api_key: ApiKeyDep,
+) -> dict:
+    """Probe a luxd node's health via RPC."""
+    _validate_k8s_name(pod_name, "pod_name")
+    kubeconfig = await _get_kubeconfig(cluster_id)
+    return await _manager.probe_node_health(kubeconfig, network, pod_name)
+
+
+# ---- Stats ----
+
+@router.get("/stats", response_model=LuxFleetStats)
+async def get_stats(_api_key: ApiKeyDep) -> LuxFleetStats:
+    """Get aggregate fleet statistics."""
+    total_fleets = len(_fleets)
+    fleets_by_network: dict[str, int] = {}
+    for fdata in _fleets.values():
+        net = fdata["network"]
+        fleets_by_network[net] = fleets_by_network.get(net, 0) + 1
+
+    return LuxFleetStats(
+        total_fleets=total_fleets,
+        fleets_by_network=fleets_by_network,
+    )
+
+
+# ---- Network info ----
+
+@router.get("/networks")
+async def list_networks() -> dict:
+    """List available Lux networks and their configurations."""
+    return {
+        "networks": {
+            net.value: {
+                "network_id": cfg.network_id,
+                "chain_id": cfg.chain_id,
+                "http_port": cfg.http_port,
+                "staking_port": cfg.staking_port,
+                "namespace": cfg.namespace,
+            }
+            for net, cfg in NETWORK_CONFIGS.items()
+        }
+    }

api/bootnode/config.py

@@ -126,6 +126,10 @@ class Settings(BaseSettings):
     deploy_k8s_namespace: str = "bootnode"
     deploy_k8s_context: str = ""
 
+    # Helm / Lux Fleet
+    helm_chart_path: str = "/opt/charts/lux"
+    helm_binary: str = "helm"
+
     @property
     def is_production(self) -> bool:
         return self.app_env == "production"

api/bootnode/core/chains/__init__.py

@@ -3,4 +3,10 @@
 from bootnode.core.chains.registry import Chain, ChainRegistry, ChainType, Network
 from bootnode.core.chains.rpc import RPCClient
 
-__all__ = ["ChainRegistry", "Chain", "Network", "ChainType", "RPCClient"]
+__all__ = [
+    "ChainRegistry",
+    "Chain",
+    "Network",
+    "ChainType",
+    "RPCClient",
+]

api/bootnode/core/chains/lux.py

@@ -0,0 +1,465 @@
+"""Lux fleet manager -- business logic for validator fleet lifecycle.
+
+Orchestrates Lux validator fleet deployments via HelmDeployer.
+Each fleet operation:
+  1. Fetches kubeconfig for the target cluster
+  2. Creates a HelmDeployer pointed at that cluster
+  3. Runs the Helm operation
+  4. Probes individual pod health via luxd RPC
+"""
+
+import asyncio
+import json
+import logging
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+from bootnode.core.chains.lux_models import (
+    NETWORK_CONFIGS,
+    LuxFleetCreate,
+    LuxFleetResponse,
+    LuxFleetStatus,
+    LuxFleetSummary,
+    LuxFleetUpdate,
+    LuxNetwork,
+    LuxNodeInfo,
+    LuxNodeStatus,
+    fleet_create_to_helm_values,
+    fleet_update_to_helm_values,
+)
+from bootnode.core.deploy.helm import HelmDeployer, HelmError
+
+logger = logging.getLogger(__name__)
+
+# Chart path defaults to the devops repo checkout
+DEFAULT_CHART_PATH = Path("/opt/charts/lux")
+
+# Network-specific values files
+NETWORK_VALUES_FILES: dict[str, str] = {
+    "mainnet": "values-mainnet.yaml",
+    "testnet": "values-testnet.yaml",
+}
+
+
+@dataclass
+class LuxFleetManager:
+    """Manages Lux validator fleet lifecycle via Helm."""
+
+    chart_path: Path = DEFAULT_CHART_PATH
+
+    def _release_name(self, network: LuxNetwork) -> str:
+        return f"luxd-{network.value}"
+
+    def _namespace(self, network: LuxNetwork) -> str:
+        return NETWORK_CONFIGS[network].namespace
+
+    def _fleet_id(self, name: str, network: LuxNetwork) -> str:
+        return f"{name}-{network.value}"
+
+    async def _get_deployer(self, kubeconfig_data: str) -> HelmDeployer:
+        """Create a HelmDeployer from raw kubeconfig YAML.
+
+        The returned deployer's kubeconfig temp file is tracked for cleanup.
+        Call deployer.cleanup_temp_files() when done.
+        """
+        tmp = tempfile.NamedTemporaryFile(
+            mode="w", suffix=".kubeconfig", delete=False
+        )
+        tmp.write(kubeconfig_data)
+        tmp.flush()
+        tmp.close()
+        kc_path = Path(tmp.name)
+        deployer = HelmDeployer(
+            chart_path=self.chart_path,
+            kubeconfig_path=kc_path,
+        )
+        # Track kubeconfig path for cleanup alongside helm values files
+        if not hasattr(deployer, "_temp_files"):
+            deployer._temp_files = []
+        deployer._temp_files.append(kc_path)
+        return deployer
+
+    def _get_values_files(self, network: LuxNetwork) -> list[Path]:
+        """Get the network-specific values file if one exists."""
+        filename = NETWORK_VALUES_FILES.get(network.value)
+        if filename:
+            full_path = self.chart_path / filename
+            if full_path.exists():
+                return [full_path]
+        return []
+
+    # ---- Fleet lifecycle ----
+
+    async def create(
+        self,
+        kubeconfig: str,
+        params: LuxFleetCreate,
+    ) -> LuxFleetResponse:
+        """Create a new Lux validator fleet."""
+        deployer = await self._get_deployer(kubeconfig)
+        try:
+            release_name = self._release_name(params.network)
+            namespace = self._namespace(params.network)
+            values = fleet_create_to_helm_values(params)
+            values_files = self._get_values_files(params.network)
+
+            logger.info(
+                "Creating Lux fleet: release=%s namespace=%s network=%s replicas=%d",
+                release_name, namespace, params.network.value, params.replicas,
+            )
+
+            try:
+                release = await deployer.install(
+                    release_name=release_name,
+                    namespace=namespace,
+                    values=values,
+                    values_files=values_files,
+                    wait=False,
+                    create_namespace=True,
+                )
+                status = LuxFleetStatus.DEPLOYING
+            except HelmError as e:
+                logger.error("Fleet creation failed: %s", e)
+                return LuxFleetResponse(
+                    id=self._fleet_id(params.name, params.network),
+                    name=params.name,
+                    cluster_id=params.cluster_id,
+                    network=params.network,
+                    status=LuxFleetStatus.ERROR,
+                    replicas=params.replicas,
+                    namespace=namespace,
+                    error=str(e),
+                )
+
+            return LuxFleetResponse(
+                id=self._fleet_id(params.name, params.network),
+                name=params.name,
+                cluster_id=params.cluster_id,
+                network=params.network,
+                status=status,
+                replicas=params.replicas,
+                namespace=namespace,
+                helm_revision=release.revision,
+                image=params.image if params.image else None,
+            )
+        finally:
+            deployer.cleanup_temp_files()
+
+    async def update(
+        self,
+        kubeconfig: str,
+        name: str,
+        network: LuxNetwork,
+        cluster_id: str,
+        params: LuxFleetUpdate,
+    ) -> LuxFleetResponse:
+        """Update an existing fleet (helm upgrade)."""
+        deployer = await self._get_deployer(kubeconfig)
+        try:
+            release_name = self._release_name(network)
+            namespace = self._namespace(network)
+            values = fleet_update_to_helm_values(params)
+
+            logger.info("Updating fleet: release=%s values=%s", release_name, values)
+
+            try:
+                release = await deployer.upgrade(
+                    release_name=release_name,
+                    namespace=namespace,
+                    values=values,
+                    reuse_values=True,
+                )
+            except HelmError as e:
+                logger.error("Fleet update failed: %s", e)
+                return LuxFleetResponse(
+                    id=self._fleet_id(name, network),
+                    name=name,
+                    cluster_id=cluster_id,
+                    network=network,
+                    status=LuxFleetStatus.ERROR,
+                    replicas=params.replicas or 0,
+                    namespace=namespace,
+                    error=str(e),
+                )
+
+            # Get pod status after upgrade
+            pods = await deployer.get_pods(namespace, "app=luxd")
+            ready_count = sum(1 for p in pods if p.ready)
+            total = len(pods)
+
+            return LuxFleetResponse(
+                id=self._fleet_id(name, network),
+                name=name,
+                cluster_id=cluster_id,
+                network=network,
+                status=LuxFleetStatus.UPDATING,
+                replicas=total,
+                ready_replicas=ready_count,
+                namespace=namespace,
+                helm_revision=release.revision,
+            )
+        finally:
+            deployer.cleanup_temp_files()
+
+    async def scale(
+        self,
+        kubeconfig: str,
+        name: str,
+        network: LuxNetwork,
+        cluster_id: str,
+        replicas: int,
+    ) -> LuxFleetResponse:
+        """Scale a fleet's replica count."""
+        return await self.update(
+            kubeconfig=kubeconfig,
+            name=name,
+            network=network,
+            cluster_id=cluster_id,
+            params=LuxFleetUpdate(replicas=replicas),
+        )
+
+    async def destroy(
+        self,
+        kubeconfig: str,
+        network: LuxNetwork,
+    ) -> bool:
+        """Destroy a fleet (helm uninstall). PVCs persist."""
+        deployer = await self._get_deployer(kubeconfig)
+        try:
+            release_name = self._release_name(network)
+            namespace = self._namespace(network)
+            return await deployer.uninstall(release_name, namespace)
+        finally:
+            deployer.cleanup_temp_files()
+
+    async def get_status(
+        self,
+        kubeconfig: str,
+        name: str,
+        network: LuxNetwork,
+        cluster_id: str,
+    ) -> LuxFleetResponse:
+        """Get full status of a fleet including pods and endpoints."""
+        deployer = await self._get_deployer(kubeconfig)
+        try:
+            release_name = self._release_name(network)
+            namespace = self._namespace(network)
+            net_cfg = NETWORK_CONFIGS[network]
+
+            # Get Helm release status
+            try:
+                release = await deployer.status(release_name, namespace)
+            except HelmError:
+                return LuxFleetResponse(
+                    id=self._fleet_id(name, network),
+                    name=name,
+                    cluster_id=cluster_id,
+                    network=network,
+                    status=LuxFleetStatus.ERROR,
+                    replicas=0,
+                    namespace=namespace,
+                    error="Helm release not found",
+                )
+
+            # Get pods
+            pods = await deployer.get_pods(namespace, "app=luxd")
+            ready_count = sum(1 for p in pods if p.ready)
+            total = len(pods)
+
+            # Get services for endpoint discovery
+            services = await deployer.get_services(namespace)
+            rpc_endpoint = None
+            staking_endpoint = None
+            for svc in services:
+                for ext_ip in svc.get("external_ips", []):
+                    if ext_ip:
+                        for port in svc.get("ports", []):
+                            if port.get("port") == net_cfg.http_port:
+                                rpc_endpoint = f"http://{ext_ip}:{net_cfg.http_port}"
+                            elif port.get("port") == net_cfg.staking_port:
+                                staking_endpoint = f"{ext_ip}:{net_cfg.staking_port}"
+
+            # Build node info
+            nodes = []
+            for i, pod in enumerate(sorted(pods, key=lambda p: p.name)):
+                node_status = LuxNodeStatus.HEALTHY if pod.ready else LuxNodeStatus.UNHEALTHY
+                if pod.status == "Pending":
+                    node_status = LuxNodeStatus.PENDING
+                elif pod.status == "Init":
+                    node_status = LuxNodeStatus.INIT
+
+                nodes.append(LuxNodeInfo(
+                    pod_name=pod.name,
+                    pod_index=i,
+                    status=node_status,
+                    http_port=net_cfg.http_port,
+                    staking_port=net_cfg.staking_port,
+                ))
+
+            # Determine fleet status
+            if total == 0:
+                fleet_status = LuxFleetStatus.DEPLOYING
+            elif ready_count == total:
+                fleet_status = LuxFleetStatus.RUNNING
+            elif ready_count > 0:
+                fleet_status = LuxFleetStatus.DEGRADED
+            else:
+                fleet_status = LuxFleetStatus.ERROR
+
+            return LuxFleetResponse(
+                id=self._fleet_id(name, network),
+                name=name,
+                cluster_id=cluster_id,
+                network=network,
+                status=fleet_status,
+                replicas=total,
+                ready_replicas=ready_count,
+                namespace=namespace,
+                helm_revision=release.revision,
+                rpc_endpoint=rpc_endpoint,
+                staking_endpoint=staking_endpoint,
+                nodes=nodes,
+            )
+        finally:
+            deployer.cleanup_temp_files()
+
+    async def list_fleets(
+        self,
+        kubeconfig: str,
+        cluster_id: str,
+    ) -> list[LuxFleetSummary]:
+        """List all Lux fleets on a cluster."""
+        deployer = await self._get_deployer(kubeconfig)
+        try:
+            releases = await deployer.list_releases(
+                all_namespaces=True,
+                filter_pattern="^luxd-",
+            )
+
+            fleets = []
+            for r in releases:
+                # Infer network from release name (luxd-mainnet -> mainnet)
+                network_str = r.name.removeprefix("luxd-")
+                try:
+                    network = LuxNetwork(network_str)
+                except ValueError:
+                    continue
+
+                pods = await deployer.get_pods(r.namespace, "app=luxd")
+                ready = sum(1 for p in pods if p.ready)
+                total = len(pods)
+
+                if total == 0:
+                    status = LuxFleetStatus.DEPLOYING
+                elif ready == total:
+                    status = LuxFleetStatus.RUNNING
+                elif ready > 0:
+                    status = LuxFleetStatus.DEGRADED
+                else:
+                    status = LuxFleetStatus.ERROR
+
+                fleets.append(LuxFleetSummary(
+                    id=f"{cluster_id}:{r.name}",
+                    name=r.name,
+                    network=network,
+                    status=status,
+                    replicas=total,
+                    ready_replicas=ready,
+                    cluster_id=cluster_id,
+                    created_at=r.updated or "",
+                ))
+
+            return fleets
+        finally:
+            deployer.cleanup_temp_files()
+
+    async def get_pod_logs(
+        self,
+        kubeconfig: str,
+        network: LuxNetwork,
+        pod_name: str,
+        tail: int = 100,
+    ) -> str:
+        """Get logs from a specific pod."""
+        deployer = await self._get_deployer(kubeconfig)
+        try:
+            namespace = self._namespace(network)
+            return await deployer.get_pod_logs(pod_name, namespace, tail=tail)
+        finally:
+            deployer.cleanup_temp_files()
+
+    async def rolling_restart(
+        self,
+        kubeconfig: str,
+        network: LuxNetwork,
+    ) -> list[str]:
+        """Rolling restart: delete pods one at a time (OnDelete strategy)."""
+        deployer = await self._get_deployer(kubeconfig)
+        try:
+            namespace = self._namespace(network)
+            pods = await deployer.get_pods(namespace, "app=luxd")
+            restarted = []
+
+            for pod in sorted(pods, key=lambda p: p.name):
+                logger.info("Rolling restart: deleting %s", pod.name)
+                await deployer.delete_pod(pod.name, namespace)
+                restarted.append(pod.name)
+                # Wait for pod to come back ready (up to 5 min)
+                for _ in range(60):
+                    await asyncio.sleep(5)
+                    new_pods = await deployer.get_pods(namespace, "app=luxd")
+                    replaced = next((p for p in new_pods if p.name == pod.name), None)
+                    if replaced and replaced.ready:
+                        break
+
+            return restarted
+        finally:
+            deployer.cleanup_temp_files()
+
+    async def probe_node_health(
+        self,
+        kubeconfig: str,
+        network: LuxNetwork,
+        pod_name: str,
+    ) -> dict:
+        """Probe a luxd node's health via kubectl exec + wget."""
+        deployer = await self._get_deployer(kubeconfig)
+        try:
+            namespace = self._namespace(network)
+            net_cfg = NETWORK_CONFIGS[network]
+            port = net_cfg.http_port
+
+            results: dict = {}
+
+            # Health check
+            try:
+                health = await deployer.exec_pod(
+                    pod_name, namespace,
+                    ["wget", "-qO-", f"http://localhost:{port}/ext/health"],
+                )
+                results["health"] = json.loads(health)
+                results["healthy"] = results["health"].get("healthy", False)
+            except Exception:
+                results["healthy"] = False
+
+            # Block height
+            try:
+                rpc_resp = await deployer.exec_pod(
+                    pod_name, namespace,
+                    [
+                        "wget", "-qO-",
+                        "--header=Content-Type: application/json",
+                        f"--post-data={{\"jsonrpc\":\"2.0\",\"method\":\"eth_blockNumber\",\"params\":[],\"id\":1}}",
+                        f"http://localhost:{port}/ext/bc/C/rpc",
+                    ],
+                )
+                data = json.loads(rpc_resp)
+                results["c_chain_height"] = int(data.get("result", "0x0"), 16)
+            except Exception:
+                results["c_chain_height"] = None
+
+            return results
+        finally:
+            deployer.cleanup_temp_files()

api/bootnode/core/chains/lux_models.py

@@ -0,0 +1,444 @@
+"""Shared Pydantic models for Lux fleet management.
+
+This module is the single source of truth for data structures shared between:
+  - HelmDeployer (core/deploy/helm.py)
+  - Lux Fleet API (api/lux/__init__.py)
+  - Lux Chain Manager (core/chains/lux.py)
+  - Dashboard (web/app/dashboard/infrastructure/lux/)
+
+All four components import from here. No model duplication.
+
+The models map 1:1 to the Helm chart values in lux/devops/charts/lux/values.yaml
+so that API request bodies can be translated directly to helm --set flags or
+values override files without lossy conversion.
+"""
+
+from datetime import datetime
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, Field, field_validator
+
+
+# ---------------------------------------------------------------------------
+# Enums
+# ---------------------------------------------------------------------------
+
+class LuxNetwork(str, Enum):
+    """Lux network identifier. Maps to values.network in the Helm chart."""
+
+    MAINNET = "mainnet"
+    TESTNET = "testnet"
+    DEVNET = "devnet"
+
+
+class LuxFleetStatus(str, Enum):
+    """Lifecycle state of a fleet (Helm release)."""
+
+    PENDING = "pending"          # Request accepted, not yet applied
+    DEPLOYING = "deploying"      # helm install/upgrade in progress
+    RUNNING = "running"          # All pods healthy
+    DEGRADED = "degraded"        # Some pods not ready
+    UPDATING = "updating"        # Rolling update in progress
+    ERROR = "error"              # Helm or K8s error
+    DESTROYING = "destroying"    # helm uninstall in progress
+    DESTROYED = "destroyed"      # Release removed
+
+
+class LuxNodeStatus(str, Enum):
+    """Status of an individual luxd pod within a fleet."""
+
+    PENDING = "pending"
+    INIT = "init"                # Init container running
+    STARTING = "starting"        # luxd process starting
+    BOOTSTRAPPING = "bootstrapping"  # RLP import / chain sync
+    HEALTHY = "healthy"          # Health endpoint responding
+    UNHEALTHY = "unhealthy"      # Failed liveness/readiness
+    TERMINATED = "terminated"
+
+
+# ---------------------------------------------------------------------------
+# Helm chart value sub-models
+# These mirror the YAML structure in charts/lux/values.yaml exactly.
+# ---------------------------------------------------------------------------
+
+class LuxNetworkConfig(BaseModel):
+    """Per-network constants. Read-only reference data, not user input."""
+
+    network_id: int
+    chain_id: int
+    http_port: int
+    staking_port: int
+    namespace: str
+
+
+# Canonical network parameters. Must match values.yaml .networks.*
+NETWORK_CONFIGS: dict[LuxNetwork, LuxNetworkConfig] = {
+    LuxNetwork.MAINNET: LuxNetworkConfig(
+        network_id=1,
+        chain_id=96369,
+        http_port=9630,
+        staking_port=9631,
+        namespace="lux-mainnet",
+    ),
+    LuxNetwork.TESTNET: LuxNetworkConfig(
+        network_id=2,
+        chain_id=96368,
+        http_port=9640,
+        staking_port=9641,
+        namespace="lux-testnet",
+    ),
+    LuxNetwork.DEVNET: LuxNetworkConfig(
+        network_id=3,
+        chain_id=96370,
+        http_port=9650,
+        staking_port=9651,
+        namespace="lux-devnet",
+    ),
+}
+
+
+class RLPImportConfig(BaseModel):
+    """C-chain RLP block import settings. Maps to values.bootstrap.rlpImport."""
+
+    enabled: bool = False
+    base_url: str = ""
+    rlp_filename: str = ""
+    multi_part: bool = False
+    parts: list[str] = Field(default_factory=list)
+    min_height: int = 1
+    timeout: int = 7200
+
+
+class BootstrapConfig(BaseModel):
+    """Bootstrap node configuration. Maps to values.bootstrap."""
+
+    node_ids: list[str] = Field(default_factory=list)
+    use_hostnames: bool = True
+    external_ips: list[str] = Field(default_factory=list)
+    rlp_import: RLPImportConfig = Field(default_factory=RLPImportConfig)
+
+    @field_validator("node_ids")
+    @classmethod
+    def validate_node_ids(cls, v: list[str]) -> list[str]:
+        for nid in v:
+            if not nid.startswith("NodeID-"):
+                raise ValueError(f"Invalid node ID format: {nid!r} (must start with 'NodeID-')")
+        return v
+
+
+class ConsensusConfig(BaseModel):
+    """Consensus tuning. Maps to values.consensus."""
+
+    sample_size: int = 5
+    quorum_size: int = 4
+    sybil_protection_enabled: bool = False
+    require_validator_to_connect: bool = False
+    allow_private_ips: bool = True
+
+
+class ChainTrackingConfig(BaseModel):
+    """Subnet/chain tracking. Maps to values.chainTracking."""
+
+    track_all_chains: bool = True
+    tracked_chains: list[str] = Field(default_factory=list)
+    aliases: list[str] = Field(
+        default_factory=lambda: ["zoo", "hanzo", "spc", "pars"],
+    )
+
+
+class ImageConfig(BaseModel):
+    """Container image reference. Maps to values.image."""
+
+    repository: str = "registry.digitalocean.com/hanzo/bootnode"
+    tag: str = "luxd-v1.23.11"
+    pull_policy: str = "Always"
+
+
+class ResourceSpec(BaseModel):
+    """K8s resource request/limit pair."""
+
+    memory: str = "1Gi"
+    cpu: str = "500m"
+
+
+class ResourceConfig(BaseModel):
+    """Maps to values.resources."""
+
+    requests: ResourceSpec = Field(default_factory=lambda: ResourceSpec(memory="1Gi", cpu="500m"))
+    limits: ResourceSpec = Field(default_factory=lambda: ResourceSpec(memory="4Gi", cpu="2"))
+
+
+class StorageConfig(BaseModel):
+    """Maps to values.storage."""
+
+    size: str = "100Gi"
+    storage_class: str = "do-block-storage"
+
+
+class NodeServicesConfig(BaseModel):
+    """Maps to values.nodeServices."""
+
+    enabled: bool = True
+    type: str = "LoadBalancer"
+    annotations: dict[str, str] = Field(default_factory=dict)
+
+
+class APIConfig(BaseModel):
+    """Maps to values.api."""
+
+    admin_enabled: bool = True
+    metrics_enabled: bool = True
+    index_enabled: bool = True
+    http_allowed_hosts: str = "*"
+
+
+# ---------------------------------------------------------------------------
+# API request/response models
+# ---------------------------------------------------------------------------
+
+class LuxFleetCreate(BaseModel):
+    """Request body: create a new Lux fleet (Helm release).
+
+    Used by: POST /lux/fleets
+    Consumed by: HelmDeployer, LuxChainManager
+    """
+
+    name: str = Field(
+        ...,
+        min_length=1,
+        max_length=63,
+        pattern=r"^[a-z][a-z0-9-]*$",
+        description="Helm release name. Must be a valid K8s name.",
+    )
+    cluster_id: str = Field(
+        ...,
+        description="Target K8s cluster ID (from /infra/clusters).",
+    )
+    network: LuxNetwork = LuxNetwork.DEVNET
+    replicas: int = Field(default=5, ge=1, le=20)
+
+    # Optional overrides -- None means use chart defaults
+    image: Optional[ImageConfig] = None
+    bootstrap: Optional[BootstrapConfig] = None
+    consensus: Optional[ConsensusConfig] = None
+    chain_tracking: Optional[ChainTrackingConfig] = None
+    resources: Optional[ResourceConfig] = None
+    storage: Optional[StorageConfig] = None
+    node_services: Optional[NodeServicesConfig] = None
+    api: Optional[APIConfig] = None
+    log_level: str = "info"
+    db_type: str = "badgerdb"
+
+    @field_validator("replicas")
+    @classmethod
+    def quorum_safety(cls, v: int) -> int:
+        """Warn-level: consensus requires >= 4 for BFT quorum of 5."""
+        # Validation only; warning handled at the API layer.
+        return v
+
+
+class LuxFleetUpdate(BaseModel):
+    """Request body: update (helm upgrade) an existing fleet.
+
+    Used by: PATCH /lux/fleets/{fleet_id}
+    Only non-None fields are applied as overrides.
+    """
+
+    replicas: Optional[int] = Field(default=None, ge=1, le=20)
+    image: Optional[ImageConfig] = None
+    consensus: Optional[ConsensusConfig] = None
+    chain_tracking: Optional[ChainTrackingConfig] = None
+    resources: Optional[ResourceConfig] = None
+    log_level: Optional[str] = None
+
+
+class LuxNodeInfo(BaseModel):
+    """Status of a single luxd pod within a fleet.
+
+    Populated by LuxChainManager from K8s pod status + RPC probes.
+    """
+
+    pod_name: str
+    pod_index: int
+    node_id: Optional[str] = None
+    status: LuxNodeStatus = LuxNodeStatus.PENDING
+    external_ip: Optional[str] = None
+    http_port: int = 9650
+    staking_port: int = 9651
+    # RPC-derived metrics (populated lazily)
+    is_bootstrapped: Optional[bool] = None
+    peers: Optional[int] = None
+    c_chain_height: Optional[int] = None
+    p_chain_height: Optional[int] = None
+    uptime_pct: Optional[float] = None
+    # K8s resource usage
+    cpu_usage: Optional[float] = None
+    memory_usage_mb: Optional[float] = None
+
+
+class LuxFleetResponse(BaseModel):
+    """Response body: full fleet state.
+
+    Used by: GET /lux/fleets/{fleet_id}, POST /lux/fleets
+    """
+
+    id: str = Field(description="Fleet ID (= Helm release name + network).")
+    name: str
+    cluster_id: str
+    network: LuxNetwork
+    status: LuxFleetStatus = LuxFleetStatus.PENDING
+    replicas: int
+    ready_replicas: int = 0
+    image: ImageConfig = Field(default_factory=ImageConfig)
+    namespace: str = ""
+    helm_revision: int = 0
+    rpc_endpoint: Optional[str] = None
+    staking_endpoint: Optional[str] = None
+    nodes: list[LuxNodeInfo] = Field(default_factory=list)
+    created_at: str = Field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
+    updated_at: Optional[str] = None
+    error: Optional[str] = None
+
+
+class LuxFleetSummary(BaseModel):
+    """Lightweight fleet summary for list endpoints.
+
+    Used by: GET /lux/fleets
+    """
+
+    id: str
+    name: str
+    network: LuxNetwork
+    status: LuxFleetStatus
+    replicas: int
+    ready_replicas: int = 0
+    cluster_id: str
+    created_at: str
+
+
+class LuxFleetStats(BaseModel):
+    """Aggregate fleet statistics.
+
+    Used by: GET /lux/stats
+    """
+
+    total_fleets: int = 0
+    total_nodes: int = 0
+    healthy_nodes: int = 0
+    fleets_by_network: dict[str, int] = Field(default_factory=dict)
+    fleets_by_status: dict[str, int] = Field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# Helm values translation
+# ---------------------------------------------------------------------------
+
+def fleet_create_to_helm_values(req: LuxFleetCreate) -> dict:
+    """Convert a LuxFleetCreate request into a flat dict suitable for
+    helm install --set or a values override YAML file.
+
+    This is the canonical translation layer. HelmDeployer calls this;
+    no other component should build Helm values dicts directly.
+    """
+    vals: dict = {
+        "network": req.network.value,
+        "replicas": req.replicas,
+        "logLevel": req.log_level,
+        "dbType": req.db_type,
+    }
+
+    if req.image:
+        vals["image.repository"] = req.image.repository
+        vals["image.tag"] = req.image.tag
+        vals["image.pullPolicy"] = req.image.pull_policy
+
+    if req.bootstrap:
+        b = req.bootstrap
+        vals["bootstrap.useHostnames"] = b.use_hostnames
+        if b.node_ids:
+            # Helm list: pass as JSON or comma-separated via --set-json
+            vals["bootstrap.nodeIDs"] = b.node_ids
+        if b.external_ips:
+            vals["bootstrap.externalIPs"] = b.external_ips
+        r = b.rlp_import
+        vals["bootstrap.rlpImport.enabled"] = r.enabled
+        vals["bootstrap.rlpImport.baseUrl"] = r.base_url
+        vals["bootstrap.rlpImport.rlpFilename"] = r.rlp_filename
+        vals["bootstrap.rlpImport.multiPart"] = r.multi_part
+        if r.parts:
+            vals["bootstrap.rlpImport.parts"] = r.parts
+        vals["bootstrap.rlpImport.minHeight"] = r.min_height
+        vals["bootstrap.rlpImport.timeout"] = r.timeout
+
+    if req.consensus:
+        c = req.consensus
+        vals["consensus.sampleSize"] = c.sample_size
+        vals["consensus.quorumSize"] = c.quorum_size
+        vals["consensus.sybilProtectionEnabled"] = c.sybil_protection_enabled
+        vals["consensus.requireValidatorToConnect"] = c.require_validator_to_connect
+        vals["consensus.allowPrivateIPs"] = c.allow_private_ips
+
+    if req.chain_tracking:
+        ct = req.chain_tracking
+        vals["chainTracking.trackAllChains"] = ct.track_all_chains
+        if ct.tracked_chains:
+            vals["chainTracking.trackedChains"] = ct.tracked_chains
+        if ct.aliases:
+            vals["chainTracking.aliases"] = ct.aliases
+
+    if req.resources:
+        vals["resources.requests.memory"] = req.resources.requests.memory
+        vals["resources.requests.cpu"] = req.resources.requests.cpu
+        vals["resources.limits.memory"] = req.resources.limits.memory
+        vals["resources.limits.cpu"] = req.resources.limits.cpu
+
+    if req.storage:
+        vals["storage.size"] = req.storage.size
+        vals["storage.storageClass"] = req.storage.storage_class
+
+    if req.node_services:
+        vals["nodeServices.enabled"] = req.node_services.enabled
+        vals["nodeServices.type"] = req.node_services.type
+
+    if req.api:
+        vals["api.adminEnabled"] = req.api.admin_enabled
+        vals["api.metricsEnabled"] = req.api.metrics_enabled
+        vals["api.indexEnabled"] = req.api.index_enabled
+        vals["api.httpAllowedHosts"] = req.api.http_allowed_hosts
+
+    return vals
+
+
+def fleet_update_to_helm_values(req: LuxFleetUpdate) -> dict:
+    """Convert a LuxFleetUpdate into a partial Helm values dict.
+
+    Only non-None fields are included.
+    """
+    vals: dict = {}
+
+    if req.replicas is not None:
+        vals["replicas"] = req.replicas
+    if req.log_level is not None:
+        vals["logLevel"] = req.log_level
+    if req.image:
+        vals["image.repository"] = req.image.repository
+        vals["image.tag"] = req.image.tag
+        vals["image.pullPolicy"] = req.image.pull_policy
+    if req.consensus:
+        c = req.consensus
+        vals["consensus.sampleSize"] = c.sample_size
+        vals["consensus.quorumSize"] = c.quorum_size
+        vals["consensus.sybilProtectionEnabled"] = c.sybil_protection_enabled
+    if req.chain_tracking:
+        ct = req.chain_tracking
+        vals["chainTracking.trackAllChains"] = ct.track_all_chains
+        if ct.tracked_chains:
+            vals["chainTracking.trackedChains"] = ct.tracked_chains
+    if req.resources:
+        vals["resources.requests.memory"] = req.resources.requests.memory
+        vals["resources.requests.cpu"] = req.resources.requests.cpu
+        vals["resources.limits.memory"] = req.resources.limits.memory
+        vals["resources.limits.cpu"] = req.resources.limits.cpu
+
+    return vals

api/bootnode/core/deploy/__init__.py

@@ -21,6 +21,7 @@ from bootnode.core.deploy.base import DeployTarget, ServiceStatus, ServiceType
 from bootnode.core.deploy.docker import DockerDeployer
 from bootnode.core.deploy.factory import clear_deployer_cache, get_deployer
 from bootnode.core.deploy.kubernetes import KubernetesDeployer
+from bootnode.core.deploy.helm import HelmDeployer
 from bootnode.core.deploy.process import ProcessDeployer
 
 __all__ = [
@@ -30,6 +31,7 @@ __all__ = [
     "ServiceType",
     # Deployers
     "DockerDeployer",
+    "HelmDeployer",
     "KubernetesDeployer",
     "ProcessDeployer",
     # Factory

api/bootnode/core/deploy/helm.py

@@ -0,0 +1,446 @@
+"""Helm CLI deployer for Lux validator fleets.
+
+Wraps the helm and kubectl CLIs via asyncio subprocess for deploying
+Helm releases to remote DOKS clusters. NOT a DeployTarget subclass --
+different interface (Helm release-oriented, not ServiceType-oriented).
+"""
+
+import asyncio
+import json
+import logging
+import tempfile
+from dataclasses import dataclass, field
+from enum import Enum
+from pathlib import Path
+from typing import Optional
+
+import yaml
+
+logger = logging.getLogger(__name__)
+
+
+class HelmReleaseStatus(str, Enum):
+    DEPLOYED = "deployed"
+    FAILED = "failed"
+    PENDING_INSTALL = "pending-install"
+    PENDING_UPGRADE = "pending-upgrade"
+    PENDING_ROLLBACK = "pending-rollback"
+    SUPERSEDED = "superseded"
+    UNINSTALLED = "uninstalled"
+    UNINSTALLING = "uninstalling"
+    UNKNOWN = "unknown"
+
+
+class HelmError(Exception):
+    """Raised when a helm/kubectl command fails."""
+
+    def __init__(self, message: str, stderr: str = "", returncode: int = 1):
+        super().__init__(message)
+        self.stderr = stderr
+        self.returncode = returncode
+
+
+@dataclass
+class HelmRelease:
+    name: str
+    namespace: str
+    revision: int
+    status: HelmReleaseStatus
+    chart: str = ""
+    app_version: str = ""
+    updated: str = ""
+
+
+@dataclass
+class PodInfo:
+    name: str
+    ready: bool
+    status: str
+    restarts: int = 0
+    node: str = ""
+    ip: str = ""
+    age: str = ""
+
+
+@dataclass
+class HelmDeployer:
+    """Deploys Helm charts to K8s clusters via CLI.
+
+    Each instance is bound to a specific kubeconfig (cluster).
+    Create per-request via LuxFleetManager._get_deployer().
+    """
+
+    chart_path: Path
+    kubeconfig_path: Optional[Path] = None
+    kube_context: Optional[str] = None
+    helm_binary: str = "helm"
+    kubectl_binary: str = "kubectl"
+    _locks: dict = field(default_factory=dict, repr=False)
+
+    def _get_lock(self, release_name: str) -> asyncio.Lock:
+        if release_name not in self._locks:
+            self._locks[release_name] = asyncio.Lock()
+        return self._locks[release_name]
+
+    def _base_helm_args(self) -> list[str]:
+        args = [self.helm_binary]
+        if self.kubeconfig_path:
+            args += ["--kubeconfig", str(self.kubeconfig_path)]
+        if self.kube_context:
+            args += ["--kube-context", self.kube_context]
+        return args
+
+    def _base_kubectl_args(self) -> list[str]:
+        args = [self.kubectl_binary]
+        if self.kubeconfig_path:
+            args += ["--kubeconfig", str(self.kubeconfig_path)]
+        if self.kube_context:
+            args += ["--context", self.kube_context]
+        return args
+
+    async def _run(self, args: list[str], timeout: int = 300) -> str:
+        logger.debug("Running: %s", " ".join(args))
+        proc = await asyncio.create_subprocess_exec(
+            *args,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        try:
+            stdout, stderr = await asyncio.wait_for(
+                proc.communicate(), timeout=timeout
+            )
+        except asyncio.TimeoutError:
+            proc.kill()
+            raise HelmError(f"Command timed out after {timeout}s: {' '.join(args[:3])}")
+
+        out = stdout.decode().strip()
+        err = stderr.decode().strip()
+
+        if proc.returncode != 0:
+            raise HelmError(
+                f"Command failed (rc={proc.returncode}): {err or out}",
+                stderr=err,
+                returncode=proc.returncode,
+            )
+        return out
+
+    def _write_values_file(self, values: dict) -> Path:
+        """Write a values dict to a temp YAML file for helm -f.
+
+        Temp files are tracked and cleaned up via cleanup_temp_files().
+        """
+        tmp = tempfile.NamedTemporaryFile(
+            mode="w", suffix=".yaml", prefix="helm-values-", delete=False
+        )
+        yaml.safe_dump(values, tmp, default_flow_style=False)
+        tmp.flush()
+        tmp.close()
+        path = Path(tmp.name)
+        if not hasattr(self, "_temp_files"):
+            self._temp_files: list[Path] = []
+        self._temp_files.append(path)
+        return path
+
+    def cleanup_temp_files(self) -> None:
+        """Remove all temp files created by this deployer."""
+        for f in getattr(self, "_temp_files", []):
+            try:
+                f.unlink(missing_ok=True)
+            except OSError:
+                pass
+        self._temp_files = []
+
+    # ---- Helm operations ----
+
+    async def install(
+        self,
+        release_name: str,
+        namespace: str,
+        values: dict | None = None,
+        values_files: list[Path] | None = None,
+        wait: bool = False,
+        create_namespace: bool = True,
+        timeout: int = 600,
+    ) -> HelmRelease:
+        """helm install."""
+        async with self._get_lock(release_name):
+            args = self._base_helm_args() + [
+                "upgrade", "--install", release_name, str(self.chart_path),
+                "--namespace", namespace,
+                "--output", "json",
+            ]
+            if create_namespace:
+                args.append("--create-namespace")
+            if wait:
+                args += ["--wait", "--timeout", f"{timeout}s"]
+
+            # Network-specific values files first (lower priority)
+            for vf in (values_files or []):
+                args += ["-f", str(vf)]
+
+            # Computed values override (highest priority)
+            if values:
+                vf_path = self._write_values_file(values)
+                args += ["-f", str(vf_path)]
+
+            out = await self._run(args, timeout=timeout + 30)
+
+            try:
+                data = json.loads(out)
+                info = data.get("info", {})
+                return HelmRelease(
+                    name=release_name,
+                    namespace=namespace,
+                    revision=data.get("version", 1),
+                    status=HelmReleaseStatus(info.get("status", "unknown")),
+                    chart=data.get("chart", {}).get("metadata", {}).get("name", ""),
+                    app_version=data.get("chart", {}).get("metadata", {}).get("appVersion", ""),
+                    updated=info.get("last_deployed", ""),
+                )
+            except (json.JSONDecodeError, KeyError):
+                # Fallback: command succeeded but output wasn't JSON
+                return HelmRelease(
+                    name=release_name,
+                    namespace=namespace,
+                    revision=1,
+                    status=HelmReleaseStatus.DEPLOYED,
+                )
+
+    async def upgrade(
+        self,
+        release_name: str,
+        namespace: str,
+        values: dict | None = None,
+        values_files: list[Path] | None = None,
+        reuse_values: bool = True,
+        wait: bool = False,
+        timeout: int = 600,
+    ) -> HelmRelease:
+        """helm upgrade (existing release)."""
+        async with self._get_lock(release_name):
+            args = self._base_helm_args() + [
+                "upgrade", release_name, str(self.chart_path),
+                "--namespace", namespace,
+                "--output", "json",
+            ]
+            if reuse_values:
+                args.append("--reuse-values")
+            if wait:
+                args += ["--wait", "--timeout", f"{timeout}s"]
+
+            for vf in (values_files or []):
+                args += ["-f", str(vf)]
+
+            if values:
+                vf_path = self._write_values_file(values)
+                args += ["-f", str(vf_path)]
+
+            out = await self._run(args, timeout=timeout + 30)
+
+            try:
+                data = json.loads(out)
+                info = data.get("info", {})
+                return HelmRelease(
+                    name=release_name,
+                    namespace=namespace,
+                    revision=data.get("version", 1),
+                    status=HelmReleaseStatus(info.get("status", "unknown")),
+                    updated=info.get("last_deployed", ""),
+                )
+            except (json.JSONDecodeError, KeyError):
+                return HelmRelease(
+                    name=release_name,
+                    namespace=namespace,
+                    revision=1,
+                    status=HelmReleaseStatus.DEPLOYED,
+                )
+
+    async def uninstall(self, release_name: str, namespace: str) -> bool:
+        """helm uninstall."""
+        async with self._get_lock(release_name):
+            args = self._base_helm_args() + [
+                "uninstall", release_name,
+                "--namespace", namespace,
+            ]
+            try:
+                await self._run(args)
+                return True
+            except HelmError as e:
+                if "not found" in e.stderr.lower():
+                    return True
+                raise
+
+    async def rollback(
+        self, release_name: str, namespace: str, revision: int = 0
+    ) -> HelmRelease:
+        """helm rollback (0 = previous revision)."""
+        async with self._get_lock(release_name):
+            args = self._base_helm_args() + [
+                "rollback", release_name, str(revision),
+                "--namespace", namespace,
+            ]
+            await self._run(args)
+            return await self.status(release_name, namespace)
+
+    async def status(self, release_name: str, namespace: str) -> HelmRelease:
+        """helm status."""
+        args = self._base_helm_args() + [
+            "status", release_name,
+            "--namespace", namespace,
+            "--output", "json",
+        ]
+        out = await self._run(args)
+        data = json.loads(out)
+        info = data.get("info", {})
+        return HelmRelease(
+            name=data.get("name", release_name),
+            namespace=namespace,
+            revision=data.get("version", 0),
+            status=HelmReleaseStatus(info.get("status", "unknown")),
+            updated=info.get("last_deployed", ""),
+        )
+
+    async def list_releases(
+        self,
+        namespace: str = "",
+        all_namespaces: bool = False,
+        filter_pattern: str = "",
+    ) -> list[HelmRelease]:
+        """helm list."""
+        args = self._base_helm_args() + ["list", "--output", "json"]
+        if all_namespaces:
+            args.append("--all-namespaces")
+        elif namespace:
+            args += ["--namespace", namespace]
+        if filter_pattern:
+            args += ["--filter", filter_pattern]
+
+        out = await self._run(args)
+        releases = []
+        for r in json.loads(out or "[]"):
+            releases.append(HelmRelease(
+                name=r.get("name", ""),
+                namespace=r.get("namespace", ""),
+                revision=int(r.get("revision", "0")),
+                status=HelmReleaseStatus(r.get("status", "unknown")),
+                chart=r.get("chart", ""),
+                app_version=r.get("app_version", ""),
+                updated=r.get("updated", ""),
+            ))
+        return releases
+
+    # ---- Kubectl operations ----
+
+    async def get_pods(
+        self, namespace: str, label_selector: str = ""
+    ) -> list[PodInfo]:
+        """kubectl get pods."""
+        args = self._base_kubectl_args() + [
+            "get", "pods",
+            "--namespace", namespace,
+            "--output", "json",
+        ]
+        if label_selector:
+            args += ["-l", label_selector]
+
+        out = await self._run(args)
+        data = json.loads(out)
+        pods = []
+        for item in data.get("items", []):
+            meta = item.get("metadata", {})
+            status = item.get("status", {})
+            conditions = {c["type"]: c["status"] for c in status.get("conditions", [])}
+            container_statuses = status.get("containerStatuses", [])
+            restarts = sum(cs.get("restartCount", 0) for cs in container_statuses)
+
+            pods.append(PodInfo(
+                name=meta.get("name", ""),
+                ready=conditions.get("Ready") == "True",
+                status=status.get("phase", "Unknown"),
+                restarts=restarts,
+                node=item.get("spec", {}).get("nodeName", ""),
+                ip=status.get("podIP", ""),
+            ))
+        return pods
+
+    async def get_services(self, namespace: str, label_selector: str = "") -> list[dict]:
+        """kubectl get services, returns raw dicts."""
+        args = self._base_kubectl_args() + [
+            "get", "services",
+            "--namespace", namespace,
+            "--output", "json",
+        ]
+        if label_selector:
+            args += ["-l", label_selector]
+
+        out = await self._run(args)
+        data = json.loads(out)
+        services = []
+        for item in data.get("items", []):
+            meta = item.get("metadata", {})
+            spec = item.get("spec", {})
+            lb_status = item.get("status", {}).get("loadBalancer", {})
+            external_ips = [
+                ing.get("ip", ing.get("hostname", ""))
+                for ing in lb_status.get("ingress", [])
+            ]
+            services.append({
+                "name": meta.get("name", ""),
+                "type": spec.get("type", ""),
+                "cluster_ip": spec.get("clusterIP", ""),
+                "external_ips": external_ips,
+                "ports": [
+                    {"name": p.get("name", ""), "port": p.get("port"), "target_port": p.get("targetPort")}
+                    for p in spec.get("ports", [])
+                ],
+            })
+        return services
+
+    async def get_pod_logs(
+        self,
+        pod_name: str,
+        namespace: str,
+        tail: int = 100,
+        container: str = "",
+    ) -> str:
+        """kubectl logs."""
+        args = self._base_kubectl_args() + [
+            "logs", pod_name,
+            "--namespace", namespace,
+            f"--tail={tail}",
+        ]
+        if container:
+            args += ["-c", container]
+        return await self._run(args, timeout=30)
+
+    async def delete_pod(self, pod_name: str, namespace: str) -> bool:
+        """kubectl delete pod (for OnDelete rolling update)."""
+        args = self._base_kubectl_args() + [
+            "delete", "pod", pod_name,
+            "--namespace", namespace,
+        ]
+        try:
+            await self._run(args)
+            return True
+        except HelmError:
+            return False
+
+    _K8S_NAME_RE = __import__("re").compile(r"^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?$")
+
+    async def exec_pod(
+        self, pod_name: str, namespace: str, command: list[str], timeout: int = 30
+    ) -> str:
+        """kubectl exec for running commands in a pod.
+
+        Validates pod_name and namespace to prevent injection.
+        """
+        if not self._K8S_NAME_RE.match(pod_name):
+            raise ValueError(f"Invalid pod name: {pod_name}")
+        if not self._K8S_NAME_RE.match(namespace):
+            raise ValueError(f"Invalid namespace: {namespace}")
+        args = self._base_kubectl_args() + [
+            "exec", pod_name,
+            "--namespace", namespace,
+            "--",
+        ] + command
+        return await self._run(args, timeout=timeout)