From df74ad8a2c743bf1f6176cd324dfa3ce761855f2 Mon Sep 17 00:00:00 2001
From: Sylvester Kaczmarek
 <16242628+sylvesterkaczmarek@users.noreply.github.com>
Date: Thu, 10 Sep 2026 13:40:05 +0100
Subject: [PATCH 1/6] fix(operator): persist corrected batch checkpoints

Signed-off-by: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com>
---
 docs/user-guide/deployment-policy.md          |   8 ++
 .../internal/controller/cluster_state_v2.go   |  27 ++--
 .../controller/cluster_state_v2_test.go       | 128 ++++++++++++++++++
 3 files changed, 153 insertions(+), 10 deletions(-)

diff --git a/docs/user-guide/deployment-policy.md b/docs/user-guide/deployment-policy.md
index b47f5474e..5df185b0e 100644
--- a/docs/user-guide/deployment-policy.md
+++ b/docs/user-guide/deployment-policy.md
@@ -274,6 +274,14 @@ When using progressive rollout strategies (linear, exponential), the operator tr
 
 However, when a rollout **completes** or a **spec version changes**, you typically want the next rollout to start fresh from batch 1 rather than continuing with scaled-up batch sizes. Batch state reset handles this automatically.
 
+### Checkpoint Corrections
+
+When completed or failed nodes leave a compartment, the operator saves corrected
+checkpoint counts once the compartment has no nodes in progress. This correction
+does not advance the batch, change the previous batch result, or clear a stop
+decision. Later batches are evaluated against the saved counts, including after
+an operator restart.
+
 ### Auto-Reset Triggers
 
 Batch state is automatically reset when **either** of these events occurs (if configured):
diff --git a/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go
index ddc40b3a1..5db26a223 100644
--- a/operator/internal/controller/cluster_state_v2.go
+++ b/operator/internal/controller/cluster_state_v2.go
@@ -1244,6 +1244,7 @@ func evaluateCompletedBatches(skyhook SkyhookNodes) bool {
 
 	changed := false
 	for _, compartment := range compartments {
+		previousBatchState := compartment.GetBatchState()
 		if isComplete, successCount, failureCount := compartment.EvaluateCurrentBatch(); isComplete {
 			batchSize := successCount + failureCount
 
@@ -1283,18 +1284,24 @@ func evaluateCompletedBatches(skyhook SkyhookNodes) bool {
 
 			// Update the compartment's batch state using strategy logic
 			compartment.EvaluateAndUpdateBatchState(batchSize, successCount, failureCount)
+		}
 
-			// Persist the updated batch state to the skyhook status immediately
-			if skyhook.GetSkyhook().Status.CompartmentStatuses == nil {
-				skyhook.GetSkyhook().Status.CompartmentStatuses = make(map[string]v1alpha1.CompartmentStatus)
-			}
-			// Build and persist the compartment status with the updated batch state
-			newStatus := buildCompartmentStatus(compartment)
-			skyhook.GetSkyhook().Status.CompartmentStatuses[compartment.GetName()] = newStatus
-
-			skyhook.GetSkyhook().Updated = true
-			changed = true
+		// Evaluation can correct checkpoints without completing a batch.
+		if compartment.GetBatchState() == previousBatchState {
+			continue
 		}
+		newStatus := buildCompartmentStatus(compartment)
+		statuses := skyhook.GetSkyhook().Status.CompartmentStatuses
+		if previous, exists := statuses[compartment.GetName()]; exists && compartmentStatusEqual(previous, newStatus) {
+			continue
+		}
+		if statuses == nil {
+			statuses = make(map[string]v1alpha1.CompartmentStatus)
+			skyhook.GetSkyhook().Status.CompartmentStatuses = statuses
+		}
+		statuses[compartment.GetName()] = newStatus
+		skyhook.GetSkyhook().Updated = true
+		changed = true
 	}
 
 	return changed
diff --git a/operator/internal/controller/cluster_state_v2_test.go b/operator/internal/controller/cluster_state_v2_test.go
index 5b21d635d..90a31d4de 100644
--- a/operator/internal/controller/cluster_state_v2_test.go
+++ b/operator/internal/controller/cluster_state_v2_test.go
@@ -2211,6 +2211,134 @@ var _ = Describe("Compartment Status Tests", func() {
 		})
 	})
 
