@@ -9,6 +9,7 @@ import (
 
 	helmv2 "github.com/fluxcd/helm-controller/api/v2"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -19,6 +20,11 @@ import (
 
 const (
 	annotationLastProcessedVersion = "flux-plunger.cozystack.io/last-processed-version"
+	// annotationSuspendedByPlunger marks a suspension that flux-plunger itself
+	// set, so recovery can be told apart from a suspension an operator or another
+	// controller owns. It is written atomically with spec.suspend=true and
+	// removed atomically with spec.suspend=false.
+	annotationSuspendedByPlunger   = "flux-plunger.cozystack.io/suspended"
 	errorMessageNoDeployedReleases = "has no deployed releases"
 	fieldManager                   = "flux-client-side-apply"
 )
@@ -41,50 +47,69 @@ func (r *FluxPlunger) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Res
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	// Check if HelmRelease is suspended
-	if hr.Spec.Suspend {
-		logger.Info("HelmRelease is suspended, checking if we need to unsuspend")
-
-		// Get the list of Helm release secrets
-		secrets, err := r.listHelmReleaseSecrets(ctx, hr.Namespace, hr.Name)
-		if err != nil {
-			logger.Error(err, "Failed to list Helm release secrets")
-			return ctrl.Result{}, err
+	// A HelmRelease that is being deleted must never stay suspended. helm-controller
+	// v1.5.x skips the Helm uninstall and drops its finalizer for a suspended
+	// HelmRelease (https://github.com/fluxcd/helm-controller/blob/v1.5.0/internal/controller/helmrelease_controller.go#L447-L475),
+	// which orphans every resource the release owns. If flux-plunger owns the
+	// suspension, release it so the uninstall proceeds; in every case do not begin
+	// recovery on a terminating object.
+	if !hr.DeletionTimestamp.IsZero() {
+		if hr.Spec.Suspend && ownsSuspension(hr) {
+			logger.Info("HelmRelease is being deleted while suspended by flux-plunger, clearing suspension so uninstall can proceed")
+			if err := r.unsuspendHelmRelease(ctx, hr); err != nil {
+				// Requeue with backoff: leaving the suspension is exactly what
+				// orphans the release, so retry rather than wait for another event.
+				logger.Info("Could not clear suspension on deleted HelmRelease, requeuing", "error", err.Error())
+				return ctrl.Result{}, err
+			}
 		}
+		return ctrl.Result{}, nil
+	}
 
-		// If no secrets, treat latest version as 0
-		latestVersion := 0
-		if len(secrets) > 0 {
-			latestSecret := getLatestSecret(secrets)
-			latestVersion = extractVersionNumber(latestSecret.Name)
-		} else {
-			logger.Info("No Helm release secrets found while suspended, treating as version 0")
+	// Check if HelmRelease is suspended
+	if hr.Spec.Suspend {
+		// Ownership is decided solely by the marker flux-plunger writes atomically
+		// with the suspension. A suspension we own but see in a fresh reconcile means
+		// recovery crashed mid-flight rather than another instance running now: the
+		// Deployment pins replicas: 1 (packages/system/flux-plunger, which is what
+		// makes this annotation-as-state safe — see the comment there), and
+		// controller-runtime serialises reconciles per key within that one process,
+		// so there is no concurrent owner to race. Releasing it un-sticks the
+		// release and lets error-recovery start over. A suspension without the marker
+		// belongs to an operator or another controller and must be left untouched —
+		// the processed-version annotation is deliberately NOT consulted here, because
+		// it is never cleared and would misread a much later, unrelated operator
+		// suspension as ours once the release's latest revision stops advancing.
+		if !ownsSuspension(hr) {
+			logger.Info("HelmRelease is suspended by external process, skipping")
+			return ctrl.Result{}, nil
 		}
 
-		// Check if version is previous to just processed (latestVersion+1 == processedVersion)
-		// This is the ONLY condition when we unsuspend
-		shouldUnsuspend := false
-		if hr.Annotations != nil {
-			if processedVersionStr, exists := hr.Annotations[annotationLastProcessedVersion]; exists {
-				processedVersion, err := strconv.Atoi(processedVersionStr)
-				if err == nil && latestVersion+1 == processedVersion {
-					shouldUnsuspend = true
-				}
-			}
+		logger.Info("Removing suspension owned by flux-plunger")
+		if err := r.unsuspendHelmRelease(ctx, hr); err != nil {
+			// Requeue with backoff so a stuck owned suspension is not left until the
+			// next informer resync, symmetric with the deletion and recovery paths.
+			logger.Info("Could not unsuspend HelmRelease, requeuing", "error", err.Error())
+			return ctrl.Result{}, err
 		}
+		return ctrl.Result{}, nil
+	}
 
-		if shouldUnsuspend {
-			// Unsuspend the HelmRelease
-			logger.Info("Secret was already deleted in previous run, removing suspend", "latest", latestVersion, "processed", latestVersion+1)
-			if err := r.unsuspendHelmRelease(ctx, hr); err != nil {
-				logger.Info("Could not unsuspend HelmRelease, will retry on next reconcile", "error", err.Error())
-				return ctrl.Result{}, nil
-			}
-			return ctrl.Result{}, nil
+	// A suspension flux-plunger set is marked with its ownership annotation and
+	// cleared atomically with it. But an external actor — an operator, `cozyhr
+	// resume`, the backup controller's own unsuspend — can lift the suspension by
+	// writing spec.suspend=false WITHOUT removing the marker, since nothing else
+	// removes it. A marker left on a non-suspended release would make a LATER,
+	// unrelated operator suspension look plunger-owned and get cleared out from
+	// under them: the one boundary this controller must never cross. Reap the stale
+	// marker the moment we observe it on a release that is no longer suspended,
+	// before it can be inherited. The watch predicate wakes us for this transition.
+	if ownsSuspension(hr) {
+		logger.Info("Reaping stale flux-plunger ownership marker on a non-suspended HelmRelease")
+		if err := r.clearOwnershipMarker(ctx, hr); err != nil {
+			logger.Info("Could not reap stale ownership marker, requeuing", "error", err.Error())
+			return ctrl.Result{}, err
 		}
-
-		// If not previous to processed, skip all actions
-		logger.Info("HelmRelease is suspended by external process, skipping", "latest", latestVersion)
 		return ctrl.Result{}, nil
 	}
 
@@ -132,10 +157,76 @@ func (r *FluxPlunger) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Res
 
 	// Suspend the HelmRelease
 	logger.Info("Suspending HelmRelease")
-	if err := r.suspendHelmRelease(ctx, hr); err != nil {
-		// Optimistic lock conflicts are normal - FluxCD also updates HelmRelease
-		// Don't return error, just log and let controller-runtime requeue on next update
-		logger.Info("Could not suspend HelmRelease, will retry on next reconcile", "error", err.Error())
+	acquired, err := r.suspendHelmRelease(ctx, hr)
+	if err != nil {
+		// An optimistic-lock conflict is expected here — helm-controller and Flux
+		// also patch the HelmRelease — and the conflicting write is itself an update
+		// event that requeues us, so a conflict needs no error return. Any OTHER
+		// error (a transient API failure) produces no such follow-up event, so
+		// return it to requeue with backoff, symmetric with every step below;
+		// swallowing it would strand a stuck release until the next informer resync.
+		if apierrors.IsConflict(err) {
+			logger.Info("Suspend conflicted with a concurrent write, will retry on the resulting update", "error", err.Error())
+			return ctrl.Result{}, nil
+		}
+		logger.Info("Could not suspend HelmRelease, requeuing", "error", err.Error())
+		return ctrl.Result{}, err
+	}
+	if !acquired {
+		// The release was suspended by someone else, or started terminating, between
+		// our read and our write, so flux-plunger does not own it. Do not delete its
+		// storage Secret; let the next reconcile handle it (the suspended branch, or
+		// the deletion branch which lets helm-controller uninstall).
+		logger.Info("HelmRelease was not acquired for recovery, deferring to next reconcile")
+		return ctrl.Result{}, nil
+	}
+
+	// From here flux-plunger owns the suspension (acquired above). Guarantee it is
+	// released before returning from any path below: a step that fails must never
+	// leave the HelmRelease suspended, because a delete arriving on a suspended
+	// release makes helm-controller skip the uninstall and orphan the release's
+	// resources. The unsuspend is idempotent, so the happy path running it
+	// explicitly and this deferred fallback are safe together.
+	completed := false
+	defer func() {
+		if completed {
+			return
+		}
+		if err := r.unsuspendHelmRelease(ctx, hr); err != nil {
+			logger.Info("Could not clear suspension after failed recovery step, will retry on next reconcile", "error", err.Error())
+		}
+	}()
+
+	// Record the version we are about to reclaim BEFORE deleting its storage
+	// Secret. The "already processed" guard above keys on this annotation, so
+	// writing it first means a failed write can never authorise deleting the NEXT
+	// revision on the following requeue: on failure we return here, the deferred
+	// unsuspend clears the suspension, and the retry re-deletes the SAME revision
+	// (idempotent) instead of walking down the release history one Secret per
+	// requeue until none is left — which would itself orphan the release. Requeue
+	// with backoff on failure, symmetric with every other step.
+	logger.Info("Updating annotation with processed version", "version", latestVersion)
+	if err := r.updateProcessedVersionAnnotation(ctx, hr, latestVersion); err != nil {
+		logger.Info("Could not update annotation, requeuing", "error", err.Error())
+		return ctrl.Result{}, err
+	}
+
+	// A delete can land between acquiring the suspension and here. Deleting the
+	// storage Secret of a terminating release removes the only revision
+	// helm-controller could uninstall; it then treats the release as already gone,
+	// drops its finalizer, and orphans the resources — the exact failure this
+	// recovery exists to prevent, reached by a narrower race than the suspended one.
+	// The top-of-Reconcile check and suspendHelmRelease's re-fetch only see a delete
+	// visible earlier, so re-check immediately before the delete. On termination,
+	// leave the Secret intact and return: the deferred unsuspend clears our
+	// suspension so helm-controller runs the uninstall against the intact revision.
+	terminating, err := r.releaseIsTerminating(ctx, hr)
+	if err != nil {
+		logger.Info("Could not re-check deletion before deleting the storage Secret, requeuing", "error", err.Error())
+		return ctrl.Result{}, err
+	}
+	if terminating {
+		logger.Info("HelmRelease started terminating during recovery, leaving its storage Secret for helm-controller to uninstall")
 		return ctrl.Result{}, nil
 	}
 
@@ -146,24 +237,30 @@ func (r *FluxPlunger) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Res
 		return ctrl.Result{}, err
 	}
 
-	// Update annotation with processed version
-	logger.Info("Updating annotation with processed version", "version", latestVersion)
-	if err := r.updateProcessedVersionAnnotation(ctx, hr, latestVersion); err != nil {
-		logger.Info("Could not update annotation, will retry on next reconcile", "error", err.Error())
-		return ctrl.Result{}, nil
-	}
-
-	// Unsuspend the HelmRelease
+	// Unsuspend the HelmRelease. Force a backoff requeue on failure, symmetric with
+	// the deletion path: a persistently failing unsuspend (e.g. an admission webhook
+	// rejecting the patch) would otherwise leave the release suspended-by-plunger
+	// until the informer's periodic resync, far longer than necessary.
 	logger.Info("Unsuspending HelmRelease")
 	if err := r.unsuspendHelmRelease(ctx, hr); err != nil {
-		logger.Info("Could not unsuspend HelmRelease, will retry on next reconcile", "error", err.Error())
-		return ctrl.Result{}, nil
+		logger.Info("Could not unsuspend HelmRelease, requeuing", "error", err.Error())
+		// The returned error requeues and retries; skip the redundant deferred attempt.
+		completed = true
+		return ctrl.Result{}, err
 	}
 
+	completed = true
 	logger.Info("Successfully processed HelmRelease", "version", latestVersion)
 	return ctrl.Result{}, nil
 }
 
+// ownsSuspension reports whether the current suspension was set by flux-plunger,
+// as opposed to an operator or another controller. Recovery and delete handling
+// must only clear a suspension flux-plunger owns.
+func ownsSuspension(hr *helmv2.HelmRelease) bool {
+	return hr.Annotations[annotationSuspendedByPlunger] == "true"
+}
+
 // hasNoDeployedReleasesError checks if the HelmRelease has the specific error
 func hasNoDeployedReleasesError(hr *helmv2.HelmRelease) bool {
 	for _, condition := range hr.Status.Conditions {
@@ -240,27 +337,84 @@ func extractVersionNumber(secretName string) int {
 	return num
 }
 
-// suspendHelmRelease sets suspend to true on the HelmRelease
-func (r *FluxPlunger) suspendHelmRelease(ctx context.Context, hr *helmv2.HelmRelease) error {
+// suspendHelmRelease sets suspend to true on the HelmRelease and marks the
+// suspension as owned by flux-plunger. It reports whether it acquired the
+// suspension: false means the release was already suspended by someone else, or
+// it started terminating after the top-of-Reconcile check, so the caller must
+// not proceed as if it owns it.
+func (r *FluxPlunger) suspendHelmRelease(ctx context.Context, hr *helmv2.HelmRelease) (bool, error) {
 	// Re-fetch the HelmRelease to get the latest state
 	key := types.NamespacedName{Namespace: hr.Namespace, Name: hr.Name}
 	latestHR := &helmv2.HelmRelease{}
 	if err := r.Get(ctx, key, latestHR); err != nil {
-		return fmt.Errorf("failed to get latest HelmRelease: %w", err)
+		return false, fmt.Errorf("failed to get latest HelmRelease: %w", err)
+	}
+
+	// Never suspend a terminating release. A delete can arrive between the
+	// top-of-Reconcile deletion check and this re-fetch, and suspending a release
+	// helm-controller is about to uninstall is exactly what makes it skip the
+	// uninstall and orphan the release's resources.
+	if !latestHR.DeletionTimestamp.IsZero() {
+		return false, nil
 	}
 
-	// If already suspended, nothing to do
+	// If already suspended, do not take ownership of a suspension we did not set.
 	if latestHR.Spec.Suspend {
-		return nil
+		return false, nil
 	}
 
 	patch := client.MergeFromWithOptions(latestHR.DeepCopy(), client.MergeFromWithOptimisticLock{})
 	latestHR.Spec.Suspend = true
+	if latestHR.Annotations == nil {
+		latestHR.Annotations = make(map[string]string)
+	}
+	latestHR.Annotations[annotationSuspendedByPlunger] = "true"
+
+	if err := r.Patch(ctx, latestHR, patch, client.FieldOwner(fieldManager)); err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// releaseIsTerminating re-fetches the HelmRelease and reports whether it has begun
+// terminating. Recovery calls this immediately before deleting the storage Secret
+// so a delete that landed after the earlier checks cannot make it remove the only
+// revision helm-controller could uninstall.
+func (r *FluxPlunger) releaseIsTerminating(ctx context.Context, hr *helmv2.HelmRelease) (bool, error) {
+	key := types.NamespacedName{Namespace: hr.Namespace, Name: hr.Name}
+	latestHR := &helmv2.HelmRelease{}
+	if err := r.Get(ctx, key, latestHR); err != nil {
+		return false, fmt.Errorf("failed to get latest HelmRelease: %w", err)
+	}
+	return !latestHR.DeletionTimestamp.IsZero(), nil
+}
 
+// clearOwnershipMarker removes a stale ownership marker from a HelmRelease that is
+// no longer suspended. It never touches spec.suspend: if the release was suspended
+// again between the caller's read and this re-fetch, the marker is left in place
+// for the suspend path to evaluate and this is a no-op. Optimistic locking makes
+// the marker delete lose to a concurrent write rather than clobber it.
+func (r *FluxPlunger) clearOwnershipMarker(ctx context.Context, hr *helmv2.HelmRelease) error {
+	key := types.NamespacedName{Namespace: hr.Namespace, Name: hr.Name}
+	latestHR := &helmv2.HelmRelease{}
+	if err := r.Get(ctx, key, latestHR); err != nil {
+		return fmt.Errorf("failed to get latest HelmRelease: %w", err)
+	}
+	if _, ok := latestHR.Annotations[annotationSuspendedByPlunger]; !ok {
+		return nil
+	}
+	if latestHR.Spec.Suspend {
+		return nil
+	}
+	patch := client.MergeFromWithOptions(latestHR.DeepCopy(), client.MergeFromWithOptimisticLock{})
+	delete(latestHR.Annotations, annotationSuspendedByPlunger)
 	return r.Patch(ctx, latestHR, patch, client.FieldOwner(fieldManager))
 }
 
-// unsuspendHelmRelease sets suspend to false on the HelmRelease
+// unsuspendHelmRelease clears a flux-plunger-owned suspension and its marker.
+// Callers already gate on ownership; the re-fetched ownership check here is
+// defense-in-depth against a future caller that forgets to, so an operator's
+// suspension is never cleared by accident.
 func (r *FluxPlunger) unsuspendHelmRelease(ctx context.Context, hr *helmv2.HelmRelease) error {
 	// Re-fetch the HelmRelease to get the latest state
 	key := types.NamespacedName{Namespace: hr.Namespace, Name: hr.Name}
@@ -269,13 +423,28 @@ func (r *FluxPlunger) unsuspendHelmRelease(ctx context.Context, hr *helmv2.HelmR
 		return fmt.Errorf("failed to get latest HelmRelease: %w", err)
 	}
 
-	// If already unsuspended, nothing to do
+	// If already unsuspended, clear any stale ownership marker and stop.
 	if !latestHR.Spec.Suspend {
+		if _, ok := latestHR.Annotations[annotationSuspendedByPlunger]; !ok {
+			return nil
+		}
+		patch := client.MergeFromWithOptions(latestHR.DeepCopy(), client.MergeFromWithOptimisticLock{})
+		delete(latestHR.Annotations, annotationSuspendedByPlunger)
+		return r.Patch(ctx, latestHR, patch, client.FieldOwner(fieldManager))
+	}
+
+	// Deliberate redundancy with the caller-side ownership gate in Reconcile: this
+	// re-checks against the freshly re-fetched object, which the caller cannot,
+	// and it is the guard pinned directly by
+	// TestUnsuspendHelmRelease_LeavesSuspensionWithoutMarker so a future refactor
+	// that drops the caller-side gate still cannot clear an operator's suspension.
+	if !ownsSuspension(latestHR) {
 		return nil
 	}
 
 	patch := client.MergeFromWithOptions(latestHR.DeepCopy(), client.MergeFromWithOptimisticLock{})
 	latestHR.Spec.Suspend = false
+	delete(latestHR.Annotations, annotationSuspendedByPlunger)
 
 	return r.Patch(ctx, latestHR, patch, client.FieldOwner(fieldManager))
 }
@@ -303,7 +472,10 @@ func (r *FluxPlunger) updateProcessedVersionAnnotation(ctx context.Context, hr *
 func (r *FluxPlunger) SetupWithManager(mgr ctrl.Manager) error {
 	// Watch HelmReleases that either:
 	// 1. Have the specific error, OR
-	// 2. Are suspended with our annotation (to handle crash recovery)
+	// 2. Carry flux-plunger's ownership marker — whether still suspended (crash
+	//    recovery, and a delete arriving mid-recovery, both need to wake a
+	//    reconcile) or already resumed by an external actor (reap the stale marker
+	//    before a later operator suspension can inherit it).
 	pred := predicate.NewPredicateFuncs(func(obj client.Object) bool {
 		hr, ok := obj.(*helmv2.HelmRelease)
 		if !ok {
@@ -315,11 +487,9 @@ func (r *FluxPlunger) SetupWithManager(mgr ctrl.Manager) error {
 			return true
 		}
 
-		// Also process suspended HelmReleases with our annotation (crash recovery)
-		if hr.Spec.Suspend && hr.Annotations != nil {
-			if _, exists := hr.Annotations[annotationLastProcessedVersion]; exists {
-				return true
-			}
+		// Also process any HelmRelease still carrying our ownership marker.
+		if ownsSuspension(hr) {
+			return true
 		}
 
 		return false