fix(operator): never resurrect a package entry on interrupt completion (#426)

docs/designs/2026-07-10-package-execution-as-jobs.md

@@ -128,6 +128,8 @@ Because completed pods now linger, pod-deletion can no longer be the processed-o
 
 The Pod watch is **evidence, not authority**: it may only update an entry that already exists at the pod's stage and is not yet complete. It must never create one. Under `restartPolicy: Never` a failing Job mints a fresh pod per attempt, indefinitely, so an unguarded write becomes a repeating one. If it could create an entry, a node-state reset would be undone by the very Job the reset is meant to clear: the resurrected entry makes the reset invisible to the not-in-node-state check in rule 1 below, so the stale Job is never invalidated, and the existence gate then blocks the package's new stage forever. The same guard also stops a retained failed-attempt archive pod (kept on purpose, below) from regressing a completion the Job path already recorded.
 
+**Neither may the Job path create an entry**, for the same reason, even though it *is* the completion authority. Authority means it decides when a stage is done, not that it may write a package entry that node state says is gone. This bites on interrupt Jobs specifically: they are package-agnostic, so a completion is recorded on the strength of a *sibling* still sitting at (interrupt, `skipped`) — which says nothing about whether this Job's own package still has an entry. A rerun, reset or uninstall landing in that window would otherwise see the entry return at (interrupt, `complete`), where the rerun predicate keeps the Job and the stage never runs again. The promotion of skipped siblings happens either way; only the self-write is gated.
+
 Recording a completion is two writes to two objects (node state, then the Job marker) and cannot be atomic; a crash between them re-serves the event. The re-processing path is guarded by per-transition postcondition checks so a re-served completion is only marked, not re-applied — detail in [Edge cases](#edge-cases-and-correctness-arguments). This is strictly better than today, which has the same two-write window with no guard.
 
 ### Retry and the failed-attempt archive

operator/internal/controller/job_controller.go

@@ -223,15 +223,32 @@ func (r *JobReconciler) shouldRecordCompletion(job *batchv1.Job, pkg *PackageSky
 	if isInterruptJob(job) {
 		// ProgressSkipped promotes only StageInterrupt-skipped packages, so re-run the record
 		// only while such a package remains, matching what the promotion can actually advance.
+		//
+		// This authorizes the write on a SIBLING's state, so it says nothing about this Job's own
+		// package — recordJobCompletion gates its self-write on entryOpenAtStage separately.
 		for _, s := range state {
 			if s.Stage == v1alpha1.StageInterrupt && s.State == v1alpha1.StateSkipped {
 				return true, nil
 			}
 		}
 	}
 
+	return entryOpenAtStage(state, pkg), nil
+}
+
+// entryOpenAtStage reports whether the package's entry is present, still at the stage the caller
+// is reporting on, and not already complete — the only shape an executor may write onto at all.
+// Absent means removed (rerun, reset, uninstall) and writing would resurrect it; a later stage or
+// an existing complete would regress or duplicate.
+//
+// Named for the shape it tests, not for one caller's use: both the completion write and the Pod
+// watch's erroring write require exactly this, and the two must not drift. It deliberately does
+// not exclude an already-erroring entry — the Pod watch re-reports erroring so a rising restart
+// count still lands, and an otherwise identical write is a no-op the Changed() check drops.
+// recordJobErroring adds that exclusion itself; see the note there.
+func entryOpenAtStage(state v1alpha1.NodeState, pkg *PackageSkyhook) bool {
 	status, present := state[pkg.GetUniqueName()]
-	return present && status.Stage == pkg.Stage && status.State != v1alpha1.StateComplete, nil
+	return present && status.Stage == pkg.Stage && status.State != v1alpha1.StateComplete
 }
 
 // patchNodeState re-reads the node, applies mutate, and patches under an optimistic-lock
@@ -318,17 +335,46 @@ func (r *JobReconciler) recordJobCompletion(ctx context.Context, job *batchv1.Jo
 		if err != nil {
 			return false, fmt.Errorf("recording completion for job %s: %w", job.Name, err)
 		}
-		if !updated {
+
+		// Read after HandleCompletePod: its interrupt branch promotes skipped packages, which can
+		// move this package's own entry.
+		state, err := skyhookNode.State()
+		if err != nil {
+			return false, fmt.Errorf("reading node state for job %s: %w", job.Name, err)
+		}
+
+		// Upsert creates, so it must never run on an entry that is not there. An interrupt Job
+		// reaches this on a sibling's behalf (see shouldRecordCompletion), so its own package may
+		// have been removed by a rerun, reset or uninstall since the Job started; re-creating it
+		// at (interrupt, complete) would then make the rerun predicate keep the Job, and the stage
+		// would never run again. The promotion above still lands either way.
+		//
+		// Both guards are kept deliberately: updated says HandleCompletePod already wrote this
+		// entry, entryOpenAtStage says there is an entry to write onto. They agree today —
+		// every branch that sets updated leaves the entry absent or at another stage — and neither
+		// is safe to drop on the strength of the other.
+		recorded := false
+		if !updated && entryOpenAtStage(state, pkg) {
 			if err := skyhookNode.Upsert(pkg.PackageRef, pkg.Image, v1alpha1.StateComplete, pkg.Stage, job.Status.Failed, pkg.ContainerSHA); err != nil {
 				return false, fmt.Errorf("upserting complete state for job %s: %w", job.Name, err)
 			}
+			recorded = true
 		}
 
 		if !skyhookNode.Changed() {
 			return false, nil
 		}
-		r.recorder.Eventf(node, nil, EventTypeNormal, EventsReasonSkyhookStateChange, "JobComplete",
-			"Package [%s:%s] state %s on [skyhook:%s]", pkg.Name, pkg.Version, v1alpha1.StateComplete, pkg.Skyhook)
+
+		// The event follows what was actually written. An interrupt Job can reach here purely on a
+		// sibling's promotion with its own entry gone, and announcing that package complete would
+		// name the one a reset just cleared.
+		if recorded || updated {
+			r.recorder.Eventf(node, nil, EventTypeNormal, EventsReasonSkyhookStateChange, "JobComplete",
+				"Package [%s:%s] state %s on [skyhook:%s]", pkg.Name, pkg.Version, v1alpha1.StateComplete, pkg.Skyhook)
+		} else {
+			r.recorder.Eventf(node, nil, EventTypeNormal, EventsReasonSkyhookStateChange, "JobComplete",
+				"Interrupt complete on [skyhook:%s]: promoted packages skipped during interrupt sequencing", pkg.Skyhook)
+		}
 		return true, nil
 	})
 }
@@ -349,14 +395,20 @@ func (r *JobReconciler) HandleCompletePod(ctx context.Context, skyhookNode wrapp
 			return false, err
 		}
 
-		upgraded, err := wrapper.Convert(skyhookNode, skyhook)
-		if err != nil {
-			return false, fmt.Errorf("error converting node wrapper: %w", err)
-		}
+		// A deleted CR reads as (nil, nil), and Convert dereferences it. That is reachable in the
+		// same window this stage's guards are about: the CR removed while a completed interrupt
+		// Job is still unprocessed. Nothing is left to promote then, so skip the branch rather
+		// than panic — matching the nil check the uninstall branch below already does.
+		if skyhook != nil {
+			upgraded, err := wrapper.Convert(skyhookNode, skyhook)
+			if err != nil {
+				return false, fmt.Errorf("error converting node wrapper: %w", err)
+			}
 
-		// progress forward any skipped packages that this interrupt completed
-		if err := upgraded.ProgressSkipped(); err != nil {
-			return false, fmt.Errorf("error progressing skipped packages: %w", err)
+			// progress forward any skipped packages that this interrupt completed
+			if err := upgraded.ProgressSkipped(); err != nil {
+				return false, fmt.Errorf("error progressing skipped packages: %w", err)
+			}
 		}
 	} else if packagePtr.Stage == v1alpha1.StageUpgrade {
 		nodeState, err := skyhookNode.State()
@@ -522,6 +574,9 @@ func (r *JobReconciler) recordJobErroring(ctx context.Context, job *batchv1.Job,
 			return false, fmt.Errorf("reading node state for job %s: %w", job.Name, err)
 		}
 
+		// Deliberately NOT entryOpenAtStage: the completion guard excludes an entry that is
+		// already complete, this one excludes an entry that is already erroring, for idempotence
+		// on a re-served terminal event. Same shape, different exclusion — do not unify them.
 		status, present := state[pkg.GetUniqueName()]
 		if !present || status.Stage != pkg.Stage || status.State == v1alpha1.StateErroring {
 			return false, nil

operator/internal/controller/job_controller_test.go

@@ -412,6 +412,108 @@ var _ = Describe("JobReconcile", func() {
 		Expect(state[sibling.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete))
 	})
 
+	It("promotes the sibling without resurrecting a package whose entry was removed", func() {
+		// A rerun/reset/uninstall clears an entry while that package's interrupt Job is
+		// completing. The Job is package-agnostic and still owes the sibling its promotion, but
+		// re-creating the cleared entry would put it back at (interrupt, complete) — the rerun
+		// predicate would then keep the Job and the stage would never run again, so the rerun
+		// the user asked for would silently do nothing.
+		node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}
+		sn, err := wrapper.NewSkyhookNodeOnly(node, skyhookName)
+		Expect(err).ToNot(HaveOccurred())
+		sibling := v1alpha1.PackageRef{Name: "other", Version: "2.0.0"}
+		Expect(sn.Upsert(sibling, image, v1alpha1.StateSkipped, v1alpha1.StageInterrupt, 0, "")).To(Succeed())
+		// deliberately no entry for pkgRef: that is the removal
+
+		scr := &v1alpha1.NodeWright{
+			ObjectMeta: metav1.ObjectMeta{Name: skyhookName},
+			Spec: v1alpha1.NodeWrightSpec{Packages: v1alpha1.Packages{
+				"tuning": {PackageRef: pkgRef, Image: image},
+				"other":  {PackageRef: sibling, Image: image},
+			}},
+		}
+		job := packageJob(v1alpha1.StageInterrupt, true, trueCondition(batchv1.JobComplete, ""))
+		r := newReconciler(node, scr, job)
+
+		_, err = r.JobReconcile(ctx, job)
+		Expect(err).ToNot(HaveOccurred())
+
+		state := getNodeState(r)
+		Expect(state).ToNot(HaveKey(pkgRef.GetUniqueName()), "the removal must stand")
+		Expect(state[sibling.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete), "the sibling is still promoted")
+	})
+
+	It("does not regress an entry that already advanced past the interrupt", func() {
+		// The non-interrupt path never reaches the write here — shouldRecordCompletion returns
+		// false on a mismatched stage. On the interrupt path a skipped sibling makes it return
+		// true, so this direction is guarded only by the completion check.
+		node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}
+		sn, err := wrapper.NewSkyhookNodeOnly(node, skyhookName)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(sn.Upsert(pkgRef, image, v1alpha1.StateComplete, v1alpha1.StagePostInterrupt, 0, "")).To(Succeed())
+		sibling := v1alpha1.PackageRef{Name: "other", Version: "2.0.0"}
+		Expect(sn.Upsert(sibling, image, v1alpha1.StateSkipped, v1alpha1.StageInterrupt, 0, "")).To(Succeed())
+
+		scr := &v1alpha1.NodeWright{
+			ObjectMeta: metav1.ObjectMeta{Name: skyhookName},
+			Spec: v1alpha1.NodeWrightSpec{Packages: v1alpha1.Packages{
+				"tuning": {PackageRef: pkgRef, Image: image},
+				"other":  {PackageRef: sibling, Image: image},
+			}},
+		}
+		job := packageJob(v1alpha1.StageInterrupt, true, trueCondition(batchv1.JobComplete, ""))
+		r := newReconciler(node, scr, job)
+
+		_, err = r.JobReconcile(ctx, job)
+		Expect(err).ToNot(HaveOccurred())
+
+		state := getNodeState(r)
+		// The sibling's promotion is what proves the interrupt completion path actually ran:
+		// without it this spec would pass on a reconcile that did nothing at all.
+		Expect(state[sibling.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete))
+		Expect(getJob(r, job.Name).Annotations).To(HaveKeyWithValue(annotationStateRecorded, annotationValueTrue))
+
+		entry := state[pkgRef.GetUniqueName()]
+		Expect(entry.Stage).To(Equal(v1alpha1.StagePostInterrupt))
+		Expect(entry.State).To(Equal(v1alpha1.StateComplete))
+	})
+
+	It("marks an interrupt completion without panicking when the CR is already gone", func() {
+		// Same window as the resurrection case: the NodeWright deleted while a completed interrupt
+		// Job is still unprocessed. GetSkyhook reads a missing CR as (nil, nil).
+		node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageInterrupt)
+		job := packageJob(v1alpha1.StageInterrupt, true, trueCondition(batchv1.JobComplete, ""))
+		r := newReconciler(node, job) // no NodeWright seeded
+
+		_, err := r.JobReconcile(ctx, job)
+		Expect(err).ToNot(HaveOccurred())
+
+		Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete))
+		Expect(getJob(r, job.Name).Annotations).To(HaveKeyWithValue(annotationStateRecorded, annotationValueTrue))
+	})
+
+	It("removes the superseded version's entry on upgrade completion", func() {
+		// The upgrade branch is the other place that calls RemoveState and then leans on the
+		// guarded fallback; NodeState is keyed name|version, so only the old key goes.
+		node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}
+		sn, err := wrapper.NewSkyhookNodeOnly(node, skyhookName)
+		Expect(err).ToNot(HaveOccurred())
+		old := v1alpha1.PackageRef{Name: "tuning", Version: "0.9.0"}
+		Expect(sn.Upsert(old, image, v1alpha1.StateComplete, v1alpha1.StageConfig, 0, "")).To(Succeed())
+		Expect(sn.Upsert(pkgRef, image, v1alpha1.StateInProgress, v1alpha1.StageUpgrade, 0, "")).To(Succeed())
+
+		job := packageJob(v1alpha1.StageUpgrade, false, trueCondition(batchv1.JobComplete, ""))
+		r := newReconciler(node, job)
+
+		_, err = r.JobReconcile(ctx, job)
+		Expect(err).ToNot(HaveOccurred())
+
+		state := getNodeState(r)
+		Expect(state).ToNot(HaveKey(old.GetUniqueName()))
+		Expect(state[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete))
+		Expect(state[pkgRef.GetUniqueName()].Stage).To(Equal(v1alpha1.StageUpgrade))
+	})
+
 	It("records erroring (state only, no marker) for a stale FailureTarget on an unreachable node", func() {
 		node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig)
 		job := packageJob(v1alpha1.StageConfig, false)

operator/internal/controller/pod_controller.go

@@ -242,8 +242,7 @@ func shouldRecordPodErroring(skyhookNode wrapper.SkyhookNodeOnly, packagePtr *Pa
 		return false, fmt.Errorf("error reading node state for package %s: %w", packagePtr.GetUniqueName(), err)
 	}
 
-	status, present := state[packagePtr.GetUniqueName()]
-	return present && status.Stage == packagePtr.Stage && status.State != v1alpha1.StateComplete, nil
+	return entryOpenAtStage(state, packagePtr), nil
 }
 
 const (