feat(eks): 6d nodegroup provisioning + NATS-gated cluster health (#300)

* feat(eks): nodegroup provisioning + K3s agent join (6d)

CreateNodegroup launches N customer-owned worker EC2 from the unified
eks-node AMI in agent role; each joins the cluster via K3S_URL + K3S_TOKEN
baked into cloud-init (first-boot.env SPINIFEX_K3S_ROLE=agent + agent.env).
Workers are customer-owned and DescribeInstances-visible (no ManagedBy,
no mgmt NIC) via the normal RunInstances path, matching real EKS.

K3S_URL targets the control-plane ENI's in-VPC private IP on :6443 — the
cluster NLB DNS endpoint is not resolvable in-VPC yet; switch to the NLB
endpoint on :443 once authoritative in-VPC DNS lands. K3S_TOKEN is the
decrypted cluster node-token persisted at server bootstrap.

Describe/List/Update(scaling)/Delete round out CRUD. Cluster<->nodegroup
SG ingress rules (apiserver 6443, flannel VXLAN 8472/udp, kubelet 10250,
intra-nodegroup) authorized idempotently. UpdateNodegroupVersion deferred.

* feat: gate EKS cluster health via NATS state reports

The K3s apiserver binds the VPC node-ip, unreachable from the host-side
spinifex daemon, so the reconciler cannot probe /healthz directly. Add a
control-plane state reporter that publishes {healthz, node_count} over the
management NATS bus the daemon already shares.

- eks-node server role runs mulga-eks-state-report as an OpenRC service,
  publishing eks.state.{account}.{cluster}.server every 30s.
- ClusterReconciler subscribes to that subject, gates CREATING -> ACTIVE on
  a healthy report, and records Ready node_count into ClusterMeta.
- node_count counts Ready nodes only; k3s retains terminated workers as
  NotReady until pruned, so a raw count never drops on scale-down.

scripts/images/eks-node/eks-node-role.sh

@@ -49,9 +49,11 @@ case "${ROLE}" in
         rc-update add eks-token-webhook default
         rc-update add k3s default
         rc-update add k3s-first-boot default
+        rc-update add mulga-eks-state-report default
         rc-service eks-token-webhook start
         rc-service k3s start
         rc-service k3s-first-boot start
+        rc-service mulga-eks-state-report start
         ;;
     agent)
         log "configuring agent role"

scripts/images/eks-node/manifest.conf

@@ -30,12 +30,13 @@ scripts/images/eks-node/k3s-agent.initd:/etc/init.d/k3s-agent \
 scripts/images/eks-node/eks-token-webhook.initd:/etc/init.d/eks-token-webhook \
 scripts/images/eks-node/k3s-first-boot.initd:/etc/init.d/k3s-first-boot \
 scripts/images/eks-node/k3s-first-boot.sh:/usr/local/sbin/k3s-first-boot \
-scripts/images/eks-node/mulga-eks-state-report.sh:/etc/periodic/15min/mulga-eks-state-report \
+scripts/images/eks-node/mulga-eks-state-report.initd:/etc/init.d/mulga-eks-state-report \
+scripts/images/eks-node/mulga-eks-state-report.sh:/usr/local/sbin/mulga-eks-state-report \
 scripts/images/eks-node/mulga-eks-etcd-snapshot.sh:/etc/periodic/daily/mulga-eks-etcd-snapshot"
 
-# Only crond (state-report) + the role selector at bake. The selector adds the
-# role-specific services (k3s/eks-token-webhook/k3s-first-boot or k3s-agent) to
-# the default runlevel on first boot.
+# Only crond (etcd snapshot) + the role selector at bake. The selector adds the
+# role-specific services (k3s/eks-token-webhook/k3s-first-boot/state-report or
+# k3s-agent) to the default runlevel on first boot.
 ENABLE_SERVICES="crond eks-node-role"
 
 SETUP_SCRIPT="scripts/images/eks-node/setup.sh"

scripts/images/eks-node/mulga-eks-state-report.initd

@@ -0,0 +1,29 @@
+#!/sbin/openrc-run
+# mulga-eks-state-report — server-role service that publishes K3s health +
+# node_count to the management NATS bus every STATE_REPORT_INTERVAL seconds. The
+# spinifex cluster reconciler consumes this to gate CREATING → ACTIVE and to
+# track nodegroup join/scale, since the apiserver (bound to the VPC node-ip) is
+# unreachable from the host. Enabled on the server role by eks-node-role.sh.
+
+description="Mulga EKS control-plane state reporter"
+command="/usr/local/sbin/mulga-eks-state-report"
+command_background="yes"
+pidfile="/run/mulga-eks-state-report.pid"
+output_log="/var/log/mulga-eks-state-report.log"
+error_log="/var/log/mulga-eks-state-report.log"
+
+# Cadence the report loop sleeps between publishes. 30s is well under the
+# reconciler's 90s stale window, so a missed tick does not flap health.
+export STATE_REPORT_INTERVAL="${STATE_REPORT_INTERVAL:-30}"
+
+depend() {
+    need net
+    after k3s
+}
+
+start_pre() {
+    if [ ! -f /etc/spinifex-eks/first-boot.env ]; then
+        eerror "/etc/spinifex-eks/first-boot.env not present — cloud-init did not seed NATS creds"
+        return 1
+    fi
+}

scripts/images/eks-node/mulga-eks-state-report.sh

@@ -1,15 +1,20 @@
 #!/bin/sh
 set -eu
 
-# mulga-eks-state-report — periodic NATS publish of K3s server health.
-# Invoked every 15 minutes via /etc/periodic/15min (Alpine crond). One-shot
-# best-effort; failures log but do not retry — the spinifex reconciler also
-# does its own NLB healthcheck so this is supplementary signal.
+# mulga-eks-state-report — NATS publish of K3s server health for the spinifex
+# cluster reconciler. The apiserver binds the VPC node-ip, unreachable from the
+# host-side daemon, so the daemon cannot probe /healthz directly — this report
+# carries {healthz, node_count} over the management NATS bus the daemon shares.
 #
 # Subject: eks.state.{accountID}.{clusterName}.server
 # Payload: JSON { healthz: "ok"|"fail", node_count: N, ts: <unix-seconds> }
+#
+# Reads NATS creds + cluster identity from the cloud-init-seeded first-boot env
+# (same vars k3s-first-boot.sh publishes its bootstrap artifacts with). Run as a
+# loop by the mulga-eks-state-report OpenRC service when STATE_REPORT_INTERVAL is
+# set; one-shot otherwise (e.g. a manual invocation).
 
-ENVFILE=/etc/spinifex-eks/state-report.env
+ENVFILE=/etc/spinifex-eks/first-boot.env
 [ -f "${ENVFILE}" ] || { logger -t mulga-eks-state-report "${ENVFILE} missing — exiting"; exit 0; }
 # shellcheck disable=SC1090
 . "${ENVFILE}"
@@ -21,27 +26,39 @@ ENVFILE=/etc/spinifex-eks/state-report.env
 KUBECONFIG=${KUBECONFIG:-/etc/rancher/k3s/k3s.yaml}
 export KUBECONFIG
 
-if curl -sk --max-time 3 https://127.0.0.1:6443/healthz | grep -q '^ok$'; then
-    HEALTH=ok
-else
-    HEALTH=fail
-fi
-
-if [ "${HEALTH}" = "ok" ]; then
-    NODE_COUNT=$(kubectl get nodes --no-headers 2>/dev/null | wc -l | tr -d ' ')
-else
-    NODE_COUNT=0
-fi
-
-TS=$(date +%s)
-PAYLOAD=$(printf '{"healthz":"%s","node_count":%s,"ts":%s}' "${HEALTH}" "${NODE_COUNT}" "${TS}")
 SUBJ="eks.state.${EKS_ACCOUNT_ID}.${EKS_CLUSTER_NAME}.server"
 
+# NATS args mirror k3s-first-boot.sh (token + TLS CA) — the proven publish path.
 NATS_ARGS="-s ${SPINIFEX_NATS_URL}"
-if [ -n "${SPINIFEX_NATS_CREDS_FILE:-}" ] && [ -f "${SPINIFEX_NATS_CREDS_FILE}" ]; then
-    NATS_ARGS="${NATS_ARGS} --creds ${SPINIFEX_NATS_CREDS_FILE}"
+if [ -n "${SPINIFEX_NATS_TOKEN:-}" ]; then
+    NATS_ARGS="${NATS_ARGS} --token ${SPINIFEX_NATS_TOKEN}"
+fi
+if [ -n "${SPINIFEX_NATS_CA:-}" ] && [ -f "${SPINIFEX_NATS_CA}" ]; then
+    NATS_ARGS="${NATS_ARGS} --tlsca ${SPINIFEX_NATS_CA}"
 fi
 
-# shellcheck disable=SC2086
-printf '%s' "${PAYLOAD}" | nats ${NATS_ARGS} pub "${SUBJ}" --stdin 2>&1 \
-    | logger -t mulga-eks-state-report
+publish_report() {
+    if curl -sk --max-time 3 https://127.0.0.1:6443/healthz | grep -q '^ok$'; then
+        health=ok
+        # Count Ready nodes only — k3s retains terminated workers as NotReady
+        # until manually pruned, so a raw node count never drops on scale-down.
+        node_count=$(kubectl get nodes --no-headers 2>/dev/null | awk '$2=="Ready"' | wc -l | tr -d ' ')
+    else
+        health=fail
+        node_count=0
+    fi
+    payload=$(printf '{"healthz":"%s","node_count":%s,"ts":%s}' "${health}" "${node_count}" "$(date +%s)")
+    # shellcheck disable=SC2086
+    printf '%s' "${payload}" | nats ${NATS_ARGS} pub "${SUBJ}" --force-stdin 2>&1 \
+        | logger -t mulga-eks-state-report
+}
+
+# STATE_REPORT_INTERVAL set → run forever (service mode); unset → one-shot.
+if [ -n "${STATE_REPORT_INTERVAL:-}" ] && [ "${STATE_REPORT_INTERVAL}" -gt 0 ] 2>/dev/null; then
+    while true; do
+        publish_report
+        sleep "${STATE_REPORT_INTERVAL}"
+    done
+else
+    publish_report
+fi

scripts/images/eks-node/setup.sh

@@ -55,9 +55,9 @@ apk del --no-cache unzip
 # Init scripts ship as 0644 from INSTALL_FILES; OpenRC requires 0755. Every
 # role's services are baked; the selector enables the right ones at first boot.
 chmod 0755 /etc/init.d/eks-node-role /etc/init.d/k3s /etc/init.d/k3s-agent \
-    /etc/init.d/eks-token-webhook /etc/init.d/k3s-first-boot
-chmod 0755 /usr/local/sbin/eks-node-role /usr/local/sbin/k3s-first-boot
-chmod 0755 /etc/periodic/15min/mulga-eks-state-report
+    /etc/init.d/eks-token-webhook /etc/init.d/k3s-first-boot /etc/init.d/mulga-eks-state-report
+chmod 0755 /usr/local/sbin/eks-node-role /usr/local/sbin/k3s-first-boot \
+    /usr/local/sbin/mulga-eks-state-report
 chmod 0755 /etc/periodic/daily/mulga-eks-etcd-snapshot
 
 # K3s server config — skeleton; cloud-init / first-boot fills in the

spinifex/daemon/daemon_handlers_eks_test.go