+	Describe("evaluateCompletedBatches checkpoint persistence", func() {
+		var record *v1alpha1.NodeWright
+		var rebuild func() SkyhookNodes
+
+		BeforeEach(func() {
+			record = &v1alpha1.NodeWright{
+				ObjectMeta: metav1.ObjectMeta{Name: "checkpoint-rollout"},
+				Spec: v1alpha1.NodeWrightSpec{
+					DeploymentPolicy: "checkpoint-policy",
+					Packages: map[string]v1alpha1.Package{
+						"demo": {
+							PackageRef: v1alpha1.PackageRef{Name: "demo", Version: "1.0.0"},
+							Image:      "example/demo:1.0.0",
+						},
+					},
+				},
+				Status: v1alpha1.NodeWrightStatus{
+					Status: v1alpha1.StatusWaiting,
+					CompartmentStatuses: map[string]v1alpha1.CompartmentStatus{
+						v1alpha1.DefaultCompartmentName: {
+							BatchState: &v1alpha1.BatchProcessingState{
+								CurrentBatch: 4, ConsecutiveFailures: 1,
+								LastBatchSize: 2, LastBatchFailed: true,
+							},
+						},
+					},
+				},
+			}
+			policies := &v1alpha1.DeploymentPolicyList{Items: []v1alpha1.DeploymentPolicy{{
+				ObjectMeta: metav1.ObjectMeta{Name: "checkpoint-policy"},
+				Spec: v1alpha1.DeploymentPolicySpec{Default: v1alpha1.PolicyDefault{
+					Budget: v1alpha1.DeploymentBudget{Count: kptr.To(3)},
+					Strategy: &v1alpha1.DeploymentStrategy{Fixed: &v1alpha1.FixedStrategy{
+						InitialBatch: kptr.To(2), BatchThreshold: kptr.To(100),
+						FailureThreshold: kptr.To(2), SafetyLimit: kptr.To(100),
+					}},
+				}},
+			}}}
+			nodes := &corev1.NodeList{Items: []corev1.Node{
+				{ObjectMeta: metav1.ObjectMeta{Name: "node-a"}},
+				{ObjectMeta: metav1.ObjectMeta{Name: "node-b"}},
+				{ObjectMeta: metav1.ObjectMeta{Name: "node-c"}},
+			}}
+			rebuild = func() SkyhookNodes {
+				state, err := BuildState(&v1alpha1.NodeWrightList{
+					Items: []v1alpha1.NodeWright{*record.DeepCopy()},
+				}, nodes.DeepCopy(), policies)
+				Expect(err).NotTo(HaveOccurred())
+				return state.skyhooks[0]
+			}
+		})
+
+		DescribeTable("persists a corrected checkpoint without evaluating a batch",
+			func(completed, failed int, stopped bool) {
+				initial := record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState
+				initial.CompletedNodes, initial.FailedNodes, initial.ShouldStop = completed, failed, stopped
+				expected := *initial
+				expected.CompletedNodes, expected.FailedNodes = 0, 0
+				rollout := rebuild()
+
+				Expect(evaluateCompletedBatches(rollout)).To(BeTrue())
+				Expect(rollout.GetSkyhook().Updated).To(BeTrue())
+				record = rollout.GetSkyhook().NodeWright.DeepCopy()
+				Expect(record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState).To(Equal(&expected))
+
+				reloaded := rebuild()
+				Expect(reloaded.GetCompartments()[v1alpha1.DefaultCompartmentName].GetBatchState()).To(Equal(expected))
+				Expect(evaluateCompletedBatches(reloaded)).To(BeFalse())
+				Expect(reloaded.GetSkyhook().Updated).To(BeFalse())
+			},
+			Entry("completed nodes left", 2, 0, false),
+			Entry("failed nodes left", 0, 2, false),
+			Entry("both counts shrank", 2, 2, false),
+			Entry("a stopped compartment retains its stop decision", 2, 2, true),
+		)
+
+		It("evaluates later failures after reloading a corrected checkpoint", func() {
+			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.FailedNodes = 88
+			rollout := rebuild()
+			Expect(evaluateCompletedBatches(rollout)).To(BeTrue())
+			record = rollout.GetSkyhook().NodeWright.DeepCopy()
+
+			reloaded := rebuild()
+			reloaded.GetNodes()[0].SetStatus(v1alpha1.StatusErroring)
+			Expect(evaluateCompletedBatches(reloaded)).To(BeTrue())
+			batch := reloaded.GetSkyhook().Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState
+			Expect(batch.CurrentBatch).To(Equal(5))
+			Expect(batch.FailedNodes).To(Equal(1))
+			Expect(batch.ConsecutiveFailures).To(Equal(2))
+			Expect(batch.ShouldStop).To(BeTrue())
+			Expect(batch.LastBatchSize).To(Equal(1))
+		})
+
+		It("persists first-batch initialization without advancing the strategy", func() {
+			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.CurrentBatch = 0
+			rollout := rebuild()
+			Expect(evaluateCompletedBatches(rollout)).To(BeTrue())
+			batch := rollout.GetSkyhook().Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState
+			Expect(batch.CurrentBatch).To(Equal(1))
+			Expect(batch.ConsecutiveFailures).To(Equal(1))
+			Expect(batch.LastBatchSize).To(Equal(2))
+		})
+
+		It("leaves completed-rollout batch state unchanged", func() {
+			record.Status.Status = v1alpha1.StatusComplete
+			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.CompletedNodes = 10
+			rollout := rebuild()
+			before := rollout.GetSkyhook().Status.DeepCopy()
+			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+			Expect(rollout.GetSkyhook().Status).To(Equal(*before))
+			Expect(rollout.GetSkyhook().Updated).To(BeFalse())
+		})
+
+		DescribeTable("does not persist an unevaluated or unchanged batch",
+			func(status v1alpha1.Status) {
+				rollout := rebuild()
+				rollout.GetNodes()[0].SetStatus(status)
+				before := rollout.GetSkyhook().Status.DeepCopy()
+				Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+				Expect(rollout.GetSkyhook().Updated).To(BeFalse())
+				Expect(rollout.GetSkyhook().Status).To(Equal(*before))
+			},
+			Entry("no progress", v1alpha1.StatusWaiting),
+			Entry("work in progress", v1alpha1.StatusInProgress),
+			Entry("blocked work", v1alpha1.StatusBlocked),
+		)
+	})
+
 	Describe("should persist compartment status to skyhook status", func() {
 		It("should persist compartment status to skyhook status in ReportState", func() {
 			skyhook := &v1alpha1.NodeWright{

From 3a83fa42f896bac5c5c6cf8cb646083d18edaaf5 Mon Sep 17 00:00:00 2001
From: Sylvester Kaczmarek
 <16242628+sylvesterkaczmarek@users.noreply.github.com>
Date: Thu, 10 Sep 2026 21:20:12 +0100
Subject: [PATCH 2/6] fix(operator): make checkpoint correction durable

Signed-off-by: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com>
---
 docs/user-guide/deployment-policy.md          |  20 +--
 .../internal/controller/cluster_state_v2.go   | 128 +++++++++---------
 .../controller/cluster_state_v2_test.go       |  73 +++++++++-
 .../internal/controller/skyhook_controller.go |  16 ++-
 operator/internal/wrapper/compartment.go      |  40 ++++--
 operator/internal/wrapper/compartment_test.go |  39 ++++++
 6 files changed, 223 insertions(+), 93 deletions(-)

diff --git a/docs/user-guide/deployment-policy.md b/docs/user-guide/deployment-policy.md
index 5df185b0e..24b756173 100644
--- a/docs/user-guide/deployment-policy.md
+++ b/docs/user-guide/deployment-policy.md
@@ -268,20 +268,24 @@ compartments:
 
 ---
 
+## Checkpoint Corrections
+
+When compartment membership changes, the operator rebaselines completed and failed
+checkpoint counts to the current members. Membership changes are bookkeeping, not
+completed batches: they do not advance the batch, change the previous batch result,
+or clear a stop decision. If only one outcome count later moves backwards, for
+example when a node recovers from `Erroring` to `Complete`, only that checkpoint is
+corrected so positive progress in the other outcome is still evaluated.
+
+Corrections are saved to status without stopping the rest of the reconciliation pass,
+and later batches continue from the saved baseline after an operator restart.
+
 ## Batch State Reset
 
 When using progressive rollout strategies (linear, exponential), the operator tracks batch processing state per compartment — current batch number, consecutive failures, completed/failed node counts, etc. This state persists across reconciliations so the rollout can scale up progressively.
 
 However, when a rollout **completes** or a **spec version changes**, you typically want the next rollout to start fresh from batch 1 rather than continuing with scaled-up batch sizes. Batch state reset handles this automatically.
 
-### Checkpoint Corrections
-
-When completed or failed nodes leave a compartment, the operator saves corrected
-checkpoint counts once the compartment has no nodes in progress. This correction
-does not advance the batch, change the previous batch result, or clear a stop
-decision. Later batches are evaluated against the saved counts, including after
-an operator restart.
-
 ### Auto-Reset Triggers
 
 Batch state is automatically reset when **either** of these events occurs (if configured):
diff --git a/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go
index 5db26a223..385366214 100644
--- a/operator/internal/controller/cluster_state_v2.go
+++ b/operator/internal/controller/cluster_state_v2.go
@@ -1205,6 +1205,7 @@ func IntrospectSkyhook(skyhook SkyhookNodes, allSkyhooks []SkyhookNodes, logger
 
 	if scrStatus != collectNodeStatus {
 		skyhook.SetStatus(collectNodeStatus)
+		change = true
 	}
 
 	for _, node := range skyhook.GetNodes() {
@@ -1222,14 +1223,15 @@ func IntrospectSkyhook(skyhook SkyhookNodes, allSkyhooks []SkyhookNodes, logger
 		change = true
 	}
 
-	skyhook.UpdateCondition(logger)
-	if skyhook.GetSkyhook().Updated {
+	if skyhook.UpdateCondition(logger) {
 		change = true
 	}
 	return change
 }
 
-// evaluateCompletedBatches checks if any compartment batches are complete and evaluates them
+// evaluateCompletedBatches checks if any compartment batches are complete and evaluates them.
+// It returns true only when rollout state advances; checkpoint/status bookkeeping is
+// persisted separately and must not force the controller to abandon the rest of a reconcile.
 func evaluateCompletedBatches(skyhook SkyhookNodes) bool {
 	compartments := skyhook.GetCompartments()
 	if len(compartments) == 0 {
@@ -1237,18 +1239,29 @@ func evaluateCompletedBatches(skyhook SkyhookNodes) bool {
 	}
 
 	// Skip batch evaluation when skyhook is Complete - this prevents overwriting
-	// the batch state that was just reset by SetStatus transitioning to Complete
+	// the batch state that was just reset by SetStatus transitioning to Complete.
 	if skyhook.Status() == v1alpha1.StatusComplete {
 		return false
 	}
 
-	changed := false
+	batchAdvanced := false
+	statuses := skyhook.GetSkyhook().Status.CompartmentStatuses
 	for _, compartment := range compartments {
-		previousBatchState := compartment.GetBatchState()
+		name := compartment.GetName()
+
+		// A membership change invalidates cumulative checkpoints as a baseline, but
+		// it is not itself a completed batch. Rebaseline first so nodes that later
+		// return cannot be mistaken for a new successful batch.
+		if previous, exists := statuses[name]; exists && previous.Matched != len(compartment.GetNodes()) {
+			compartment.RebaselineBatchCheckpoints()
+			persistCompartmentStatus(skyhook, compartment)
+			continue
+		}
+
 		if isComplete, successCount, failureCount := compartment.EvaluateCurrentBatch(); isComplete {
 			batchSize := successCount + failureCount
 
-			// Count blocked nodes to determine if we should skip batch evaluation
+			// Count blocked nodes to determine if this is real rollout progress.
 			blockedCount := 0
 			for _, node := range compartment.GetNodes() {
 				if node.Status() == v1alpha1.StatusBlocked {
@@ -1256,55 +1269,34 @@ func evaluateCompletedBatches(skyhook SkyhookNodes) bool {
 				}
 			}
 
-			// If batchSize is 0 but batch is complete, check if all nodes are blocked
-			// If all nodes are blocked, don't advance the batch - wait for them to become unblocked
-			// Blocked nodes are not failures, they're just temporarily unable to proceed
+			shouldAdvance := true
 			if batchSize == 0 {
-				// If all nodes in the compartment are blocked, skip batch evaluation
-				// The batch will be re-evaluated when nodes become unblocked
 				if blockedCount > 0 && blockedCount == len(compartment.GetNodes()) {
-					continue // Skip this compartment - all nodes blocked, wait for them to unblock
-				}
-				// If some nodes are blocked but not all, use blocked count as batch size
-				// This handles mixed batches (some blocked, some completed/failed)
-				if blockedCount > 0 {
+					shouldAdvance = false
+				} else if blockedCount > 0 {
 					batchSize = blockedCount
 				} else if compartment.GetBatchState().LastBatchSize > 0 {
 					batchSize = compartment.GetBatchState().LastBatchSize
 				}
 			}
 
-			// If batch has blocked nodes but no successes/failures, don't treat as failure
-			// Blocked nodes should not increment consecutive failures
-			// Only evaluate if we have actual progress (successes or failures)
+			// Blocked nodes are temporary and should not become a failed batch.
 			if batchSize > 0 && successCount == 0 && failureCount == 0 && blockedCount == batchSize {
-				// All nodes in batch are blocked - skip evaluation to avoid false failures
-				continue
+				shouldAdvance = false
 			}
 
-			// Update the compartment's batch state using strategy logic
-			compartment.EvaluateAndUpdateBatchState(batchSize, successCount, failureCount)
+			if shouldAdvance {
+				compartment.EvaluateAndUpdateBatchState(batchSize, successCount, failureCount)
+				batchAdvanced = true
+			}
 		}
 
-		// Evaluation can correct checkpoints without completing a batch.
-		if compartment.GetBatchState() == previousBatchState {
-			continue
-		}
-		newStatus := buildCompartmentStatus(compartment)
-		statuses := skyhook.GetSkyhook().Status.CompartmentStatuses
-		if previous, exists := statuses[compartment.GetName()]; exists && compartmentStatusEqual(previous, newStatus) {
-			continue
-		}
-		if statuses == nil {
-			statuses = make(map[string]v1alpha1.CompartmentStatus)
-			skyhook.GetSkyhook().Status.CompartmentStatuses = statuses
-		}
-		statuses[compartment.GetName()] = newStatus
-		skyhook.GetSkyhook().Updated = true
-		changed = true
+		// Persist any checkpoint change, including first-batch initialization and
+		// field-specific negative-delta corrections, through the shared status path.
+		persistCompartmentStatus(skyhook, compartment)
 	}
 
-	return changed
+	return batchAdvanced
 }
 
 func IntrospectNode(node wrapper.SkyhookNode, skyhook SkyhookNodes, allSkyhooks []SkyhookNodes) bool {
@@ -1443,18 +1435,16 @@ func buildCompartmentStatus(compartment *wrapper.Compartment) v1alpha1.Compartme
 	// Get batch state
 	batchState := compartment.GetBatchState()
 
-	// Copy batch state for status
-	var batchStateCopy *v1alpha1.BatchProcessingState
-	if compartment.Strategy != nil {
-		batchStateCopy = &v1alpha1.BatchProcessingState{
-			CurrentBatch:        batchState.CurrentBatch,
-			ConsecutiveFailures: batchState.ConsecutiveFailures,
-			CompletedNodes:      batchState.CompletedNodes,
-			FailedNodes:         batchState.FailedNodes,
-			ShouldStop:          batchState.ShouldStop,
-			LastBatchSize:       batchState.LastBatchSize,
-			LastBatchFailed:     batchState.LastBatchFailed,
-		}
+	// Batch state is persistent bookkeeping even when a compartment has no
+	// rollout strategy, so always publish it.
+	batchStateCopy := &v1alpha1.BatchProcessingState{
+		CurrentBatch:        batchState.CurrentBatch,
+		ConsecutiveFailures: batchState.ConsecutiveFailures,
+		CompletedNodes:      batchState.CompletedNodes,
+		FailedNodes:         batchState.FailedNodes,
+		ShouldStop:          batchState.ShouldStop,
+		LastBatchSize:       batchState.LastBatchSize,
+		LastBatchFailed:     batchState.LastBatchFailed,
 	}
 
 	return v1alpha1.CompartmentStatus{
@@ -1700,21 +1690,29 @@ func (skyhook *skyhookNodes) AssignNodeToCompartment(node wrapper.SkyhookNode) (
 	return matches[0].name, nil
 }
 
-// updateCompartmentStatuses updates compartment statuses for all current compartments
-func updateCompartmentStatuses(skyhook *skyhookNodes) {
-	if len(skyhook.compartments) == 0 {
-		return
+// persistCompartmentStatus writes a compartment status only when it changed.
+func persistCompartmentStatus(skyhook SkyhookNodes, compartment *wrapper.Compartment) bool {
+	statuses := skyhook.GetSkyhook().Status.CompartmentStatuses
+	if statuses == nil {
+		statuses = make(map[string]v1alpha1.CompartmentStatus)
+		skyhook.GetSkyhook().Status.CompartmentStatuses = statuses
 	}
-	if skyhook.skyhook.Status.CompartmentStatuses == nil {
-		skyhook.skyhook.Status.CompartmentStatuses = make(map[string]v1alpha1.CompartmentStatus)
+
+	name := compartment.GetName()
+	newStatus := buildCompartmentStatus(compartment)
+	if existing, ok := statuses[name]; ok && compartmentStatusEqual(existing, newStatus) {
+		return false
 	}
 
-	for name, compartment := range skyhook.compartments {
-		newStatus := buildCompartmentStatus(compartment)
-		if existing, ok := skyhook.skyhook.Status.CompartmentStatuses[name]; !ok || !compartmentStatusEqual(existing, newStatus) {
-			skyhook.skyhook.Status.CompartmentStatuses[name] = newStatus
-			skyhook.skyhook.Updated = true
-		}
+	statuses[name] = newStatus
+	skyhook.GetSkyhook().Updated = true
+	return true
+}
+
+// updateCompartmentStatuses updates compartment statuses for all current compartments.
+func updateCompartmentStatuses(skyhook SkyhookNodes) {
+	for _, compartment := range skyhook.GetCompartments() {
+		persistCompartmentStatus(skyhook, compartment)
 	}
 }
 
diff --git a/operator/internal/controller/cluster_state_v2_test.go b/operator/internal/controller/cluster_state_v2_test.go
index 90a31d4de..61f8470a6 100644
--- a/operator/internal/controller/cluster_state_v2_test.go
+++ b/operator/internal/controller/cluster_state_v2_test.go
@@ -2123,7 +2123,8 @@ var _ = Describe("Compartment Status Tests", func() {
 			Expect(status.InProgress).To(Equal(0))
 			Expect(status.Completed).To(Equal(0))
 			Expect(status.ProgressPercent).To(Equal(0))
-			Expect(status.BatchState).To(BeNil())
+			Expect(status.BatchState).NotTo(BeNil())
+			Expect(status.BatchState.CurrentBatch).To(Equal(1))
 		})
 
 		It("should build status for compartment with strategy and batch state", func() {
@@ -2231,6 +2232,7 @@ var _ = Describe("Compartment Status Tests", func() {
 					Status: v1alpha1.StatusWaiting,
 					CompartmentStatuses: map[string]v1alpha1.CompartmentStatus{
 						v1alpha1.DefaultCompartmentName: {
+							Matched: 3,
 							BatchState: &v1alpha1.BatchProcessingState{
 								CurrentBatch: 4, ConsecutiveFailures: 1,
 								LastBatchSize: 2, LastBatchFailed: true,
@@ -2271,7 +2273,7 @@ var _ = Describe("Compartment Status Tests", func() {
 				expected.CompletedNodes, expected.FailedNodes = 0, 0
 				rollout := rebuild()
 
-				Expect(evaluateCompletedBatches(rollout)).To(BeTrue())
+				Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
 				Expect(rollout.GetSkyhook().Updated).To(BeTrue())
 				record = rollout.GetSkyhook().NodeWright.DeepCopy()
 				Expect(record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState).To(Equal(&expected))
@@ -2290,7 +2292,7 @@ var _ = Describe("Compartment Status Tests", func() {
 		It("evaluates later failures after reloading a corrected checkpoint", func() {
 			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.FailedNodes = 88
 			rollout := rebuild()
-			Expect(evaluateCompletedBatches(rollout)).To(BeTrue())
+			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
 			record = rollout.GetSkyhook().NodeWright.DeepCopy()
 
 			reloaded := rebuild()
@@ -2307,13 +2309,69 @@ var _ = Describe("Compartment Status Tests", func() {
 		It("persists first-batch initialization without advancing the strategy", func() {
 			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.CurrentBatch = 0
 			rollout := rebuild()
-			Expect(evaluateCompletedBatches(rollout)).To(BeTrue())
+			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
 			batch := rollout.GetSkyhook().Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState
 			Expect(batch.CurrentBatch).To(Equal(1))
 			Expect(batch.ConsecutiveFailures).To(Equal(1))
 			Expect(batch.LastBatchSize).To(Equal(2))
 		})
 
+		It("publishes batch state for a compartment without a strategy", func() {
+			compartment := wrapper.NewCompartmentWrapper(&v1alpha1.Compartment{
+				Name:   "no-strategy",
+				Budget: v1alpha1.DeploymentBudget{Count: kptr.To(2)},
+			}, &v1alpha1.BatchProcessingState{CurrentBatch: 3, CompletedNodes: 1})
+
+			status := buildCompartmentStatus(compartment)
+			Expect(status.BatchState).NotTo(BeNil())
+			Expect(*status.BatchState).To(Equal(compartment.GetBatchState()))
+		})
+
+		It("rebaselines returning members without creating a phantom batch", func() {
+			strategy := &v1alpha1.DeploymentStrategy{Exponential: &v1alpha1.ExponentialStrategy{
+				InitialBatch: kptr.To(2), GrowthFactor: kptr.To(2), BatchThreshold: kptr.To(100), SafetyLimit: kptr.To(100),
+			}}
+			state := v1alpha1.BatchProcessingState{CurrentBatch: 4, CompletedNodes: 4, LastBatchSize: 2}
+			nodewright := &v1alpha1.NodeWright{Status: v1alpha1.NodeWrightStatus{
+				Status: v1alpha1.StatusWaiting,
+				CompartmentStatuses: map[string]v1alpha1.CompartmentStatus{
+					"moving": {Matched: 10, BatchState: &state},
+				},
+			}}
+			compartment := wrapper.NewCompartmentWrapper(&v1alpha1.Compartment{
+				Name: "moving", Budget: v1alpha1.DeploymentBudget{Count: kptr.To(10)}, Strategy: strategy,
+			}, &state)
+			rollout := &skyhookNodes{
+				skyhook:      wrapper.NewSkyhookWrapper(nodewright),
+				compartments: map[string]*wrapper.Compartment{"moving": compartment},
+			}
+
+			// All members leave. Persist a new zero baseline without advancing the strategy.
+			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+			saved := rollout.GetSkyhook().Status.CompartmentStatuses["moving"]
+			Expect(saved.Matched).To(Equal(0))
+			Expect(saved.BatchState.CompletedNodes).To(Equal(0))
+			Expect(saved.BatchState.CurrentBatch).To(Equal(4))
+			Expect(saved.BatchState.LastBatchSize).To(Equal(2))
+
+			// Four already-complete nodes return with six pending nodes. Their return is
+			// another membership rebaseline, not a new four-node successful batch.
+			returned := wrapper.NewCompartmentWrapper(&compartment.Compartment, saved.BatchState)
+			for index := 0; index < 10; index++ {
+				node := wrapperMock.NewMockSkyhookNode(GinkgoT())
+				node.EXPECT().IsComplete().Return(index < 4).Maybe()
+				node.EXPECT().Status().Return(v1alpha1.StatusWaiting).Maybe()
+				returned.Nodes = append(returned.Nodes, node)
+			}
+			rollout.compartments["moving"] = returned
+			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+			rebased := rollout.GetSkyhook().Status.CompartmentStatuses["moving"].BatchState
+			Expect(rebased.CompletedNodes).To(Equal(4))
+			Expect(rebased.CurrentBatch).To(Equal(4))
+			Expect(rebased.LastBatchSize).To(Equal(2))
+			Expect(strategy.CalculateBatchSize(10, rebased)).To(Equal(4))
+		})
+
 		It("leaves completed-rollout batch state unchanged", func() {
 			record.Status.Status = v1alpha1.StatusComplete
 			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.CompletedNodes = 10
@@ -2328,10 +2386,11 @@ var _ = Describe("Compartment Status Tests", func() {
 			func(status v1alpha1.Status) {
 				rollout := rebuild()
 				rollout.GetNodes()[0].SetStatus(status)
-				before := rollout.GetSkyhook().Status.DeepCopy()
+				// This table isolates batch persistence; SetStatus itself marks the Skyhook dirty.
+				rollout.GetSkyhook().Updated = false
+				beforeBatch := rollout.GetCompartments()[v1alpha1.DefaultCompartmentName].GetBatchState()
 				Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
-				Expect(rollout.GetSkyhook().Updated).To(BeFalse())
-				Expect(rollout.GetSkyhook().Status).To(Equal(*before))
+				Expect(rollout.GetCompartments()[v1alpha1.DefaultCompartmentName].GetBatchState()).To(Equal(beforeBatch))
 			},
 			Entry("no progress", v1alpha1.StatusWaiting),
 			Entry("work in progress", v1alpha1.StatusInProgress),
diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go
index c2469d8e8..6307f1fbd 100644
--- a/operator/internal/controller/skyhook_controller.go
+++ b/operator/internal/controller/skyhook_controller.go
@@ -504,11 +504,10 @@ func (r *SkyhookReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 			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.ReportState(ctx, clusterState, skyhook)); yes {
+				return result, err
+			}
 			if err := r.suspendUnfinishedJobs(ctx, skyhook); err != nil {
 				return ctrl.Result{RequeueAfter: time.Second * 2}, fmt.Errorf("suspending jobs for paused skyhook %s: %w", skyhook.GetSkyhook().Name, err)
 			}
@@ -533,11 +532,18 @@ func (r *SkyhookReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		}
 
 		changed := IntrospectSkyhook(skyhook, clusterState.skyhooks, logger)
-		if changed {
+
+		// Report after introspection so compartment status reflects the state just
+		// evaluated. Persist status-only bookkeeping without turning it into an
+		// operational change that aborts processing of later NodeWrights.
+		skyhook.ReportState()
+		if changed || skyhook.GetSkyhook().Updated {
 			_, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
 			if len(errs) > 0 {
 				return ctrl.Result{RequeueAfter: time.Second * 2}, utilerrors.NewAggregate(errs)
 			}
+		}
+		if changed {
 			return ctrl.Result{RequeueAfter: time.Second * 2}, nil
 		}
 
diff --git a/operator/internal/wrapper/compartment.go b/operator/internal/wrapper/compartment.go
index a5e63d727..b0932c0e7 100644
--- a/operator/internal/wrapper/compartment.go
+++ b/operator/internal/wrapper/compartment.go
@@ -202,6 +202,25 @@ func (c *Compartment) IsBatchComplete() bool {
 	return c.getInProgressCount() == 0
 }
 
+// RebaselineBatchCheckpoints aligns cumulative checkpoints with the current
+// compartment membership without treating the membership change as a completed batch.
+func (c *Compartment) RebaselineBatchCheckpoints() bool {
+	currentCompleted := 0
+	currentFailed := 0
+	for _, node := range c.Nodes {
+		if node.IsComplete() {
+			currentCompleted++
+		} else if node.Status() == v1alpha1.StatusErroring {
+			currentFailed++
+		}
+	}
+
+	changed := c.BatchState.CompletedNodes != currentCompleted || c.BatchState.FailedNodes != currentFailed
+	c.BatchState.CompletedNodes = currentCompleted
+	c.BatchState.FailedNodes = currentFailed
+	return changed
+}
+
 // EvaluateCurrentBatch evaluates the current batch result if it's complete
 // Uses delta-based tracking: compares current state to last checkpoint
 func (c *Compartment) EvaluateCurrentBatch() (bool, int, int) {
@@ -230,18 +249,23 @@ func (c *Compartment) EvaluateCurrentBatch() (bool, int, int) {
 		}
 	}
 
-	// Calculate delta from last checkpoint
+	// Calculate delta from last checkpoint. A count can decrease without the other
+	// count becoming invalid (for example, Erroring -> Complete recovery), so
+	// rebaseline only the field that moved backwards and preserve positive progress.
 	deltaCompleted := currentCompleted - c.BatchState.CompletedNodes
 	deltaFailed := currentFailed - c.BatchState.FailedNodes
-
-	// Handle negative deltas: this happens when nodes move between compartments mid-rollout
-	// When nodes leave a compartment, the checkpoint becomes invalid, so we reset it
-	if deltaCompleted < 0 || deltaFailed < 0 {
-		// Nodes moved compartments - reset checkpoints to current state
-		// This prevents negative batch sizes and incorrect evaluations
+	corrected := false
+	if deltaCompleted < 0 {
 		c.BatchState.CompletedNodes = currentCompleted
+		deltaCompleted = 0
+		corrected = true
+	}
+	if deltaFailed < 0 {
 		c.BatchState.FailedNodes = currentFailed
-		// Don't evaluate this batch since we just reset - wait for next reconcile
+		deltaFailed = 0
+		corrected = true
+	}
+	if corrected && deltaCompleted == 0 && deltaFailed == 0 {
 		return false, 0, 0
 	}
 
diff --git a/operator/internal/wrapper/compartment_test.go b/operator/internal/wrapper/compartment_test.go
index 871cb792b..6f1902f74 100644
--- a/operator/internal/wrapper/compartment_test.go
+++ b/operator/internal/wrapper/compartment_test.go
@@ -135,6 +135,45 @@ var _ = Describe("Compartment", func() {
 		})
 	})
 
+	Context("EvaluateCurrentBatch checkpoint corrections", func() {
+		It("preserves positive completion progress when failures decrease", func() {
+			skyhook := &wrapper.Skyhook{NodeWright: &v1alpha1.NodeWright{}}
+			compartment := wrapper.NewCompartmentWrapper(&v1alpha1.Compartment{Name: "recovery"}, &v1alpha1.BatchProcessingState{
+				CurrentBatch: 3, CompletedNodes: 1, FailedNodes: 2,
+			})
+			compartment.Nodes = []wrapper.SkyhookNode{
+				newMockNode(GinkgoT(), "complete-a", v1alpha1.StatusComplete, true, skyhook),
+				newMockNode(GinkgoT(), "complete-b", v1alpha1.StatusComplete, true, skyhook),
+				newMockNode(GinkgoT(), "error", v1alpha1.StatusErroring, false, skyhook),
+			}
+
+			complete, successes, failures := compartment.EvaluateCurrentBatch()
+			Expect(complete).To(BeTrue())
+			Expect(successes).To(Equal(1))
+			Expect(failures).To(Equal(0))
+			Expect(compartment.GetBatchState().CompletedNodes).To(Equal(2))
+			Expect(compartment.GetBatchState().FailedNodes).To(Equal(1))
+		})
+
+		It("rebaselines only the checkpoint that moved backwards", func() {
+			skyhook := &wrapper.Skyhook{NodeWright: &v1alpha1.NodeWright{}}
+			compartment := wrapper.NewCompartmentWrapper(&v1alpha1.Compartment{Name: "correction"}, &v1alpha1.BatchProcessingState{
+				CurrentBatch: 3, CompletedNodes: 2, FailedNodes: 1,
+			})
+			compartment.Nodes = []wrapper.SkyhookNode{
+				newMockNode(GinkgoT(), "pending", v1alpha1.StatusWaiting, false, skyhook),
+				newMockNode(GinkgoT(), "error", v1alpha1.StatusErroring, false, skyhook),
+			}
+
+			complete, successes, failures := compartment.EvaluateCurrentBatch()
+			Expect(complete).To(BeFalse())
+			Expect(successes).To(BeZero())
+			Expect(failures).To(BeZero())
+			Expect(compartment.GetBatchState().CompletedNodes).To(Equal(0))
+			Expect(compartment.GetBatchState().FailedNodes).To(Equal(1))
+		})
+	})
+
 	Context("EvaluateAndUpdateBatchState", func() {
 		It("should update basic state without strategy", func() {
 			compartment := wrapper.NewCompartmentWrapper(&v1alpha1.Compartment{

From 768eee18981a2fe15b6dd821d6518b9b22885319 Mon Sep 17 00:00:00 2001
From: Sylvester Kaczmarek
 <16242628+sylvesterkaczmarek@users.noreply.github.com>
Date: Sat, 12 Sep 2026 13:20:09 +0100
Subject: [PATCH 3/6] fix(operator): preserve progress during membership churn

Signed-off-by: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com>
---
 .../internal/controller/cluster_state_v2.go   | 39 +++------------
 .../controller/cluster_state_v2_test.go       | 41 ++++++++++++++++
 operator/internal/wrapper/compartment.go      | 49 ++++++++++++++++---
 3 files changed, 92 insertions(+), 37 deletions(-)

diff --git a/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go
index 385366214..fa930a57f 100644
--- a/operator/internal/controller/cluster_state_v2.go
+++ b/operator/internal/controller/cluster_state_v2.go
@@ -1249,43 +1249,20 @@ func evaluateCompletedBatches(skyhook SkyhookNodes) bool {
 	for _, compartment := range compartments {
 		name := compartment.GetName()
 
-		// A membership change invalidates cumulative checkpoints as a baseline, but
-		// it is not itself a completed batch. Rebaseline first so nodes that later
-		// return cannot be mistaken for a new successful batch.
+		// Membership churn can carry terminal outcomes into or out of the compartment.
+		// Absorb only the number of outcome changes explainable by that churn, then
+		// evaluate any residual progress in this same reconcile.
 		if previous, exists := statuses[name]; exists && previous.Matched != len(compartment.GetNodes()) {
-			compartment.RebaselineBatchCheckpoints()
-			persistCompartmentStatus(skyhook, compartment)
-			continue
+			membershipDelta := len(compartment.GetNodes()) - previous.Matched
+			compartment.RebaselineBatchCheckpoints(membershipDelta)
 		}
 
 		if isComplete, successCount, failureCount := compartment.EvaluateCurrentBatch(); isComplete {
 			batchSize := successCount + failureCount
 
-			// Count blocked nodes to determine if this is real rollout progress.
-			blockedCount := 0
-			for _, node := range compartment.GetNodes() {
-				if node.Status() == v1alpha1.StatusBlocked {
-					blockedCount++
-				}
-			}
-
-			shouldAdvance := true
-			if batchSize == 0 {
-				if blockedCount > 0 && blockedCount == len(compartment.GetNodes()) {
-					shouldAdvance = false
-				} else if blockedCount > 0 {
-					batchSize = blockedCount
-				} else if compartment.GetBatchState().LastBatchSize > 0 {
-					batchSize = compartment.GetBatchState().LastBatchSize
-				}
-			}
-
-			// Blocked nodes are temporary and should not become a failed batch.
-			if batchSize > 0 && successCount == 0 && failureCount == 0 && blockedCount == batchSize {
-				shouldAdvance = false
-			}
-
-			if shouldAdvance {
+			// A completed batch with no terminal outcomes is blocked bookkeeping, not
+			// rollout progress, so it must not advance or count as a failed batch.
+			if batchSize > 0 {
 				compartment.EvaluateAndUpdateBatchState(batchSize, successCount, failureCount)
 				batchAdvanced = true
 			}
diff --git a/operator/internal/controller/cluster_state_v2_test.go b/operator/internal/controller/cluster_state_v2_test.go
index 61f8470a6..7a741a8fb 100644
--- a/operator/internal/controller/cluster_state_v2_test.go
+++ b/operator/internal/controller/cluster_state_v2_test.go
@@ -2372,6 +2372,47 @@ var _ = Describe("Compartment Status Tests", func() {
 			Expect(strategy.CalculateBatchSize(10, rebased)).To(Equal(4))
 		})
 
+		It("evaluates residual failures when membership changes in the same reconcile", func() {
+			strategy := &v1alpha1.DeploymentStrategy{Fixed: &v1alpha1.FixedStrategy{
+				InitialBatch: kptr.To(2), BatchThreshold: kptr.To(100),
+				FailureThreshold: kptr.To(2), SafetyLimit: kptr.To(100),
+			}}
+			state := v1alpha1.BatchProcessingState{
+				CurrentBatch: 4, ConsecutiveFailures: 1, LastBatchSize: 2,
+			}
+			nodewright := &v1alpha1.NodeWright{Status: v1alpha1.NodeWrightStatus{
+				Status: v1alpha1.StatusWaiting,
+				CompartmentStatuses: map[string]v1alpha1.CompartmentStatus{
+					"moving": {Matched: 3, BatchState: &state},
+				},
+			}}
+			compartment := wrapper.NewCompartmentWrapper(&v1alpha1.Compartment{
+				Name: "moving", Budget: v1alpha1.DeploymentBudget{Count: kptr.To(4)}, Strategy: strategy,
+			}, &state)
+			for index := 0; index < 4; index++ {
+				node := wrapperMock.NewMockSkyhookNode(GinkgoT())
+				node.EXPECT().IsComplete().Return(false).Maybe()
+				status := v1alpha1.StatusWaiting
+				if index < 2 {
+					status = v1alpha1.StatusErroring
+				}
+				node.EXPECT().Status().Return(status).Maybe()
+				compartment.Nodes = append(compartment.Nodes, node)
+			}
+			rollout := &skyhookNodes{
+				skyhook:      wrapper.NewSkyhookWrapper(nodewright),
+				compartments: map[string]*wrapper.Compartment{"moving": compartment},
+			}
+
+			Expect(evaluateCompletedBatches(rollout)).To(BeTrue())
+			batch := rollout.GetSkyhook().Status.CompartmentStatuses["moving"].BatchState
+			Expect(batch.CurrentBatch).To(Equal(5))
+			Expect(batch.FailedNodes).To(Equal(2))
+			Expect(batch.ConsecutiveFailures).To(Equal(2))
+			Expect(batch.ShouldStop).To(BeTrue())
+			Expect(batch.LastBatchSize).To(Equal(1))
+		})
+
 		It("leaves completed-rollout batch state unchanged", func() {
 			record.Status.Status = v1alpha1.StatusComplete
 			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.CompletedNodes = 10
diff --git a/operator/internal/wrapper/compartment.go b/operator/internal/wrapper/compartment.go
index b0932c0e7..526092ae6 100644
--- a/operator/internal/wrapper/compartment.go
+++ b/operator/internal/wrapper/compartment.go
@@ -202,9 +202,9 @@ func (c *Compartment) IsBatchComplete() bool {
 	return c.getInProgressCount() == 0
 }
 
-// RebaselineBatchCheckpoints aligns cumulative checkpoints with the current
-// compartment membership without treating the membership change as a completed batch.
-func (c *Compartment) RebaselineBatchCheckpoints() bool {
+// RebaselineBatchCheckpoints absorbs terminal-count changes that can be
+// explained by compartment membership churn without hiding additional progress.
+func (c *Compartment) RebaselineBatchCheckpoints(membershipDelta int) bool {
 	currentCompleted := 0
 	currentFailed := 0
 	for _, node := range c.Nodes {
@@ -215,9 +215,46 @@ func (c *Compartment) RebaselineBatchCheckpoints() bool {
 		}
 	}
 
-	changed := c.BatchState.CompletedNodes != currentCompleted || c.BatchState.FailedNodes != currentFailed
-	c.BatchState.CompletedNodes = currentCompleted
-	c.BatchState.FailedNodes = currentFailed
+	remaining := membershipDelta
+	if remaining < 0 {
+		remaining = -remaining
+	}
+	if remaining == 0 {
+		return false
+	}
+
+	changed := false
+	absorb := func(checkpoint *int, current int, increasing bool) {
+		if remaining == 0 {
+			return
+		}
+		delta := current - *checkpoint
+		if (!increasing && delta >= 0) || (increasing && delta <= 0) {
+			return
+		}
+		if delta < 0 {
+			delta = -delta
+		}
+		amount := min(remaining, delta)
+		if increasing {
+			*checkpoint += amount
+		} else {
+			*checkpoint -= amount
+		}
+		remaining -= amount
+		changed = changed || amount > 0
+	}
+
+	// Added members can only explain increases; removed members can only explain
+	// decreases. Prefer completed-node attribution when both outcomes changed so
+	// membership churn does not suppress a safety-relevant failure unnecessarily.
+	if membershipDelta > 0 {
+		absorb(&c.BatchState.CompletedNodes, currentCompleted, true)
+		absorb(&c.BatchState.FailedNodes, currentFailed, true)
+	} else {
+		absorb(&c.BatchState.CompletedNodes, currentCompleted, false)
+		absorb(&c.BatchState.FailedNodes, currentFailed, false)
+	}
 	return changed
 }
 

From 978e7e2f5c03043cba927a26d1c7fd32a72753b1 Mon Sep 17 00:00:00 2001
From: Sylvester Kaczmarek
 <16242628+sylvesterkaczmarek@users.noreply.github.com>
Date: Sat, 12 Sep 2026 14:18:39 +0100
Subject: [PATCH 4/6] fix(operator): preserve failure attribution across
 membership churn

Signed-off-by: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com>
---
 .../internal/controller/cluster_state_v2.go   | 11 +++--
 .../controller/cluster_state_v2_test.go       | 34 ++++++++++-----
 operator/internal/wrapper/compartment.go      | 42 +++++++++++++------
 3 files changed, 61 insertions(+), 26 deletions(-)

diff --git a/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go
index fa930a57f..c31d4a888 100644
--- a/operator/internal/controller/cluster_state_v2.go
+++ b/operator/internal/controller/cluster_state_v2.go
@@ -1203,6 +1203,11 @@ func IntrospectSkyhook(skyhook SkyhookNodes, allSkyhooks []SkyhookNodes, logger
 		collectNodeStatus = v1alpha1.StatusBlocked
 	}
 
+	previousNodeStatus := make(map[string]v1alpha1.Status, len(skyhook.GetSkyhook().Status.NodeStatus))
+	for name, status := range skyhook.GetSkyhook().Status.NodeStatus {
+		previousNodeStatus[name] = status
+	}
+
 	if scrStatus != collectNodeStatus {
 		skyhook.SetStatus(collectNodeStatus)
 		change = true
@@ -1219,7 +1224,7 @@ func IntrospectSkyhook(skyhook SkyhookNodes, allSkyhooks []SkyhookNodes, logger
 	}
 
 	// Evaluate completed batches for compartments with deployment policies
-	if evaluateCompletedBatches(skyhook) {
+	if evaluateCompletedBatches(skyhook, previousNodeStatus) {
 		change = true
 	}
 
@@ -1232,7 +1237,7 @@ func IntrospectSkyhook(skyhook SkyhookNodes, allSkyhooks []SkyhookNodes, logger
 // evaluateCompletedBatches checks if any compartment batches are complete and evaluates them.
 // It returns true only when rollout state advances; checkpoint/status bookkeeping is
 // persisted separately and must not force the controller to abandon the rest of a reconcile.
-func evaluateCompletedBatches(skyhook SkyhookNodes) bool {
+func evaluateCompletedBatches(skyhook SkyhookNodes, previousNodeStatus map[string]v1alpha1.Status) bool {
 	compartments := skyhook.GetCompartments()
 	if len(compartments) == 0 {
 		return false // No compartments to evaluate
@@ -1254,7 +1259,7 @@ func evaluateCompletedBatches(skyhook SkyhookNodes) bool {
 		// evaluate any residual progress in this same reconcile.
 		if previous, exists := statuses[name]; exists && previous.Matched != len(compartment.GetNodes()) {
 			membershipDelta := len(compartment.GetNodes()) - previous.Matched
-			compartment.RebaselineBatchCheckpoints(membershipDelta)
+			compartment.RebaselineBatchCheckpoints(membershipDelta, previousNodeStatus)
 		}
 
 		if isComplete, successCount, failureCount := compartment.EvaluateCurrentBatch(); isComplete {
diff --git a/operator/internal/controller/cluster_state_v2_test.go b/operator/internal/controller/cluster_state_v2_test.go
index 7a741a8fb..4d5b4b819 100644
--- a/operator/internal/controller/cluster_state_v2_test.go
+++ b/operator/internal/controller/cluster_state_v2_test.go
@@ -2273,14 +2273,14 @@ var _ = Describe("Compartment Status Tests", func() {
 				expected.CompletedNodes, expected.FailedNodes = 0, 0
 				rollout := rebuild()
 
-				Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+				Expect(evaluateCompletedBatches(rollout, nil)).To(BeFalse())
 				Expect(rollout.GetSkyhook().Updated).To(BeTrue())
 				record = rollout.GetSkyhook().NodeWright.DeepCopy()
 				Expect(record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState).To(Equal(&expected))
 
 				reloaded := rebuild()
 				Expect(reloaded.GetCompartments()[v1alpha1.DefaultCompartmentName].GetBatchState()).To(Equal(expected))
-				Expect(evaluateCompletedBatches(reloaded)).To(BeFalse())
+				Expect(evaluateCompletedBatches(reloaded, nil)).To(BeFalse())
 				Expect(reloaded.GetSkyhook().Updated).To(BeFalse())
 			},
 			Entry("completed nodes left", 2, 0, false),
@@ -2292,12 +2292,12 @@ var _ = Describe("Compartment Status Tests", func() {
 		It("evaluates later failures after reloading a corrected checkpoint", func() {
 			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.FailedNodes = 88
 			rollout := rebuild()
-			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+			Expect(evaluateCompletedBatches(rollout, nil)).To(BeFalse())
 			record = rollout.GetSkyhook().NodeWright.DeepCopy()
 
 			reloaded := rebuild()
 			reloaded.GetNodes()[0].SetStatus(v1alpha1.StatusErroring)
-			Expect(evaluateCompletedBatches(reloaded)).To(BeTrue())
+			Expect(evaluateCompletedBatches(reloaded, nil)).To(BeTrue())
 			batch := reloaded.GetSkyhook().Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState
 			Expect(batch.CurrentBatch).To(Equal(5))
 			Expect(batch.FailedNodes).To(Equal(1))
@@ -2309,7 +2309,7 @@ var _ = Describe("Compartment Status Tests", func() {
 		It("persists first-batch initialization without advancing the strategy", func() {
 			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.CurrentBatch = 0
 			rollout := rebuild()
-			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+			Expect(evaluateCompletedBatches(rollout, nil)).To(BeFalse())
 			batch := rollout.GetSkyhook().Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState
 			Expect(batch.CurrentBatch).To(Equal(1))
 			Expect(batch.ConsecutiveFailures).To(Equal(1))
@@ -2347,7 +2347,7 @@ var _ = Describe("Compartment Status Tests", func() {
 			}
 
 			// All members leave. Persist a new zero baseline without advancing the strategy.
-			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+			Expect(evaluateCompletedBatches(rollout, nil)).To(BeFalse())
 			saved := rollout.GetSkyhook().Status.CompartmentStatuses["moving"]
 			Expect(saved.Matched).To(Equal(0))
 			Expect(saved.BatchState.CompletedNodes).To(Equal(0))
@@ -2357,14 +2357,22 @@ var _ = Describe("Compartment Status Tests", func() {
 			// Four already-complete nodes return with six pending nodes. Their return is
 			// another membership rebaseline, not a new four-node successful batch.
 			returned := wrapper.NewCompartmentWrapper(&compartment.Compartment, saved.BatchState)
+			previousNodeStatus := make(map[string]v1alpha1.Status, 10)
 			for index := 0; index < 10; index++ {
+				name := fmt.Sprintf("node-%d", index)
 				node := wrapperMock.NewMockSkyhookNode(GinkgoT())
+				node.EXPECT().GetNode().Return(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: name}}).Maybe()
 				node.EXPECT().IsComplete().Return(index < 4).Maybe()
 				node.EXPECT().Status().Return(v1alpha1.StatusWaiting).Maybe()
+				if index < 4 {
+					previousNodeStatus[name] = v1alpha1.StatusComplete
+				} else {
+					previousNodeStatus[name] = v1alpha1.StatusWaiting
+				}
 				returned.Nodes = append(returned.Nodes, node)
 			}
 			rollout.compartments["moving"] = returned
-			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+			Expect(evaluateCompletedBatches(rollout, previousNodeStatus)).To(BeFalse())
 			rebased := rollout.GetSkyhook().Status.CompartmentStatuses["moving"].BatchState
 			Expect(rebased.CompletedNodes).To(Equal(4))
 			Expect(rebased.CurrentBatch).To(Equal(4))
@@ -2389,13 +2397,17 @@ var _ = Describe("Compartment Status Tests", func() {
 			compartment := wrapper.NewCompartmentWrapper(&v1alpha1.Compartment{
 				Name: "moving", Budget: v1alpha1.DeploymentBudget{Count: kptr.To(4)}, Strategy: strategy,
 			}, &state)
+			previousNodeStatus := make(map[string]v1alpha1.Status, 4)
 			for index := 0; index < 4; index++ {
+				name := fmt.Sprintf("node-%d", index)
 				node := wrapperMock.NewMockSkyhookNode(GinkgoT())
+				node.EXPECT().GetNode().Return(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: name}}).Maybe()
 				node.EXPECT().IsComplete().Return(false).Maybe()
 				status := v1alpha1.StatusWaiting
 				if index < 2 {
 					status = v1alpha1.StatusErroring
 				}
+				previousNodeStatus[name] = v1alpha1.StatusWaiting
 				node.EXPECT().Status().Return(status).Maybe()
 				compartment.Nodes = append(compartment.Nodes, node)
 			}
@@ -2404,13 +2416,13 @@ var _ = Describe("Compartment Status Tests", func() {
 				compartments: map[string]*wrapper.Compartment{"moving": compartment},
 			}
 
-			Expect(evaluateCompletedBatches(rollout)).To(BeTrue())
+			Expect(evaluateCompletedBatches(rollout, previousNodeStatus)).To(BeTrue())
 			batch := rollout.GetSkyhook().Status.CompartmentStatuses["moving"].BatchState
 			Expect(batch.CurrentBatch).To(Equal(5))
 			Expect(batch.FailedNodes).To(Equal(2))
 			Expect(batch.ConsecutiveFailures).To(Equal(2))
 			Expect(batch.ShouldStop).To(BeTrue())
-			Expect(batch.LastBatchSize).To(Equal(1))
+			Expect(batch.LastBatchSize).To(Equal(2))
 		})
 
 		It("leaves completed-rollout batch state unchanged", func() {
@@ -2418,7 +2430,7 @@ var _ = Describe("Compartment Status Tests", func() {
 			record.Status.CompartmentStatuses[v1alpha1.DefaultCompartmentName].BatchState.CompletedNodes = 10
 			rollout := rebuild()
 			before := rollout.GetSkyhook().Status.DeepCopy()
-			Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+			Expect(evaluateCompletedBatches(rollout, nil)).To(BeFalse())
 			Expect(rollout.GetSkyhook().Status).To(Equal(*before))
 			Expect(rollout.GetSkyhook().Updated).To(BeFalse())
 		})
@@ -2430,7 +2442,7 @@ var _ = Describe("Compartment Status Tests", func() {
 				// This table isolates batch persistence; SetStatus itself marks the Skyhook dirty.
 				rollout.GetSkyhook().Updated = false
 				beforeBatch := rollout.GetCompartments()[v1alpha1.DefaultCompartmentName].GetBatchState()
-				Expect(evaluateCompletedBatches(rollout)).To(BeFalse())
+				Expect(evaluateCompletedBatches(rollout, nil)).To(BeFalse())
 				Expect(rollout.GetCompartments()[v1alpha1.DefaultCompartmentName].GetBatchState()).To(Equal(beforeBatch))
 			},
 			Entry("no progress", v1alpha1.StatusWaiting),
diff --git a/operator/internal/wrapper/compartment.go b/operator/internal/wrapper/compartment.go
index 526092ae6..1336ca3a3 100644
--- a/operator/internal/wrapper/compartment.go
+++ b/operator/internal/wrapper/compartment.go
@@ -202,16 +202,24 @@ func (c *Compartment) IsBatchComplete() bool {
 	return c.getInProgressCount() == 0
 }
 
-// RebaselineBatchCheckpoints absorbs terminal-count changes that can be
-// explained by compartment membership churn without hiding additional progress.
-func (c *Compartment) RebaselineBatchCheckpoints(membershipDelta int) bool {
+// RebaselineBatchCheckpoints absorbs terminal-count changes attributable to
+// compartment membership churn without hiding progress from existing members.
+func (c *Compartment) RebaselineBatchCheckpoints(membershipDelta int, previousNodeStatus map[string]v1alpha1.Status) bool {
 	currentCompleted := 0
 	currentFailed := 0
+	previouslyCompleted := 0
+	previouslyFailed := 0
 	for _, node := range c.Nodes {
 		if node.IsComplete() {
 			currentCompleted++
+			if previousNodeStatus != nil && previousNodeStatus[node.GetNode().Name] == v1alpha1.StatusComplete {
+				previouslyCompleted++
+			}
 		} else if node.Status() == v1alpha1.StatusErroring {
 			currentFailed++
+			if previousNodeStatus != nil && previousNodeStatus[node.GetNode().Name] == v1alpha1.StatusErroring {
+				previouslyFailed++
+			}
 		}
 	}
 
@@ -224,8 +232,8 @@ func (c *Compartment) RebaselineBatchCheckpoints(membershipDelta int) bool {
 	}
 
 	changed := false
-	absorb := func(checkpoint *int, current int, increasing bool) {
-		if remaining == 0 {
+	absorb := func(checkpoint *int, current int, increasing bool, limit int) {
+		if remaining == 0 || limit == 0 {
 			return
 		}
 		delta := current - *checkpoint
@@ -236,6 +244,9 @@ func (c *Compartment) RebaselineBatchCheckpoints(membershipDelta int) bool {
 			delta = -delta
 		}
 		amount := min(remaining, delta)
+		if limit > 0 {
+			amount = min(amount, limit)
+		}
 		if increasing {
 			*checkpoint += amount
 		} else {
@@ -245,15 +256,22 @@ func (c *Compartment) RebaselineBatchCheckpoints(membershipDelta int) bool {
 		changed = changed || amount > 0
 	}
 
-	// Added members can only explain increases; removed members can only explain
-	// decreases. Prefer completed-node attribution when both outcomes changed so
-	// membership churn does not suppress a safety-relevant failure unnecessarily.
 	if membershipDelta > 0 {
-		absorb(&c.BatchState.CompletedNodes, currentCompleted, true)
-		absorb(&c.BatchState.FailedNodes, currentFailed, true)
+		// Only absorb terminal states that were already terminal before this reconcile.
+		// This keeps failures that happened to existing members in the current batch.
+		completedLimit := currentCompleted - c.BatchState.CompletedNodes
+		failedLimit := currentFailed - c.BatchState.FailedNodes
+		if previousNodeStatus != nil {
+			completedLimit = max(0, previouslyCompleted-c.BatchState.CompletedNodes)
+			failedLimit = max(0, previouslyFailed-c.BatchState.FailedNodes)
+		}
+		absorb(&c.BatchState.CompletedNodes, currentCompleted, true, completedLimit)
+		absorb(&c.BatchState.FailedNodes, currentFailed, true, failedLimit)
 	} else {
-		absorb(&c.BatchState.CompletedNodes, currentCompleted, false)
-		absorb(&c.BatchState.FailedNodes, currentFailed, false)
+		// Removed members can only explain decreases; field-specific correction in
+		// EvaluateCurrentBatch preserves positive progress in the other outcome.
+		absorb(&c.BatchState.CompletedNodes, currentCompleted, false, remaining)
+		absorb(&c.BatchState.FailedNodes, currentFailed, false, remaining)
 	}
 	return changed
 }

From ed639a36df0192852df28bd0a53f106e542a4c6e Mon Sep 17 00:00:00 2001
From: Sylvester Kaczmarek
 <16242628+sylvesterkaczmarek@users.noreply.github.com>
Date: Tue, 15 Sep 2026 01:45:46 +0100
Subject: [PATCH 5/6] fix(operator): satisfy controller lint

Signed-off-by: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com>
---
 .../internal/controller/cluster_state_v2.go   |  5 ++-
 .../internal/controller/skyhook_controller.go | 32 +++++++++++--------
 2 files changed, 21 insertions(+), 16 deletions(-)

diff --git a/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go
index d081f1044..2da83df0e 100644
--- a/operator/internal/controller/cluster_state_v2.go
+++ b/operator/internal/controller/cluster_state_v2.go
@@ -1679,7 +1679,7 @@ func (skyhook *skyhookNodes) AssignNodeToCompartment(node wrapper.SkyhookNode) (
 }
 
 // persistCompartmentStatus writes a compartment status only when it changed.
-func persistCompartmentStatus(skyhook SkyhookNodes, compartment *wrapper.Compartment) bool {
+func persistCompartmentStatus(skyhook SkyhookNodes, compartment *wrapper.Compartment) {
 	statuses := skyhook.GetSkyhook().Status.CompartmentStatuses
 	if statuses == nil {
 		statuses = make(map[string]v1alpha1.CompartmentStatus)
@@ -1689,12 +1689,11 @@ func persistCompartmentStatus(skyhook SkyhookNodes, compartment *wrapper.Compart
 	name := compartment.GetName()
 	newStatus := buildCompartmentStatus(compartment)
 	if existing, ok := statuses[name]; ok && compartmentStatusEqual(existing, newStatus) {
-		return false
+		return
 	}
 
 	statuses[name] = newStatus
 	skyhook.GetSkyhook().Updated = true
-	return true
 }
 
 // updateCompartmentStatuses updates compartment statuses for all current compartments.
diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go
index 6307f1fbd..b9463a5e9 100644
--- a/operator/internal/controller/skyhook_controller.go
+++ b/operator/internal/controller/skyhook_controller.go
@@ -532,19 +532,8 @@ func (r *SkyhookReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		}
 
 		changed := IntrospectSkyhook(skyhook, clusterState.skyhooks, logger)
-
-		// Report after introspection so compartment status reflects the state just
-		// evaluated. Persist status-only bookkeeping without turning it into an
-		// operational change that aborts processing of later NodeWrights.
-		skyhook.ReportState()
-		if changed || skyhook.GetSkyhook().Updated {
-			_, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
-			if len(errs) > 0 {
-				return ctrl.Result{RequeueAfter: time.Second * 2}, utilerrors.NewAggregate(errs)
-			}
-		}
-		if changed {
-			return ctrl.Result{RequeueAfter: time.Second * 2}, nil
+		if yes, result, err := shouldReturn(r.persistIntrospectedSkyhook(ctx, clusterState, skyhook, changed)); yes {
+			return result, err
 		}
 
 		_, err := HandleVersionChange(skyhook)
@@ -683,6 +672,23 @@ func hasReadyNodesForSkyhook(skyhook SkyhookNodes, allSkyhooks []SkyhookNodes) (
 	return false, nil
 }
 
+// persistIntrospectedSkyhook reports and saves state after introspection.
+// Status-only bookkeeping is persisted without forcing an early requeue.
+func (r *SkyhookReconciler) persistIntrospectedSkyhook(
+	ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes, changed bool,
+) (bool, error) {
+	skyhook.ReportState()
+	if !changed && !skyhook.GetSkyhook().Updated {
+		return false, nil
+	}
+
+	_, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
+	if len(errs) > 0 {
+		return false, utilerrors.NewAggregate(errs)
+	}
+	return changed, nil
+}
+
 func shouldReturn(updates bool, err error) (bool, ctrl.Result, error) {
 	if err != nil {
 		return true, ctrl.Result{}, err

From 1bbcbb8f2dce6d932e1f46c4c66e14da81a86cce Mon Sep 17 00:00:00 2001
From: Sylvester Kaczmarek
 <16242628+sylvesterkaczmarek@users.noreply.github.com>
Date: Wed, 16 Sep 2026 22:43:38 +0100
Subject: [PATCH 6/6] fix(operator): handle unobserved churn state

Signed-off-by: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com>
---
 docs/user-guide/deployment-policy.md          | 15 ++++---
 .../internal/controller/cluster_state_v2.go   | 29 ++++++++++++--
 .../controller/cluster_state_v2_test.go       | 40 +++++++++++++++++--
 operator/internal/wrapper/compartment.go      | 14 ++++++-
 4 files changed, 84 insertions(+), 14 deletions(-)

diff --git a/docs/user-guide/deployment-policy.md b/docs/user-guide/deployment-policy.md
index 8748fbad2..2f6ecb97d 100644
--- a/docs/user-guide/deployment-policy.md
+++ b/docs/user-guide/deployment-policy.md
@@ -372,12 +372,15 @@ compartments:
 
 ## Checkpoint Corrections
 
-When compartment membership changes, the operator rebaselines completed and failed
-checkpoint counts to the current members. Membership changes are bookkeeping, not
-completed batches: they do not advance the batch, change the previous batch result,
-or clear a stop decision. If only one outcome count later moves backwards, for
-example when a node recovers from `Erroring` to `Complete`, only that checkpoint is
-corrected so positive progress in the other outcome is still evaluated.
+When compartment membership changes, the operator absorbs completed and failed
+checkpoint changes only up to the terminal outcomes that the membership churn can
+explain. Any residual genuine progress is evaluated in the same reconciliation pass,
+so one pass can both rebaseline for churn and advance a completed batch. A returning
+terminal node with no trustworthy prior observation, including after pause or disable,
+is treated as already terminal rather than as new batch progress. If only one outcome
+count later moves backwards, for example when a node recovers from `Erroring` to
+`Complete`, only that checkpoint is corrected so positive progress in the other outcome
+is still evaluated.
 
 Corrections are saved to status without stopping the rest of the reconciliation pass,
 and later batches continue from the saved baseline after an operator restart.
diff --git a/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go
index 2da83df0e..e3eaea832 100644
--- a/operator/internal/controller/cluster_state_v2.go
+++ b/operator/internal/controller/cluster_state_v2.go
@@ -1264,9 +1264,32 @@ func evaluateCompletedBatches(skyhook SkyhookNodes, previousNodeStatus map[strin
 		if isComplete, successCount, failureCount := compartment.EvaluateCurrentBatch(); isComplete {
 			batchSize := successCount + failureCount
 
-			// A completed batch with no terminal outcomes is blocked bookkeeping, not
-			// rollout progress, so it must not advance or count as a failed batch.
-			if batchSize > 0 {
+			// Preserve the existing blocked-node semantics while membership churn is
+			// rebaselined above. Blocked nodes are temporary and must not become a
+			// failed batch or advance the rollout on their own.
+			blockedCount := 0
+			for _, node := range compartment.GetNodes() {
+				if node.Status() == v1alpha1.StatusBlocked {
+					blockedCount++
+				}
+			}
+
+			shouldAdvance := true
+			if batchSize == 0 {
+				if blockedCount > 0 && blockedCount == len(compartment.GetNodes()) {
+					shouldAdvance = false
+				} else if blockedCount > 0 {
+					batchSize = blockedCount
+				} else if compartment.GetBatchState().LastBatchSize > 0 {
+					batchSize = compartment.GetBatchState().LastBatchSize
+				}
+			}
+
+			if batchSize > 0 && successCount == 0 && failureCount == 0 && blockedCount == batchSize {
+				shouldAdvance = false
+			}
+
+			if shouldAdvance {
 				compartment.EvaluateAndUpdateBatchState(batchSize, successCount, failureCount)
 				batchAdvanced = true
 			}
diff --git a/operator/internal/controller/cluster_state_v2_test.go b/operator/internal/controller/cluster_state_v2_test.go
index c375c6cf9..5a12743d2 100644
--- a/operator/internal/controller/cluster_state_v2_test.go
+++ b/operator/internal/controller/cluster_state_v2_test.go
@@ -2556,9 +2556,9 @@ var _ = Describe("Compartment Status Tests", func() {
 				node.EXPECT().GetNode().Return(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: name}}).Maybe()
 				node.EXPECT().IsComplete().Return(index < 4).Maybe()
 				node.EXPECT().Status().Return(v1alpha1.StatusWaiting).Maybe()
-				if index < 4 {
-					previousNodeStatus[name] = v1alpha1.StatusComplete
-				} else {
+				// Departed nodes are pruned from production NodeStatus. Only nodes that
+				// remained observable have a trustworthy prior status here.
+				if index >= 4 {
 					previousNodeStatus[name] = v1alpha1.StatusWaiting
 				}
 				returned.Nodes = append(returned.Nodes, node)
@@ -2572,6 +2572,40 @@ var _ = Describe("Compartment Status Tests", func() {
 			Expect(strategy.CalculateBatchSize(10, rebased)).To(Equal(4))
 		})
 
+		DescribeTable("absorbs returning completions without a trustworthy prior observation",
+			func(previousNodeStatus map[string]v1alpha1.Status) {
+				strategy := &v1alpha1.DeploymentStrategy{Fixed: &v1alpha1.FixedStrategy{
+					InitialBatch: kptr.To(1), BatchThreshold: kptr.To(100), SafetyLimit: kptr.To(100),
+				}}
+				state := v1alpha1.BatchProcessingState{CurrentBatch: 4, LastBatchSize: 1}
+				nodewright := &v1alpha1.NodeWright{Status: v1alpha1.NodeWrightStatus{
+					Status: v1alpha1.StatusWaiting,
+					CompartmentStatuses: map[string]v1alpha1.CompartmentStatus{
+						"moving": {Matched: 0, BatchState: &state},
+					},
+				}}
+				compartment := wrapper.NewCompartmentWrapper(&v1alpha1.Compartment{
+					Name: "moving", Budget: v1alpha1.DeploymentBudget{Count: kptr.To(1)}, Strategy: strategy,
+				}, &state)
+				node := wrapperMock.NewMockSkyhookNode(GinkgoT())
+				node.EXPECT().GetNode().Return(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-0"}}).Maybe()
+				node.EXPECT().IsComplete().Return(true).Maybe()
+				node.EXPECT().Status().Return(v1alpha1.StatusComplete).Maybe()
+				compartment.Nodes = append(compartment.Nodes, node)
+				rollout := &skyhookNodes{
+					skyhook:      wrapper.NewSkyhookWrapper(nodewright),
+					compartments: map[string]*wrapper.Compartment{"moving": compartment},
+				}
+
+				Expect(evaluateCompletedBatches(rollout, previousNodeStatus)).To(BeFalse())
+				batch := rollout.GetSkyhook().Status.CompartmentStatuses["moving"].BatchState
+				Expect(batch.CompletedNodes).To(Equal(1))
+				Expect(batch.CurrentBatch).To(Equal(4))
+			},
+			Entry("when the node is absent", map[string]v1alpha1.Status{}),
+			Entry("when pause overwrote the node status", map[string]v1alpha1.Status{"node-0": v1alpha1.StatusPaused}),
+		)
+
 		It("evaluates residual failures when membership changes in the same reconcile", func() {
 			strategy := &v1alpha1.DeploymentStrategy{Fixed: &v1alpha1.FixedStrategy{
 				InitialBatch: kptr.To(2), BatchThreshold: kptr.To(100),
diff --git a/operator/internal/wrapper/compartment.go b/operator/internal/wrapper/compartment.go
index e68e814e8..8752364a4 100644
--- a/operator/internal/wrapper/compartment.go
+++ b/operator/internal/wrapper/compartment.go
@@ -211,15 +211,25 @@ func (c *Compartment) RebaselineBatchCheckpoints(membershipDelta int, previousNo
 	currentFailed := 0
 	previouslyCompleted := 0
 	previouslyFailed := 0
+	wasTerminal := func(name string, want v1alpha1.Status) bool {
+		prev, ok := previousNodeStatus[name]
+		// Missing or operator-imposed placeholder state means there is no trustworthy
+		// prior observation. Treat a terminal node as already terminal so returning
+		// members are absorbed as churn instead of counted as new batch progress.
+		if !ok || prev == v1alpha1.StatusPaused || prev == v1alpha1.StatusDisabled {
+			return true
+		}
+		return prev == want
+	}
 	for _, node := range c.Nodes {
 		if node.IsComplete() {
 			currentCompleted++
-			if previousNodeStatus != nil && previousNodeStatus[node.GetNode().Name] == v1alpha1.StatusComplete {
+			if wasTerminal(node.GetNode().Name, v1alpha1.StatusComplete) {
 				previouslyCompleted++
 			}
 		} else if node.Status() == v1alpha1.StatusErroring {
 			currentFailed++
-			if previousNodeStatus != nil && previousNodeStatus[node.GetNode().Name] == v1alpha1.StatusErroring {
+			if wasTerminal(node.GetNode().Name, v1alpha1.StatusErroring) {
 				previouslyFailed++
 			}
 		}