diff --git a/docs/user-guide/deployment-policy.md b/docs/user-guide/deployment-policy.md
index 0fa231da..2f6ecb97 100644
--- a/docs/user-guide/deployment-policy.md
+++ b/docs/user-guide/deployment-policy.md
@@ -370,6 +370,21 @@ compartments:
 
 ---
 
+## Checkpoint Corrections
+
+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.
+
 ## 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.
diff --git a/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go
index 6104f828..af2c05ac 100644
--- a/operator/internal/controller/cluster_state_v2.go
+++ b/operator/internal/controller/cluster_state_v2.go
@@ -1208,8 +1208,14 @@ 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
 	}
 
 	for _, node := range skyhook.GetNodes() {
@@ -1223,36 +1229,50 @@ func IntrospectSkyhook(skyhook SkyhookNodes, allSkyhooks []SkyhookNodes, logger
 	}
 
 	// Evaluate completed batches for compartments with deployment policies
-	if evaluateCompletedBatches(skyhook) {
+	if evaluateCompletedBatches(skyhook, previousNodeStatus) {
 		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
-func evaluateCompletedBatches(skyhook SkyhookNodes) bool {
+// 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, previousNodeStatus map[string]v1alpha1.Status) bool {
 	compartments := skyhook.GetCompartments()
 	if len(compartments) == 0 {
 		return false // No compartments to evaluate
 	}
 
 	// 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 {
+		name := compartment.GetName()
+
+		// 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()) {
+			membershipDelta := len(compartment.GetNodes()) - previous.Matched
+			compartment.RebaselineBatchCheckpoints(membershipDelta, previousNodeStatus)
+		}
+
 		if isComplete, successCount, failureCount := compartment.EvaluateCurrentBatch(); isComplete {
 			batchSize := successCount + failureCount
 
-			// Count blocked nodes to determine if we should skip batch evaluation
+			// 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 {
@@ -1260,49 +1280,33 @@ 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)
 			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)
-
-			// 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)
+			if shouldAdvance {
+				compartment.EvaluateAndUpdateBatchState(batchSize, successCount, failureCount)
+				batchAdvanced = true
 			}
-			// 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
 		}
+
+		// 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 {
@@ -1448,18 +1452,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{
@@ -1705,21 +1707,28 @@ 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) {
+	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
 	}
 
-	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
+}
+
+// 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 47fc6482..5a12743d 100644
--- a/operator/internal/controller/cluster_state_v2_test.go
+++ b/operator/internal/controller/cluster_state_v2_test.go
@@ -2315,7 +2315,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() {
@@ -2403,6 +2404,279 @@ 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: {
+							Matched: 3,
+							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, 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, nil)).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, nil)).To(BeFalse())
+			record = rollout.GetSkyhook().NodeWright.DeepCopy()
+
+			reloaded := rebuild()
+			reloaded.GetNodes()[0].SetStatus(v1alpha1.StatusErroring)
+			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))
+			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, nil)).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, nil)).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)
+			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()
+				// 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)
+			}
+			rollout.compartments["moving"] = returned
+			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))
+			Expect(rebased.LastBatchSize).To(Equal(2))
+			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),
+				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)
+			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)
+			}
+			rollout := &skyhookNodes{
+				skyhook:      wrapper.NewSkyhookWrapper(nodewright),
+				compartments: map[string]*wrapper.Compartment{"moving": compartment},
+			}
+
+			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(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, nil)).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)
+				// This table isolates batch persistence; SetStatus itself marks the Skyhook dirty.
+				rollout.GetSkyhook().Updated = false
+				beforeBatch := rollout.GetCompartments()[v1alpha1.DefaultCompartmentName].GetBatchState()
+				Expect(evaluateCompletedBatches(rollout, nil)).To(BeFalse())
+				Expect(rollout.GetCompartments()[v1alpha1.DefaultCompartmentName].GetBatchState()).To(Equal(beforeBatch))
+			},
+			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{
diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go
index c7174dd0..f91a30c9 100644
--- a/operator/internal/controller/skyhook_controller.go
+++ b/operator/internal/controller/skyhook_controller.go
@@ -505,11 +505,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)
 			}
@@ -534,12 +533,8 @@ func (r *SkyhookReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		}
 
 		changed := IntrospectSkyhook(skyhook, clusterState.skyhooks, logger)
-		if changed {
-			_, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
-			if len(errs) > 0 {
-				return ctrl.Result{RequeueAfter: time.Second * 2}, utilerrors.NewAggregate(errs)
-			}
-			return ctrl.Result{RequeueAfter: time.Second * 2}, nil
+		if yes, result, err := shouldReturn(r.persistIntrospectedSkyhook(ctx, clusterState, skyhook, changed)); yes {
+			return result, err
 		}
 
 		_, err := HandleVersionChange(skyhook)
@@ -789,6 +784,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
diff --git a/operator/internal/wrapper/compartment.go b/operator/internal/wrapper/compartment.go
index 0c9ea6ce..8752364a 100644
--- a/operator/internal/wrapper/compartment.go
+++ b/operator/internal/wrapper/compartment.go
@@ -204,6 +204,90 @@ func (c *Compartment) IsBatchComplete() bool {
 	return c.getInProgressCount() == 0
 }
 
+// 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
+	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 wasTerminal(node.GetNode().Name, v1alpha1.StatusComplete) {
+				previouslyCompleted++
+			}
+		} else if node.Status() == v1alpha1.StatusErroring {
+			currentFailed++
+			if wasTerminal(node.GetNode().Name, v1alpha1.StatusErroring) {
+				previouslyFailed++
+			}
+		}
+	}
+
+	remaining := membershipDelta
+	if remaining < 0 {
+		remaining = -remaining
+	}
+	if remaining == 0 {
+		return false
+	}
+
+	changed := false
+	absorb := func(checkpoint *int, current int, increasing bool, limit int) {
+		if remaining == 0 || limit == 0 {
+			return
+		}
+		delta := current - *checkpoint
+		if (!increasing && delta >= 0) || (increasing && delta <= 0) {
+			return
+		}
+		if delta < 0 {
+			delta = -delta
+		}
+		amount := min(remaining, delta)
+		if limit > 0 {
+			amount = min(amount, limit)
+		}
+		if increasing {
+			*checkpoint += amount
+		} else {
+			*checkpoint -= amount
+		}
+		remaining -= amount
+		changed = changed || amount > 0
+	}
+
+	if membershipDelta > 0 {
+		// 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 {
+		// 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
+}
+
 // 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) {
@@ -232,18 +316,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 c1cc2335..7d22c2ad 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{