@@ -61,6 +61,11 @@ func TestDaemonHandleEKS_AllHandlersDispatchToService(t *testing.T) {
 	// no error, so wantCode is "" (success — assert a non-error payload).
 	notImpl := awserrors.ErrorNotImplemented
 	invalid := awserrors.ErrorInvalidParameterValue
+	// Nodegroup mutators gate on orchestration deps, which the shim service
+	// lacks → ServiceUnavailable. The read paths reach input validation and an
+	// empty body fails with InvalidParameterValue. UpdateNodegroupVersion stays
+	// NotImplemented (v1 doesn't do AMI upgrades).
+	unavailable := awserrors.ErrorServiceUnavailable
 	cases := []struct {
 		subject  string
 		handler  nats.MsgHandler
@@ -68,12 +73,12 @@ func TestDaemonHandleEKS_AllHandlersDispatchToService(t *testing.T) {
 	}{
 		{"eks.UpdateClusterConfig", d.handleEKSUpdateClusterConfig, notImpl},
 		{"eks.UpdateClusterVersion", d.handleEKSUpdateClusterVersion, notImpl},
-		{"eks.CreateNodegroup", d.handleEKSCreateNodegroup, notImpl},
-		{"eks.DescribeNodegroup", d.handleEKSDescribeNodegroup, notImpl},
-		{"eks.ListNodegroups", d.handleEKSListNodegroups, notImpl},
-		{"eks.UpdateNodegroupConfig", d.handleEKSUpdateNodegroupConfig, notImpl},
+		{"eks.CreateNodegroup", d.handleEKSCreateNodegroup, unavailable},
+		{"eks.DescribeNodegroup", d.handleEKSDescribeNodegroup, invalid},
+		{"eks.ListNodegroups", d.handleEKSListNodegroups, invalid},
+		{"eks.UpdateNodegroupConfig", d.handleEKSUpdateNodegroupConfig, unavailable},
 		{"eks.UpdateNodegroupVersion", d.handleEKSUpdateNodegroupVersion, notImpl},
-		{"eks.DeleteNodegroup", d.handleEKSDeleteNodegroup, notImpl},
+		{"eks.DeleteNodegroup", d.handleEKSDeleteNodegroup, unavailable},
 		{"eks.CreateAccessEntry", d.handleEKSCreateAccessEntry, invalid},
 		{"eks.DescribeAccessEntry", d.handleEKSDescribeAccessEntry, invalid},
 		{"eks.ListAccessEntries", d.handleEKSListAccessEntries, invalid},

spinifex/daemon/eks_deps.go

@@ -59,6 +59,7 @@ func (d *Daemon) buildEKSServiceDeps() handlers_eks.EKSServiceDeps {
 		Instance:       d,
 		Image:          d.imageService,
 		EIP:            d.eipService,
+		Worker:         d,
 	}
 }
 

spinifex/daemon/eks_worker_launch.go

@@ -0,0 +1,67 @@
+package daemon
+
+import (
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"log/slog"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks"
+	spxtypes "github.com/mulgadc/spinifex/spinifex/types"
+)
+
+// Compile-time check that Daemon satisfies the EKS worker-launch surface.
+var _ handlers_eks.WorkerLauncher = (*Daemon)(nil)
+
+// RunWorkerInstance launches a nodegroup worker through the normal customer
+// RunInstances path: customer-owned, primary ENI auto-created in input.SubnetId
+// with input.SecurityGroupIds, no ManagedBy tag, no mgmt NIC — so the worker is
+// visible in DescribeInstances and reclaimed by TerminateInstances like any
+// other customer EC2. The EKS service hands us raw cloud-config YAML in
+// input.UserData; RunInstances expects it base64-encoded, so wrap it here.
+func (d *Daemon) RunWorkerInstance(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) {
+	if d.instanceService == nil {
+		return nil, errors.New("eks worker: instance service not initialized")
+	}
+	if input == nil {
+		return nil, errors.New("eks worker: nil RunInstancesInput")
+	}
+	if input.UserData != nil && *input.UserData != "" {
+		input.UserData = aws.String(base64.StdEncoding.EncodeToString([]byte(*input.UserData)))
+	}
+	return d.instanceService.RunInstances(input, accountID)
+}
+
+// TerminateWorkerInstances terminates each nodegroup worker via the same path
+// the EC2 NATS terminate handler uses. A worker no longer tracked by the local
+// vmMgr is treated as already-gone (idempotent success) so a retried
+// DeleteNodegroup / scale-down does not wedge on instances that already drained.
+func (d *Daemon) TerminateWorkerInstances(instanceIDs []string, accountID string) error {
+	if d.instanceService == nil {
+		return errors.New("eks worker: instance service not initialized")
+	}
+	var errs []error
+	for _, id := range instanceIDs {
+		if id == "" {
+			continue
+		}
+		instance, ok := d.vmMgr.Get(id)
+		if !ok {
+			slog.Debug("TerminateWorkerInstances: instance already gone", "instanceId", id)
+			continue
+		}
+		cmd := spxtypes.EC2InstanceCommand{
+			ID:         id,
+			Attributes: spxtypes.EC2CommandAttributes{TerminateInstance: true},
+		}
+		if err := d.instanceService.StopOrTerminateInstance(instance, cmd); err != nil {
+			errs = append(errs, fmt.Errorf("terminate worker %s: %w", id, err))
+		}
+	}
+	if len(errs) > 0 {
+		return errors.Join(errs...)
+	}
+	return nil
+}

spinifex/daemon/eks_worker_launch_test.go

@@ -0,0 +1,37 @@
+package daemon
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
+	"github.com/stretchr/testify/require"
+)
+
+// A worker whose ID is not tracked by the local vmMgr is already-gone, so
+// terminating it is an idempotent no-op success (a retried DeleteNodegroup /
+// scale-down must not wedge on drained instances).
+func TestTerminateWorkerInstances_NotFoundIsIdempotent(t *testing.T) {
+	d := newDaemonWithVMs()
+	// Non-nil service passes the init guard; no VM matches the IDs so
+	// StopOrTerminateInstance is never reached — the not-found branch returns
+	// success.
+	d.instanceService = &handlers_ec2_instance.InstanceServiceImpl{}
+
+	require.NoError(t, d.TerminateWorkerInstances([]string{"i-gone1", "i-gone2"}, "111122223333"))
+	// Empty / blank IDs are skipped.
+	require.NoError(t, d.TerminateWorkerInstances([]string{"", ""}, "111122223333"))
+	require.NoError(t, d.TerminateWorkerInstances(nil, "111122223333"))
+}
+
+// Both worker methods must guard against an uninitialized instance service
+// rather than nil-panic.
+func TestWorkerLauncher_NilInstanceService(t *testing.T) {
+	d := &Daemon{}
+
+	_, err := d.RunWorkerInstance(&ec2.RunInstancesInput{ImageId: aws.String("ami-1")}, "111122223333")
+	require.Error(t, err)
+
+	require.Error(t, d.TerminateWorkerInstances([]string{"i-1"}, "111122223333"))
+}

spinifex/handlers/eks/cluster_reconciler.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"log/slog"
 	"net/http"
+	"sync/atomic"
 	"time"
 
 	"github.com/nats-io/nats.go"
@@ -21,8 +22,18 @@ const (
 	// (absolute), so a VM that never boots reaches a terminal state even across
 	// daemon restarts. Plan Risk #1.
 	defaultCreateTimeout = 15 * time.Minute
+	// defaultStateStaleAfter bounds how old the control plane's last NATS state
+	// report may be before health is treated as unknown/failed. The CP publishes
+	// every ~30s; 3x that tolerates a missed tick without flapping.
+	defaultStateStaleAfter = 90 * time.Second
 )
 
+// natsSubscriber is the minimum surface the reconciler needs to consume the
+// control-plane state report; *nats.Conn satisfies it, tests stub it.
+type natsSubscriber interface {
+	Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error)
+}
+
 // ErrReconcilerLeaseLost is returned by Run when the leader lease was lost
 // mid-loop. Callers can re-Acquire and re-Run if they want to keep driving
 // reconcile from the same node.
@@ -58,6 +69,14 @@ type ClusterReconciler struct {
 	healthTimeout time.Duration
 	createTimeout time.Duration
 	httpClient    HTTPDoer
+
+	// State-report health source. When stateSub is non-nil the reconciler gates
+	// on the CP's NATS self-report (latest) rather than the HTTP /healthz probe —
+	// the apiserver is bound to the VPC node-ip and unreachable from the host.
+	stateSub        natsSubscriber
+	stateSubject    string
+	stateStaleAfter time.Duration
+	latest          atomic.Pointer[ServerStateReport]
 }
 
 // ReconcilerOption tunes a ClusterReconciler. Tests use the With* helpers to
@@ -91,6 +110,22 @@ func WithHTTPClient(c HTTPDoer) ReconcilerOption {
 	return func(r *ClusterReconciler) { r.httpClient = c }
 }
 
+// WithStateSource configures the reconciler to gate health on the control
+// plane's NATS state report (subject) consumed via sub, instead of the HTTP
+// /healthz probe. The subscription is opened in Run and closed when it returns.
+func WithStateSource(sub natsSubscriber, subject string) ReconcilerOption {
+	return func(r *ClusterReconciler) {
+		r.stateSub = sub
+		r.stateSubject = subject
+	}
+}
+
+// WithStateStaleAfter overrides how old the last state report may be before
+// health is treated as failing.
+func WithStateStaleAfter(d time.Duration) ReconcilerOption {
+	return func(r *ClusterReconciler) { r.stateStaleAfter = d }
+}
+
 // NewClusterReconciler constructs a ClusterReconciler. healthURL is the
 // fully-qualified probe target (e.g. https://nlb-dns:443/healthz) or empty
 // to skip /healthz probing (CREATING transitions on bootstrap state alone).
