skyhook_controller.go

/*
 * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 * SPDX-License-Identifier: Apache-2.0
 *
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package controller

import (
	"cmp"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"slices"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/NVIDIA/nodewright/operator/api/nodewright/v1alpha1"
	"github.com/NVIDIA/nodewright/operator/internal/dal"
	"github.com/NVIDIA/nodewright/operator/internal/drain"
	"github.com/NVIDIA/nodewright/operator/internal/version"
	"github.com/NVIDIA/nodewright/operator/internal/wrapper"
	"github.com/go-logr/logr"

	corev1 "k8s.io/api/core/v1"
	policyv1 "k8s.io/api/policy/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	utilerrors "k8s.io/apimachinery/pkg/util/errors"
	"k8s.io/client-go/tools/events"
	"k8s.io/kubernetes/pkg/util/taints"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
)

const (
	EventsReasonSkyhookApply       = "Apply"
	EventsReasonSkyhookInterrupt   = "Interrupt"
	EventsReasonSkyhookDrain       = "Drain"
	EventsReasonSkyhookStateChange = "State"
	EventsReasonNodeReboot         = "Reboot"
	EventTypeNormal                = "Normal"
	// EventTypeWarning = "Warning"
	TaintUnschedulable     = corev1.TaintNodeUnschedulable
	InterruptContainerName = "interrupt"

	SkyhookFinalizer = "nodewright.nvidia.com/nodewright"

	// Annotation values used as truthy/falsy strings on Skyhook and Node objects.
	annotationTrueValue  = "true"
	annotationFalseValue = "false"

	// Field selector keys used when filtering pod lists by node.
	fieldSelectorNodeName = "spec.nodeName"

	// Volume + mountpath shared by every package container's host-root mount.
	volumeNameRootMount = "root-mount"
	mountPathRoot       = "/root"

	// Directory inside package containers where the SCR's configMap is projected.
	mountPathConfigMaps = "/skyhook-package/configmaps"

	// Environment variable names propagated into package containers.
	envSkyhookResourceID = "SKYHOOK_RESOURCE_ID"
	envSkyhookNodeOrder  = "SKYHOOK_NODE_ORDER"

	// globalReconcileName is both the controller's registered name (driving the
	// reconcile metric and log labels) and the .Name of the sentinel request the
	// heavy reconcile path collapses Skyhook and Node events onto. The sentinel
	// value is arbitrary and ignored by Reconcile (which grabs the whole world);
	// it only has to be constant and must not collide with the "pod---" dispatch
	// prefix.
	globalReconcileName = "nodewright"

	// globalReconcileDelay is how long Skyhook and Node events wait before the
	// global key becomes ready. The bulk of coalescing is already free: a burst
	// arriving while a pass is in flight dedups onto one follow-up via the
	// priority queue's locked-key handling. This short window only lets a pass's
	// own writes propagate to the cache and near-simultaneous events land before
	// the follow-up runs. It is kept small on purpose: the interrupt flow
	// advances one stage per Node event, so a larger delay (we started at 500ms)
	// adds up across stages and blows the interrupt e2e timing budget.
	globalReconcileDelay = 50 * time.Millisecond
)

type SkyhookOperatorOptions struct {
	Namespace            string        `env:"NAMESPACE, default=skyhook"`
	MaxInterval          time.Duration `env:"DEFAULT_INTERVAL, default=10m"`
	ImagePullSecret      string        `env:"IMAGE_PULL_SECRET"`
	CopyDirRoot          string        `env:"COPY_DIR_ROOT, default=/var/lib/skyhook"`
	ReapplyOnReboot      bool          `env:"REAPPLY_ON_REBOOT, default=false"`
	RuntimeRequiredTaint string        `env:"RUNTIME_REQUIRED_TAINT, default=skyhook.nvidia.com=runtime-required:NoSchedule"`
	PauseImage           string        `env:"PAUSE_IMAGE, default=registry.k8s.io/pause:3.10"`
	AgentImage           string        `env:"AGENT_IMAGE, default=ghcr.io/nvidia/nodewright/agent:latest"` // TODO: pin a released agent version instead of :latest
	AgentLogRoot         string        `env:"AGENT_LOG_ROOT, default=/var/log/skyhook"`
	// MIGRATION-SHIM: transition-only for the skyhook.nvidia.com -> nodewright.nvidia.com
	// rename. LegacyCleanupDelay is how long after a Skyhook finishes migrating the
	// operator keeps its legacy skyhook.nvidia.com node state / pods / ConfigMap labels
	// around (a rollback window) before pruning them. 0 or less prunes immediately (no
	// rollback window). Remove with the legacy group at the removal release.
	LegacyCleanupDelay time.Duration `env:"LEGACY_CLEANUP_DELAY, default=24h"`
}

func (o *SkyhookOperatorOptions) Validate() error {

	messages := make([]string, 0)
	if o.Namespace == "" {
		messages = append(messages, "namespace must be set")
	}
	if o.CopyDirRoot == "" {
		messages = append(messages, "copy dir root must be set")
	}
	if o.RuntimeRequiredTaint == "" {
		messages = append(messages, "runtime required taint must be set")
	}
	if o.MaxInterval < time.Minute {
		messages = append(messages, "max interval must be at least 1 minute")
	}

	// CopyDirRoot must start with /
	if !strings.HasPrefix(o.CopyDirRoot, "/") {
		messages = append(messages, "copy dir root must start with /")
	}

	// RuntimeRequiredTaint must be parsable and must not be a deletion
	_, delete, err := taints.ParseTaints([]string{o.RuntimeRequiredTaint})
	if err != nil {
		messages = append(messages, fmt.Sprintf("runtime required taint is invalid: %s", err.Error()))
	}
	if len(delete) > 0 {
		messages = append(messages, "runtime required taint must not be a deletion")
	}

	if o.AgentImage == "" {
		messages = append(messages, "agent image must be set")
	}

	if !strings.Contains(o.AgentImage, ":") {
		messages = append(messages, "agent image must contain a tag")
	}

	if o.PauseImage == "" {
		messages = append(messages, "pause image must be set")
	}

	if !strings.Contains(o.PauseImage, ":") {
		messages = append(messages, "pause image must contain a tag")
	}

	if len(messages) > 0 {
		return errors.New(strings.Join(messages, ", "))
	}

	return nil
}

// AgentVersion returns the image tag portion of AgentImage
func (o *SkyhookOperatorOptions) AgentVersion() string {
	parts := strings.Split(o.AgentImage, ":")
	return parts[len(parts)-1]
}

func (o *SkyhookOperatorOptions) GetRuntimeRequiredTaint() corev1.Taint {
	to_add, _, _ := taints.ParseTaints([]string{o.RuntimeRequiredTaint})
	return to_add[0]
}

func (o *SkyhookOperatorOptions) GetRuntimeRequiredToleration() corev1.Toleration {
	taint := o.GetRuntimeRequiredTaint()
	return corev1.Toleration{
		Key:      taint.Key,
		Operator: corev1.TolerationOpEqual,
		Value:    taint.Value,
		Effect:   taint.Effect,
	}
}

// force type checking against this interface
var _ reconcile.Reconciler = &SkyhookReconciler{}

func NewSkyhookReconciler(schema *runtime.Scheme, c client.Client, recorder events.EventRecorder, opts SkyhookOperatorOptions) (*SkyhookReconciler, error) {

	err := opts.Validate()
	if err != nil {
		return nil, fmt.Errorf("invalid skyhook operator options: %w", err)
	}

	return &SkyhookReconciler{
		Client:   c,
		scheme:   schema,
		recorder: recorder,
		opts:     opts,
		dal:      dal.New(c),
	}, nil
}

// SkyhookReconciler reconciles a Skyhook object
type SkyhookReconciler struct {
	client.Client
	scheme   *runtime.Scheme
	recorder events.EventRecorder
	opts     SkyhookOperatorOptions
	dal      dal.DAL
}

// SetupWithManager sets up the controller with the Manager.
func (r *SkyhookReconciler) SetupWithManager(mgr ctrl.Manager) error {

	// indexes allow for query on fields to use the local cache
	indexer := mgr.GetFieldIndexer()
	err := indexer.
		IndexField(context.TODO(), &corev1.Pod{}, fieldSelectorNodeName, func(o client.Object) []string {
			pod, ok := o.(*corev1.Pod)
			if !ok {
				return nil
			}
			return []string{pod.Spec.NodeName}
		})

	if err != nil {
		return err
	}

	globalHandler := &globalDelayHandler{
		logger: mgr.GetLogger(),
		dal:    dal.New(r.Client),
		delay:  globalReconcileDelay,
	}

	return ctrl.NewControllerManagedBy(mgr).
		Named(globalReconcileName).
		WithOptions(controller.Options{
			// Only one heavy "grab the world" pass may run at a time: it is a
			// centralized scheduler reading across every SCR, Node and Pod, so
			// concurrent passes would race each other's writes. This makes the
			// global key the single in-flight reconcile.
			MaxConcurrentReconciles: 1,
		}).
		// Heavy path: Skyhook and Node events collapse onto the global key.
		Watches(
			&v1alpha1.NodeWright{},
			globalHandler,
		).
		Watches(
			&corev1.Node{},
			globalHandler,
		).
		// Cheap path: Pod events keep their targeted "pod---<name>" routing,
		// dispatched to PodReconcile in Reconcile below.
		Watches(
			&corev1.Pod{},
			handler.EnqueueRequestsFromMapFunc(podHandlerFunc),
		).
		Complete(r)
}

// CRD Permissions
//+kubebuilder:rbac:groups=skyhook.nvidia.com,resources=skyhooks,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=skyhook.nvidia.com,resources=skyhooks/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=skyhook.nvidia.com,resources=skyhooks/finalizers,verbs=update
//+kubebuilder:rbac:groups=skyhook.nvidia.com,resources=deploymentpolicies,verbs=get;list;watch

// core permissions
//+kubebuilder:rbac:groups=core,resources=nodes,verbs=get;list;update;patch;watch
//+kubebuilder:rbac:groups=core,resources=nodes/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=core,resources=pods/eviction,verbs=create
//+kubebuilder:rbac:groups=core,resources=events,verbs=create;patch
// The event recorder writes via the events.k8s.io/v1 API (client-go tools/events,
// wired through mgr.GetEventRecorder), so the core rule above is not sufficient on
// its own — without this rule every recorded event is rejected as forbidden.
//+kubebuilder:rbac:groups=events.k8s.io,resources=events,verbs=create;patch
//+kubebuilder:rbac:groups=core,resources=configmaps,verbs=get;list;watch;create;update;patch;delete

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.16.3/pkg/reconcile
func (r *SkyhookReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	// split off requests for pods
	if strings.HasPrefix(req.Name, "pod---") {
		name := strings.Split(req.Name, "pod---")[1]
		pod, err := r.dal.GetPod(ctx, req.Namespace, name)
		if err == nil && pod != nil { // if pod, then call other wise not a pod
			return r.PodReconcile(ctx, pod)
		}
		return ctrl.Result{}, err
	}

	// Migration safety interlock: while any legacy skyhook.nvidia.com Skyhook is still
	// mid-rollout, hold this NodeWright reconcile and requeue rather than take over a
	// node the pre-rename operator may still be mutating. Clears once the legacy
	// Skyhooks read complete; fresh clusters and post-migration installs never hold.
	if hold := r.legacyMigrationHold(ctx); hold != nil {
		return *hold, nil
	}

	// get all skyhooks (SCR)
	skyhooks, err := r.dal.GetSkyhooks(ctx)
	if err != nil {
		// error, going to requeue and backoff
		logger.Error(err, "error getting skyhooks")
		return ctrl.Result{}, err
	}

	// if there are no skyhooks, so actually nothing to do, so don't requeue
	if skyhooks == nil || len(skyhooks.Items) == 0 {
		return ctrl.Result{}, nil
	}

	// get all nodes
	nodes, err := r.dal.GetNodes(ctx)
	if err != nil {
		// error, going to requeue and backoff
		logger.Error(err, "error getting nodes")
		return ctrl.Result{}, err
	}

	// if no nodes, well not work to do either
	if nodes == nil || len(nodes.Items) == 0 {
		// no nodes, so nothing to do
		return ctrl.Result{}, nil
	}

	// get all deployment policies
	deploymentPolicies, err := r.dal.GetDeploymentPolicies(ctx)
	if err != nil {
		logger.Error(err, "error getting deployment policies")
		return ctrl.Result{}, err
	}

	// TODO: this build state could error in a lot of ways, and I think we might want to move towards partial state
	// mean if we cant get on SCR state, great, process that one and error

	// BUILD cluster state from all skyhooks, and all nodes
	// this filters and pairs up nodes to skyhooks, also provides help methods for introspection and mutation
	clusterState, err := BuildState(skyhooks, nodes, deploymentPolicies)
	if err != nil {
		// error, going to requeue and backoff
		logger.Error(err, "error building cluster state")
		return ctrl.Result{}, err
	}

	// handle auto-tainting new nodes first so it
	if yes, result, err := shouldReturn(r.HandleAutoTaint(ctx, clusterState)); yes {
		return result, err
	}

	if yes, result, err := shouldReturn(r.HandleMigrations(ctx, clusterState)); yes {
		return result, err
	}

	if yes, result, err := shouldReturn(r.TrackReboots(ctx, clusterState)); yes {
		return result, err
	}

	// node picker is for selecting nodes to do work, tries maintain a prior of nodes between SCRs
	nodePicker := NewNodePicker(logger, r.opts.GetRuntimeRequiredToleration())

	errs := make([]error, 0)
	var result *ctrl.Result
	configSyncPending := false

	for _, skyhook := range clusterState.skyhooks {
		if err := r.refreshSkyhookConditions(ctx, clusterState, skyhook); err != nil {
			return ctrl.Result{RequeueAfter: time.Second * 2}, err
		}

		if yes, result, err := shouldReturn(r.HandleFinalizer(ctx, skyhook, clusterState)); yes {
			return result, err
		}

		if yes, result, err := shouldReturn(r.ReportState(ctx, clusterState, skyhook)); yes {
			return result, err
		}

		if skyhook.IsPaused() {
			if yes, result, err := shouldReturn(r.UpdatePauseStatus(ctx, clusterState, skyhook)); yes {
				return result, err
			}
			continue
		}

		if yes, pendingSync, result, err := r.validateAndUpsertSkyhookData(ctx, skyhook, clusterState); yes {
			return result, err
		} else if pendingSync {
			configSyncPending = true
		}

		changed := IntrospectSkyhook(skyhook, clusterState.skyhooks, logger)
		if changed {
			_, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
			if len(errs) > 0 {
				return ctrl.Result{RequeueAfter: time.Second * 2}, utilerrors.NewAggregate(errs)
			}
			return ctrl.Result{RequeueAfter: time.Second * 2}, nil
		}

		_, err := HandleVersionChange(skyhook)
		if err != nil {
			return ctrl.Result{RequeueAfter: time.Second * 2}, fmt.Errorf("error getting packages to uninstall: %w", err)
		}
	}

	// Process all non-complete, non-disabled skyhooks (in priority order)
	// Each skyhook is processed only for nodes that are ready (all higher-priority skyhooks complete on that node)
	// This enables per-node priority ordering: nodes can progress independently
	result, err = r.processSkyhooksPerNode(ctx, clusterState, nodePicker, logger)
	if err != nil {
		errs = append(errs, err)
	}

	err = r.HandleRuntimeRequired(ctx, clusterState)
	if err != nil {
		errs = append(errs, err)
	}

	if len(errs) > 0 {
		err := utilerrors.NewAggregate(errs)
		return ctrl.Result{}, err
	}

	return reconcileResult(result, configSyncPending, r.opts.MaxInterval), nil
}

// reconcileResult picks the requeue for a completed reconcile pass. Active work
// supplies its own (shorter) result, which is returned untouched. When the pass is
// otherwise idle but an owned ConfigMap write was deferred because the completedNodes
// gate was closed (configSyncPending), retry after configSyncRetryInterval instead of
// the much longer maxInterval so the CM converges promptly rather than appearing stuck
// while status reads complete (issue #245). Otherwise fall back to maxInterval.
func reconcileResult(result *ctrl.Result, configSyncPending bool, maxInterval time.Duration) ctrl.Result {
	if result != nil {
		return *result
	}
	if configSyncPending {
		return ctrl.Result{RequeueAfter: configSyncRetryInterval}
	}
	return ctrl.Result{RequeueAfter: maxInterval}
}

// refreshSkyhookConditions updates and persists the per-Skyhook conditions
// that have to stay accurate regardless of pause / disable / delete state:
//
//   - NodeStateMalformed surfaces unreadable nodeState annotations BEFORE
//     any handler that reads node.State() runs and aborts on parse errors.
//   - Blocked + UninstallInProgress + UninstallFailed mirror node state so
//     paused or disabled Skyhooks (which short-circuit
//     processSkyhooksPerNode) still get current conditions, and so
//     HandleFinalizer's deletion-gating logic stays focused on its own
//     concern instead of duplicating condition-mirroring work.
//
// UpdateBlockedCondition / UpdateUninstallConditions are tolerant to
// per-node state read errors — they skip unreadable nodes and let
// UpdateNodeStateMalformedCondition (set above) be the user-visible signal.
// This keeps HandleFinalizer's malformed-state branch reachable so its
// DeletionBlocked condition + Warning event fire on CR deletion.
func (r *SkyhookReconciler) refreshSkyhookConditions(ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes) error {
	skyhook.UpdateNodeStateMalformedCondition()
	if _, saveErrs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook); len(saveErrs) > 0 {
		return utilerrors.NewAggregate(saveErrs)
	}
	if err := skyhook.UpdateBlockedCondition(); err != nil {
		return fmt.Errorf("error updating blocked condition: %w", err)
	}
	if err := skyhook.UpdateUninstallConditions(); err != nil {
		return fmt.Errorf("error updating uninstall conditions: %w", err)
	}
	return nil
}

// processSkyhooksPerNode processes all skyhooks for nodes that are ready (per-node priority ordering).
// A node is ready for a skyhook if all higher-priority skyhooks are complete on that specific node.
func (r *SkyhookReconciler) processSkyhooksPerNode(ctx context.Context, clusterState *clusterState, nodePicker *NodePicker, logger logr.Logger) (*ctrl.Result, error) {
	var result *ctrl.Result
	var errs []error

	for _, skyhook := range clusterState.skyhooks {
		if skyhook.IsDisabled() || skyhook.IsPaused() {
			continue
		}
		hasWork, err := skyhook.HasUninstallWork()
		if err != nil {
			errs = append(errs, fmt.Errorf("error checking uninstall work for skyhook %s: %w", skyhook.GetSkyhook().Name, err))
			continue
		}
		if skyhook.IsComplete() && !hasWork {
			continue
		}

		// Check if any nodes are ready for this skyhook
		ready, err := hasReadyNodesForSkyhook(skyhook, clusterState.skyhooks)
		if err != nil {
			errs = append(errs, fmt.Errorf("error checking ready nodes for skyhook %s: %w", skyhook.GetSkyhook().Name, err))
			continue
		}
		if !ready {
			continue
		}

		res, err := r.RunSkyhookPackages(ctx, clusterState, nodePicker, skyhook)
		if err != nil {
			logger.Error(err, "error processing skyhook", "skyhook", skyhook.GetSkyhook().Name)
			errs = append(errs, err)
		}
		if res != nil {
			result = res
		}
	}

	if len(errs) > 0 {
		return result, utilerrors.NewAggregate(errs)
	}
	return result, nil
}

// hasReadyNodesForSkyhook checks if any nodes are ready to process this skyhook.
// A node is ready if it's not complete and all higher-priority skyhooks are complete on that node.
func hasReadyNodesForSkyhook(skyhook SkyhookNodes, allSkyhooks []SkyhookNodes) (bool, error) {
	pendingUninstall, err := skyhook.HasUninstallWork()
	if err != nil {
		return false, err
	}
	for _, node := range skyhook.GetNodes() {
		if node.IsComplete() && !pendingUninstall {
			continue
		}
		if IsNodeReadyForSkyhook(node.GetNode().Name, skyhook, allSkyhooks) {
			return true, nil
		}
	}
	return false, nil
}

func shouldReturn(updates bool, err error) (bool, ctrl.Result, error) {
	if err != nil {
		return true, ctrl.Result{}, err
	}
	if updates {
		return true, ctrl.Result{RequeueAfter: time.Second * 2}, nil
	}
	return false, ctrl.Result{}, nil
}

func (r *SkyhookReconciler) HandleMigrations(ctx context.Context, clusterState *clusterState) (bool, error) {

	updates := false

	if version.VERSION == "" {
		// this means the binary was complied without version information
		return false, nil
	}

	logger := log.FromContext(ctx)
	errors := make([]error, 0)
	for _, skyhook := range clusterState.skyhooks {

		err := skyhook.Migrate(logger)
		if err != nil {
			return false, fmt.Errorf("error migrating skyhook [%s]: %w", skyhook.GetSkyhook().Name, err)
		}

		if err := skyhook.GetSkyhook().NodeWright.Validate(); err != nil {
			return false, fmt.Errorf("error validating skyhook [%s]: %w", skyhook.GetSkyhook().Name, err)
		}

		// MIGRATION-SHIM: rollback-safe legacy cleanup. skyhook.Migrate above adopts
		// legacy node state ADDITIVELY (keeps the skyhook.nvidia.com keys). We only
		// prune those legacy keys/pods/labels once the rollback window has elapsed,
		// tracked by the legacy-migrated-at stamp on the NodeWright. See
		// docs/plans/2026-07-20-legacy-cleanup-ttl-design.md.
		nw := skyhook.GetSkyhook().NodeWright
		prune := legacyCleanupShouldPrune(nw.GetAnnotations()[legacyMigratedAtAnnotation], r.opts.LegacyCleanupDelay, time.Now())

		if prune {
			for _, node := range skyhook.GetNodes() {
				node.PruneLegacyMetadata()
			}
		}

		for _, node := range skyhook.GetNodes() {
			if node.Changed() {
				err := r.Status().Patch(ctx, node.GetNode(), client.MergeFrom(clusterState.tracker.GetOriginal(node.GetNode())))
				if err != nil {
					errors = append(errors, fmt.Errorf("error patching node [%s]: %w", node.GetNode().Name, err))
				}

				err = r.Patch(ctx, node.GetNode(), client.MergeFrom(clusterState.tracker.GetOriginal(node.GetNode())))
				if err != nil {
					errors = append(errors, fmt.Errorf("error patching node [%s]: %w", node.GetNode().Name, err))
				}
				updates = true
			}
		}

		// Converge (add nodewright labels, keep legacy pods) or prune (delete legacy
		// pods, drop legacy labels) the workloads the pre-rename operator created under
		// the legacy skyhook.nvidia.com labels.
		hadLegacyWorkloads, workloadsChanged, err := r.reconcileLegacyLabeledWorkloads(ctx, skyhook.GetSkyhook().Name, prune)
		if err != nil {
			return false, fmt.Errorf("error reconciling legacy-labeled workloads for skyhook [%s]: %w", skyhook.GetSkyhook().Name, err)
		}
		if workloadsChanged {
			updates = true
		}

		if skyhook.GetSkyhook().Updated {
			// need to do this because SaveNodesAndSkyhook only saves skyhook status, not the main skyhook object where the annotations are
			// additionally it needs to be an update, a patch nils out the annotations for some reason, which the save function does a patch

			if err = r.Status().Update(ctx, skyhook.GetSkyhook().NodeWright); err != nil {
				return false, fmt.Errorf("error updating during migration skyhook status [%s]: %w", skyhook.GetSkyhook().Name, err)
			}

			// because of conflict issues (409) we need to do things a bit differently here.
			// We might be able to use server side apply in the future, but for now we need to do this
			// https://kubernetes.io/docs/reference/using-api/server-side-apply/
			// https://github.com/kubernetes-sigs/controller-runtime/issues/347

			// work around for now is to grab a new copy of the object, and then patch it

			newskyhook, err := r.dal.GetSkyhook(ctx, skyhook.GetSkyhook().Name)
			if err != nil {
				return false, fmt.Errorf("error getting skyhook to migrate [%s]: %w", skyhook.GetSkyhook().Name, err)
			}
			newPatch := client.MergeFrom(newskyhook.DeepCopy())

			// set version
			wrapper.NewSkyhookWrapper(newskyhook).SetVersion()

			if err = r.Patch(ctx, newskyhook, newPatch); err != nil {
				return false, fmt.Errorf("error updating during migration skyhook [%s]: %w", skyhook.GetSkyhook().Name, err)
			}

			updates = true
		}

		// Manage the rollback-window stamp LAST: it re-gets and patches the NodeWright,
		// bumping its resourceVersion. Running it after the status/version writes above
		// (which submit the in-memory copy) avoids a stale-RV 409 on the first migration
		// reconcile. Set the stamp on the first converge that finds legacy artifacts
		// (unless pruning immediately); clear it once a prune has removed everything.
		if stampChanged, err := r.reconcileLegacyMigratedStamp(ctx, skyhook, prune, hadLegacyWorkloads); err != nil {
			return false, err
		} else if stampChanged {
			updates = true
		}
	}

	if len(errors) > 0 {
		return false, utilerrors.NewAggregate(errors)
	}

	return updates, nil
}

// MIGRATION-SHIM: transition-only for the skyhook.nvidia.com -> nodewright.nvidia.com
// rename. Delete everything tagged MIGRATION-SHIM together with the legacy
// skyhook.nvidia.com group in the removal release (see docs/plans Phase 10).
//
// legacyMetadataPrefix is the metadata prefix the pre-rename operator stamped on
// package pods and per-node metadata ConfigMaps. It is intentionally hardcoded (a
// one-shot migration constant whose value can never change) so the controller keeps
// depending only on the new nodewright api group.
const legacyMetadataPrefix = "skyhook.nvidia.com"

// legacyMigratedAtAnnotation is stamped on a NodeWright (RFC3339) the first time its
// legacy skyhook.nvidia.com artifacts are adopted. The prune of those artifacts is
// deferred until LegacyCleanupDelay has elapsed since this time, giving a rollback
// window. MIGRATION-SHIM: remove with the legacy group.
const legacyMigratedAtAnnotation = v1alpha1.METADATA_PREFIX + "/legacy-migrated-at"

// legacyCleanupShouldPrune reports whether the legacy skyhook.nvidia.com artifacts for
// a NodeWright may be pruned yet. delay <= 0 prunes immediately (no rollback window).
// Otherwise a NodeWright is prunable only once delay has elapsed since its stamp; an
// absent stamp means "not yet adopted, converge first", and an unparseable stamp fails
// toward prune so a corrupt value cannot pin legacy state forever.
func legacyCleanupShouldPrune(stamp string, delay time.Duration, now time.Time) bool {
	if delay <= 0 {
		return true
	}
	if stamp == "" {
		return false
	}
	t, err := time.Parse(time.RFC3339, stamp)
	if err != nil {
		return true
	}
	return !now.Before(t.Add(delay))
}

// MIGRATION-SHIM (see legacyMetadataPrefix): remove with the legacy group.
// reconcileLegacyLabeledWorkloads converges or prunes the workloads the pre-rename
// operator created under the legacy skyhook.nvidia.com labels for the named skyhook.
// Converge (prune=false) adds the nodewright label to the per-node metadata ConfigMaps
// alongside the legacy one and leaves the legacy package pods in place, so a rolled-back
// pre-rename operator still owns its workloads. Prune (prune=true) graceful-deletes the
// legacy package pods and drops the legacy ConfigMap label. Returns whether any legacy
// artifact was found and whether anything was written. Level-triggered and idempotent.
func (r *SkyhookReconciler) reconcileLegacyLabeledWorkloads(ctx context.Context, skyhookName string, prune bool) (bool, bool, error) {
	hadLegacy := false
	changed := false

	pods := &corev1.PodList{}
	if err := r.List(ctx, pods, client.InNamespace(r.opts.Namespace),
		client.MatchingLabels{fmt.Sprintf("%s/name", legacyMetadataPrefix): skyhookName}); err != nil {
		return false, false, fmt.Errorf("listing legacy-labeled pods for skyhook [%s]: %w", skyhookName, err)
	}
	if len(pods.Items) > 0 {
		hadLegacy = true
	}
	if prune {
		for i := range pods.Items {
			// Graceful delete: honor the pod's terminationGracePeriodSeconds. Single-writer
			// safety comes from the migration hold (the legacy Skyhook is complete, so its
			// pods are no longer mutating the host), not from the grace period.
			if err := r.Delete(ctx, &pods.Items[i]); err != nil && !apierrors.IsNotFound(err) {
				return false, false, fmt.Errorf("deleting legacy-labeled pod [%s]: %w", pods.Items[i].Name, err)
			}
			changed = true
		}
	}

	cms := &corev1.ConfigMapList{}
	if err := r.List(ctx, cms, client.InNamespace(r.opts.Namespace),
		client.MatchingLabels{fmt.Sprintf("%s/skyhook-node-meta", legacyMetadataPrefix): skyhookName}); err != nil {
		return false, false, fmt.Errorf("listing legacy-labeled configmaps for skyhook [%s]: %w", skyhookName, err)
	}
	if len(cms.Items) > 0 {
		hadLegacy = true
	}
	for i := range cms.Items {
		cm := &cms.Items[i]
		var mutated bool
		if prune {
			mutated = relabelLegacyMetadataPrefix(cm.Labels)
		} else {
			mutated = addNodeWrightMetaLabel(cm.Labels)
		}
		if mutated {
			if err := r.Update(ctx, cm); err != nil {
				return false, false, fmt.Errorf("updating legacy configmap labels [%s]: %w", cm.Name, err)
			}
			changed = true
		}
	}

	return hadLegacy, changed, nil
}

// MIGRATION-SHIM (see legacyMetadataPrefix): remove with the legacy group.
// reconcileLegacyMigratedStamp stamps the NodeWright's legacy-migrated-at annotation
// the first time a converge finds legacy artifacts (starting the rollback window), and
// clears it once a prune has removed everything legacy. Returns whether it wrote.
func (r *SkyhookReconciler) reconcileLegacyMigratedStamp(ctx context.Context, skyhook SkyhookNodes, prune, hadLegacyWorkloads bool) (bool, error) {
	nw := skyhook.GetSkyhook().NodeWright
	stamp := nw.GetAnnotations()[legacyMigratedAtAnnotation]
	name := skyhook.GetSkyhook().Name

	if prune {
		if stamp != "" && !hadLegacyWorkloads && !anyNodeHasLegacyMetadata(skyhook.GetNodes()) {
			return true, r.patchLegacyMigratedStamp(ctx, name, "", true)
		}
		return false, nil
	}

	if stamp == "" && (hadLegacyWorkloads || anyNodeHasLegacyMetadata(skyhook.GetNodes())) {
		return true, r.patchLegacyMigratedStamp(ctx, name, time.Now().Format(time.RFC3339), false)
	}
	return false, nil
}

// patchLegacyMigratedStamp sets or removes the legacy-migrated-at annotation on the
// NodeWright with a focused merge patch (re-read then patch, to avoid clobbering other
// concurrent writers and the annotation-nilling gotcha of a full update).
func (r *SkyhookReconciler) patchLegacyMigratedStamp(ctx context.Context, name, value string, remove bool) error {
	nw, err := r.dal.GetSkyhook(ctx, name)
	if err != nil {
		return fmt.Errorf("getting nodewright to update legacy-migrated-at stamp [%s]: %w", name, err)
	}
	before := nw.DeepCopy()
	annotations := nw.GetAnnotations()
	if remove {
		if annotations == nil {
			return nil
		}
		delete(annotations, legacyMigratedAtAnnotation)
	} else {
		if annotations == nil {
			annotations = map[string]string{}
		}
		annotations[legacyMigratedAtAnnotation] = value
	}
	nw.SetAnnotations(annotations)
	if err := r.Patch(ctx, nw, client.MergeFrom(before)); err != nil {
		return fmt.Errorf("patching legacy-migrated-at stamp on [%s]: %w", name, err)
	}
	return nil
}

// anyNodeHasLegacyMetadata reports whether any node still carries a legacy
// skyhook.nvidia.com-prefixed annotation, label, or condition.
func anyNodeHasLegacyMetadata(nodes []wrapper.SkyhookNode) bool {
	for _, node := range nodes {
		n := node.GetNode()
		for k := range n.Annotations {
			if strings.HasPrefix(k, legacyMetadataPrefix+"/") {
				return true
			}
		}
		for k := range n.Labels {
			if strings.HasPrefix(k, legacyMetadataPrefix+"/") {
				return true
			}
		}
		for _, c := range n.Status.Conditions {
			if strings.HasPrefix(string(c.Type), legacyMetadataPrefix+"/") {
				return true
			}
		}
	}
	return false
}

// addNodeWrightMetaLabel adds a nodewright.nvidia.com-prefixed copy of each legacy
// skyhook.nvidia.com label, keeping the legacy label. Returns true if it added any.
func addNodeWrightMetaLabel(labels map[string]string) bool {
	changed := false
	var legacy []string
	for k := range labels {
		if strings.HasPrefix(k, legacyMetadataPrefix+"/") {
			legacy = append(legacy, k)
		}
	}
	for _, k := range legacy {
		suffix := strings.TrimPrefix(k, legacyMetadataPrefix+"/")
		newKey := fmt.Sprintf("%s/%s", v1alpha1.METADATA_PREFIX, suffix)
		if _, exists := labels[newKey]; !exists {
			labels[newKey] = labels[k]
			changed = true
		}
	}
	return changed
}

// MIGRATION-SHIM (see legacyMetadataPrefix): remove with the legacy group.
// relabelLegacyMetadataPrefix rewrites, in place, any label key under the legacy
// skyhook.nvidia.com prefix to the current nodewright.nvidia.com prefix, preserving
// the value. It returns true if it changed anything. Safe to mutate the map during
// the range: rewritten keys carry the new prefix, so CutPrefix skips them if the
// iteration happens to visit them.
func relabelLegacyMetadataPrefix(labels map[string]string) bool {
	changed := false
	for k, v := range labels {
		suffix, ok := strings.CutPrefix(k, legacyMetadataPrefix+"/")
		if !ok {
			continue
		}
		delete(labels, k)
		labels[fmt.Sprintf("%s/%s", v1alpha1.METADATA_PREFIX, suffix)] = v
		changed = true
	}
	return changed
}

// ReportState computes and puts important information into the skyhook status so that monitoring tools such as k9s
// can see the information at a glance. For example, the number of completed nodes and the list of packages in the skyhook.
func (r *SkyhookReconciler) ReportState(ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes) (bool, error) {

	// save updated state to skyhook status
	skyhook.ReportState()

	if skyhook.GetSkyhook().Updated {
		_, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
		if len(errs) > 0 {
			return false, utilerrors.NewAggregate(errs)
		}
		return true, nil
	}

	return false, nil
}

func (r *SkyhookReconciler) UpdatePauseStatus(ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes) (bool, error) {
	changed := UpdateSkyhookPauseStatus(skyhook, log.FromContext(ctx))

	if changed {
		_, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
		if len(errs) > 0 {
			return false, utilerrors.NewAggregate(errs)
		}
		return true, nil
	}

	return false, nil
}

func (r *SkyhookReconciler) TrackReboots(ctx context.Context, clusterState *clusterState) (bool, error) {

	updates := false
	errs := make([]error, 0)

	for _, skyhook := range clusterState.skyhooks {
		if skyhook.GetSkyhook().Status.NodeBootIds == nil {
			skyhook.GetSkyhook().Status.NodeBootIds = make(map[string]string)
		}

		for _, node := range skyhook.GetNodes() {
			id, ok := skyhook.GetSkyhook().Status.NodeBootIds[node.GetNode().Name]

			if !ok { // new node
				skyhook.GetSkyhook().Status.NodeBootIds[node.GetNode().Name] = node.GetNode().Status.NodeInfo.BootID
				skyhook.GetSkyhook().Updated = true
			}

			if id != "" && id != node.GetNode().Status.NodeInfo.BootID { // node rebooted
				if r.opts.ReapplyOnReboot {
					r.recorder.Eventf(skyhook.GetSkyhook().NodeWright, nil, EventTypeNormal, EventsReasonNodeReboot, "ResetNodeState", "detected reboot, resetting node [%s] to be reapplied", node.GetNode().Name)
					r.recorder.Eventf(node.GetNode(), nil, EventTypeNormal, EventsReasonNodeReboot, "ResetNodeState", "detected reboot, resetting node for [%s] to be reapplied", node.GetSkyhook().Name)
					node.Reset()

					// Re-apply the runtime-required taint so workloads cannot schedule on the
					// rebooted node until Skyhook finishes re-applying. The original auto-taint
					// annotation survives Reset() and remains the record that this taint is
					// operator-managed; no annotation update is needed.
					if skyhook.GetSkyhook().Spec.RuntimeRequired && skyhook.GetSkyhook().Spec.AutoTaintNewNodes {
						taintToAdd := r.opts.GetRuntimeRequiredTaint()
						newNode, updated, _ := taints.AddOrUpdateTaint(node.GetNode(), &taintToAdd)
						if updated {
							node.GetNode().Spec.Taints = newNode.Spec.Taints
							log.FromContext(ctx).Info("re-applying runtime-required taint after reboot", "node", node.GetNode().Name, "taint", taintToAdd.Key)
						}
					}

					// Persist the reset before recording the new boot id. We Patch rather than
					// Update because a busy node's resourceVersion churns constantly under other
					// controllers, and a full Update would lose that optimistic-concurrency race; a
					// strategic merge of only our annotation/label changes does not. And we advance
					// NodeBootIds only after the write is durable: if the reset never lands, leaving
					// the boot id unchanged keeps the reboot pending so it is re-detected and retried
					// next reconcile, instead of being silently consumed while the node's stale
					// "complete" state remains and the package is never reapplied.
					if node.Changed() {
						updates = true
						patch := client.StrategicMergeFrom(clusterState.tracker.GetOriginal(node.GetNode()))
						if err := r.Patch(ctx, node.GetNode(), patch); err != nil {
							errs = append(errs, fmt.Errorf("error patching node after reboot [%s]: %w", node.GetNode().Name, err))
							continue
						}
					}
				}
				skyhook.GetSkyhook().Status.NodeBootIds[node.GetNode().Name] = node.GetNode().Status.NodeInfo.BootID
				skyhook.GetSkyhook().Updated = true
			}
		}
		if skyhook.GetSkyhook().Updated { // update
			updates = true
			err := r.Status().Update(ctx, skyhook.GetSkyhook().NodeWright)
			if err != nil {
				errs = append(errs, fmt.Errorf("error updating skyhook status after reboot [%s]: %w", skyhook.GetSkyhook().Name, err))
			}
		}
	}

	return updates, utilerrors.NewAggregate(errs)
}

// RunSkyhookPackages runs all skyhook packages then saves and requeues if changes were made
func (r *SkyhookReconciler) RunSkyhookPackages(ctx context.Context, clusterState *clusterState, nodePicker *NodePicker, skyhook SkyhookNodes) (*ctrl.Result, error) {

	logger := log.FromContext(ctx)
	requeue := false
	beingDeleted := !skyhook.GetSkyhook().DeletionTimestamp.IsZero()

	toExplicitUninstall, err := HandleUninstallRequests(skyhook)
	if err != nil {
		return nil, fmt.Errorf("error handling uninstall requests: %w", err)
	}

	err = HandleCancelledUninstalls(skyhook)
	if err != nil {
		return nil, fmt.Errorf("error handling cancelled uninstalls: %w", err)
	}

	toUninstall, err := HandleVersionChange(skyhook)
	if err != nil {
		return nil, fmt.Errorf("error getting packages to uninstall: %w", err)
	}

	toUninstall = append(toExplicitUninstall, toUninstall...)

	// UpdateBlockedCondition / UpdateUninstallConditions run at the top of
	// Reconcile (before the pause/disable short-circuit) so paused and
	// disabled Skyhooks get the same conditions as running ones.

	changed := IntrospectSkyhook(skyhook, clusterState.skyhooks, logger)
	if !changed && skyhook.IsComplete() {
		return nil, nil
	}

	selectedNode := nodePicker.SelectNodes(skyhook)

	for _, node := range selectedNode {
		// Skip nodes that are waiting on higher-priority skyhooks
		// This enables per-node priority ordering
		if !IsNodeReadyForSkyhook(node.GetNode().Name, skyhook, clusterState.skyhooks) {
			continue
		}

		if node.IsComplete() && !node.Changed() {
			continue
		}

		toRun, err := node.RunNext()
		if err != nil {
			return nil, fmt.Errorf("error getting next packages to run: %w", err)
		}

		// Filter out packages where uninstall is in progress or already
		// completed on this node. A package absent from nodeState with
		// uninstall requested (IsUninstalling, or finalizer-driven via
		// beingDeleted && UninstallEnabled) means uninstall finished — skip
		// apply. Absent + never-requested means never installed yet — allow
		// apply.
		//
		// A State() error here would silently produce a nil nodeState, and the
		// IsUninstallCycleInProgress check below is nil-safe (returns false) — so
		// we'd queue an apply pod while the node might actually be mid-uninstall.
		// Propagate the error instead; the user-visible NodeStateMalformed
		// condition is already set at the top of Reconcile.
		nodeState, err := node.State()
		if err != nil {
			return nil, fmt.Errorf("node %s: reading state while filtering runnable packages: %w",
				node.GetNode().Name, err)
		}
		filtered := make([]*v1alpha1.Package, 0, len(toRun))
		for _, pkg := range toRun {
			if shouldSkipApplyForUninstall(pkg, nodeState, beingDeleted) {
				continue
			}
			filtered = append(filtered, pkg)
		}
		toRun = filtered

		// prepend the uninstall packages so they are ran first.
		// filterUninstallForNode drops entries that aren't in this node's
		// state — toUninstall is global across all nodes, so a package can
		// be pending uninstall on node B while already absent on node A.
		toRun = append(filterUninstallForNode(toUninstall, nodeState), toRun...)

		interrupt, pack := fudgeInterruptWithPriority(toRun, skyhook.GetSkyhook().GetConfigUpdates(), skyhook.GetSkyhook().GetConfigInterrupts())

		for _, f := range toRun {

			ok, err := r.ProcessInterrupt(ctx, node, f, interrupt, interrupt != nil && f.Name == pack)
			if err != nil {
				// TODO: error handle
				return nil, fmt.Errorf("error processing if we should interrupt [%s:%s]: %w", f.Name, f.Version, err)
			}
			if !ok {
				requeue = true
				continue
			}

			err = r.ApplyPackage(ctx, logger, clusterState, node, f, interrupt != nil && f.Name == pack)
			if err != nil {
				return nil, fmt.Errorf("error applying package [%s:%s]: %w", f.Name, f.Version, err)
			}

			// process one package at a time
			if skyhook.GetSkyhook().Spec.Serial {
				return &ctrl.Result{Requeue: true}, nil
			}
		}
	}

	saved, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
	if len(errs) > 0 {
		return &ctrl.Result{}, utilerrors.NewAggregate(errs)
	}
	if saved {
		requeue = true
	}

	if !skyhook.IsComplete() || requeue {
		return &ctrl.Result{RequeueAfter: time.Second * 2}, nil // not sure this is better then just requeue bool
	}

	return nil, utilerrors.NewAggregate(errs)
}

// SaveNodesAndSkyhook saves nodes and skyhook and will update the events if the skyhook status changes
func (r *SkyhookReconciler) SaveNodesAndSkyhook(ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes) (bool, []error) {
	saved := false
	errs := make([]error, 0)
	logger := log.FromContext(ctx)

	for _, node := range skyhook.GetNodes() {
		patch := client.StrategicMergeFrom(clusterState.tracker.GetOriginal(node.GetNode()))
		if node.Changed() {
			err := r.Patch(ctx, node.GetNode(), patch)
			if err != nil {
				errs = append(errs, fmt.Errorf("error patching node [%s]: %w", node.GetNode().Name, err))
			}
			saved = true

			err = r.UpsertNodeLabelsAnnotationsPackages(ctx, skyhook.GetSkyhook(), node.GetNode())
			if err != nil {
				errs = append(errs, fmt.Errorf("error upserting labels, annotations, and packages config map for node [%s]: %w", node.GetNode().Name, err))
			}

			if node.IsComplete() {
				r.recorder.Eventf(node.GetNode(), nil, EventTypeNormal, EventsReasonSkyhookStateChange, "MarkComplete", "Skyhook [%s] complete.", skyhook.GetSkyhook().Name)

				// since node is complete remove from priority
				skyhook.GetSkyhook().RemoveNodePriority(node.GetNode().Name)
			}
		}

		// updates node's condition
		node.UpdateCondition()
		if node.Changed() {
			// conditions are in status
			err := r.Status().Patch(ctx, node.GetNode(), patch)
			if err != nil {
				errs = append(errs, fmt.Errorf("error patching node status [%s]: %w", node.GetNode().Name, err))
			}
			saved = true
		}

		if node.GetSkyhook() != nil && node.GetSkyhook().Updated {
			skyhook.GetSkyhook().Updated = true
		}
	}

	if len(errs) == 0 {
		skyhook.UpdateCondition(logger)
	}

	if len(errs) == 0 && skyhook.GetSkyhook().Updated {
		patch := client.MergeFrom(clusterState.tracker.GetOriginal(skyhook.GetSkyhook().NodeWright))
		err := r.Status().Patch(ctx, skyhook.GetSkyhook().NodeWright, patch)
		if err != nil {
			errs = append(errs, err)
		}
		saved = true

		if skyhook.GetPriorStatus() != "" && skyhook.GetPriorStatus() != skyhook.Status() {
			// we transitioned, fire event
			r.recorder.Eventf(skyhook.GetSkyhook(), nil, EventTypeNormal, EventsReasonSkyhookStateChange, "Transition", "Skyhook transitioned [%s] -> [%s]", skyhook.GetPriorStatus(), skyhook.Status())
		}
	}

	if len(errs) > 0 {
		saved = false
	}
	return saved, errs
}

// HandleUninstallRequests checks for packages that need uninstall and triggers
// StageUninstall on nodes where the package is at a complete install stage. Returns the
// list of packages that need uninstall pods created.
// A package needs uninstall when:
//   - IsUninstalling()==true (explicit: user set apply=true), OR
//   - UninstallEnabled()==true and the CR is being deleted (finalizer-driven)
func HandleUninstallRequests(skyhook SkyhookNodes) ([]*v1alpha1.Package, error) {
	toUninstall := make([]*v1alpha1.Package, 0)
	beingDeleted := !skyhook.GetSkyhook().DeletionTimestamp.IsZero()
	for _, node := range skyhook.GetNodes() {
		nodeState, err := node.State()
		if err != nil {
			return nil, fmt.Errorf("node %s: reading state in HandleUninstallRequests: %w",
				node.GetNode().Name, err)
		}
		for name, pkg := range skyhook.GetSkyhook().Spec.Packages {
			status, exists := nodeState[pkg.GetUniqueName()]
			if !exists {
				continue // already uninstalled on this node (absent = done)
			}

			needsUninstall := pkg.IsUninstalling() || (beingDeleted && pkg.UninstallEnabled())
			if !needsUninstall && !nodeState.IsUninstallCycleInProgress(pkg.GetUniqueName()) {
				continue
			}

			// Handle packages progressing through the uninstall cycle.
			switch status.Stage {
			case v1alpha1.StageUninstall:
				// All states re-add so ApplyPackage / ProcessInterrupt sees the
				// package. StageUninstall/Complete is not used under the new rules
				// (HandleCompletePod transitions directly to either
				// StageUninstallInterrupt/InProgress or RemoveState), but we re-add
				// it defensively.
				p := skyhook.GetSkyhook().Spec.Packages[name]
				toUninstall = appendIfNotPresent(toUninstall, &p)
				continue

			case v1alpha1.StageUninstallInterrupt:
				if status.State == v1alpha1.StateComplete {
					if err := node.RemoveState(pkg.PackageRef); err != nil {
						return nil, fmt.Errorf("error removing state after uninstall interrupt for %s: %w", name, err)
					}
					zeroOutSkyhookPackageMetrics(skyhook.GetSkyhook().Name, pkg.Name, pkg.Version)
					node.SetStatus(v1alpha1.StatusInProgress)
				} else {
					p := skyhook.GetSkyhook().Spec.Packages[name]
					toUninstall = appendIfNotPresent(toUninstall, &p)
				}
				continue
			}

			// Install-cycle stages (Apply, Config, Interrupt, PostInterrupt, Upgrade):
			// only kick off uninstall from a terminal Complete state. Don't interrupt
			// an install mid-flight — wait for it.
			if status.State != v1alpha1.StateComplete {
				continue
			}
			if err := node.Upsert(pkg.PackageRef, pkg.Image,
				v1alpha1.StateInProgress, v1alpha1.StageUninstall, 0, pkg.ContainerSHA); err != nil {
				return nil, fmt.Errorf("error triggering uninstall for %s: %w", name, err)
			}
			node.SetStatus(v1alpha1.StatusInProgress)
			p := skyhook.GetSkyhook().Spec.Packages[name]
			toUninstall = appendIfNotPresent(toUninstall, &p)
		}
	}
	return toUninstall, nil
}

// HandleCancelledUninstalls resets packages at StageUninstall back to the install pipeline
// when uninstall.apply has been set to false (cancel). For packages where uninstall already
// completed (absent from node state), RunNext will naturally re-apply them.
// Skips cancellation during CR deletion — the finalizer drives uninstall via
// HandleUninstallRequests and must not be interfered with.
func HandleCancelledUninstalls(skyhook SkyhookNodes) error {
	// During CR deletion, the finalizer drives uninstall for enabled packages.
	// Do not cancel those — they must complete for the finalizer to proceed.
	beingDeleted := !skyhook.GetSkyhook().DeletionTimestamp.IsZero()

	for _, node := range skyhook.GetNodes() {
		nodeState, err := node.State()
		if err != nil {
			return fmt.Errorf("node %s: reading state in HandleCancelledUninstalls: %w",
				node.GetNode().Name, err)
		}
		for _, status := range nodeState {
			if status.Stage != v1alpha1.StageUninstall {
				continue
			}
			pkg, exists := skyhook.GetSkyhook().Spec.Packages[status.Name]
			if !exists {
				continue // removed from spec — HandleVersionChange handles
			}
			if pkg.IsUninstalling() {
				continue // still actively uninstalling — not cancelled
			}
			if beingDeleted && pkg.UninstallEnabled() {
				continue // finalizer-driven uninstall — do not cancel
			}
			// Package is at StageUninstall but apply is false → cancelled
			if status.IsActive() {
				// Reset to re-enter install pipeline
				err := node.Upsert(pkg.PackageRef, pkg.Image,
					v1alpha1.StateInProgress, v1alpha1.StageApply, 0, pkg.ContainerSHA)
				if err != nil {
					return fmt.Errorf("error resetting cancelled uninstall for %s: %w", status.Name, err)
				}
				node.SetStatus(v1alpha1.StatusInProgress)
			}
		}
	}
	return nil
}

// appendIfNotPresent appends a package to the list if not already present (by name+version).
func appendIfNotPresent(list []*v1alpha1.Package, pkg *v1alpha1.Package) []*v1alpha1.Package {
	for _, existing := range list {
		if existing.Name == pkg.Name && existing.Version == pkg.Version {
			return list
		}
	}
	return append(list, pkg)
}

// filterUninstallForNode keeps only the packages from toUninstall that are
// still tracked in nodeState. HandleUninstallRequests and HandleVersionChange
// build toUninstall globally across all of a Skyhook's nodes, so a package
// may be pending uninstall on one node while already absent on another;
// feeding those absent entries into ApplyPackage would fall through to
// StageApply and re-install a package the user explicitly uninstalled.
func filterUninstallForNode(toUninstall []*v1alpha1.Package, nodeState v1alpha1.NodeState) []*v1alpha1.Package {
	filtered := make([]*v1alpha1.Package, 0, len(toUninstall))
	for _, pkg := range toUninstall {
		if _, inState := nodeState[pkg.GetUniqueName()]; !inState {
			continue
		}
		filtered = append(filtered, pkg)
	}
	return filtered
}

// shouldSkipApplyForUninstall reports whether a package should be filtered
// out of a node's runnable set because uninstall is either in progress on
// this node or has already completed. Mirrors HandleUninstallRequests'
// needsUninstall predicate: a finalizer-driven delete (beingDeleted &&
// UninstallEnabled) counts as "uninstall requested" even when the spec's
// Uninstall.Apply is false, which IsUninstalling alone would miss.
func shouldSkipApplyForUninstall(pkg *v1alpha1.Package, nodeState v1alpha1.NodeState, beingDeleted bool) bool {
	if nodeState.IsUninstallCycleInProgress(pkg.GetUniqueName()) {
		return true
	}
	uninstallRequested := pkg.IsUninstalling() || (beingDeleted && pkg.UninstallEnabled())
	return uninstallRequested && nodeState.IsUninstalled(pkg.GetUniqueName())
}

// HandleVersionChange updates the state for the node or skyhook if a version is changed on a package
func HandleVersionChange(skyhook SkyhookNodes) ([]*v1alpha1.Package, error) {
	toUninstall := make([]*v1alpha1.Package, 0)
	versionChangeDetected := false

	for _, node := range skyhook.GetNodes() {
		nodeState, err := node.State()
		if err != nil {
			return nil, fmt.Errorf("node %s: reading state in HandleVersionChange: %w",
				node.GetNode().Name, err)
		}

		for _, packageStatus := range nodeState {
			_package, exists := skyhook.GetSkyhook().Spec.Packages[packageStatus.Name]

			// Skip packages where uninstall has started on this node — handled by HandleUninstallRequests.
			// Uses node annotation (StageUninstall) as source of truth, not the spec's apply flag.
			if exists && nodeState.IsUninstallCycleInProgress(_package.GetUniqueName()) {
				continue
			}

			if exists && _package.Version == packageStatus.Version {
				continue // no uninstall needed for package
			}

			if !exists {
				// Package removed from spec. The webhook blocks removal of
				// enabled=true packages unless the package is already fully
				// uninstalled on all nodes, so if we reach here the package
				// was enabled=false (or unset). Per D2 semantics, leave the
				// node state entry in place — non-absent signals to the user
				// that the package's files are still on the node (no
				// uninstall.sh ran). The operator no longer tracks it; any
				// future spec changes are what would clean it up.
				skyhook.GetSkyhook().RemoveConfigUpdates(packageStatus.Name)
				continue
			} else if exists && _package.Version != packageStatus.Version {
				versionChangeDetected = true
				comparison := version.Compare(_package.Version, packageStatus.Version)
				if comparison == -2 {
					return nil, errors.New("error comparing package versions: invalid version string provided enabling webhooks validates versions before being applied")
				}

				if comparison == 1 {
					// Upgrade path.
					_packageStatus, found := node.PackageStatus(_package.GetUniqueName())
					if found && _packageStatus.Stage == v1alpha1.StageUpgrade {
						continue
					}

					// start upgrade of package
					err := node.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateInProgress, v1alpha1.StageUpgrade, 0, _package.ContainerSHA)
					if err != nil {
						return nil, fmt.Errorf("error updating node status: %w", err)
					}
				} else {
					// Downgrade: no-op. Webhook rejects downgrades of enabled=true packages
					// unless the package is already fully uninstalled. For enabled=false
					// packages, the old-version state stays in node state per D2 semantics
					// (non-absent = "not cleanly uninstalled, just superseded").
					continue
				}
			}

			// remove all config updates for the package since it's being uninstalled or
			// upgraded. NOTE: The config updates must be removed whenever the version changes
			// or else the package interrupt may be skipped if there is one.
			// Use packageStatus.Name (from node state) since _package may be zero-value when
			// the package has been removed from spec.
			skyhook.GetSkyhook().RemoveConfigUpdates(packageStatus.Name)

			// set the node and skyhook status to in progress
			node.SetStatus(v1alpha1.StatusInProgress)
		}
	}

	// Auto-reset batch state when version changes are detected (if configured)
	if versionChangeDetected {
		resetSkyhookBatchState(skyhook)
	}

	return toUninstall, nil
}

// helper for get a point to a ref
func ptr[E any](e E) *E {
	return &e
}

// generateSafeName generates a consistent name for Kubernetes resources that is unique
// while staying within the specified character limit
func generateSafeName(maxLen int, nameParts ...string) string {
	name := strings.Join(nameParts, "-")
	// Replace dots with dashes as they're not allowed in resource names
	name = strings.ReplaceAll(name, ".", "-")

	unique := sha256.Sum256([]byte(name))
	uniqueStr := hex.EncodeToString(unique[:])[:8]

	maxlen := maxLen - len(uniqueStr) - 1
	if len(name) > maxlen {
		name = name[:maxlen]
	}

	return strings.ToLower(fmt.Sprintf("%s-%s", name, uniqueStr))
}

func (r *SkyhookReconciler) UpsertNodeLabelsAnnotationsPackages(ctx context.Context, skyhook *wrapper.Skyhook, node *corev1.Node) error {
	// No work to do if there is no labels or annotations for node
	if len(node.Labels) == 0 && len(node.Annotations) == 0 {
		return nil
	}

	annotations, err := json.Marshal(node.Annotations)
	if err != nil {
		return fmt.Errorf("error converting annotations into byte array: %w", err)
	}

	labels, err := json.Marshal(node.Labels)
	if err != nil {
		return fmt.Errorf("error converting labels into byte array: %w", err)
	}

	// marshal intermediary package metadata for the agent
	metadata := NewSkyhookMetadata(r.opts, skyhook)
	packages, err := metadata.Marshal()
	if err != nil {
		return fmt.Errorf("error converting packages into byte array: %w", err)
	}

	configMapName := generateSafeName(253, skyhook.Name, node.Name, "metadata")
	newCM := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      configMapName,
			Namespace: r.opts.Namespace,
			Labels: map[string]string{
				fmt.Sprintf("%s/skyhook-node-meta", v1alpha1.METADATA_PREFIX): skyhook.Name,
			},
			Annotations: map[string]string{
				fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):      skyhook.Name,
				fmt.Sprintf("%s/Node.name", v1alpha1.METADATA_PREFIX): node.Name,
			},
		},
		Data: map[string]string{
			"annotations.json": string(annotations),
			"labels.json":      string(labels),
			"packages.json":    string(packages),
		},
	}

	if err := ctrl.SetControllerReference(skyhook.NodeWright, newCM, r.scheme); err != nil {
		return fmt.Errorf("error setting ownership: %w", err)
	}

	existingConfigMap := &corev1.ConfigMap{}
	err = r.Get(ctx, client.ObjectKey{Namespace: r.opts.Namespace, Name: configMapName}, existingConfigMap)
	if err != nil {
		if apierrors.IsNotFound(err) {
			// create
			err := r.Create(ctx, newCM)
			if err != nil {
				return fmt.Errorf("error creating config map [%s]: %w", newCM.Name, err)
			}
		} else {
			return fmt.Errorf("error getting config map: %w", err)
		}
	} else {
		if !reflect.DeepEqual(existingConfigMap.Data, newCM.Data) {
			// update
			err := r.Update(ctx, newCM)
			if err != nil {
				return fmt.Errorf("error updating config map [%s]: %w", newCM.Name, err)
			}
		}
	}

	return nil
}

// configSyncRetryInterval is the requeue delay used when HandleConfigUpdates
// observes a ConfigMap diff it cannot yet apply because the completedNodes gate
// is closed. Without it the only fallback is the 10m MaxInterval requeue, which
// can leave an owned ConfigMap diverged from spec for minutes while status reads
// complete (issue #245). Short enough to heal quickly, long enough not to spin
// the grab-the-world reconcile while a node works through an interrupt cycle.
const configSyncRetryInterval = 30 * time.Second

// HandleConfigUpdates checks whether the configMap on a package was updated and if it was the configmap will
// be updated and the package will be put into config mode if the package is complete or erroring.
//
// The second return value (pendingSync) reports that a ConfigMap diff was observed
// but could not be applied this reconcile because the completedNodes gate was
// closed; the caller should requeue so the write is retried once the gate opens.
func (r *SkyhookReconciler) HandleConfigUpdates(ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes, _package v1alpha1.Package, oldConfigMap, newConfigMap *corev1.ConfigMap) (bool, bool, error) {
	completedNodes, nodeCount := 0, len(skyhook.GetNodes())
	erroringNode := false

	// if configmap changed
	if !reflect.DeepEqual(oldConfigMap.Data, newConfigMap.Data) {
		for _, node := range skyhook.GetNodes() {
			exists, err := r.PodExists(ctx, node.GetNode().Name, skyhook.GetSkyhook().Name, &_package)
			if err != nil {
				return false, false, fmt.Errorf("checking package pod existence on node %s for package %s: %w",
					node.GetNode().Name, _package.GetUniqueName(), err)
			}

			if !exists && node.IsPackageComplete(_package) {
				completedNodes++
			}

			// if we have an erroring node in the config, interrupt, or post-interrupt mode
			// then we will restart the config changes
			if packageStatus, found := node.PackageStatus(_package.GetUniqueName()); found {
				switch packageStatus.Stage {
				case v1alpha1.StageConfig, v1alpha1.StageInterrupt, v1alpha1.StagePostInterrupt:
					if packageStatus.State == v1alpha1.StateErroring {
						erroringNode = true

						// delete the erroring pod from the node so that it can be recreated
						// with the updated configmap
						pods, err := r.dal.GetPods(ctx,
							client.MatchingFields{
								fieldSelectorNodeName: node.GetNode().Name,
							},
							client.MatchingLabels{
								fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):    skyhook.GetSkyhook().Name,
								fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX): fmt.Sprintf("%s-%s", _package.Name, _package.Version),
							},
						)
						if err != nil {
							return false, false, fmt.Errorf("listing package pods on node %s for package %s: %w",
								node.GetNode().Name, _package.GetUniqueName(), err)
						}

						if pods != nil {
							for _, pod := range pods.Items {
								err := r.Delete(ctx, &pod)
								if err != nil {
									return false, false, fmt.Errorf("deleting erroring pod %s/%s on node %s: %w",
										pod.Namespace, pod.Name, node.GetNode().Name, err)
								}
							}
						}
					}
				}
			}
		}

		// if the update is complete or there is an erroring node put the package back into
		// the config mode and update the config map
		if completedNodes == nodeCount || erroringNode {
			// get the keys in the configmap that changed
			newConfigUpdates := make([]string, 0)
			for key, new_val := range newConfigMap.Data {
				if old_val, exists := oldConfigMap.Data[key]; !exists || old_val != new_val {
					newConfigUpdates = append(newConfigUpdates, key)
				}
			}

			// if updates completed then clear out old config updates as they are finished
			if completedNodes == nodeCount {
				skyhook.GetSkyhook().RemoveConfigUpdates(_package.Name)
			}

			// Add the new changed keys to the config updates
			skyhook.GetSkyhook().AddConfigUpdates(_package.Name, newConfigUpdates...)

			for _, node := range skyhook.GetNodes() {
				err := node.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateInProgress, v1alpha1.StageConfig, 0, _package.ContainerSHA)
				if err != nil {
					return false, false, fmt.Errorf("error upserting node status [%s]: %w", node.GetNode().Name, err)
				}

				node.SetStatus(v1alpha1.StatusInProgress)
			}

			_, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
			if len(errs) > 0 {
				return false, false, utilerrors.NewAggregate(errs)
			}

			// update config map
			err := r.Update(ctx, newConfigMap)
			if err != nil {
				return false, false, fmt.Errorf("error updating config map [%s]: %w", newConfigMap.Name, err)
			}

			return true, false, nil
		}

		// Diff observed but the gate is closed (no node has completed the package
		// and none is erroring). Defer the CM write and ask the caller to requeue
		// so we retry once the gate opens, rather than relying on the 10m fallback.
		return false, true, nil
	}

	return false, false, nil
}

func (r *SkyhookReconciler) UpsertConfigmaps(ctx context.Context, skyhook SkyhookNodes, clusterState *clusterState) (bool, bool, error) {
	updated := false
	pendingSync := false

	var list corev1.ConfigMapList
	err := r.List(ctx, &list, client.InNamespace(r.opts.Namespace), client.MatchingLabels{fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX): skyhook.GetSkyhook().Name})
	if err != nil {
		return false, false, fmt.Errorf("error listing config maps while upserting: %w", err)
	}

	existingCMs := make(map[string]corev1.ConfigMap)
	for _, cm := range list.Items {
		existingCMs[cm.Name] = cm
	}

	// clean up from an update
	shouldExist := make(map[string]struct{})
	for _, _package := range skyhook.GetSkyhook().Spec.Packages {
		shouldExist[strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.GetSkyhook().Name, _package.Name, _package.Version))] = struct{}{}
	}

	for k, v := range existingCMs {
		if _, ok := shouldExist[k]; !ok {
			// delete
			err := r.Delete(ctx, &v)
			if err != nil {
				return false, false, fmt.Errorf("error deleting existing config map [%s] while upserting: %w", v.Name, err)
			}
		}
	}

	// Build set of packages that are being uninstalled on any node (source of truth).
	// A failed state read on any node must propagate: if we silently skip an
	// unreadable node, we might miss a package mid-uninstall on that node and
	// then blow away its ConfigMap in the loop below, interfering with the
	// in-progress uninstall. Surface the error so the reconcile requeues and
	// the user-visible NodeStateMalformed condition (set at the top of
	// Reconcile) stays accurate.
	uninstallingPkgs := make(map[string]bool)
	for _, node := range skyhook.GetNodes() {
		ns, err := node.State()
		if err != nil {
			return false, false, fmt.Errorf("node %s: reading state while upserting configmaps: %w",
				node.GetNode().Name, err)
		}
		for _, _package := range skyhook.GetSkyhook().Spec.Packages {
			if ns.IsUninstallCycleInProgress(_package.GetUniqueName()) {
				uninstallingPkgs[_package.Name] = true
			}
		}
	}

	for _, _package := range skyhook.GetSkyhook().Spec.Packages {
		if uninstallingPkgs[_package.Name] {
			continue // config changes should not interfere with an in-progress uninstall
		}
		if len(_package.ConfigMap) > 0 {

			newCM := &corev1.ConfigMap{
				ObjectMeta: metav1.ObjectMeta{
					Name:      strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.GetSkyhook().Name, _package.Name, _package.Version)),
					Namespace: r.opts.Namespace,
					Labels: map[string]string{
						fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX): skyhook.GetSkyhook().Name,
					},
					Annotations: map[string]string{
						fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):            skyhook.GetSkyhook().Name,
						fmt.Sprintf("%s/Package.Name", v1alpha1.METADATA_PREFIX):    _package.Name,
						fmt.Sprintf("%s/Package.Version", v1alpha1.METADATA_PREFIX): _package.Version,
					},
				},
				Data: _package.ConfigMap,
			}
			// set owner of CM to the SCR, which will clean up the CM in delete of the SCR
			if err := ctrl.SetControllerReference(skyhook.GetSkyhook().NodeWright, newCM, r.scheme); err != nil {
				return false, false, fmt.Errorf("error setting ownership of cm: %w", err)
			}

			if existingCM, ok := existingCMs[strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.GetSkyhook().Name, _package.Name, _package.Version))]; ok {
				updatedConfigMap, pendingConfigSync, err := r.HandleConfigUpdates(ctx, clusterState, skyhook, _package, &existingCM, newCM)
				if err != nil {
					return false, false, fmt.Errorf("error updating config map [%s]: %w", newCM.Name, err)
				}
				if updatedConfigMap {
					updated = true
				}
				if pendingConfigSync {
					pendingSync = true
				}
			} else {
				// create
				err := r.Create(ctx, newCM)
				if err != nil {
					return false, false, fmt.Errorf("error creating config map [%s]: %w", newCM.Name, err)
				}
			}
		}
	}

	return updated, pendingSync, nil
}

func (r *SkyhookReconciler) IsDrained(ctx context.Context, skyhookNode wrapper.SkyhookNode) (bool, error) {

	pods, err := r.dal.GetPods(ctx, client.MatchingFields{
		fieldSelectorNodeName: skyhookNode.GetNode().Name,
	})
	if err != nil {
		return false, err
	}

	if pods == nil || len(pods.Items) == 0 {
		return true, nil
	}

	options := drain.OptionsFromConfig(skyhookNode.GetSkyhook().Spec.DrainConfig)
	options.PackageNamespace = r.opts.Namespace
	for _, pod := range pods.Items {
		if drain.DecidePod(&pod, options).BlocksDrain() {
			return false, nil
		}
	}

	return true, nil
}

// HandleFinalizer returns true only if we container is deleted and we handled it completely, else false.
// For Skyhooks with UninstallEnabled packages, this uses a multi-reconcile flow:
// Phase 1: trigger uninstall for enabled packages, return false to requeue
// Phase 2: check completion, requeue if still in progress
// Phase 3: cleanup (zero metrics, uncordon, remove SCR metadata, remove finalizer)
func (r *SkyhookReconciler) HandleFinalizer(ctx context.Context, skyhook SkyhookNodes, clusterState *clusterState) (bool, error) {
	if skyhook.GetSkyhook().DeletionTimestamp.IsZero() { // if not deleted, and does not have our finalizer, add it
		if !controllerutil.ContainsFinalizer(skyhook.GetSkyhook().NodeWright, SkyhookFinalizer) {
			controllerutil.AddFinalizer(skyhook.GetSkyhook().NodeWright, SkyhookFinalizer)

			if err := r.Update(ctx, skyhook.GetSkyhook().NodeWright); err != nil {
				return false, fmt.Errorf("error updating skyhook to add finalizer: %w", err)
			}
		}
	} else { // being deleted
		if controllerutil.ContainsFinalizer(skyhook.GetSkyhook().NodeWright, SkyhookFinalizer) {

			// Phase 2: scan uninstall-enabled packages across all nodes to
			// decide whether to block, wait, or proceed to Phase 3. Capture
			// both a state-read error (if any) and whether any uninstall-
			// enabled package is still tracked in nodeState.
			var stateErr error
			hasPendingUninstall := false
			for _, pkg := range skyhook.GetSkyhook().Spec.Packages {
				if !pkg.UninstallEnabled() {
					continue
				}
				for _, node := range skyhook.GetNodes() {
					nodeState, err := node.State()
					if err != nil {
						stateErr = fmt.Errorf(
							"error reading node state for finalizer on node %s: %w",
							node.GetNode().Name, err,
						)
						break
					}
					if _, inState := nodeState[pkg.GetUniqueName()]; inState {
						hasPendingUninstall = true
					}
				}
				if stateErr != nil {
					break
				}
			}

			switch {
			case stateErr != nil:
				// Malformed nodeState: we can't safely decide what to
				// preserve or what still needs uninstalling. NodeStateMalformed
				// is set separately at the top of reconcile; surface
				// DeletionBlocked too so the impact on deletion is explicit.
				// Return the error so controller-runtime retries with backoff —
				// the user must repair the annotation to proceed.
				wrapper.AddSkyhookCondition(skyhook.GetSkyhook(), metav1.Condition{
					Type:               wrapper.SkyhookConditionDeletionBlocked,
					Status:             metav1.ConditionTrue,
					ObservedGeneration: skyhook.GetSkyhook().Generation,
					LastTransitionTime: metav1.Now(),
					Reason:             "MalformedNodeState",
					Message: "Cannot safely delete Skyhook: malformed nodeState on one or more nodes. " +
						"Repair the nodeState annotation before deletion.",
				})
				r.recorder.Eventf(
					skyhook.GetSkyhook().NodeWright,
					nil,
					corev1.EventTypeWarning,
					"DeletionBlocked",
					"BlockDelete",
					"Cannot delete Skyhook %s: malformed nodeState. Repair and retry.",
					skyhook.GetSkyhook().Name,
				)
				if _, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook); len(errs) > 0 {
					return false, utilerrors.NewAggregate(errs)
				}
				return false, stateErr

			case skyhook.IsPaused() && hasPendingUninstall:
				// Paused Skyhooks cannot drive uninstall (processSkyhooksPerNode
				// short-circuits). Block deletion so the user unpauses and
				// lets uninstall run rather than silently leaving host-side
				// remnants — pause is a temporary "resume later" signal.
				wrapper.AddSkyhookCondition(skyhook.GetSkyhook(), metav1.Condition{
					Type:               wrapper.SkyhookConditionDeletionBlocked,
					Status:             metav1.ConditionTrue,
					ObservedGeneration: skyhook.GetSkyhook().Generation,
					LastTransitionTime: metav1.Now(),
					Reason:             "PausedWithPendingUninstall",
					Message: "Skyhook is paused with uninstall-enabled packages still tracked in nodeState. " +
						"Unpause to let uninstall complete before deletion.",
				})
				r.recorder.Eventf(
					skyhook.GetSkyhook().NodeWright,
					nil,
					corev1.EventTypeWarning,
					"DeletionBlocked",
					"BlockDelete",
					"Cannot delete Skyhook %s: paused with uninstall work pending. Unpause to proceed.",
					skyhook.GetSkyhook().Name,
				)
				if _, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook); len(errs) > 0 {
					return false, utilerrors.NewAggregate(errs)
				}
				return false, nil

			case skyhook.IsDisabled() && hasPendingUninstall:
				// Disabled Skyhooks also can't drive uninstall
				// (processSkyhooksPerNode short-circuits on disable).
				// uninstall.enabled=true packages are an explicit request to
				// run uninstall scripts before the CR goes away; allowing
				// deletion here would silently leave host-side state that
				// should have been cleaned up. Block and require the user
				// to re-enable the Skyhook so the uninstall flow can run.
				wrapper.AddSkyhookCondition(skyhook.GetSkyhook(), metav1.Condition{
					Type:               wrapper.SkyhookConditionDeletionBlocked,
					Status:             metav1.ConditionTrue,
					ObservedGeneration: skyhook.GetSkyhook().Generation,
					LastTransitionTime: metav1.Now(),
					Reason:             "DisabledWithPendingUninstall",
					Message: "Skyhook is disabled with uninstall-enabled packages still tracked in nodeState. " +
						"Re-enable the Skyhook to let uninstall complete before deletion.",
				})
				r.recorder.Eventf(
					skyhook.GetSkyhook().NodeWright,
					nil,
					corev1.EventTypeWarning,
					"DeletionBlocked",
					"BlockDelete",
					"Cannot delete Skyhook %s: disabled with uninstall work pending. Re-enable to proceed.",
					skyhook.GetSkyhook().Name,
				)
				if _, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook); len(errs) > 0 {
					return false, utilerrors.NewAggregate(errs)
				}
				return false, nil

			case hasPendingUninstall:
				// Normal path: wait for the main reconcile to drive packages
				// out of nodeState via the uninstall flow
				// (processSkyhooksPerNode → RunSkyhookPackages →
				// HandleUninstallRequests → ApplyPackage). The finalizer only
				// gates cleanup.
				wrapper.RemoveSkyhookConditionTypes(skyhook.GetSkyhook(), wrapper.SkyhookConditionDeletionBlocked)
				return false, nil

			default:
				// No pending uninstall — proceed to Phase 3.
				wrapper.RemoveSkyhookConditionTypes(skyhook.GetSkyhook(), wrapper.SkyhookConditionDeletionBlocked)
			}

			// Phase 3: All enabled packages uninstalled (or none exist). Cleanup.
			errs := make([]error, 0)

			// zero out all the metrics related to this skyhook both skyhook and packages
			zeroOutSkyhookMetrics(skyhook)

			for _, node := range skyhook.GetNodes() {
				patch := client.StrategicMergeFrom(node.GetNode().DeepCopy())

				node.Uncordon()
				node.CleanupSCRMetadata()

				// if this doesn't change the node then don't patch
				if !node.Changed() {
					continue
				}

				err := r.Patch(ctx, node.GetNode(), patch)
				if err != nil {
					errs = append(errs, fmt.Errorf("error patching node [%s] in finalizer: %w", node.GetNode().Name, err))
				}
			}

			if len(errs) > 0 { // we errored, so we need to return error, otherwise we would release the skyhook when we didnt finish
				return false, utilerrors.NewAggregate(errs)
			}

			// Bump ObservedGeneration and write status BEFORE removing the finalizer:
			// removing the last finalizer lets the apiserver delete the object
			// immediately, so a status update afterward would race that deletion into a
			// spurious NotFound. (The bump keeps the finalizer-removal Update below from
			// re-triggering add-finalizer logic.)
			skyhook.GetSkyhook().Status.ObservedGeneration = skyhook.GetSkyhook().Status.ObservedGeneration + 1
			if err := r.Status().Update(ctx, skyhook.GetSkyhook().NodeWright); err != nil {
				return false, fmt.Errorf("error updating skyhook status: %w", err)
			}

			controllerutil.RemoveFinalizer(skyhook.GetSkyhook().NodeWright, SkyhookFinalizer)
			if err := r.Update(ctx, skyhook.GetSkyhook().NodeWright); err != nil {
				return false, fmt.Errorf("error updating skyhook removing finalizer: %w", err)
			}

			return true, nil
		}
	}
	return false, nil
}

// HasNonInterruptWork returns true if pods are running on the node that are either packages, or matches the SCR selector
func (r *SkyhookReconciler) HasNonInterruptWork(ctx context.Context, skyhookNode wrapper.SkyhookNode) (bool, error) {

	selector, err := metav1.LabelSelectorAsSelector(&skyhookNode.GetSkyhook().Spec.PodNonInterruptLabels)
	if err != nil {
		return false, fmt.Errorf("error creating selector: %w", err)
	}

	if selector.Empty() { // when selector is empty it does not do any selecting, ie will return all pods on node.
		return false, nil
	}

	pods, err := r.dal.GetPods(ctx,
		client.MatchingLabelsSelector{Selector: selector},
		client.MatchingFields{
			fieldSelectorNodeName: skyhookNode.GetNode().Name,
		},
	)
	if err != nil {
		return false, fmt.Errorf("error getting pods: %w", err)
	}

	if pods == nil || len(pods.Items) == 0 {
		return false, nil
	}

	for _, pod := range pods.Items {
		switch pod.Status.Phase {
		case corev1.PodRunning, corev1.PodPending:
			return true, nil
		}
	}

	return false, nil
}

func (r *SkyhookReconciler) HasRunningPackages(ctx context.Context, skyhookNode wrapper.SkyhookNode) (bool, error) {
	pods, err := r.dal.GetPods(ctx,
		client.HasLabels{fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX)},
		client.MatchingFields{
			fieldSelectorNodeName: skyhookNode.GetNode().Name,
		},
	)
	if err != nil {
		return false, fmt.Errorf("error getting pods: %w", err)
	}

	return pods != nil && len(pods.Items) > 0, nil
}

func (r *SkyhookReconciler) DrainNode(ctx context.Context, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package) (bool, error) {
	drained, err := r.IsDrained(ctx, skyhookNode)
	if err != nil {
		return false, err
	}
	if drained {
		skyhookNode.ClearDrainStart()
		return true, nil
	}

	drainStartedAt, err := skyhookNode.DrainStartedAt()
	if err != nil {
		return false, fmt.Errorf("error reading drain start for node [%s]: %w", skyhookNode.GetNode().Name, err)
	}

	drainConfig := skyhookNode.GetSkyhook().Spec.DrainConfig
	now := metav1.Now()
	if drainStartedAt == nil {
		skyhookNode.StartDrain(now)
		skyhookNode.SetStatus(v1alpha1.StatusInProgress)
	} else if drainConfig != nil && drain.TimedOut(drainStartedAt, drainConfig.Timeout, now.Time) {
		r.recorder.Eventf(skyhookNode.GetNode(), nil, corev1.EventTypeWarning, EventsReasonSkyhookDrain, "DrainTimeout",
			"drain timed out after [%s] for node [%s] package [%s:%s] from [skyhook:%s]",
			drainConfig.Timeout.Duration,
			skyhookNode.GetNode().Name,
			_package.Name,
			_package.Version,
			skyhookNode.GetSkyhook().Name,
		)
		r.recorder.Eventf(skyhookNode.GetSkyhook().NodeWright, nil, corev1.EventTypeWarning, EventsReasonSkyhookDrain, "DrainTimeout",
			"drain timed out after [%s] for node [%s] package [%s:%s]",
			drainConfig.Timeout.Duration,
			skyhookNode.GetNode().Name,
			_package.Name,
			_package.Version,
		)
		skyhookNode.SetStatus(v1alpha1.StatusErroring)
		return false, nil
	}

	pods, err := r.dal.GetPods(ctx, client.MatchingFields{
		fieldSelectorNodeName: skyhookNode.GetNode().Name,
	})
	if err != nil {
		return false, err
	}

	if pods == nil || len(pods.Items) == 0 {
		return true, nil
	}

	r.recorder.Eventf(skyhookNode.GetNode(), nil, EventTypeNormal, EventsReasonSkyhookDrain, "DrainNode",
		"draining node [%s] package [%s:%s] from [skyhook:%s]",
		skyhookNode.GetNode().Name,
		_package.Name,
		_package.Version,
		skyhookNode.GetSkyhook().Name,
	)

	options := drain.OptionsFromConfig(skyhookNode.GetSkyhook().Spec.DrainConfig)
	options.PackageNamespace = r.opts.Namespace
	errs := make([]error, 0)
	waitingForPods := false
	for _, pod := range pods.Items {
		decision := drain.DecidePod(&pod, options)
		switch decision.Action {
		case drain.ActionBlock:
			waitingForPods = true
		case drain.ActionEvict:
			waitingForPods = true
			eviction := policyv1.Eviction{DeleteOptions: options.EvictionDeleteOptions()}
			err := r.Client.SubResource("eviction").Create(ctx, &pod, &eviction)
			if err != nil {
				errs = append(errs, fmt.Errorf("error evicting pod [%s:%s]: %w", pod.Namespace, pod.Name, err))
			}
		case drain.ActionDelete:
			waitingForPods = true
			err := r.Delete(ctx, &pod, options.DeleteOptions()...)
			if err != nil {
				errs = append(errs, fmt.Errorf("error deleting pod [%s:%s]: %w", pod.Namespace, pod.Name, err))
			}
		}
	}

	if len(errs) > 0 {
		return false, utilerrors.NewAggregate(errs)
	}

	return !waitingForPods, nil
}

// Interrupt should not be called unless safe to do so, IE already cordoned and drained
func (r *SkyhookReconciler) Interrupt(ctx context.Context, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package, _interrupt *v1alpha1.Interrupt, stage v1alpha1.Stage) error {

	hasPackagesRunning, err := r.HasRunningPackages(ctx, skyhookNode)
	if err != nil {
		return err
	}

	if hasPackagesRunning { // keep waiting...
		return nil
	}

	exists, err := r.PodExists(ctx, skyhookNode.GetNode().Name, skyhookNode.GetSkyhook().Name, _package)
	if err != nil {
		return err
	}
	if exists {
		// nothing to do here, already running
		return nil
	}

	// Ensure the node metadata configmap exists before creating the pod
	// This prevents a race where the pod starts before its required configmap is created
	if err := r.UpsertNodeLabelsAnnotationsPackages(ctx, skyhookNode.GetSkyhook(), skyhookNode.GetNode()); err != nil {
		return fmt.Errorf("error upserting node metadata configmap: %w", err)
	}

	argEncode, err := _interrupt.ToArgs()
	if err != nil {
		return fmt.Errorf("error creating interrupt args: %w", err)
	}

	pod := createInterruptPodForPackage(r.opts, _interrupt, argEncode, _package, skyhookNode.GetSkyhook(), skyhookNode.GetNode().Name, stage)

	if err := SetPackages(pod, skyhookNode.GetSkyhook().NodeWright, _package.Image, stage, _package); err != nil {
		return fmt.Errorf("error setting package on interrupt: %w", err)
	}

	if err := ctrl.SetControllerReference(skyhookNode.GetSkyhook().NodeWright, pod, r.scheme); err != nil {
		return fmt.Errorf("error setting ownership: %w", err)
	}

	if err := r.Create(ctx, pod); err != nil {
		return fmt.Errorf("error creating interruption pod: %w", err)
	}

	_ = skyhookNode.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateInProgress, stage, 0, _package.ContainerSHA)

	r.recorder.Eventf(skyhookNode.GetSkyhook().NodeWright, nil, EventTypeNormal, EventsReasonSkyhookInterrupt, "InterruptNode",
		"Interrupting node [%s] package [%s:%s] from [skyhook:%s]",
		skyhookNode.GetNode().Name,
		_package.Name,
		_package.Version,
		skyhookNode.GetSkyhook().Name)

	return nil
}

// fudgeInterruptWithPriority takes a list of packages, interrupts, and configUpdates and returns the correct merged interrupt to run to handle all the packages
func fudgeInterruptWithPriority(next []*v1alpha1.Package, configUpdates map[string][]string, interrupts map[string][]*v1alpha1.Interrupt) (*v1alpha1.Interrupt, string) {
	var ret *v1alpha1.Interrupt
	var pack string

	// map interrupt to priority
	// A lower priority value means a higher priority and will be used in favor of anything with a higher value
	var priorities = map[v1alpha1.InterruptType]int{
		v1alpha1.REBOOT:               0,
		v1alpha1.RESTART_ALL_SERVICES: 1,
		v1alpha1.SERVICE:              2,
		v1alpha1.NOOP:                 3,
	}

	for _, _package := range next {

		if len(configUpdates[_package.Name]) == 0 {
			interrupts[_package.Name] = []*v1alpha1.Interrupt{}
			if _package.HasInterrupt() {
				interrupts[_package.Name] = append(interrupts[_package.Name], _package.Interrupt)
			}
		}
	}

	packageNames := make([]string, 0, len(next))
	for _, pkg := range next {
		packageNames = append(packageNames, pkg.Name)
	}
	sort.Strings(packageNames)

	for _, _package := range packageNames {
		_interrupts, ok := interrupts[_package]
		if !ok {
			continue
		}

		for _, interrupt := range _interrupts {
			if ret == nil { // prime ret, base case
				ret = interrupt
				pack = _package
			}

			// short circuit, reboot has highest priority
			switch interrupt.Type {
			case v1alpha1.REBOOT:
				return interrupt, _package
			}

			// check if interrupt is higher priority using the priority_order
			// A lower priority value means a higher priority
			if priorities[interrupt.Type] < priorities[ret.Type] {
				ret = interrupt
				pack = _package
			} else if priorities[interrupt.Type] == priorities[ret.Type] {
				mergeInterrupt(ret, interrupt)
			}
		}
	}

	return ret, pack // return merged interrupt and package
}

func mergeInterrupt(left, right *v1alpha1.Interrupt) {

	// make sure both are of type service
	if left.Type != v1alpha1.SERVICE || right.Type != v1alpha1.SERVICE {
		return
	}

	left.Services = merge(left.Services, right.Services)
}

func merge[T cmp.Ordered](left, right []T) []T {
	for _, r := range right {
		if !slices.Contains(left, r) {
			left = append(left, r)
		}
	}
	slices.Sort(left)
	return left
}

// ValidateNodeConfigmaps validates that there are no orphaned or stale config maps for a node
func (r *SkyhookReconciler) ValidateNodeConfigmaps(ctx context.Context, skyhookName string, nodes []wrapper.SkyhookNode) (bool, error) {
	var list corev1.ConfigMapList
	err := r.List(ctx, &list, client.InNamespace(r.opts.Namespace), client.MatchingLabels{fmt.Sprintf("%s/skyhook-node-meta", v1alpha1.METADATA_PREFIX): skyhookName})
	if err != nil {
		return false, fmt.Errorf("error listing config maps: %w", err)
	}

	// No configmaps created by this skyhook, no work needs to be done
	if len(list.Items) == 0 {
		return false, nil
	}

	existingCMs := make(map[string]corev1.ConfigMap)
	for _, cm := range list.Items {
		existingCMs[cm.Name] = cm
	}

	shouldExist := make(map[string]struct{})
	for _, node := range nodes {
		shouldExist[generateSafeName(253, skyhookName, node.GetNode().Name, "metadata")] = struct{}{}
	}

	update := false
	errs := make([]error, 0)
	for k, v := range existingCMs {
		if _, ok := shouldExist[k]; !ok {
			update = true
			err := r.Delete(ctx, &v)
			if err != nil {
				errs = append(errs, fmt.Errorf("error deleting existing config map [%s]: %w", v.Name, err))
			}
		}
	}

	// Ensure packages.json is present and up-to-date for expected configmaps
	skyhookCR, err := r.dal.GetSkyhook(ctx, skyhookName)
	if err != nil {
		return update, fmt.Errorf("error getting skyhook for metadata validation: %w", err)
	}
	skyhookWrapper := wrapper.NewSkyhookWrapper(skyhookCR)
	metadata := NewSkyhookMetadata(r.opts, skyhookWrapper)
	expectedBytes, err := metadata.Marshal()
	if err != nil {
		return update, fmt.Errorf("error marshalling metadata for validation: %w", err)
	}
	expected := string(expectedBytes)

	for i := range list.Items {
		cm := &list.Items[i]
		if _, ok := shouldExist[cm.Name]; !ok {
			continue
		}
		if cm.Data == nil {
			cm.Data = make(map[string]string)
		}
		if cm.Data["packages.json"] != expected {
			cm.Data["packages.json"] = expected
			if err := r.Update(ctx, cm); err != nil {
				errs = append(errs, fmt.Errorf("error updating packages.json on config map [%s]: %w", cm.Name, err))
			} else {
				update = true
			}
		}
	}

	return update, utilerrors.NewAggregate(errs)
}

// PodExists tests if this package is exists on a node.
func (r *SkyhookReconciler) PodExists(ctx context.Context, nodeName, skyhookName string, _package *v1alpha1.Package) (bool, error) {

	pods, err := r.dal.GetPods(ctx,
		client.MatchingFields{
			fieldSelectorNodeName: nodeName,
		},
		client.MatchingLabels{
			fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):    skyhookName,
			fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX): fmt.Sprintf("%s-%s", _package.Name, _package.Version),
		},
	)
	if err != nil {
		return false, fmt.Errorf("error check from existing pods: %w", err)
	}

	if pods == nil || len(pods.Items) == 0 {
		return false, nil
	}
	return true, nil
}

// createInterruptPodForPackage returns the pod spec for an interrupt pod given an package
func createInterruptPodForPackage(opts SkyhookOperatorOptions, _interrupt *v1alpha1.Interrupt, argEncode string, _package *v1alpha1.Package, skyhook *wrapper.Skyhook, nodeName string, stage v1alpha1.Stage) *corev1.Pod {
	copyDir := fmt.Sprintf("%s/%s/%s-%s-%s-%d",
		opts.CopyDirRoot,
		skyhook.Name,
		_package.Name,
		_package.Version,
		skyhook.UID,
		skyhook.Generation,
	)

	volumes := []corev1.Volume{
		{
			Name: volumeNameRootMount,
			VolumeSource: corev1.VolumeSource{
				HostPath: &corev1.HostPathVolumeSource{
					Path: "/",
				},
			},
		},
		{
			// node names in different CSPs might include dots which isn't allowed in volume names
			// so we have to replace all dots with dashes
			Name: generateSafeName(63, skyhook.Name, nodeName, "metadata"),
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{
						Name: strings.ReplaceAll(fmt.Sprintf("%s-%s-metadata", skyhook.Name, nodeName), ".", "-"),
					},
				},
			},
		},
	}
	volumeMounts := []corev1.VolumeMount{
		{
			Name:             volumeNameRootMount,
			MountPath:        mountPathRoot,
			MountPropagation: ptr(corev1.MountPropagationHostToContainer),
		},
	}

	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      generateSafeName(63, skyhook.Name, string(stage), string(_interrupt.Type), nodeName),
			Namespace: opts.Namespace,
			Labels: map[string]string{
				fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):      skyhook.Name,
				fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX):   fmt.Sprintf("%s-%s", _package.Name, _package.Version),
				fmt.Sprintf("%s/interrupt", v1alpha1.METADATA_PREFIX): "True",
			},
		},
		Spec: corev1.PodSpec{
			NodeName:      nodeName,
			RestartPolicy: corev1.RestartPolicyOnFailure,
			InitContainers: []corev1.Container{
				{
					Name:  InterruptContainerName,
					Image: getAgentImage(opts, _package),
					Args:  []string{"interrupt", mountPathRoot, copyDir, argEncode},
					Env:   getAgentConfigEnvVars(opts, _package.Name, _package.Version, skyhook.ResourceID(), skyhook.Name, skyhook.NodeOrder(nodeName)),
					SecurityContext: &corev1.SecurityContext{
						Privileged: ptr(true),
					},
					VolumeMounts: volumeMounts,
					Resources: corev1.ResourceRequirements{
						Limits: corev1.ResourceList{
							corev1.ResourceCPU:    resource.MustParse("500m"),
							corev1.ResourceMemory: resource.MustParse("64Mi"),
						},
						Requests: corev1.ResourceList{
							corev1.ResourceCPU:    resource.MustParse("500m"),
							corev1.ResourceMemory: resource.MustParse("64Mi"),
						},
					},
				},
			},
			Containers: []corev1.Container{
				{
					Name:  "pause",
					Image: opts.PauseImage,
					Resources: corev1.ResourceRequirements{
						Limits: corev1.ResourceList{
							corev1.ResourceCPU:    resource.MustParse("100m"),
							corev1.ResourceMemory: resource.MustParse("20Mi"),
						},
						Requests: corev1.ResourceList{
							corev1.ResourceCPU:    resource.MustParse("100m"),
							corev1.ResourceMemory: resource.MustParse("20Mi"),
						},
					},
				},
			},
			HostPID:     true,
			HostNetwork: true,
			// If you change these go change the SelectNode toleration in cluster_state.go
			Tolerations: append([]corev1.Toleration{ // tolerate all cordon
				{
					Key:      TaintUnschedulable,
					Operator: corev1.TolerationOpExists,
				},
				opts.GetRuntimeRequiredToleration(),
			}, skyhook.Spec.AdditionalTolerations...),
			Volumes: volumes,
		},
	}
	if opts.ImagePullSecret != "" {
		pod.Spec.ImagePullSecrets = []corev1.LocalObjectReference{
			{
				Name: opts.ImagePullSecret,
			},
		}
	}
	return pod
}

func trunstr(str string, length int) string {
	if len(str) > length {
		return str[:length]
	}
	return str
}

func getAgentImage(opts SkyhookOperatorOptions, _package *v1alpha1.Package) string {
	if _package.AgentImageOverride != "" {
		return _package.AgentImageOverride
	}
	return opts.AgentImage
}

// getPackageImage returns the full image reference for a package, using the digest if specified
func getPackageImage(_package *v1alpha1.Package) string {
	if _package.ContainerSHA != "" {
		// When containerSHA is specified, use it instead of the version tag for immutable image reference
		return fmt.Sprintf("%s@%s", _package.Image, _package.ContainerSHA)
	}
	// Fall back to version tag
	return fmt.Sprintf("%s:%s", _package.Image, _package.Version)
}

func getAgentConfigEnvVars(opts SkyhookOperatorOptions, packageName string, packageVersion string, resourceID string, skyhookName string, nodeOrder int) []corev1.EnvVar {
	return []corev1.EnvVar{
		{
			Name:  "SKYHOOK_LOG_DIR",
			Value: fmt.Sprintf("%s/%s", opts.AgentLogRoot, skyhookName),
		},
		{
			Name:  "SKYHOOK_ROOT_DIR",
			Value: fmt.Sprintf("%s/%s", opts.CopyDirRoot, skyhookName),
		},
		{
			Name:  "COPY_RESOLV",
			Value: annotationFalseValue,
		},
		{
			Name:  envSkyhookResourceID,
			Value: fmt.Sprintf("%s_%s_%s", resourceID, packageName, packageVersion),
		},
		{
			Name:  envSkyhookNodeOrder,
			Value: strconv.Itoa(nodeOrder),
		},
	}
}

// createPodFromPackage creates a pod spec for a skyhook pod for a given package
func createPodFromPackage(opts SkyhookOperatorOptions, _package *v1alpha1.Package, skyhook *wrapper.Skyhook, nodeName string, stage v1alpha1.Stage) *corev1.Pod {
	// Generate consistent names that won't exceed k8s limits
	volumeName := generateSafeName(63, "metadata", nodeName)
	configMapName := generateSafeName(253, skyhook.Name, nodeName, "metadata")

	volumes := []corev1.Volume{
		{
			Name: volumeNameRootMount,
			VolumeSource: corev1.VolumeSource{
				HostPath: &corev1.HostPathVolumeSource{
					Path: "/",
				},
			},
		},
		{
			Name: volumeName,
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{
						Name: configMapName,
					},
				},
			},
		},
	}

	volumeMounts := []corev1.VolumeMount{
		{
			Name:             volumeNameRootMount,
			MountPath:        mountPathRoot,
			MountPropagation: ptr(corev1.MountPropagationHostToContainer),
		},
		{
			Name:      volumeName,
			MountPath: "/skyhook-package/node-metadata",
		},
	}

	if len(_package.ConfigMap) > 0 {
		volumes = append(volumes, corev1.Volume{
			Name: _package.Name,
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{
						Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.Name, _package.Name, _package.Version)),
					},
				},
			},
		})

		// Mount each configMap key as its own subPath rather than mounting the
		// whole configMap as a directory. A directory mount replaces the entire
		// path, hiding any files the package image baked in under
		// mountPathConfigMaps; per-key subPath mounts overlay individual files
		// on top of the image content instead. Trade-off: subPath mounts do not
		// receive live configMap updates, which is fine here because package
		// pods are recreated per stage / version bump. Keys are iterated in
		// sorted order so the generated pod spec is deterministic.
		configMapKeys := make([]string, 0, len(_package.ConfigMap))
		for key := range _package.ConfigMap {
			configMapKeys = append(configMapKeys, key)
		}
		sort.Strings(configMapKeys)

		for _, key := range configMapKeys {
			volumeMounts = append(volumeMounts, corev1.VolumeMount{
				Name:      _package.Name,
				MountPath: fmt.Sprintf("%s/%s", mountPathConfigMaps, key),
				SubPath:   key,
				ReadOnly:  true,
			})
		}
	}

	copyDir := fmt.Sprintf("%s/%s/%s-%s-%s-%d",
		opts.CopyDirRoot,
		skyhook.Name,
		_package.Name,
		_package.Version,
		skyhook.UID,
		skyhook.Generation,
	)
	applyargs := []string{strings.ToLower(string(stage)), mountPathRoot, copyDir}
	checkargs := []string{strings.ToLower(string(stage) + "-check"), mountPathRoot, copyDir}

	agentEnvs := append(
		_package.Env,
		getAgentConfigEnvVars(opts, _package.Name, _package.Version, skyhook.ResourceID(), skyhook.Name, skyhook.NodeOrder(nodeName))...,
	)

	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      generateSafeName(63, skyhook.Name, _package.Name, _package.Version, string(stage), nodeName),
			Namespace: opts.Namespace,
			Labels: map[string]string{
				fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):    skyhook.Name,
				fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX): fmt.Sprintf("%s-%s", _package.Name, _package.Version),
			},
		},
		Spec: corev1.PodSpec{
			NodeName:      nodeName,
			RestartPolicy: corev1.RestartPolicyOnFailure,
			InitContainers: []corev1.Container{
				{
					Name:            fmt.Sprintf("%s-init", trunstr(_package.Name, 43)),
					Image:           getPackageImage(_package),
					ImagePullPolicy: corev1.PullAlways,
					Command:         []string{"/bin/sh"},
					Args: []string{
						"-c",
						"mkdir -p /root/${SKYHOOK_DIR} && cp -r /skyhook-package/* /root/${SKYHOOK_DIR}",
					},
					Env: []corev1.EnvVar{
						{
							Name:  "SKYHOOK_DIR",
							Value: copyDir,
						},
					},
					SecurityContext: &corev1.SecurityContext{
						Privileged: ptr(true),
					},
					VolumeMounts: volumeMounts,
				},
				{
					Name:            fmt.Sprintf("%s-%s", trunstr(_package.Name, 43), stage),
					Image:           getAgentImage(opts, _package),
					ImagePullPolicy: corev1.PullAlways,
					Args:            applyargs,
					Env:             agentEnvs,
					SecurityContext: &corev1.SecurityContext{
						Privileged: ptr(true),
					},
					VolumeMounts: volumeMounts,
				},
				{
					Name:            fmt.Sprintf("%s-%scheck", trunstr(_package.Name, 43), stage),
					Image:           getAgentImage(opts, _package),
					ImagePullPolicy: corev1.PullAlways,
					Args:            checkargs,
					Env:             agentEnvs,
					SecurityContext: &corev1.SecurityContext{
						Privileged: ptr(true),
					},
					VolumeMounts: volumeMounts,
				},
			},
			Containers: []corev1.Container{
				{
					Name:  "pause",
					Image: opts.PauseImage,
					Resources: corev1.ResourceRequirements{
						Limits: corev1.ResourceList{
							corev1.ResourceCPU:    resource.MustParse("100m"),
							corev1.ResourceMemory: resource.MustParse("20Mi"),
						},
						Requests: corev1.ResourceList{
							corev1.ResourceCPU:    resource.MustParse("100m"),
							corev1.ResourceMemory: resource.MustParse("20Mi"),
						},
					},
				},
			},
			Volumes:     volumes,
			HostPID:     true,
			HostNetwork: true,
			// If you change these go change the SelectNode toleration in cluster_state.go
			Tolerations: append([]corev1.Toleration{ // tolerate all cordon
				{
					Key:      TaintUnschedulable,
					Operator: corev1.TolerationOpExists,
				},
				opts.GetRuntimeRequiredToleration(),
			}, skyhook.Spec.AdditionalTolerations...),
		},
	}
	if opts.ImagePullSecret != "" {
		pod.Spec.ImagePullSecrets = []corev1.LocalObjectReference{
			{
				Name: opts.ImagePullSecret,
			},
		}
	}
	if _package.GracefulShutdown != nil {
		pod.Spec.TerminationGracePeriodSeconds = ptr(int64(_package.GracefulShutdown.Duration.Seconds()))
	}
	setPodResources(pod, _package.Resources)
	return pod
}

// FilterEnv removes the environment variables passed into exlude
func FilterEnv(envs []corev1.EnvVar, exclude ...string) []corev1.EnvVar {
	var filteredEnv []corev1.EnvVar

	// build map of exclude strings for faster lookup
	excludeMap := make(map[string]struct{})
	for _, name := range exclude {
		excludeMap[name] = struct{}{}
	}

	// If the environment variable name is in the exclude list, skip it
	// otherwise append it to the final list
	for _, env := range envs {
		if _, found := excludeMap[env.Name]; !found {
			filteredEnv = append(filteredEnv, env)
		}
	}

	return filteredEnv
}

// PodMatchesPackage asserts that a given pod matches the given pod spec
func podMatchesPackage(opts SkyhookOperatorOptions, _package *v1alpha1.Package, pod corev1.Pod, skyhook *wrapper.Skyhook, stage v1alpha1.Stage) bool {
	var expectedPod *corev1.Pod

	// need to differentiate whether the pod is for an interrupt or not so we know
	// what to expect and how to compare them
	isInterrupt := false
	_, limitRange := pod.Annotations["kubernetes.io/limit-ranger"]

	if pod.Labels[fmt.Sprintf("%s/interrupt", v1alpha1.METADATA_PREFIX)] == "True" {
		expectedPod = createInterruptPodForPackage(opts, &v1alpha1.Interrupt{}, "", _package, skyhook, "", stage)
		isInterrupt = true
	} else {
		expectedPod = createPodFromPackage(opts, _package, skyhook, "", stage)
	}

	actualPod := pod.DeepCopy()

	// check to see whether the name or the version of the package changed
	packageLabel := fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX)
	if actualPod.Labels[packageLabel] != expectedPod.Labels[packageLabel] {
		return false
	}

	// compare initContainers since this is where a lot of the important info lives
	for i := range actualPod.Spec.InitContainers {
		expectedContainer := expectedPod.Spec.InitContainers[i]
		actualContainer := actualPod.Spec.InitContainers[i]

		if expectedContainer.Name != actualContainer.Name {
			return false
		}

		if expectedContainer.Image != actualContainer.Image {
			return false
		}

		// compare the containers env vars except for the ones that are inserted
		// by the operator by default as the SKYHOOK_RESOURCE_ID will change every
		// time the skyhook is updated and would cause every pod to be removed
		// TODO: This is ignoring all the static env vars that are set by operator config.
		// It probably should be just SKYHOOK_RESOURCE_ID that is ignored. Otherwise,
		// a user will have to manually delete the pod to update the package when operator is updated.
		dummyAgentEnv := getAgentConfigEnvVars(opts, "", "", "", "", 0)
		excludedEnvs := make([]string, len(dummyAgentEnv))
		for i, env := range dummyAgentEnv {
			excludedEnvs[i] = env.Name
		}
		expectedFilteredEnv := FilterEnv(expectedContainer.Env, excludedEnvs...)
		actualFilteredEnv := FilterEnv(actualContainer.Env, excludedEnvs...)
		if !reflect.DeepEqual(expectedFilteredEnv, actualFilteredEnv) {
			return false
		}

		if !isInterrupt { // dont compare these since they are not configured on interrupt
			// compare resource requests and limits (CPU, memory, etc.)
			expectedResources := expectedContainer.Resources
			actualResources := actualContainer.Resources
			if skyhook.Spec.Packages[_package.Name].Resources != nil {
				// If CR has resources specified, they should match exactly
				if !reflect.DeepEqual(expectedResources, actualResources) {
					return false
				}
			} else {
				// If CR has no resources specified, ensure pod has no resource overrides
				if !limitRange {
					if actualResources.Requests != nil || actualResources.Limits != nil {
						return false
					}
				}
			}
		}
	}

	return true
}

// ValidateRunningPackages deletes pods that don't match the current spec and checks if there are pods running
// that don't match the node state and removes them if they exist
func (r *SkyhookReconciler) ValidateRunningPackages(ctx context.Context, skyhook SkyhookNodes) (bool, error) {

	update := false
	errs := make([]error, 0)
	// get all pods for this skyhook packages
	pods, err := r.dal.GetPods(ctx,
		client.MatchingLabels{
			fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX): skyhook.GetSkyhook().Name,
		},
	)
	if err != nil {
		return false, fmt.Errorf("error getting pods while validating packages: %w", err)
	}
	if pods == nil || len(pods.Items) == 0 {
		return false, nil // nothing running for this skyhook on this node
	}

	// Initialize metrics for each stage
	stages := make(map[string]map[string]map[v1alpha1.Stage]int)

	// group pods by node
	podsbyNode := make(map[string][]corev1.Pod)
	for _, pod := range pods.Items {
		podsbyNode[pod.Spec.NodeName] = append(podsbyNode[pod.Spec.NodeName], pod)
	}

	for _, node := range skyhook.GetNodes() {
		nodeState, err := node.State()
		if err != nil {
			return false, fmt.Errorf("node %s: reading state while initializing stage metrics: %w",
				node.GetNode().Name, err)
		}

		for _, pod := range podsbyNode[node.GetNode().Name] {
			found := false

			runningPackage, err := GetPackage(&pod)
			if err != nil {
				errs = append(errs, fmt.Errorf("error getting package from pod [%s:%s] while validating packages: %w", pod.Namespace, pod.Name, err))
			}

			// check if the package is part of the skyhook spec, if not we need to delete it
			for _, v := range skyhook.GetSkyhook().Spec.Packages {
				if podMatchesPackage(r.opts, &v, pod, skyhook.GetSkyhook(), runningPackage.Stage) {
					found = true
				}
			}

			// Increment the stage count for metrics
			if _, ok := stages[runningPackage.Name]; !ok {
				stages[runningPackage.Name] = make(map[string]map[v1alpha1.Stage]int)
				if _, ok := stages[runningPackage.Name][runningPackage.Version]; !ok {
					stages[runningPackage.Name][runningPackage.Version] = make(map[v1alpha1.Stage]int)
					for _, stage := range v1alpha1.Stages {
						stages[runningPackage.Name][runningPackage.Version][stage] = 0
					}
				}
			}
			stages[runningPackage.Name][runningPackage.Version][runningPackage.Stage]++

			// For uninstall pods, check whether this is a legacy downgrade uninstall
			// (package removed from spec or version changed) vs an explicit uninstall
			// (package still in spec). Legacy uninstall pods can't be validated against
			// the spec since the package was removed/changed, so mark them as found.
			// Explicit uninstall pods ARE in spec and should be validated — if the spec
			// changed (e.g. user fixed a bad configmap), podMatchesPackage returns false
			// and the pod gets recreated with the new config.
			if runningPackage.Stage == v1alpha1.StageUninstall && !found {
				specPkg, inSpec := skyhook.GetSkyhook().Spec.Packages[runningPackage.Name]
				if !inSpec || specPkg.Version != runningPackage.Version {
					// Legacy downgrade or removed-from-spec uninstall — can't validate
					found = true
				}
				// else: explicit uninstall — leave found as-is so podMatchesPackage decides
			}

			if !found {
				update = true

				err := r.InvalidPackage(ctx, &pod)
				if err != nil {
					errs = append(errs, fmt.Errorf("error invalidating package: %w", err))
				}
				continue
			}

			// Check if package exists in node state, ie a package running that the node state doesn't know about
			// something that is often done to try to fix bad node state is to clear the node state completely
			// which if a package is running, we want to terminate it gracefully. Ofthen what leads to this is
			// the package is in a crashloop and the operator want to restart it the whole package.
			// when we apply a package it just check if there is a running package on the node for the state of the package
			// this can cause to leave a pod running in say config mode, and it there is a depends on you might not correctly
			// run thins in the correct order.
			deleteMe := false
			packageStatus, exists := nodeState[runningPackage.GetUniqueName()]
			if !exists { // package not in node state, so we need to delete it
				deleteMe = true
			} else { // package in node state, so we need to check if it's running
				// need check if the stats match, if not we need to delete it
				if packageStatus.Stage != runningPackage.Stage {
					deleteMe = true
				}
			}

			if deleteMe {
				update = true
				err := r.InvalidPackage(ctx, &pod)
				if err != nil {
					errs = append(errs, fmt.Errorf("error invalidating package: %w", err))
				}
			}
		}
	}

	return update, utilerrors.NewAggregate(errs)
}

// InvalidPackage invalidates a package and updates the pod, which will trigger the pod to be deleted
func (r *SkyhookReconciler) InvalidPackage(ctx context.Context, pod *corev1.Pod) error {
	err := InvalidatePackage(pod)
	if err != nil {
		return fmt.Errorf("error invalidating package: %w", err)
	}

	err = r.Update(ctx, pod)
	if err != nil {
		return fmt.Errorf("error updating pod: %w", err)
	}

	return nil
}

// ProcessInterrupt will check and do the interrupt if need, and returns
// false means we are waiting
// true means we are good to proceed
func (r *SkyhookReconciler) ProcessInterrupt(ctx context.Context, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package, interrupt *v1alpha1.Interrupt, runInterrupt bool) (bool, error) {

	if !skyhookNode.HasInterrupt(*_package) {
		return true, nil
	}

	// default starting stage
	stage := v1alpha1.StageApply
	nextStage := skyhookNode.NextStage(_package)
	if nextStage != nil {
		stage = *nextStage
	}

	// wait tell this is done if its happening
	status, found := skyhookNode.PackageStatus(_package.GetUniqueName())
	if found && status.IsSkipped() {
		// Level-triggered backstop. This package was skipped because a higher-priority
		// interrupt won the node's single interrupt slot (e.g. a reboot that also satisfies
		// it). The pod controller promotes skipped packages when that interrupt pod completes,
		// but a skipped package has no pod of its own, so if that edge is missed (the pod is
		// already GC'd, or the skip was written after it completed) nothing else ever promotes
		// it and the node never finishes. runInterrupt is true only once this is the highest-
		// priority interrupt left to run, meaning the preempting interrupt has completed and
		// left the runnable set: promote then, so the reconcile converges on its own rather
		// than depending on a pod event that may never arrive. While a higher-priority
		// interrupt is still pending (runInterrupt false) the package keeps waiting.
		if runInterrupt {
			if err := skyhookNode.ProgressSkipped(); err != nil {
				return false, fmt.Errorf("error progressing skipped package [%s]: %w", _package.GetUniqueName(), err)
			}
		}
		return false, nil
	}

	// Theres is a race condition when a node reboots and api cleans up the interrupt pod
	// so we need to check if the pod exists and if it does, we need to recreate it
	if status != nil && status.IsActive() && status.IsInterruptStage() {
		// call interrupt to recreate the pod if missing
		err := r.Interrupt(ctx, skyhookNode, _package, interrupt, status.Stage)
		if err != nil {
			return false, err
		}
	}

	// drain and cordon node before applying package that has an interrupt
	if stage == v1alpha1.StageApply || stage == v1alpha1.StageUninstall {
		ready, err := r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, _package)
		if err != nil {
			return false, err
		}

		if !ready {
			return false, nil
		}
	}

	// time to interrupt (once other packages have finished)
	if stage == v1alpha1.StageInterrupt && runInterrupt {
		err := r.Interrupt(ctx, skyhookNode, _package, interrupt, v1alpha1.StageInterrupt)
		if err != nil {
			return false, err
		}

		return false, nil
	}

	//skipping
	if stage == v1alpha1.StageInterrupt && !runInterrupt {
		err := skyhookNode.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateSkipped, stage, 0, _package.ContainerSHA)
		if err != nil {
			return false, fmt.Errorf("error upserting to skip interrupt: %w", err)
		}
		return false, nil
	}

	// Uninstall-cycle interrupt: HandleCompletePod set StageUninstallInterrupt/InProgress;
	// fire the interrupt pod (idempotent — r.Interrupt bails if pod exists).
	// Always runs — once uninstall has started, the interrupt must run to completion.
	if status != nil && status.Stage == v1alpha1.StageUninstallInterrupt && status.State != v1alpha1.StateComplete {
		err := r.Interrupt(ctx, skyhookNode, _package, interrupt, v1alpha1.StageUninstallInterrupt)
		if err != nil {
			return false, err
		}
		return false, nil
	}

	// wait tell this is done if its happening
	if status != nil && status.IsInterruptStage() && status.State != v1alpha1.StateComplete {
		return false, nil
	}

	return true, nil
}

func (r *SkyhookReconciler) EnsureNodeIsReadyForInterrupt(ctx context.Context, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package) (bool, error) {
	// cordon node
	skyhookNode.Cordon()

	hasWork, err := r.HasNonInterruptWork(ctx, skyhookNode)
	if err != nil {
		return false, err
	}
	if hasWork { // keep waiting...
		return false, nil
	}

	ready, err := r.DrainNode(ctx, skyhookNode, _package)
	if err != nil {
		return false, fmt.Errorf("error draining node [%s]: %w", skyhookNode.GetNode().Name, err)
	}

	return ready, nil
}

// ApplyPackage starts a pod on node for the package
func (r *SkyhookReconciler) ApplyPackage(ctx context.Context, logger logr.Logger, clusterState *clusterState, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package, runInterrupt bool) error {

	if _package == nil {
		return errors.New("can not apply nil package")
	}

	// default starting stage
	stage := v1alpha1.StageApply

	// These modes don't have anything that comes before them so we must specify them as the
	// starting point. The next stage function will return nil until these modes complete.
	// Config is a special case as sometimes apply will come before it and other times it wont
	// which is why it needs to be here as well
	if packageStatus, found := skyhookNode.PackageStatus(_package.GetUniqueName()); found {
		switch packageStatus.Stage {
		case v1alpha1.StageConfig, v1alpha1.StageUpgrade, v1alpha1.StageUninstall,
			v1alpha1.StageUninstallInterrupt:
			stage = packageStatus.Stage
		}
	}

	// if stage != v1alpha1.StageApply {
	// 	// If a node gets rest by a user, the about method will return the wrong node state. Above sources it from the skyhook status.
	// 	// check if the node has nothing, reset it then apply the package.
	// 	nodeState, err := skyhookNode.State()
	// 	if err != nil {
	// 		return fmt.Errorf("error getting node state: %w", err)
	// 	}

	// 	_, found := nodeState[_package.GetUniqueName()]
	// 	if !found {
	// 		stage = v1alpha1.StageApply
	// 	}
	// }

	nextStage := skyhookNode.NextStage(_package)
	if nextStage != nil {
		stage = *nextStage
	}

	// Uninstall-cycle interrupt pods are controller-created by ProcessInterrupt via
	// r.Interrupt (type-based pod, not a stage-script pod). ApplyPackage has no
	// work to do here.
	if stage == v1alpha1.StageUninstallInterrupt {
		return nil
	}

	// test if pod exists, if so, bailout
	exists, err := r.PodExists(ctx, skyhookNode.GetNode().Name, skyhookNode.GetSkyhook().Name, _package)
	if err != nil {
		return err
	}

	// wait tell this is done if its happening
	status, found := skyhookNode.PackageStatus(_package.GetUniqueName())

	if found && status.IsSkipped() { // skipped, so nothing to do
		return nil
	}

	if found && status.State == v1alpha1.StateInProgress { // running, so do nothing atm
		if exists {
			return nil
		}
	}

	if exists {
		// nothing to do here, already running
		return nil
	}

	// Ensure the node metadata configmap exists before creating the pod
	// This prevents a race where the pod starts before its required configmap is created
	if err := r.UpsertNodeLabelsAnnotationsPackages(ctx, skyhookNode.GetSkyhook(), skyhookNode.GetNode()); err != nil {
		return fmt.Errorf("error upserting node metadata configmap: %w", err)
	}

	pod := createPodFromPackage(r.opts, _package, skyhookNode.GetSkyhook(), skyhookNode.GetNode().Name, stage)

	if err := SetPackages(pod, skyhookNode.GetSkyhook().NodeWright, _package.Image, stage, _package); err != nil {
		return fmt.Errorf("error setting package on pod: %w", err)
	}

	// setup ownership of the pod we created
	// helps run time know what to do when something happens to this pod we are about to create
	if err := ctrl.SetControllerReference(skyhookNode.GetSkyhook().NodeWright, pod, r.scheme); err != nil {
		return fmt.Errorf("error setting ownership: %w", err)
	}

	if err := r.Create(ctx, pod); err != nil {
		return fmt.Errorf("error creating pod: %w", err)
	}

	if err = skyhookNode.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateInProgress, stage, 0, _package.ContainerSHA); err != nil {
		err = fmt.Errorf("error upserting package: %w", err) // want to keep going in this case, but don't want to lose the err
	}

	skyhookNode.SetStatus(v1alpha1.StatusInProgress)

	wrapper.AddSkyhookConditionWithLegacy(skyhookNode.GetSkyhook(), metav1.Condition{
		Type:               wrapper.SkyhookConditionApplyPackage,
		Status:             metav1.ConditionTrue,
		ObservedGeneration: skyhookNode.GetSkyhook().Generation,
		LastTransitionTime: metav1.Now(),
		Reason:             "ApplyPackage",
		Message:            fmt.Sprintf("Applying package [%s:%s] to node [%s]", _package.Name, _package.Version, skyhookNode.GetNode().Name),
	})

	r.recorder.Eventf(skyhookNode.GetNode(), nil, EventTypeNormal, EventsReasonSkyhookApply, "ApplyPackage", "Applying package [%s:%s] from [skyhook:%s] stage [%s]", _package.Name, _package.Version, skyhookNode.GetSkyhook().Name, stage)
	r.recorder.Eventf(skyhookNode.GetSkyhook(), nil, EventTypeNormal, EventsReasonSkyhookApply, "ApplyPackage", "Applying package [%s:%s] to node [%s] stage [%s]", _package.Name, _package.Version, skyhookNode.GetNode().Name, stage)

	skyhookNode.GetSkyhook().Updated = true

	return err
}

// HandleRuntimeRequired finds any nodes for which all runtime required Skyhooks are complete and remove their runtime required taint
// Will return an error if the patching of the nodes is not possible
func (r *SkyhookReconciler) HandleRuntimeRequired(ctx context.Context, clusterState *clusterState) error {
	node_to_skyhooks, skyhook_node_map := groupSkyhooksByNode(clusterState)
	to_remove := getRuntimeRequiredTaintCompleteNodes(node_to_skyhooks, skyhook_node_map)
	// Remove the runtime required taint from nodes in to_remove
	taint_to_remove := r.opts.GetRuntimeRequiredTaint()
	errs := make([]error, 0)
	for _, node := range to_remove {
		// check before removing taint that it even exists to begin with
		if !taints.TaintExists(node.Spec.Taints, &taint_to_remove) {
			continue
		}
		// RemoveTaint will ALWAYS return nil for its error so no need to check it
		new_node, updated, _ := taints.RemoveTaint(node, &taint_to_remove)
		if updated {
			// If any runtime-required Skyhook sets runtimeRequiredCordonAfter to true, the cordon and
			// runtimeRequiredCordon annotation are only applied if the runtime-required taint exists. This means that a
			// runtime-required Skyhook with runtimeRequiredCordonAfter true will not apply the persistent cordon if the
			// runtime-required taint was already removed. Removing the taint and applying the node cordon in the same patch
			// request ensures a scheduling gate is always applied to the targeted node.
			if runtimeRequiredCordonAfterEnabled(node_to_skyhooks[node.UID]) {
				if new_node.Annotations == nil {
					new_node.Annotations = make(map[string]string)
				}
				new_node.Annotations[v1alpha1.RuntimeRequiredCordonAnnotation] = annotationTrueValue
				new_node.Spec.Unschedulable = true
			}
			if err := r.Patch(ctx, new_node, client.MergeFrom(node)); err != nil {
				errs = append(errs, err)
			}
		}
	}
	if len(errs) > 0 {
		return utilerrors.NewAggregate(errs)
	}
	return nil
}

func runtimeRequiredCordonAfterEnabled(skyhooks []SkyhookNodes) bool {
	for _, skyhook := range skyhooks {
		spec := skyhook.GetSkyhook().Spec
		if spec.RuntimeRequired && spec.RuntimeRequiredCordonAfter {
			return true
		}
	}
	return false
}

// Group Skyhooks by what node they target
func groupSkyhooksByNode(clusterState *clusterState) (map[types.UID][]SkyhookNodes, map[types.UID]*corev1.Node) {
	node_to_skyhooks := make(map[types.UID][]SkyhookNodes)
	nodes := make(map[types.UID]*corev1.Node)
	for _, skyhook := range clusterState.skyhooks {
		// Ignore skyhooks that don't have runtime required
		if !skyhook.GetSkyhook().Spec.RuntimeRequired {
			continue
		}
		for _, node := range skyhook.GetNodes() {
			if _, ok := node_to_skyhooks[node.GetNode().UID]; !ok {
				node_to_skyhooks[node.GetNode().UID] = make([]SkyhookNodes, 0)
				nodes[node.GetNode().UID] = node.GetNode()
			}
			node_to_skyhooks[node.GetNode().UID] = append(node_to_skyhooks[node.GetNode().UID], skyhook)
		}

	}
	return node_to_skyhooks, nodes
}

// Get the nodes to remove runtime required taint from node that all skyhooks targeting that node have completed
// Note: This checks per-node completion, not skyhook-level completion. A node's taint is removed when all
// runtime-required skyhooks are complete ON THAT SPECIFIC NODE, regardless of other nodes' completion status.
func getRuntimeRequiredTaintCompleteNodes(node_to_skyhooks map[types.UID][]SkyhookNodes, nodes map[types.UID]*corev1.Node) []*corev1.Node {
	to_remove := make([]*corev1.Node, 0)
	for node_uid, skyhooks := range node_to_skyhooks {
		node := nodes[node_uid]
		all_complete := true
		for _, skyhook := range skyhooks {
			// Check if THIS specific node is complete for this skyhook (not all nodes)
			_, nodeWrapper := skyhook.GetNode(node.Name)
			if nodeWrapper == nil || !nodeWrapper.IsComplete() {
				all_complete = false
				break
			}
		}
		if all_complete {
			to_remove = append(to_remove, node)
		}
	}
	return to_remove
}

// HandleAutoTaint applies the runtime-required taint to new nodes matching runtime-required
// Skyhooks that have AutoTaintNewNodes enabled.
func (r *SkyhookReconciler) HandleAutoTaint(ctx context.Context, clusterState *clusterState) (bool, error) {
	taint_to_add := r.opts.GetRuntimeRequiredTaint()
	to_taint := clusterState.getAutoTaintNodes(taint_to_add)
	errs := make([]error, 0)
	changed := false
	for _, node := range to_taint {
		newNode, updated, _ := taints.AddOrUpdateTaint(node, &taint_to_add)
		if !updated {
			continue
		}
		// add annotation to indicate that the node was auto-tainted
		if newNode.Annotations == nil {
			newNode.Annotations = make(map[string]string)
		}
		newNode.Annotations[fmt.Sprintf("%s/autoTaint_%s", v1alpha1.METADATA_PREFIX, taint_to_add.Key)] = annotationTrueValue

		if err := r.Patch(ctx, newNode, client.MergeFrom(node)); err != nil {
			errs = append(errs, err)
		}
		changed = true
	}
	if len(errs) > 0 {
		return changed, utilerrors.NewAggregate(errs)
	}
	return changed, nil
}

// setPodResources sets resources for all containers and init containers in the pod if override is set, else leaves empty for LimitRange
func setPodResources(pod *corev1.Pod, res *v1alpha1.ResourceRequirements) {
	if res == nil {
		return
	}
	if !res.CPURequest.IsZero() || !res.CPULimit.IsZero() || !res.MemoryRequest.IsZero() || !res.MemoryLimit.IsZero() {
		for i := range pod.Spec.InitContainers {
			pod.Spec.InitContainers[i].Resources = corev1.ResourceRequirements{
				Limits: corev1.ResourceList{
					corev1.ResourceCPU:    res.CPULimit,
					corev1.ResourceMemory: res.MemoryLimit,
				},
				Requests: corev1.ResourceList{
					corev1.ResourceCPU:    res.CPURequest,
					corev1.ResourceMemory: res.MemoryRequest,
				},
			}
		}
	}
}

// PartitionNodesIntoCompartments partitions nodes for each skyhook that uses deployment policies.
func partitionNodesIntoCompartments(clusterState *clusterState) error {
	for _, skyhook := range clusterState.skyhooks {
		// Skip skyhooks without a deployment policy (they use the default compartment created in BuildState)
		if skyhook.GetSkyhook().Spec.DeploymentPolicy == "" {
			continue
		}

		// Skip if no compartments exist (e.g., deployment policy not found)
		// The webhook should prevent this at admission time, and the controller sets a condition at runtime,
		// but we guard here to prevent panics if the policy goes missing
		if len(skyhook.GetCompartments()) == 0 {
			continue
		}

		// Clear all compartments before reassigning nodes to prevent stale nodes
		// This ensures nodes are only in their current compartment based on current labels
		for _, compartment := range skyhook.GetCompartments() {
			compartment.ClearNodes()
		}

		for _, node := range skyhook.GetNodes() {
			compartmentName, err := skyhook.AssignNodeToCompartment(node)
			if err != nil {
				return fmt.Errorf("error assigning node %s: %w", node.GetNode().Name, err)
			}
			if err := skyhook.AddCompartmentNode(compartmentName, node); err != nil {
				return fmt.Errorf("error adding node %s to compartment %s: %w", node.GetNode().Name, compartmentName, err)
			}
		}
	}

	return nil
}

// validateAndUpsertSkyhookData performs validation and configmap operations for a skyhook.
//
// The second return value (pendingSync) reports that an owned ConfigMap diverged from
// spec but the write was deferred because the completedNodes gate was closed. It is
// deliberately decoupled from the first (shouldReturn): a deferred sync must NOT
// short-circuit the reconcile, because progression toward the gate opening happens in
// processSkyhooksPerNode. The caller uses pendingSync only to shorten the otherwise
// MaxInterval idle requeue so the deferred write is retried promptly (issue #245).
func (r *SkyhookReconciler) validateAndUpsertSkyhookData(ctx context.Context, skyhook SkyhookNodes, clusterState *clusterState) (bool, bool, ctrl.Result, error) {
	if yes, result, err := shouldReturn(r.ValidateRunningPackages(ctx, skyhook)); yes {
		return yes, false, result, err
	}

	if yes, result, err := shouldReturn(r.ValidateNodeConfigmaps(ctx, skyhook.GetSkyhook().Name, skyhook.GetNodes())); yes {
		return yes, false, result, err
	}

	updated, pendingSync, err := r.UpsertConfigmaps(ctx, skyhook, clusterState)
	if err != nil {
		return true, false, ctrl.Result{}, err
	}
	if updated {
		return true, false, ctrl.Result{RequeueAfter: time.Second * 2}, nil
	}

	return false, pendingSync, ctrl.Result{}, nil
}