@@ -111,16 +146,17 @@ func NewClusterReconciler(leaderKV, acctKV nats.KeyValue, accountID, clusterName
 		return nil, errors.New("eks: NewClusterReconciler empty holderID")
 	}
 	r := &ClusterReconciler{
-		leaderKV:      leaderKV,
-		acctKV:        acctKV,
-		accountID:     accountID,
-		clusterName:   clusterName,
-		holderID:      holderID,
-		healthURL:     healthURL,
-		leaseRefresh:  defaultReconcileLeaseRefresh,
-		interval:      defaultReconcileInterval,
-		healthTimeout: defaultHealthzTimeout,
-		createTimeout: defaultCreateTimeout,
+		leaderKV:        leaderKV,
+		acctKV:          acctKV,
+		accountID:       accountID,
+		clusterName:     clusterName,
+		holderID:        holderID,
+		healthURL:       healthURL,
+		leaseRefresh:    defaultReconcileLeaseRefresh,
+		interval:        defaultReconcileInterval,
+		healthTimeout:   defaultHealthzTimeout,
+		createTimeout:   defaultCreateTimeout,
+		stateStaleAfter: defaultStateStaleAfter,
 		httpClient: &http.Client{
 			Timeout: defaultHealthzTimeout,
 			Transport: &http.Transport{
@@ -186,6 +222,22 @@ func (r *ClusterReconciler) RefreshLease() bool {
 // cluster reaches a terminal state. Caller is expected to AcquireLease first;
 // Run does not re-acquire.
 func (r *ClusterReconciler) Run(ctx context.Context) error {
+	if r.stateSub != nil && r.stateSubject != "" {
+		sub, err := r.stateSub.Subscribe(r.stateSubject, func(m *nats.Msg) {
+			report, perr := unmarshalServerStateReport(m.Data)
+			if perr != nil {
+				slog.Warn("ClusterReconciler: bad state report",
+					"cluster", r.clusterName, "subject", m.Subject, "err", perr)
+				return
+			}
+			r.latest.Store(&report)
+		})
+		if err != nil {
+			return fmt.Errorf("subscribe state report %s: %w", r.stateSubject, err)
+		}
+		defer func() { _ = sub.Unsubscribe() }()
+	}
+
 	refreshT := time.NewTicker(r.leaseRefresh)
 	defer refreshT.Stop()
 	reconcileT := time.NewTicker(r.interval)
@@ -238,24 +290,30 @@ func (r *ClusterReconciler) reconcileOnce(ctx context.Context) error {
 				"cluster", r.clusterName, "reason", reason)
 			return r.failIfCreateTimedOut(meta, "bootstrap not ready: "+reason)
 		}
-		if err := r.probeHealthz(ctx); err != nil {
-			slog.Debug("ClusterReconciler: /healthz still failing",
-				"cluster", r.clusterName, "err", err)
-			return r.failIfCreateTimedOut(meta, "healthz not ready: "+err.Error())
+		issue, nodeCount := r.observe(ctx)
+		if issue != "" {
+			slog.Debug("ClusterReconciler: health not ready",
+				"cluster", r.clusterName, "issue", issue)
+			return r.failIfCreateTimedOut(meta, "healthz not ready: "+issue)
 		}
 		if err := SetClusterStatus(r.acctKV, r.clusterName, ClusterStatusActive); err != nil {
 			return err
 		}
-		slog.Info("ClusterReconciler: cluster transitioned to ACTIVE",
-			"cluster", r.clusterName)
-	case ClusterStatusActive:
-		issue := ""
-		if err := r.probeHealthz(ctx); err != nil {
-			slog.Warn("ClusterReconciler: /healthz failing for ACTIVE cluster",
-				"cluster", r.clusterName, "err", err)
-			issue = err.Error()
+		if err := SetClusterHealthState(r.acctKV, r.clusterName, "", nodeCount); err != nil {
+			if errors.Is(err, ErrClusterNotFound) {
+				return ErrClusterNotFound
+			}
+			return fmt.Errorf("record cluster health: %w", err)
 		}
-		if err := SetClusterHealth(r.acctKV, r.clusterName, issue); err != nil {
+		slog.Info("ClusterReconciler: cluster transitioned to ACTIVE",
+			"cluster", r.clusterName, "nodes", nodeCount)
+	case ClusterStatusActive:
+		issue, nodeCount := r.observe(ctx)
+		if issue != "" {
+			slog.Warn("ClusterReconciler: health failing for ACTIVE cluster",
+				"cluster", r.clusterName, "issue", issue)
+		}
+		if err := SetClusterHealthState(r.acctKV, r.clusterName, issue, nodeCount); err != nil {
 			if errors.Is(err, ErrClusterNotFound) {
 				return ErrClusterNotFound
 			}
@@ -269,6 +327,31 @@ func (r *ClusterReconciler) reconcileOnce(ctx context.Context) error {
 	return nil
 }
 
+// observe returns the current health issue ("" when healthy) and the reported
+// node count. With a NATS state source it reads the control plane's latest
+// self-report (rejecting a stale or unhealthy one); otherwise it falls back to
+// the HTTP /healthz probe (node count is then 0, the probe carries no count).
+func (r *ClusterReconciler) observe(ctx context.Context) (issue string, nodeCount int) {
+	if r.stateSub != nil && r.stateSubject != "" {
+		report := r.latest.Load()
+		if report == nil {
+			return "no control-plane state report received yet", 0
+		}
+		age := time.Since(time.Unix(report.TS, 0))
+		if age > r.stateStaleAfter {
+			return fmt.Sprintf("control-plane state report stale (%s old)", age.Round(time.Second)), report.NodeCount
+		}
+		if !report.Healthy() {
+			return fmt.Sprintf("apiserver healthz=%q", report.Healthz), report.NodeCount
+		}
+		return "", report.NodeCount
+	}
+	if err := r.probeHealthz(ctx); err != nil {
+		return err.Error(), 0
+	}
+	return "", 0
+}
+
 // failIfCreateTimedOut transitions a still-CREATING cluster to FAILED once it
 // has exceeded createTimeout since meta.CreatedAt, and returns
 // ErrReconcilerClusterFailed so Run exits the loop. While still within the

spinifex/handlers/eks/cluster_reconciler_state_test.go

@@ -0,0 +1,154 @@
+package handlers_eks
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newStateReconcilerHarness builds a reconciler wired to a real NATS conn as its
+// state source (so observe() reads r.latest) and returns the conn + account KV.
+// It does NOT start a subscription — tests drive r.latest directly for
+// determinism; the integration test below exercises the real Subscribe path.
+func newStateReconcilerHarness(t *testing.T, opts ...ReconcilerOption) (*ClusterReconciler, *nats.Conn, nats.KeyValue) {
+	t.Helper()
+	_, nc, js := testutil.StartTestJetStream(t)
+	leaderKV, err := InitLeaderBucket(js)
+	require.NoError(t, err)
+	acctKV, err := GetOrCreateAccountBucket(js, testAccountID)
+	require.NoError(t, err)
+	require.NoError(t, PutClusterMeta(acctKV, sampleClusterMeta("alpha")))
+
+	base := []ReconcilerOption{WithStateSource(nc, StateSubject(testAccountID, "alpha"))}
+	r, err := NewClusterReconciler(leaderKV, acctKV, testAccountID, "alpha", "holder-1", "", append(base, opts...)...)
+	require.NoError(t, err)
+	return r, nc, acctKV
+}
+
+func freshReport(healthz string, nodes int) *ServerStateReport {
+	return &ServerStateReport{Healthz: healthz, NodeCount: nodes, TS: time.Now().Unix()}
+}
+
+func TestClusterReconciler_CreatingTransitionsToActiveOnStateReport(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t)
+	freshenClusterCreatedAt(t, acctKV)
+	seedBootstrapState(t, acctKV)
+	r.latest.Store(freshReport("ok", 3))
+
+	require.NoError(t, r.reconcileOnce(context.Background()))
+
+	meta, err := GetClusterMeta(acctKV, "alpha")
+	require.NoError(t, err)
+	assert.Equal(t, ClusterStatusActive, meta.Status)
+	assert.Equal(t, 3, meta.NodeCount)
+	assert.Empty(t, meta.HealthIssue)
+}
+
+func TestClusterReconciler_CreatingStaysWithoutStateReport(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t)
+	freshenClusterCreatedAt(t, acctKV)
+	seedBootstrapState(t, acctKV)
+	// No report stored.
+
+	require.NoError(t, r.reconcileOnce(context.Background()))
+
+	meta, err := GetClusterMeta(acctKV, "alpha")
+	require.NoError(t, err)
+	assert.Equal(t, ClusterStatusCreating, meta.Status, "no state report yet → stays CREATING")
+}
+
+func TestClusterReconciler_CreatingStaysOnUnhealthyReport(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t)
+	freshenClusterCreatedAt(t, acctKV)
+	seedBootstrapState(t, acctKV)
+	r.latest.Store(freshReport("fail", 0))
+
+	require.NoError(t, r.reconcileOnce(context.Background()))
+
+	meta, err := GetClusterMeta(acctKV, "alpha")
+	require.NoError(t, err)
+	assert.Equal(t, ClusterStatusCreating, meta.Status, "unhealthy apiserver → stays CREATING")
+}
+
+func TestClusterReconciler_ActiveRecordsNodeCountAndClearsIssue(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t)
+	require.NoError(t, SetClusterStatus(acctKV, "alpha", ClusterStatusActive))
+	require.NoError(t, SetClusterHealthState(acctKV, "alpha", "stale", 0))
+	r.latest.Store(freshReport("ok", 4))
+
+	require.NoError(t, r.reconcileOnce(context.Background()))
+
+	meta, err := GetClusterMeta(acctKV, "alpha")
+	require.NoError(t, err)
+	assert.Equal(t, ClusterStatusActive, meta.Status)
+	assert.Equal(t, 4, meta.NodeCount)
+	assert.Empty(t, meta.HealthIssue, "healthy report clears the prior issue")
+}
+
+func TestClusterReconciler_ActiveFlagsStaleReport(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t, WithStateStaleAfter(90*time.Second))
+	require.NoError(t, SetClusterStatus(acctKV, "alpha", ClusterStatusActive))
+	r.latest.Store(&ServerStateReport{Healthz: "ok", NodeCount: 2, TS: time.Now().Add(-5 * time.Minute).Unix()})
+
+	require.NoError(t, r.reconcileOnce(context.Background()))
+
+	meta, err := GetClusterMeta(acctKV, "alpha")
+	require.NoError(t, err)
+	assert.Contains(t, meta.HealthIssue, "stale", "report older than stale window flags an issue")
+	assert.Equal(t, 2, meta.NodeCount, "last known node count still recorded")
+}
+
+func TestClusterReconciler_StateReportSubscriptionDrivesActive(t *testing.T) {
+	r, nc, acctKV := newStateReconcilerHarness(t,
+		WithReconcileInterval(10*time.Millisecond),
+		WithLeaseRefresh(10*time.Second),
+	)
+	freshenClusterCreatedAt(t, acctKV)
+	seedBootstrapState(t, acctKV)
+
+	release, ok := r.AcquireLease()
+	require.True(t, ok)
+	defer release()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	runErr := make(chan error, 1)
+	go func() { runErr <- r.Run(ctx) }()
+
+	// Publish healthy reports until the cluster goes ACTIVE (core NATS only
+	// delivers post-subscribe, so keep emitting while Run's subscription opens).
+	stop := make(chan struct{})
+	defer close(stop)
+	go func() {
+		payload, _ := json.Marshal(ServerStateReport{Healthz: "ok", NodeCount: 3, TS: time.Now().Unix()})
+		tick := time.NewTicker(20 * time.Millisecond)
+		defer tick.Stop()
+		for {
+			select {
+			case <-stop:
+				return
+			case <-tick.C:
+				_ = nc.Publish(StateSubject(testAccountID, "alpha"), payload)
+			}
+		}
+	}()
+
+	require.Eventually(t, func() bool {
+		meta, err := GetClusterMeta(acctKV, "alpha")
+		return err == nil && meta.Status == ClusterStatusActive && meta.NodeCount == 3
+	}, 1500*time.Millisecond, 10*time.Millisecond, "state reports should drive CREATING → ACTIVE")
+
+	cancel()
+	select {
+	case err := <-runErr:
+		assert.ErrorIs(t, err, context.Canceled)
+	case <-time.After(2 * time.Second):
+		t.Fatal("Run did not return after cancel")
+	}
+}

spinifex/handlers/eks/cluster_state.go

@@ -69,6 +69,10 @@ type ClusterMeta struct {
 	// only) and never treats a time.Time struct as empty, so this field always
 	// serializes. The tag states that honestly rather than implying it elides.
 	LastHealthProbe time.Time `json:"lastHealthProbe"`
+	// NodeCount is the node total from the control plane's last NATS state
+	// report (server + joined agents). Surfaces nodegroup join/scale progress
+	// without the host needing apiserver reachability.
+	NodeCount int `json:"nodeCount,omitempty"`
 }
 
 // ErrClusterNotFound is returned by GetClusterMeta / SetClusterStatus /
@@ -231,6 +235,27 @@ func SetClusterHealth(kv nats.KeyValue, name, issue string) error {
 	})
 }
 
+// SetClusterHealthState records the latest health outcome together with the
+// node count from the control-plane state report. issue is the failure reason
+// (empty clears it / marks healthy). LastHealthProbe is stamped whenever health
+// or node count changes; a steady-state healthy cluster with an unchanging node
+// count produces no KV write. Returns ErrClusterNotFound if the meta record was
+// deleted underneath us.
+func SetClusterHealthState(kv nats.KeyValue, name, issue string, nodeCount int) error {
+	if name == "" {
+		return errors.New("eks: SetClusterHealthState empty name")
+	}
+	return casUpdateMeta(kv, name, func(m *ClusterMeta) bool {
+		if m.HealthIssue == issue && m.NodeCount == nodeCount {
+			return false
+		}
+		m.HealthIssue = issue
+		m.NodeCount = nodeCount
+		m.LastHealthProbe = time.Now().UTC()
+		return true
+	})
+}
+
 // DeleteClusterPrefix removes every KV key under clusters/{name}/. Called
 // from DeleteCluster after the OIDC private key has been zeroized.
 // Returns the first delete error encountered but continues sweeping so

spinifex/handlers/eks/delete_cluster_test.go

@@ -27,13 +27,15 @@ func deleteInput(name string) *eks.DeleteClusterInput {
 // and an embedded JetStream KV. Tests poke the fakes' *Err fields to force
 // failures at specific lifecycle steps.
 type eksServiceFixture struct {
-	svc  *EKSServiceImpl
-	kv   nats.KeyValue
-	nlb  *fakeNLBProvisioner
-	inst *fakeK3sInst
-	vpc  *fakeK3sVPC
-	ami  *fakeK3sAMI
-	eip  *fakeEIPProvisioner
+	svc    *EKSServiceImpl
+	kv     nats.KeyValue
+	nlb    *fakeNLBProvisioner
+	inst   *fakeK3sInst
+	vpc    *fakeK3sVPC
+	ami    *fakeK3sAMI
+	eip    *fakeEIPProvisioner
+	sg     *fakeSGProvisioner
+	worker *fakeWorkerLauncher
 }
 
 func newEKSServiceFixture(t *testing.T) *eksServiceFixture {
@@ -47,6 +49,8 @@ func newEKSServiceFixture(t *testing.T) *eksServiceFixture {
 	vpc := &fakeK3sVPC{}
 	ami := &fakeK3sAMI{}
 	eip := newFakeEIPProvisioner()
+	sg := newFakeSGProvisioner()
+	worker := newFakeWorkerLauncher()
 
 	svc, err := NewEKSServiceImpl(EKSServiceDeps{
 		NATSConn:       nc,
@@ -57,18 +61,19 @@ func newEKSServiceFixture(t *testing.T) *eksServiceFixture {
 		NATSURL:        "nats://gw.local:4222",
 		NATSToken:      "s3cr3t-token",
 		NATSCACert:     "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n",
-		VPCSG:          newFakeSGProvisioner(),
+		VPCSG:          sg,
 		VPCK3s:         vpc,
 		VPCSubnet:      fakeSubnetResolver{},
 		NLB:            nlb,
 		Instance:       inst,
 		Image:          ami,
 		EIP:            eip,
+		Worker:         worker,
 	})
 	require.NoError(t, err)
 	t.Cleanup(svc.Shutdown)
 
-	return &eksServiceFixture{svc: svc, kv: kv, nlb: nlb, inst: inst, vpc: vpc, ami: ami, eip: eip}
+	return &eksServiceFixture{svc: svc, kv: kv, nlb: nlb, inst: inst, vpc: vpc, ami: ami, eip: eip, sg: sg, worker: worker}
 }
 
 // deleteClusterFixture is an eksServiceFixture pre-seeded with a CREATING

spinifex/handlers/eks/k3s_server_vm.go

@@ -71,6 +71,11 @@ const (
 	// (see scripts/images/eks-server/k3s-first-boot.sh ENVFILE).
 	k3sFirstBootEnvPath = "/etc/spinifex-eks/first-boot.env"
 
+	// agentEnvPath is the env file the k3s-agent OpenRC service sources for its
+	// K3S_URL/K3S_TOKEN/K3S_NODE_NAME/K3S_NODE_LABEL. Path matches the AGENT_ENVFILE
+	// in scripts/images/eks-node/eks-node-role.sh and k3s-agent.initd.
+	agentEnvPath = "/etc/spinifex-eks/agent.env"
+
 	// k3sNATSCAPath is the on-VM destination for the NATS CA cert PEM. The K3s
 	// VM uses it to verify the daemon's NATS TLS when publishing bootstrap
 	// messages. Path matches k3s-first-boot.sh SPINIFEX_NATS_CA.

spinifex/handlers/eks/nodegroup.go

@@ -0,0 +1,605 @@
+package handlers_eks
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"maps"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/aws/aws-sdk-go/service/eks"
+	"github.com/google/uuid"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
+	"github.com/nats-io/nats.go"
+)
+
+// defaultNodegroupInstanceType mirrors the AWS EKS managed-nodegroup default
+// when the caller omits instanceTypes.
+const defaultNodegroupInstanceType = "t3.medium"
+
+// ErrNodegroupNotFound is returned by GetNodegroupRecord when the record key is
+// absent. Callers translate to the AWS shape (ResourceNotFoundException) at the
+// service boundary.
+var ErrNodegroupNotFound = errors.New("eks: nodegroup not found")
+
+// NodegroupARN composes a nodegroup ARN matching the AWS shape
+// (.../nodegroup/{cluster}/{ng}/{uuid}). The trailing UUID is the per-nodegroup
+// discriminator AWS appends; it is generated once at create time and persisted.
+func NodegroupARN(region, accountID, cluster, ng, id string) string {
+	return fmt.Sprintf("arn:aws:eks:%s:%s:nodegroup/%s/%s/%s", region, accountID, cluster, ng, id)
+}
+
+// PutNodegroupRecord writes the record unconditionally.
+func PutNodegroupRecord(kv nats.KeyValue, rec *NodegroupRecord) error {
+	if rec == nil {
+		return errors.New("eks: PutNodegroupRecord nil record")
+	}
+	if rec.ClusterName == "" || rec.Name == "" {
+		return errors.New("eks: PutNodegroupRecord missing cluster or name")
+	}
+	data, err := json.Marshal(rec)
+	if err != nil {
+		return fmt.Errorf("marshal nodegroup %s: %w", rec.Name, err)
+	}
+	key := NodegroupKey(rec.ClusterName, rec.Name)
+	if _, err := kv.Put(key, data); err != nil {
+		return fmt.Errorf("kv put %s: %w", key, err)
+	}
+	return nil
+}
+
+// GetNodegroupRecord reads one record. Returns ErrNodegroupNotFound if absent.
+func GetNodegroupRecord(kv nats.KeyValue, cluster, ng string) (*NodegroupRecord, error) {
+	if cluster == "" || ng == "" {
+		return nil, errors.New("eks: GetNodegroupRecord empty cluster or name")
+	}
+	entry, err := kv.Get(NodegroupKey(cluster, ng))
+	if err != nil {
+		if errors.Is(err, nats.ErrKeyNotFound) {
+			return nil, ErrNodegroupNotFound
+		}
+		return nil, fmt.Errorf("kv get nodegroup: %w", err)
+	}
+	var rec NodegroupRecord
+	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+		return nil, fmt.Errorf("unmarshal nodegroup %s: %w", ng, err)
+	}
+	return &rec, nil
+}
+
+// ListNodegroupRecords returns every nodegroup record under a cluster, sorted
+// by name for stable output.
+func ListNodegroupRecords(kv nats.KeyValue, cluster string) ([]*NodegroupRecord, error) {
+	if cluster == "" {
+		return nil, errors.New("eks: ListNodegroupRecords empty cluster")
+	}
+	keys, err := kv.Keys()
+	if err != nil {
+		if errors.Is(err, nats.ErrNoKeysFound) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("kv keys: %w", err)
+	}
+	prefix := NodegroupsPrefix(cluster)
+	out := make([]*NodegroupRecord, 0)
+	for _, k := range keys {
+		if !strings.HasPrefix(k, prefix) {
+			continue
+		}
+		entry, err := kv.Get(k)
+		if err != nil {
+			if errors.Is(err, nats.ErrKeyNotFound) {
+				continue
+			}
+			return nil, fmt.Errorf("kv get %s: %w", k, err)
+		}
+		var rec NodegroupRecord
+		if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+			return nil, fmt.Errorf("unmarshal nodegroup %s: %w", k, err)
+		}
+		out = append(out, &rec)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
+	return out, nil
+}
+
+// DeleteNodegroupRecord removes one record. A missing key is a no-op so
+// DeleteNodegroup stays idempotent.
+func DeleteNodegroupRecord(kv nats.KeyValue, cluster, ng string) error {
+	key := NodegroupKey(cluster, ng)
+	if err := kv.Delete(key); err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
+		return fmt.Errorf("kv delete %s: %w", key, err)
+	}
+	return nil
+}
+
+// nodegroupAcctKV opens the per-account bucket for nodegroup handlers.
+func (s *EKSServiceImpl) nodegroupAcctKV(accountID string) (nats.KeyValue, error) {
+	js, err := s.deps.NATSConn.JetStream()
+	if err != nil {
+		return nil, fmt.Errorf("jetstream: %w", err)
+	}
+	acctKV, err := GetOrCreateAccountBucket(js, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("get account bucket: %w", err)
+	}
+	return acctKV, nil
+}
+
+func (s *EKSServiceImpl) createNodegroup(acctKV nats.KeyValue, input *eks.CreateNodegroupInput, accountID string) (*eks.CreateNodegroupOutput, error) {
+	if input == nil {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	cluster := aws.StringValue(input.ClusterName)
+	ng := aws.StringValue(input.NodegroupName)
+	if cluster == "" || ng == "" {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	subnets := aws.StringValueSlice(input.Subnets)
+	if len(subnets) == 0 {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+
+	meta, err := GetClusterMeta(acctKV, cluster)
+	if err != nil {
+		if errors.Is(err, ErrClusterNotFound) {
+			return nil, errors.New(awserrors.ErrorEKSResourceNotFound)
+		}
+		return nil, err
+	}
+	if meta.Status != ClusterStatusActive {
+		slog.Warn("createNodegroup: cluster not ACTIVE", "cluster", cluster, "status", meta.Status)
+		return nil, errors.New(awserrors.ErrorInvalidRequest)
+	}
+	if meta.ControlPlaneENIIP == "" {
+		slog.Error("createNodegroup: cluster has no control-plane ENI IP", "cluster", cluster)
+		return nil, errors.New(awserrors.ErrorInvalidRequest)
+	}
+	if meta.ResourcesVpcConfig == nil || meta.ResourcesVpcConfig.VpcId == "" {
+		slog.Error("createNodegroup: cluster has no VPC", "cluster", cluster)
+		return nil, errors.New(awserrors.ErrorInvalidRequest)
+	}
+
+	if _, err := GetNodegroupRecord(acctKV, cluster, ng); err == nil {
+		return nil, errors.New(awserrors.ErrorEKSResourceInUse)
+	} else if !errors.Is(err, ErrNodegroupNotFound) {
+		return nil, err
+	}
+
+	instanceTypes := aws.StringValueSlice(input.InstanceTypes)
+	if len(instanceTypes) == 0 {
+		instanceTypes = []string{defaultNodegroupInstanceType}
+	}
+	minSize, maxSize, desired := scalingFromInput(input.ScalingConfig)
+
+	amiType := aws.StringValue(input.AmiType)
+	if amiType == "" {
+		amiType = eks.AMITypesAl2X8664
+	}
+	version := aws.StringValue(input.Version)
+	if version == "" {
+		version = meta.Version
+	}
+
+	now := time.Now().UTC()
+	rec := &NodegroupRecord{
+		ClusterName:    cluster,
+		Name:           ng,
+		Arn:            NodegroupARN(s.deps.Region, accountID, cluster, ng, uuid.NewString()),
+		Status:         eks.NodegroupStatusCreating,
+		Subnets:        subnets,
+		InstanceTypes:  instanceTypes,
+		AMIType:        amiType,
+		DiskSize:       aws.Int64Value(input.DiskSize),
+		ScalingMin:     minSize,
+		ScalingMax:     maxSize,
+		ScalingDesired: desired,
+		Version:        version,
+		NodeRole:       aws.StringValue(input.NodeRole),
+		Labels:         aws.StringValueMap(input.Labels),
+		CreatedAt:      now,
+		ModifiedAt:     now,
+	}
+
+	cpSGID, ngSGID, err := EnsureClusterSGs(s.deps.VPCSG, accountID, cluster, meta.ResourcesVpcConfig.VpcId)
+	if err != nil {
+		return nil, fmt.Errorf("ensure cluster SGs: %w", err)
+	}
+	if err := EnsureNodegroupSGRules(s.deps.VPCSG, accountID, cluster, cpSGID, ngSGID); err != nil {
+		return nil, fmt.Errorf("ensure nodegroup SG rules: %w", err)
+	}
+
+	if err := PutNodegroupRecord(acctKV, rec); err != nil {
+		return nil, err
+	}
+
+	token, err := s.decryptNodeToken(acctKV, cluster)
+	if err != nil {
+		s.markNodegroupFailed(acctKV, cluster, ng, "decrypt node token: "+err.Error())
+		return nil, fmt.Errorf("decrypt node token: %w", err)
+	}
+
+	amiID, err := lookupEKSServerAMI(s.deps.Image, accountID)
+	if err != nil {
+		s.markNodegroupFailed(acctKV, cluster, ng, "resolve eks-node AMI: "+err.Error())
+		if errors.Is(err, ErrEKSServerAMINotFound) {
+			return nil, errors.New(awserrors.ErrorServiceUnavailable)
+		}
+		return nil, fmt.Errorf("resolve eks-node AMI: %w", err)
+	}
+
+	instanceIDs, err := s.launchWorkers(accountID, rec, meta, ngSGID, amiID, token, int(desired))
+	rec.InstanceIDs = instanceIDs
+	if err != nil {
+		// Persist whatever launched so DeleteNodegroup can reclaim it, then fail.
+		rec.Status = eks.NodegroupStatusCreateFailed
+		rec.StatusReason = "launch workers: " + err.Error()
+		rec.ModifiedAt = time.Now().UTC()
+		if perr := PutNodegroupRecord(acctKV, rec); perr != nil {
+			slog.Error("createNodegroup: persist CREATE_FAILED record", "cluster", cluster, "nodegroup", ng, "err", perr)
+		}
+		return nil, fmt.Errorf("launch workers: %w", err)
+	}
+
+	// v1 marks ACTIVE on successful RunInstances (instance-level), not on
+	// k8s node-Ready — node-Ready gating needs a kubectl probe from the
+	// reconciler and lands in a follow-up.
+	rec.Status = eks.NodegroupStatusActive
+	rec.ModifiedAt = time.Now().UTC()
+	if err := PutNodegroupRecord(acctKV, rec); err != nil {
+		return nil, err
+	}
+
+	return &eks.CreateNodegroupOutput{Nodegroup: nodegroupRecordToAWS(rec)}, nil
+}
+
+// launchWorkers issues count customer-owned RunInstances calls (one per worker
+// so each gets a distinct node name + node label in its user-data) and returns
+// the launched instance IDs. On a partial failure it returns the IDs that did
+// launch plus the error so the caller can persist them for teardown.
+func (s *EKSServiceImpl) launchWorkers(accountID string, rec *NodegroupRecord, meta *ClusterMeta, ngSGID, amiID, token string, count int) ([]string, error) {
+	instanceType := defaultNodegroupInstanceType
+	if len(rec.InstanceTypes) > 0 {
+		instanceType = rec.InstanceTypes[0]
+	}
+	subnet := rec.Subnets[0]
+
+	ids := make([]string, 0, count)
+	for i := range count {
+		shortID := uuid.NewString()[:8]
+		userData := buildAgentUserData(agentUserDataInput{
+			ClusterName:       rec.ClusterName,
+			NodegroupName:     rec.Name,
+			ControlPlaneENIIP: meta.ControlPlaneENIIP,
+			JoinToken:         token,
+			NodeName:          fmt.Sprintf("%s-%s-%s", rec.ClusterName, rec.Name, shortID),
+		})
+		runInput := &ec2.RunInstancesInput{
+			ImageId:          aws.String(amiID),
+			InstanceType:     aws.String(instanceType),
+			MinCount:         aws.Int64(1),
+			MaxCount:         aws.Int64(1),
+			SubnetId:         aws.String(subnet),
+			SecurityGroupIds: aws.StringSlice([]string{ngSGID}),
+			UserData:         aws.String(userData),
+			TagSpecifications: []*ec2.TagSpecification{{
+				ResourceType: aws.String("instance"),
+				Tags: []*ec2.Tag{
+					{Key: aws.String(clusterEKSClusterTagKey), Value: aws.String(rec.ClusterName)},
+					{Key: aws.String(clusterEKSNodegroupTagKey), Value: aws.String(rec.Name)},
+				},
+			}},
+		}
+		res, err := s.deps.Worker.RunWorkerInstance(runInput, accountID)
+		if err != nil {
+			return ids, fmt.Errorf("run worker %d/%d: %w", i+1, count, err)
+		}
+		for _, inst := range res.Instances {
+			if id := aws.StringValue(inst.InstanceId); id != "" {
+				ids = append(ids, id)
+			}
+		}
+	}
+	return ids, nil
+}
+
+func (s *EKSServiceImpl) describeNodegroup(acctKV nats.KeyValue, input *eks.DescribeNodegroupInput) (*eks.DescribeNodegroupOutput, error) {
+	cluster := aws.StringValue(input.ClusterName)
+	ng := aws.StringValue(input.NodegroupName)
+	if cluster == "" || ng == "" {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	rec, err := GetNodegroupRecord(acctKV, cluster, ng)
+	if err != nil {
+		if errors.Is(err, ErrNodegroupNotFound) {
+			return nil, errors.New(awserrors.ErrorEKSResourceNotFound)
+		}
+		return nil, err
+	}
+	return &eks.DescribeNodegroupOutput{Nodegroup: nodegroupRecordToAWS(rec)}, nil
+}
+
+func (s *EKSServiceImpl) listNodegroups(acctKV nats.KeyValue, input *eks.ListNodegroupsInput) (*eks.ListNodegroupsOutput, error) {
+	cluster := aws.StringValue(input.ClusterName)
+	if cluster == "" {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	if _, err := GetClusterMeta(acctKV, cluster); err != nil {
+		if errors.Is(err, ErrClusterNotFound) {
+			return nil, errors.New(awserrors.ErrorEKSResourceNotFound)
+		}
+		return nil, err
+	}
+	recs, err := ListNodegroupRecords(acctKV, cluster)
+	if err != nil {
+		return nil, err
+	}
+	names := make([]string, 0, len(recs))
+	for _, rec := range recs {
+		names = append(names, rec.Name)
+	}
+	return &eks.ListNodegroupsOutput{Nodegroups: aws.StringSlice(names)}, nil
+}
+
+func (s *EKSServiceImpl) updateNodegroupConfig(acctKV nats.KeyValue, input *eks.UpdateNodegroupConfigInput, accountID string) (*eks.UpdateNodegroupConfigOutput, error) {
+	if input == nil {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	cluster := aws.StringValue(input.ClusterName)
+	ng := aws.StringValue(input.NodegroupName)
+	if cluster == "" || ng == "" {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	meta, err := GetClusterMeta(acctKV, cluster)
+	if err != nil {
+		if errors.Is(err, ErrClusterNotFound) {
+			return nil, errors.New(awserrors.ErrorEKSResourceNotFound)
+		}
+		return nil, err
+	}
+	rec, err := GetNodegroupRecord(acctKV, cluster, ng)
+	if err != nil {
+		if errors.Is(err, ErrNodegroupNotFound) {
+			return nil, errors.New(awserrors.ErrorEKSResourceNotFound)
+		}
+		return nil, err
+	}
+
+	if input.ScalingConfig != nil {
+		if v := input.ScalingConfig.MinSize; v != nil {
+			rec.ScalingMin = *v
+		}
+		if v := input.ScalingConfig.MaxSize; v != nil {
+			rec.ScalingMax = *v
+		}
+		if v := input.ScalingConfig.DesiredSize; v != nil {
+			rec.ScalingDesired = *v
+		}
+	}
+	if input.Labels != nil {
+		rec.Labels = applyLabelUpdate(rec.Labels, input.Labels)
+	}
+
+	if err := s.reconcileWorkerCount(acctKV, accountID, rec, meta); err != nil {
+		// reconcileWorkerCount may have launched some workers before failing; their
+		// IDs are already on rec. Persist so DeleteNodegroup can reclaim them rather
+		// than leaking the partial scale-up.
+		rec.ModifiedAt = time.Now().UTC()
+		if perr := PutNodegroupRecord(acctKV, rec); perr != nil {
+			slog.Error("updateNodegroupConfig: persist after reconcile failure",
+				"cluster", cluster, "nodegroup", ng, "err", perr)
+		}
+		return nil, err
+	}
+
+	rec.ModifiedAt = time.Now().UTC()
+	if err := PutNodegroupRecord(acctKV, rec); err != nil {
+		return nil, err
+	}
+
+	return &eks.UpdateNodegroupConfigOutput{Update: &eks.Update{
+		Id:        aws.String(uuid.NewString()),
+		Status:    aws.String(eks.UpdateStatusSuccessful),
+		Type:      aws.String(eks.UpdateTypeConfigUpdate),
+		CreatedAt: aws.Time(rec.ModifiedAt),
+	}}, nil
+}
+
+// reconcileWorkerCount launches or terminates workers so len(InstanceIDs)
+// matches ScalingDesired. Surplus removal terminates the last (highest-index)
+// instance IDs so scale-down is deterministic.
+func (s *EKSServiceImpl) reconcileWorkerCount(acctKV nats.KeyValue, accountID string, rec *NodegroupRecord, meta *ClusterMeta) error {
+	current := len(rec.InstanceIDs)
+	desired := int(rec.ScalingDesired)
+
+	switch {
+	case desired > current:
+		token, err := s.decryptNodeToken(acctKV, rec.ClusterName)
+		if err != nil {
+			return fmt.Errorf("decrypt node token: %w", err)
+		}
+		amiID, err := lookupEKSServerAMI(s.deps.Image, accountID)
+		if err != nil {
+			if errors.Is(err, ErrEKSServerAMINotFound) {
+				return errors.New(awserrors.ErrorServiceUnavailable)
+			}
+			return fmt.Errorf("resolve eks-node AMI: %w", err)
+		}
+		_, ngSGID, err := EnsureClusterSGs(s.deps.VPCSG, accountID, rec.ClusterName, meta.ResourcesVpcConfig.VpcId)
+		if err != nil {
+			return fmt.Errorf("ensure cluster SGs: %w", err)
+		}
+		newIDs, err := s.launchWorkers(accountID, rec, meta, ngSGID, amiID, token, desired-current)
+		rec.InstanceIDs = append(rec.InstanceIDs, newIDs...)
+		if err != nil {
+			return fmt.Errorf("launch workers: %w", err)
+		}
+	case desired < current:
+		surplus := rec.InstanceIDs[desired:]
+		if err := s.deps.Worker.TerminateWorkerInstances(surplus, accountID); err != nil {
+			return fmt.Errorf("terminate surplus workers: %w", err)
+		}
+		rec.InstanceIDs = rec.InstanceIDs[:desired]
+	}
+	return nil
+}
+
+func (s *EKSServiceImpl) updateNodegroupVersion(input *eks.UpdateNodegroupVersionInput) (*eks.UpdateNodegroupVersionOutput, error) {
+	// Worker AMI version upgrades (drain + replace) land in a follow-up; v1
+	// nodegroups are fixed at their create-time version.
+	if input != nil {
+		slog.Info("UpdateNodegroupVersion not implemented in v1",
+			"cluster", aws.StringValue(input.ClusterName), "nodegroup", aws.StringValue(input.NodegroupName))
+	}
+	return nil, notImpl()
+}
+
+func (s *EKSServiceImpl) deleteNodegroup(acctKV nats.KeyValue, input *eks.DeleteNodegroupInput, accountID string) (*eks.DeleteNodegroupOutput, error) {
+	if input == nil {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	cluster := aws.StringValue(input.ClusterName)
+	ng := aws.StringValue(input.NodegroupName)
+	if cluster == "" || ng == "" {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	rec, err := GetNodegroupRecord(acctKV, cluster, ng)
+	if err != nil {
+		if errors.Is(err, ErrNodegroupNotFound) {
+			return nil, errors.New(awserrors.ErrorEKSResourceNotFound)
+		}
+		return nil, err
+	}
+
+	if len(rec.InstanceIDs) > 0 {
+		if err := s.deps.Worker.TerminateWorkerInstances(rec.InstanceIDs, accountID); err != nil {
+			return nil, fmt.Errorf("terminate workers: %w", err)
+		}
+	}
+	if err := DeleteNodegroupRecord(acctKV, cluster, ng); err != nil {
+		return nil, err
+	}
+
+	rec.Status = eks.NodegroupStatusDeleting
+	return &eks.DeleteNodegroupOutput{Nodegroup: nodegroupRecordToAWS(rec)}, nil
+}
+
+// decryptNodeToken reads + decrypts the cluster's K3s join token from KV.
+func (s *EKSServiceImpl) decryptNodeToken(kv nats.KeyValue, cluster string) (string, error) {
+	if kv == nil {
+		return "", errors.New("eks: nil KV for node token")
+	}
+	entry, err := kv.Get(NodeTokenKey(cluster))
+	if err != nil {
+		if errors.Is(err, nats.ErrKeyNotFound) {
+			return "", errors.New("eks: cluster join token not yet provisioned")
+		}
+		return "", fmt.Errorf("kv get node token: %w", err)
+	}
+	token, err := handlers_iam.DecryptSecret(string(entry.Value()), s.deps.MasterKey)
+	if err != nil {
+		return "", fmt.Errorf("decrypt node token: %w", err)
+	}
+	if token == "" {
+		return "", errors.New("eks: decrypted node token is empty")
+	}
+	return token, nil
+}
+
+func (s *EKSServiceImpl) markNodegroupFailed(kv nats.KeyValue, cluster, ng, reason string) {
+	rec, err := GetNodegroupRecord(kv, cluster, ng)
+	if err != nil {
+		return
+	}
+	rec.Status = eks.NodegroupStatusCreateFailed
+	rec.StatusReason = reason
+	rec.ModifiedAt = time.Now().UTC()
+	if err := PutNodegroupRecord(kv, rec); err != nil {
+		slog.Warn("markNodegroupFailed: persist failed", "cluster", cluster, "nodegroup", ng, "err", err)
+	}
+}
+
+// scalingFromInput derives min/max/desired from a NodegroupScalingConfig,
+// defaulting to a single node when the caller omits the config or fields.
+func scalingFromInput(cfg *eks.NodegroupScalingConfig) (minSize, maxSize, desired int64) {
+	minSize, maxSize, desired = 1, 1, 1
+	if cfg == nil {
+		return minSize, maxSize, desired
+	}
+	if cfg.MinSize != nil {
+		minSize = *cfg.MinSize
+	}
+	if cfg.DesiredSize != nil {
+		desired = *cfg.DesiredSize
+	} else {
+		desired = minSize
+	}
+	if cfg.MaxSize != nil {
+		maxSize = *cfg.MaxSize
+	} else if desired > maxSize {
+		maxSize = desired
+	}
+	return minSize, maxSize, desired
+}
+
+// applyLabelUpdate applies an UpdateLabelsPayload (addOrUpdate + removeLabels)
+// onto the record's label map.
+func applyLabelUpdate(existing map[string]string, payload *eks.UpdateLabelsPayload) map[string]string {
+	if payload == nil {
+		return existing
+	}
+	out := map[string]string{}
+	maps.Copy(out, existing)
+	for k, v := range payload.AddOrUpdateLabels {
+		out[k] = aws.StringValue(v)
+	}
+	for _, k := range payload.RemoveLabels {
+		delete(out, aws.StringValue(k))
+	}
+	return out
+}
+
+func nodegroupRecordToAWS(rec *NodegroupRecord) *eks.Nodegroup {
+	if rec == nil {
+		return nil
+	}
+	out := &eks.Nodegroup{
+		ClusterName:   aws.String(rec.ClusterName),
+		NodegroupName: aws.String(rec.Name),
+		NodegroupArn:  aws.String(rec.Arn),
+		Status:        aws.String(rec.Status),
+		Subnets:       aws.StringSlice(rec.Subnets),
+		InstanceTypes: aws.StringSlice(rec.InstanceTypes),
+		AmiType:       aws.String(rec.AMIType),
+		NodeRole:      aws.String(rec.NodeRole),
+		Version:       aws.String(rec.Version),
+		CreatedAt:     aws.Time(rec.CreatedAt),
+		ModifiedAt:    aws.Time(rec.ModifiedAt),
+		ScalingConfig: &eks.NodegroupScalingConfig{
+			MinSize:     aws.Int64(rec.ScalingMin),
+			MaxSize:     aws.Int64(rec.ScalingMax),
+			DesiredSize: aws.Int64(rec.ScalingDesired),
+		},
+		Resources: &eks.NodegroupResources{},
+	}
+	if rec.DiskSize > 0 {
+		out.DiskSize = aws.Int64(rec.DiskSize)
+	}
+	if len(rec.Labels) > 0 {
+		out.Labels = aws.StringMap(rec.Labels)
+	}
+	if rec.StatusReason != "" {
+		out.Health = &eks.NodegroupHealth{Issues: []*eks.Issue{{
+			Code:        aws.String(eks.NodegroupIssueCodeNodeCreationFailure),
+			Message:     aws.String(rec.StatusReason),
+			ResourceIds: aws.StringSlice(rec.InstanceIDs),
+		}}}
+	}
+	return out
+}

spinifex/handlers/eks/nodegroup_test.go

@@ -0,0 +1,321 @@
+package handlers_eks
+
+import (
+	"errors"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/aws/aws-sdk-go/service/eks"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// fakeWorkerLauncher records RunWorkerInstance / TerminateWorkerInstances calls
+// and hands back sequential instance IDs so tests can assert on the IDs a
+// nodegroup tracks.
+type fakeWorkerLauncher struct {
+	runCalls       []*ec2.RunInstancesInput
+	terminateCalls [][]string
+
+	nextID  int
+	runErr  error
+	termErr error
+}
+
+var _ WorkerLauncher = (*fakeWorkerLauncher)(nil)
+
+func newFakeWorkerLauncher() *fakeWorkerLauncher {
+	return &fakeWorkerLauncher{}
+}
+
+func (f *fakeWorkerLauncher) RunWorkerInstance(input *ec2.RunInstancesInput, _ string) (*ec2.Reservation, error) {
+	f.runCalls = append(f.runCalls, input)
+	if f.runErr != nil {
+		return nil, f.runErr
+	}
+	f.nextID++
+	id := "i-worker" + strconv.Itoa(f.nextID)
+	return &ec2.Reservation{Instances: []*ec2.Instance{{InstanceId: aws.String(id)}}}, nil
+}
+
+func (f *fakeWorkerLauncher) TerminateWorkerInstances(ids []string, _ string) error {
+	f.terminateCalls = append(f.terminateCalls, ids)
+	return f.termErr
+}
+
+// seedActiveClusterWithToken lays down an ACTIVE cluster meta with the control-
+// plane ENI IP + VPC populated and an encrypted K3s join token, so the
+// nodegroup path can run end-to-end.
+func seedActiveClusterWithToken(t *testing.T, f *eksServiceFixture, cluster string) {
+	t.Helper()
+	meta := &ClusterMeta{
+		Name:              cluster,
+		Status:            ClusterStatusActive,
+		Version:           "1.32",
+		ControlPlaneENIIP: "10.0.1.42",
+		ResourcesVpcConfig: &ClusterVpcConfig{
+			SubnetIds: []string{"subnet-aaa"},
+			VpcId:     "vpc-aaa",
+		},
+	}
+	require.NoError(t, PutClusterMeta(f.kv, meta))
+
+	ct, err := handlers_iam.EncryptSecret("K10node-join-token::server:abc", bootstrapTestMasterKey)
+	require.NoError(t, err)
+	_, err = f.kv.Put(NodeTokenKey(cluster), []byte(ct))
+	require.NoError(t, err)
+}
+
+func createNGInput(cluster, ng string, desired int64) *eks.CreateNodegroupInput {
+	return &eks.CreateNodegroupInput{
+		ClusterName:   aws.String(cluster),
+		NodegroupName: aws.String(ng),
+		NodeRole:      aws.String("arn:aws:iam::111122223333:role/eks-node"),
+		Subnets:       aws.StringSlice([]string{"subnet-aaa"}),
+		ScalingConfig: &eks.NodegroupScalingConfig{
+			MinSize:     aws.Int64(1),
+			MaxSize:     aws.Int64(3),
+			DesiredSize: aws.Int64(desired),
+		},
+	}
+}
+
+func TestNodegroupRecord_CRUDRoundTrip(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	kv, err := GetOrCreateAccountBucket(js, testAccountID)
+	require.NoError(t, err)
+
+	rec := &NodegroupRecord{
+		ClusterName:    "c1",
+		Name:           "ng1",
+		Arn:            "arn:aws:eks:us-east-1:111122223333:nodegroup/c1/ng1/abc",
+		Status:         eks.NodegroupStatusActive,
+		Subnets:        []string{"subnet-aaa"},
+		InstanceTypes:  []string{"t3.medium"},
+		ScalingMin:     1,
+		ScalingMax:     3,
+		ScalingDesired: 2,
+		InstanceIDs:    []string{"i-1", "i-2"},
+	}
+	require.NoError(t, PutNodegroupRecord(kv, rec))
+
+	got, err := GetNodegroupRecord(kv, "c1", "ng1")
+	require.NoError(t, err)
+	assert.Equal(t, rec.InstanceIDs, got.InstanceIDs)
+	assert.Equal(t, int64(2), got.ScalingDesired)
+
+	list, err := ListNodegroupRecords(kv, "c1")
+	require.NoError(t, err)
+	require.Len(t, list, 1)
+	assert.Equal(t, "ng1", list[0].Name)
+
+	require.NoError(t, DeleteNodegroupRecord(kv, "c1", "ng1"))
+	_, err = GetNodegroupRecord(kv, "c1", "ng1")
+	assert.ErrorIs(t, err, ErrNodegroupNotFound)
+
+	// Delete of an absent record is idempotent.
+	require.NoError(t, DeleteNodegroupRecord(kv, "c1", "ng1"))
+}
+
+func TestBuildAgentUserData_Shape(t *testing.T) {
+	ud := buildAgentUserData(agentUserDataInput{
+		ClusterName:       "c1",
+		NodegroupName:     "ng1",
+		ControlPlaneENIIP: "10.0.1.42",
+		JoinToken:         "K10secret::server:xyz",
+		NodeName:          "c1-ng1-abc123de",
+	})
+
+	assert.True(t, strings.HasPrefix(ud, "#cloud-config\n"))
+	assert.Contains(t, ud, "SPINIFEX_K3S_ROLE=agent")
+	assert.Contains(t, ud, "K3S_URL=https://10.0.1.42:6443")
+	assert.Contains(t, ud, "K3S_TOKEN=K10secret::server:xyz")
+	assert.Contains(t, ud, "K3S_NODE_NAME=c1-ng1-abc123de")
+	assert.Contains(t, ud, "eks.amazonaws.com/nodegroup=ng1")
+	assert.Contains(t, ud, agentEnvPath)
+	// IMDS on-link route line present.
+	assert.Contains(t, ud, "ip route replace "+imdsServerIP+"/32")
+	// Exactly one write_files block (single "write_files:" key).
+	assert.Equal(t, 1, strings.Count(ud, "write_files:"))
+}
+
+func TestEnsureNodegroupSGRules_AuthorizesExpectedRules(t *testing.T) {
+	sg := newFakeSGProvisioner()
+
+	require.NoError(t, EnsureNodegroupSGRules(sg, testAccountID, "c1", "sg-cp", "sg-ng"))
+
+	require.Len(t, sg.authorizeCalls, 6)
+
+	// Verify the apiserver rule: 6443/tcp on cpSG sourced from ngSG.
+	var found6443 bool
+	for _, call := range sg.authorizeCalls {
+		require.Len(t, call.IpPermissions, 1)
+		perm := call.IpPermissions[0]
+		if aws.StringValue(call.GroupId) == "sg-cp" &&
+			aws.StringValue(perm.IpProtocol) == "tcp" &&
+			aws.Int64Value(perm.FromPort) == 6443 {
+			require.Len(t, perm.UserIdGroupPairs, 1)
+			assert.Equal(t, "sg-ng", aws.StringValue(perm.UserIdGroupPairs[0].GroupId))
+			assert.Empty(t, perm.IpRanges, "rules must be group-to-group, not CIDR")
+			found6443 = true
+		}
+	}
+	assert.True(t, found6443, "expected agent→apiserver 6443/tcp rule")
+}
+
+func TestEnsureNodegroupSGRules_DuplicateRuleTolerated(t *testing.T) {
+	sg := newFakeSGProvisioner()
+	sg.authorizeErr = errors.New(awserrors.ErrorInvalidPermissionDuplicate)
+
+	require.NoError(t, EnsureNodegroupSGRules(sg, testAccountID, "c1", "sg-cp", "sg-ng"),
+		"duplicate-rule error must be treated as success")
+}
+
+func TestCreateNodegroup_HappyPath(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	seedActiveClusterWithToken(t, f, "c1")
+
+	out, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 2), testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, out.Nodegroup)
+	assert.Equal(t, eks.NodegroupStatusActive, aws.StringValue(out.Nodegroup.Status))
+	assert.Equal(t, int64(2), aws.Int64Value(out.Nodegroup.ScalingConfig.DesiredSize))
+	assert.Contains(t, aws.StringValue(out.Nodegroup.NodegroupArn), ":nodegroup/c1/ng1/")
+
+	// Two workers launched and tracked.
+	assert.Len(t, f.worker.runCalls, 2)
+	rec, err := GetNodegroupRecord(f.kv, "c1", "ng1")
+	require.NoError(t, err)
+	assert.Len(t, rec.InstanceIDs, 2)
+
+	// Workers go through the customer path: no ManagedBy tag, SG = nodegroup SG.
+	for _, call := range f.worker.runCalls {
+		require.NotEmpty(t, call.SecurityGroupIds)
+		assert.Equal(t, "subnet-aaa", aws.StringValue(call.SubnetId))
+		for _, ts := range call.TagSpecifications {
+			for _, tg := range ts.Tags {
+				assert.NotEqual(t, "spinifex:managed-by", aws.StringValue(tg.Key))
+			}
+		}
+	}
+
+	// SG ingress rules were authorized.
+	assert.NotEmpty(t, f.sg.authorizeCalls)
+
+	// Duplicate create → ResourceInUseException.
+	_, err = f.svc.CreateNodegroup(createNGInput("c1", "ng1", 2), testAccountID)
+	require.EqualError(t, err, awserrors.ErrorEKSResourceInUse)
+}
+
+func TestCreateNodegroup_ClusterNotActive(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	require.NoError(t, PutClusterMeta(f.kv, &ClusterMeta{
+		Name:               "c1",
+		Status:             ClusterStatusCreating,
+		ControlPlaneENIIP:  "10.0.1.42",
+		ResourcesVpcConfig: &ClusterVpcConfig{VpcId: "vpc-aaa"},
+	}))
+
+	_, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 1), testAccountID)
+	require.EqualError(t, err, awserrors.ErrorInvalidRequest)
+}
+
+func TestCreateNodegroup_UnknownCluster(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	_, err := f.svc.CreateNodegroup(createNGInput("ghost", "ng1", 1), testAccountID)
+	require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound)
+}
+
+func TestDescribeAndListNodegroups(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	seedActiveClusterWithToken(t, f, "c1")
+	_, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 1), testAccountID)
+	require.NoError(t, err)
+
+	desc, err := f.svc.DescribeNodegroup(&eks.DescribeNodegroupInput{
+		ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1"),
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "ng1", aws.StringValue(desc.Nodegroup.NodegroupName))
+	assert.Equal(t, "c1", aws.StringValue(desc.Nodegroup.ClusterName))
+
+	list, err := f.svc.ListNodegroups(&eks.ListNodegroupsInput{ClusterName: aws.String("c1")}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"ng1"}, aws.StringValueSlice(list.Nodegroups))
+
+	_, err = f.svc.DescribeNodegroup(&eks.DescribeNodegroupInput{
+		ClusterName: aws.String("c1"), NodegroupName: aws.String("ghost"),
+	}, testAccountID)
+	require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound)
+}
+
+func TestUpdateNodegroupConfig_ScaleUpAndDown(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	seedActiveClusterWithToken(t, f, "c1")
+	_, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 1), testAccountID)
+	require.NoError(t, err)
+	require.Len(t, f.worker.runCalls, 1)
+
+	// Scale up 1 → 3: two new workers launched.
+	upOut, err := f.svc.UpdateNodegroupConfig(&eks.UpdateNodegroupConfigInput{
+		ClusterName:   aws.String("c1"),
+		NodegroupName: aws.String("ng1"),
+		ScalingConfig: &eks.NodegroupScalingConfig{DesiredSize: aws.Int64(3)},
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, eks.UpdateStatusSuccessful, aws.StringValue(upOut.Update.Status))
+	assert.Len(t, f.worker.runCalls, 3)
+	rec, err := GetNodegroupRecord(f.kv, "c1", "ng1")
+	require.NoError(t, err)
+	assert.Len(t, rec.InstanceIDs, 3)
+
+	// Scale down 3 → 1: surplus (last two) terminated.
+	_, err = f.svc.UpdateNodegroupConfig(&eks.UpdateNodegroupConfigInput{
+		ClusterName:   aws.String("c1"),
+		NodegroupName: aws.String("ng1"),
+		ScalingConfig: &eks.NodegroupScalingConfig{DesiredSize: aws.Int64(1)},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, f.worker.terminateCalls, 1)
+	assert.Len(t, f.worker.terminateCalls[0], 2)
+	rec, err = GetNodegroupRecord(f.kv, "c1", "ng1")
+	require.NoError(t, err)
+	assert.Len(t, rec.InstanceIDs, 1)
+}
+
+func TestDeleteNodegroup_TerminatesAndIsIdempotent(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	seedActiveClusterWithToken(t, f, "c1")
+	_, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 2), testAccountID)
+	require.NoError(t, err)
+
+	_, err = f.svc.DeleteNodegroup(&eks.DeleteNodegroupInput{
+		ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1"),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, f.worker.terminateCalls, 1)
+	assert.Len(t, f.worker.terminateCalls[0], 2)
+
+	_, err = GetNodegroupRecord(f.kv, "c1", "ng1")
+	assert.ErrorIs(t, err, ErrNodegroupNotFound)
+
+	// Second delete → ResourceNotFoundException (record already gone).
+	_, err = f.svc.DeleteNodegroup(&eks.DeleteNodegroupInput{
+		ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1"),
+	}, testAccountID)
+	require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound)
+}
+
+func TestUpdateNodegroupVersion_NotImplemented(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	_, err := f.svc.UpdateNodegroupVersion(&eks.UpdateNodegroupVersionInput{
+		ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1"),
+	}, testAccountID)
+	require.EqualError(t, err, awserrors.ErrorNotImplemented)
+}

spinifex/handlers/eks/nodegroup_userdata.go

@@ -0,0 +1,90 @@
+package handlers_eks
+
+import (
+	"fmt"
+	"net"
+	"strings"
+)
+
+// agentUserDataInput is the input shape for buildAgentUserData. ControlPlaneENIIP
+// is the cluster's in-VPC control-plane ENI private IP (the K3s join target);
+// JoinToken is the decrypted K3s node token.
+type agentUserDataInput struct {
+	ClusterName       string
+	NodegroupName     string
+	ControlPlaneENIIP string
+	JoinToken         string
+	NodeName          string
+}
+
+// buildAgentUserData renders the cloud-config YAML a nodegroup worker consumes
+// at first boot. It mirrors buildK3sUserData: a bootcmd resolver fix, a single
+// write_files block (first-boot.env + agent.env + the IMDS on-link route
+// script), and a runcmd that enables the OpenRC `local` service. Output is
+// unencoded YAML; the caller base64-wraps it for the RunInstances UserData
+// field. The eks-node-role first-boot selector reads SPINIFEX_K3S_ROLE=agent
+// (and the presence of agent.env) to enable the k3s-agent service, which sources
+// agent.env for K3S_URL/K3S_TOKEN.
+func buildAgentUserData(in agentUserDataInput) string {
+	// K3S_URL targets the control-plane ENI's in-VPC private IP on :6443
+	// because the cluster NLB DNS endpoint is not resolvable in-VPC yet.
+	// Workers share the VPC with the server, so this is directly reachable and
+	// TLS-pinned by the join token's CA hash (SAN/hostname irrelevant for the
+	// k3s agent join). Switch to the NLB endpoint on :443 once authoritative
+	// in-VPC DNS is available.
+	k3sURL := "https://" + net.JoinHostPort(in.ControlPlaneENIIP, "6443")
+	nodeLabel := "eks.amazonaws.com/nodegroup=" + in.NodegroupName
+
+	envBody := strings.Join([]string{
+		"SPINIFEX_K3S_ROLE=agent",
+	}, "\n")
+
+	agentBody := strings.Join([]string{
+		"K3S_URL=" + k3sURL,
+		"K3S_TOKEN=" + in.JoinToken,
+		"K3S_NODE_NAME=" + in.NodeName,
+		"K3S_NODE_LABEL=" + nodeLabel,
+	}, "\n")
+
+	files := []userDataFile{
+		{Path: k3sFirstBootEnvPath, Perms: "0600", Body: envBody},
+		{Path: agentEnvPath, Perms: "0600", Body: agentBody},
+		{
+			Path:  "/etc/local.d/imds-onlink-route.start",
+			Perms: "0755",
+			Body: "#!/bin/sh\n" +
+				"dev=$(ip route show default | awk '{print $5; exit}')\n" +
+				"[ -n \"$dev\" ] && ip route replace " + imdsServerIP + "/32 dev \"$dev\" scope link",
+		},
+	}
+
+	var buf strings.Builder
+	buf.WriteString("#cloud-config\n")
+
+	// Resolver fix via bootcmd (not write_files) for the same reason as the
+	// server path: /etc/resolv.conf is a dangling symlink on the Alpine AMI, so
+	// write_files would follow the dead link and abort the whole block.
+	// Containerd needs a working resolver to pull the worker's system images.
+	buf.WriteString("bootcmd:\n")
+	buf.WriteString("  - rm -f " + k3sResolvConfPath + "\n")
+	fmt.Fprintf(&buf, "  - printf '%s\\n' > %s\n",
+		strings.ReplaceAll(k3sResolvConf, "\n", "\\n"), k3sResolvConfPath)
+
+	buf.WriteString("write_files:\n")
+	for _, f := range files {
+		fmt.Fprintf(&buf, "  - path: %s\n", f.Path)
+		fmt.Fprintf(&buf, "    permissions: '%s'\n", f.Perms)
+		buf.WriteString("    content: |\n")
+		for line := range strings.SplitSeq(f.Body, "\n") {
+			buf.WriteString("      ")
+			buf.WriteString(line)
+			buf.WriteString("\n")
+		}
+	}
+
+	buf.WriteString("runcmd:\n")
+	buf.WriteString("  - [ rc-update, add, local, default ]\n")
+	buf.WriteString("  - [ rc-service, local, start ]\n")
+
+	return buf.String()
+}

spinifex/handlers/eks/security_groups.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/tags"
 )
 
@@ -17,6 +18,7 @@ type sgProvisioner interface {
 	CreateSecurityGroup(input *ec2.CreateSecurityGroupInput, accountID string) (*ec2.CreateSecurityGroupOutput, error)
 	DescribeSecurityGroups(input *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error)
 	DeleteSecurityGroup(input *ec2.DeleteSecurityGroupInput, accountID string) (*ec2.DeleteSecurityGroupOutput, error)
+	AuthorizeSecurityGroupIngress(input *ec2.AuthorizeSecurityGroupIngressInput, accountID string) (*ec2.AuthorizeSecurityGroupIngressOutput, error)
 }
 
 // clusterEKSClusterTagKey is stamped on every cluster-scoped resource so the
@@ -27,6 +29,10 @@ const clusterEKSClusterTagKey = "spinifex:eks-cluster"
 // SG when both live in the same VPC.
 const clusterEKSRoleTagKey = "spinifex:eks-role"
 
+// clusterEKSNodegroupTagKey is stamped on worker instances so the nodegroup
+// they belong to is recoverable from EC2 tags alone.
+const clusterEKSNodegroupTagKey = "spinifex:eks-nodegroup"
+
 const (
 	clusterEKSRoleControlPlane = "control-plane"
 	clusterEKSRoleNodegroup    = "nodegroup"
@@ -142,6 +148,66 @@ func ensureClusterSG(sgp sgProvisioner, accountID, vpcID, clusterName, name, des
 	return *out.GroupId, nil
 }
 
+// EnsureNodegroupSGRules authorizes the ingress rules a nodegroup's workers
+// need to join the K3s control plane and form the flannel overlay. All rules
+// are group-to-group (UserIdGroupPairs) so they stay correct as instances come
+// and go. Idempotent: the InvalidPermission.Duplicate error a re-run triggers
+// for an already-authorized rule is treated as success.
+//
+// Rules:
+//   - agent → server apiserver/supervisor: 6443/tcp (ngSG → cpSG)
+//   - flannel VXLAN: 8472/udp both directions between cpSG and ngSG, and
+//     ngSG ↔ ngSG
+//   - kubelet: 10250/tcp (cpSG → ngSG)
+//   - intra-nodegroup all traffic: ngSG self-reference
+func EnsureNodegroupSGRules(sgp sgProvisioner, accountID, clusterName, cpSGID, ngSGID string) error {
+	if cpSGID == "" || ngSGID == "" {
+		return errors.New("eks: EnsureNodegroupSGRules empty SG id")
+	}
+
+	type rule struct {
+		targetSG string
+		sourceSG string
+		proto    string
+		from     int64
+		to       int64
+		desc     string
+	}
+	rules := []rule{
+		{cpSGID, ngSGID, "tcp", 6443, 6443, "EKS agent to apiserver/supervisor"},
+		{cpSGID, ngSGID, "udp", 8472, 8472, "EKS flannel VXLAN node to control-plane"},
+		{ngSGID, cpSGID, "udp", 8472, 8472, "EKS flannel VXLAN control-plane to node"},
+		{ngSGID, ngSGID, "udp", 8472, 8472, "EKS flannel VXLAN node to node"},
+		{ngSGID, cpSGID, "tcp", 10250, 10250, "EKS kubelet control-plane to node"},
+		{ngSGID, ngSGID, "-1", -1, -1, "EKS intra-nodegroup all traffic"},
+	}
+
+	for _, r := range rules {
+		perm := &ec2.IpPermission{
+			IpProtocol: aws.String(r.proto),
+			UserIdGroupPairs: []*ec2.UserIdGroupPair{{
+				GroupId:     aws.String(r.sourceSG),
+				Description: aws.String(r.desc),
+			}},
+		}
+		if r.proto != "-1" {
+			perm.FromPort = aws.Int64(r.from)
+			perm.ToPort = aws.Int64(r.to)
+		}
+		_, err := sgp.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{
+			GroupId:       aws.String(r.targetSG),
+			IpPermissions: []*ec2.IpPermission{perm},
+		}, accountID)
+		if err != nil {
+			if awserrors.IsErrorCode(err, awserrors.ErrorInvalidPermissionDuplicate) {
+				continue
+			}
+			return fmt.Errorf("authorize %s ingress on %s: %w", r.desc, r.targetSG, err)
+		}
+	}
+	return nil
+}
+
 func lookupSGByName(sgp sgProvisioner, accountID, vpcID, name string) (string, error) {
 	out, err := sgp.DescribeSecurityGroups(&ec2.DescribeSecurityGroupsInput{
 		Filters: []*ec2.Filter{

spinifex/handlers/eks/security_groups_test.go

@@ -12,9 +12,10 @@ import (
 )
 
 type fakeSGProvisioner struct {
-	createCalls   []*ec2.CreateSecurityGroupInput
-	describeCalls []*ec2.DescribeSecurityGroupsInput
-	deleteCalls   []*ec2.DeleteSecurityGroupInput
+	createCalls    []*ec2.CreateSecurityGroupInput
+	describeCalls  []*ec2.DescribeSecurityGroupsInput
+	deleteCalls    []*ec2.DeleteSecurityGroupInput
+	authorizeCalls []*ec2.AuthorizeSecurityGroupIngressInput
 
 	// existing maps "name|vpcId" → groupId for DescribeSecurityGroups lookup.
 	existing map[string]string
@@ -24,9 +25,10 @@ type fakeSGProvisioner struct {
 	// nodegroup SG IDs.
 	createIDs []string
 
-	createErr   error
-	describeErr error
-	deleteErr   error
+	createErr    error
+	describeErr  error
+	deleteErr    error
+	authorizeErr error
 }
 
 var _ sgProvisioner = (*fakeSGProvisioner)(nil)
@@ -84,6 +86,14 @@ func (f *fakeSGProvisioner) DeleteSecurityGroup(input *ec2.DeleteSecurityGroupIn
 	return &ec2.DeleteSecurityGroupOutput{}, nil
 }
 
+func (f *fakeSGProvisioner) AuthorizeSecurityGroupIngress(input *ec2.AuthorizeSecurityGroupIngressInput, _ string) (*ec2.AuthorizeSecurityGroupIngressOutput, error) {
+	f.authorizeCalls = append(f.authorizeCalls, input)
+	if f.authorizeErr != nil {
+		return nil, f.authorizeErr
+	}
+	return &ec2.AuthorizeSecurityGroupIngressOutput{}, nil
+}
+
 func TestEnsureClusterSGs_EmptyInputsRejected(t *testing.T) {
 	sgp := newFakeSGProvisioner()
 

spinifex/handlers/eks/service_impl.go

@@ -55,6 +55,19 @@ type EKSServiceDeps struct {
 	Instance  k3sInstanceLauncher
 	Image     k3sAMIResolver
 	EIP       eipProvisioner
+	Worker    WorkerLauncher
+}
+
+// WorkerLauncher is the narrow EC2 surface CreateNodegroup needs to launch and
+// terminate nodegroup worker instances. Workers go through the normal
+// customer-owned RunInstances path (no ManagedBy tag, no mgmt NIC) so they are
+// visible in DescribeInstances and reclaimed by TerminateInstances like any
+// other customer EC2. The daemon adapts its concrete instance service onto this.
+// Exported so the daemon can assert *Daemon satisfies it with a compile-time
+// check.
+type WorkerLauncher interface {
+	RunWorkerInstance(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error)
+	TerminateWorkerInstances(instanceIDs []string, accountID string) error
 }
 
 // eipProvisioner is the narrow EIP surface CreateCluster needs to give the
@@ -226,6 +239,9 @@ func (s *EKSServiceImpl) missingOrchestrationDeps() []string {
 	if s.deps.EIP == nil {
 		missing = append(missing, "EIP")
 	}
+	if s.deps.Worker == nil {
+		missing = append(missing, "Worker")
+	}
 	if len(s.deps.MasterKey) == 0 {
 		missing = append(missing, "MasterKey")
 	}
@@ -682,28 +698,63 @@ func (s *EKSServiceImpl) UpdateClusterVersion(_ *eks.UpdateClusterVersionInput,
 
 // --- Nodegroup ---
 
-func (s *EKSServiceImpl) CreateNodegroup(_ *eks.CreateNodegroupInput, _ string) (*eks.CreateNodegroupOutput, error) {
-	return nil, notImpl()
+func (s *EKSServiceImpl) CreateNodegroup(input *eks.CreateNodegroupInput, accountID string) (*eks.CreateNodegroupOutput, error) {
+	if err := s.requireOrchestrationDeps("CreateNodegroup"); err != nil {
+		return nil, err
+	}
+	acctKV, err := s.nodegroupAcctKV(accountID)
+	if err != nil {
+		return nil, err
+	}
+	return s.createNodegroup(acctKV, input, accountID)
 }
 
-func (s *EKSServiceImpl) DescribeNodegroup(_ *eks.DescribeNodegroupInput, _ string) (*eks.DescribeNodegroupOutput, error) {
-	return nil, notImpl()
+func (s *EKSServiceImpl) DescribeNodegroup(input *eks.DescribeNodegroupInput, accountID string) (*eks.DescribeNodegroupOutput, error) {
+	if input == nil {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	acctKV, err := s.nodegroupAcctKV(accountID)
+	if err != nil {
+		return nil, err
+	}
+	return s.describeNodegroup(acctKV, input)
 }
 
-func (s *EKSServiceImpl) ListNodegroups(_ *eks.ListNodegroupsInput, _ string) (*eks.ListNodegroupsOutput, error) {
-	return nil, notImpl()
+func (s *EKSServiceImpl) ListNodegroups(input *eks.ListNodegroupsInput, accountID string) (*eks.ListNodegroupsOutput, error) {
+	if input == nil {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	acctKV, err := s.nodegroupAcctKV(accountID)
+	if err != nil {
+		return nil, err
+	}
+	return s.listNodegroups(acctKV, input)
 }
 
-func (s *EKSServiceImpl) UpdateNodegroupConfig(_ *eks.UpdateNodegroupConfigInput, _ string) (*eks.UpdateNodegroupConfigOutput, error) {
-	return nil, notImpl()
+func (s *EKSServiceImpl) UpdateNodegroupConfig(input *eks.UpdateNodegroupConfigInput, accountID string) (*eks.UpdateNodegroupConfigOutput, error) {
+	if err := s.requireOrchestrationDeps("UpdateNodegroupConfig"); err != nil {
+		return nil, err
+	}
+	acctKV, err := s.nodegroupAcctKV(accountID)
+	if err != nil {
+		return nil, err
+	}
+	return s.updateNodegroupConfig(acctKV, input, accountID)
 }
 
-func (s *EKSServiceImpl) UpdateNodegroupVersion(_ *eks.UpdateNodegroupVersionInput, _ string) (*eks.UpdateNodegroupVersionOutput, error) {
-	return nil, notImpl()
+func (s *EKSServiceImpl) UpdateNodegroupVersion(input *eks.UpdateNodegroupVersionInput, accountID string) (*eks.UpdateNodegroupVersionOutput, error) {
+	return s.updateNodegroupVersion(input)
 }
 
-func (s *EKSServiceImpl) DeleteNodegroup(_ *eks.DeleteNodegroupInput, _ string) (*eks.DeleteNodegroupOutput, error) {
-	return nil, notImpl()
+func (s *EKSServiceImpl) DeleteNodegroup(input *eks.DeleteNodegroupInput, accountID string) (*eks.DeleteNodegroupOutput, error) {
+	if err := s.requireOrchestrationDeps("DeleteNodegroup"); err != nil {
+		return nil, err
+	}
+	acctKV, err := s.nodegroupAcctKV(accountID)
+	if err != nil {
+		return nil, err
+	}
+	return s.deleteNodegroup(acctKV, input, accountID)
 }
 
 // --- AccessEntry + AccessPolicy ---
@@ -1120,21 +1171,7 @@ func (s *EKSServiceImpl) spawnBootstrap(accountID, clusterName string, kv nats.K
 	}()
 }
 
-func (s *EKSServiceImpl) spawnReconciler(accountID, clusterName string, meta *ClusterMeta) {
-	healthURL := ""
-	switch {
-	case meta.ControlPlaneMgmtIP != "":
-		// Workaround until authoritative DNS (Eclipso/Route53) lands: the
-		// reconciler runs on the host, outside the VPC overlay, and can neither
-		// resolve the NLB DNS name nor route to the VPC-internal NLB front-end.
-		// The control-plane VM's br-mgmt NIC is the one apiserver address
-		// reachable from the daemon, so probe :6443/healthz there directly. The
-		// probe client sets InsecureSkipVerify, so the mgmt IP need not be in the
-		// apiserver cert SAN.
-		healthURL = "https://" + net.JoinHostPort(meta.ControlPlaneMgmtIP, "6443") + "/healthz"
-	case meta.Endpoint != "":
-		healthURL = strings.TrimRight(meta.Endpoint, "/") + "/healthz"
-	}
+func (s *EKSServiceImpl) spawnReconciler(accountID, clusterName string, _ *ClusterMeta) {
 	js, err := s.deps.NATSConn.JetStream()
 	if err != nil {
 		slog.Error("spawnReconciler: jetstream", "err", err)
@@ -1145,8 +1182,13 @@ func (s *EKSServiceImpl) spawnReconciler(accountID, clusterName string, meta *Cl
 		slog.Error("spawnReconciler: account bucket", "err", err)
 		return
 	}
+	// Health gates on the control plane's NATS self-report, not an HTTP probe:
+	// k3s binds the apiserver to the VPC node-ip, unreachable from the host. The
+	// CP publishes {healthz,node_count} on the mgmt bus the daemon already shares.
+	stateSubject := StateSubject(accountID, clusterName)
 	spawn := func(ctx context.Context, _, _ string) (func(), error) {
-		return RunClusterReconciler(ctx, s.leaderKV, acctKV, accountID, clusterName, s.deps.HolderID, healthURL)
+		return RunClusterReconciler(ctx, s.leaderKV, acctKV, accountID, clusterName, s.deps.HolderID, "",
+			WithStateSource(s.deps.NATSConn, stateSubject))
 	}
 	if err := s.registry.Spawn(s.bgCtx, accountID, clusterName, spawn); err != nil {
 		slog.Error("spawnReconciler: registry spawn", "cluster", clusterName, "err", err)

spinifex/handlers/eks/service_impl_test.go

@@ -132,26 +132,30 @@ func TestDeleteCluster_ZeroizesOIDCKeyBeforeTeardown(t *testing.T) {
 	assert.Equal(t, ClusterStatusDeleting, meta.Status)
 }
 
-func TestEKSServiceImpl_NodegroupMethodsReturnNotImplemented(t *testing.T) {
+// In shim mode (orchestration deps absent) the mutating nodegroup methods
+// short-circuit to ServiceUnavailable, the read methods reach an empty
+// per-account bucket and surface ResourceNotFoundException, and
+// UpdateNodegroupVersion stays NotImplemented (v1 doesn't do AMI upgrades).
+func TestEKSServiceImpl_NodegroupMethodsShimMode(t *testing.T) {
 	svc := setupTestService(t)
 
 	_, err := svc.CreateNodegroup(&eks.CreateNodegroupInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID)
-	require.EqualError(t, err, awserrors.ErrorNotImplemented)
+	require.EqualError(t, err, awserrors.ErrorServiceUnavailable)
 
 	_, err = svc.DescribeNodegroup(&eks.DescribeNodegroupInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID)
-	require.EqualError(t, err, awserrors.ErrorNotImplemented)
+	require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound)
 
 	_, err = svc.ListNodegroups(&eks.ListNodegroupsInput{ClusterName: aws.String("c1")}, testAccountID)
-	require.EqualError(t, err, awserrors.ErrorNotImplemented)
+	require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound)
 
 	_, err = svc.UpdateNodegroupConfig(&eks.UpdateNodegroupConfigInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID)
-	require.EqualError(t, err, awserrors.ErrorNotImplemented)
+	require.EqualError(t, err, awserrors.ErrorServiceUnavailable)
 
 	_, err = svc.UpdateNodegroupVersion(&eks.UpdateNodegroupVersionInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID)
 	require.EqualError(t, err, awserrors.ErrorNotImplemented)
 
 	_, err = svc.DeleteNodegroup(&eks.DeleteNodegroupInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID)
-	require.EqualError(t, err, awserrors.ErrorNotImplemented)
+	require.EqualError(t, err, awserrors.ErrorServiceUnavailable)
 }
 
 // seedTestCluster lays down a minimal ACTIVE cluster meta in the per-account

spinifex/handlers/eks/state_report.go

@@ -0,0 +1,37 @@
+package handlers_eks
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+// StateSubject returns the NATS subject the control-plane VM publishes its
+// periodic health self-report to: "eks.state.{accountID}.{clusterName}.server".
+// The CP VM is on the management NATS bus, so this report reaches the host-side
+// reconciler without any route into the VPC overlay — unlike an HTTP /healthz
+// probe to the apiserver, which k3s binds to the unreachable VPC node-ip.
+func StateSubject(accountID, clusterName string) string {
+	return fmt.Sprintf("eks.state.%s.%s.server", accountID, clusterName)
+}
+
+// ServerStateReport is the JSON payload of a control-plane state report, emitted
+// by scripts/images/eks-node/mulga-eks-state-report.sh. Healthz mirrors the
+// apiserver's /healthz body ("ok" when serving); NodeCount is the number of
+// nodes the apiserver reports (server + joined agents); TS is the publish time
+// in unix seconds, used to reject stale reports from a CP that stopped emitting.
+type ServerStateReport struct {
+	Healthz   string `json:"healthz"`
+	NodeCount int    `json:"node_count"`
+	TS        int64  `json:"ts"`
+}
+
+// Healthy reports whether the apiserver was serving at publish time.
+func (s ServerStateReport) Healthy() bool { return s.Healthz == "ok" }
+
+func unmarshalServerStateReport(data []byte) (ServerStateReport, error) {
+	var r ServerStateReport
+	if err := json.Unmarshal(data, &r); err != nil {
+		return ServerStateReport{}, fmt.Errorf("eks: unmarshal server state report: %w", err)
+	}
+	return r, nil
+}

spinifex/handlers/eks/store.go

@@ -56,9 +56,15 @@ func ClusterMetaKey(name string) string {
 	return fmt.Sprintf("clusters/%s/meta", name)
 }
 
+// NodegroupsPrefix returns the KV key prefix under which all of a cluster's
+// nodegroup records live. Used by ListNodegroups to enumerate.
+func NodegroupsPrefix(cluster string) string {
+	return fmt.Sprintf("clusters/%s/nodegroups/", cluster)
+}
+
 // NodegroupKey returns the KV key for a nodegroup record under a cluster.
 func NodegroupKey(cluster, ng string) string {
-	return fmt.Sprintf("clusters/%s/nodegroups/%s", cluster, ng)
+	return NodegroupsPrefix(cluster) + ng
 }
 
 // AccessEntriesPrefix returns the KV key prefix under which all of a cluster's

spinifex/handlers/eks/types.go

@@ -27,14 +27,29 @@ type ClusterRecord struct {
 }
 
 // NodegroupRecord is the persisted-state envelope for an EKS managed
-// nodegroup.
+// nodegroup. It is the source of truth for Describe/ListNodegroups and tracks
+// the worker EC2 instance IDs the nodegroup owns so Update/DeleteNodegroup can
+// scale and tear them down.
 type NodegroupRecord struct {
-	ClusterName   string    `json:"clusterName"`
-	Name          string    `json:"name"`
-	Status        string    `json:"status"`
-	DesiredSize   int64     `json:"desiredSize"`
-	InstanceTypes []string  `json:"instanceTypes,omitempty"`
-	CreatedAt     time.Time `json:"createdAt"`
+	ClusterName    string            `json:"clusterName"`
+	Name           string            `json:"name"`
+	Arn            string            `json:"arn"`
+	Status         string            `json:"status"`
+	StatusReason   string            `json:"statusReason,omitempty"`
+	Subnets        []string          `json:"subnets,omitempty"`
+	InstanceTypes  []string          `json:"instanceTypes,omitempty"`
+	AMIType        string            `json:"amiType,omitempty"`
+	DiskSize       int64             `json:"diskSize,omitempty"`
+	ScalingMin     int64             `json:"scalingMin"`
+	ScalingMax     int64             `json:"scalingMax"`
+	ScalingDesired int64             `json:"scalingDesired"`
+	Version        string            `json:"version,omitempty"`
+	NodeRole       string            `json:"nodeRole,omitempty"`
+	Labels         map[string]string `json:"labels,omitempty"`
+	InstanceIDs    []string          `json:"instanceIds,omitempty"`
+	Health         string            `json:"health,omitempty"`
+	CreatedAt      time.Time         `json:"createdAt"`
+	ModifiedAt     time.Time         `json:"modifiedAt"`
 }
 
 // AccessEntryRecord is the persisted-state envelope for an EKS API-mode