From ca30d295136074bafe068670913ad32a475e37dd Mon Sep 17 00:00:00 2001
From: "Dr. Stefan Schimanski" <sschimanski@nvidia.com>
Date: Fri, 23 Oct 2026 18:00:21 +0200
Subject: [PATCH 1/2] operator/dgdr: harden run publisher reconciliation

Signed-off-by: Dr. Stefan Schimanski <sschimanski@nvidia.com>
---
 .../operator/cmd/dgdr-run-publisher/main.go   |  22 ++++
 .../cmd/dgdr-run-publisher/main_test.go       |  43 +++++++
 .../operator/internal/dgdrrunpublisher/doc.go |   6 +-
 .../internal/dgdrrunpublisher/manager.go      |  13 +-
 .../internal/dgdrrunpublisher/outcome.go      |  38 ++++--
 .../internal/dgdrrunpublisher/outcome_test.go |  15 ++-
 .../internal/dgdrrunpublisher/plan.go         |  10 +-
 .../internal/dgdrrunpublisher/plan_test.go    |   5 +-
 .../internal/dgdrrunpublisher/reconciler.go   |  31 ++++-
 .../dgdrrunpublisher/reconciler_test.go       | 119 +++++++++++++-----
 .../dgdrrunpublisher/snapshot_test.go         |  16 +++
 .../internal/dgdrrunpublisher/writer.go       |  14 +++
 .../dgdrrunpublisher/writer_envtest_test.go   | 116 +++++++++++++++++
 13 files changed, 388 insertions(+), 60 deletions(-)
 create mode 100644 deploy/operator/cmd/dgdr-run-publisher/main_test.go

diff --git a/deploy/operator/cmd/dgdr-run-publisher/main.go b/deploy/operator/cmd/dgdr-run-publisher/main.go
index 05c5c8f1cbd8..8307523b6a01 100644
--- a/deploy/operator/cmd/dgdr-run-publisher/main.go
+++ b/deploy/operator/cmd/dgdr-run-publisher/main.go
@@ -29,12 +29,18 @@ import (
 	"os/signal"
 	"syscall"
 	"time"
+	"unicode/utf8"
 
 	"github.com/ai-dynamo/dynamo/deploy/operator/internal/dgdrrunpublisher"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 )
 
+const (
+	terminationMessagePath = "/dev/termination-log"
+	maxTerminationMessage  = 4 * 1024
+)
+
 func main() {
 	os.Exit(run())
 }
@@ -68,6 +74,22 @@ func run() int {
 	err := dgdrrunpublisher.Run(ctx, ctrl.GetConfigOrDie(), cfg)
 	if err != nil {
 		ctrl.Log.WithName("dgdr-run-publisher").Error(err, "publisher stopped")
+		if writeErr := writeTerminationMessage(terminationMessagePath, err); writeErr != nil {
+			ctrl.Log.WithName("dgdr-run-publisher").Error(writeErr, "writing termination message")
+		}
 	}
 	return dgdrrunpublisher.ExitCode(err)
 }
+
+// writeTerminationMessage exposes the publisher's bounded failure detail through the
+// standard Kubernetes container termination-message channel.
+func writeTerminationMessage(path string, err error) error {
+	message := []byte(err.Error())
+	if len(message) > maxTerminationMessage {
+		message = message[:maxTerminationMessage]
+		for !utf8.Valid(message) {
+			message = message[:len(message)-1]
+		}
+	}
+	return os.WriteFile(path, message, 0o600)
+}
diff --git a/deploy/operator/cmd/dgdr-run-publisher/main_test.go b/deploy/operator/cmd/dgdr-run-publisher/main_test.go
new file mode 100644
index 000000000000..85f550903537
--- /dev/null
+++ b/deploy/operator/cmd/dgdr-run-publisher/main_test.go
@@ -0,0 +1,43 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"unicode/utf8"
+)
+
+func TestWriteTerminationMessage(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "termination.log")
+	message := strings.Repeat("failure-🙂", maxTerminationMessage)
+
+	if err := writeTerminationMessage(path, errors.New(message)); err != nil {
+		t.Fatal(err)
+	}
+	got, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) > maxTerminationMessage || !utf8.Valid(got) || !strings.HasPrefix(string(got), "failure-🙂") {
+		t.Fatalf("termination message is not a bounded UTF-8 prefix: %d bytes, %q", len(got), got[:min(len(got), 32)])
+	}
+}
diff --git a/deploy/operator/internal/dgdrrunpublisher/doc.go b/deploy/operator/internal/dgdrrunpublisher/doc.go
index 18b02a637102..308a5b2ec289 100644
--- a/deploy/operator/internal/dgdrrunpublisher/doc.go
+++ b/deploy/operator/internal/dgdrrunpublisher/doc.go
@@ -37,8 +37,7 @@
 // The publisher's own writes come back as watch events and the cache can trail them. The
 // Reconciler keeps no record of what it wrote: it decides from the current observations,
 // constrains its status writes to the resourceVersion it observed (a stale decision
-// conflicts and is retried through the workqueue), and ignores candidate update events;
-// see Reconciler.
+// conflicts and is retried through the workqueue); see Reconciler.
 //
 // The manager's cache covers only this Run, its candidates (by the run-UID label) and the
 // publisher's own Pod, so there is no API-server polling and no spec comparison:
@@ -47,7 +46,8 @@
 // The sidecar is a regular container of the same Job pod as the Sweeper. The Job only
 // completes once the publisher has exited, so a successful exit acknowledges that the
 // final snapshot was reconciled. The run controller derives the terminal Completed
-// condition from the Job result; the publisher never writes a phase.
+// condition from the Job result and the publisher's bounded container termination
+// message; the publisher never writes a phase.
 //
 // Termination (see Reconciler):
 //
diff --git a/deploy/operator/internal/dgdrrunpublisher/manager.go b/deploy/operator/internal/dgdrrunpublisher/manager.go
index 8c67862a1684..a99fc47724f9 100644
--- a/deploy/operator/internal/dgdrrunpublisher/manager.go
+++ b/deploy/operator/internal/dgdrrunpublisher/manager.go
@@ -34,7 +34,6 @@ import (
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -42,7 +41,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
@@ -104,15 +102,6 @@ func cacheOptions(c Config) cache.Options {
 	}
 }
 
-// ignoreCandidateUpdates drops candidate update events. The publisher is the only writer
-// of a candidate: its spec is immutable and its status is written once, so an update is
-// the echo of our own write and carries nothing to act on. Creates (including the
-// initial list) and deletes still enqueue the run key, so a deleted candidate is
-// repaired.
-var ignoreCandidateUpdates = predicate.Funcs{
-	UpdateFunc: func(event.UpdateEvent) bool { return false },
-}
-
 // Run starts the controller manager and returns the publisher's outcome (see the package
 // comment for its meaning and ExitCode for the exit code), or ctx's error if the process
 // is told to stop first.
@@ -181,7 +170,7 @@ func Run(ctx context.Context, restConfig *rest.Config, cfg Config) error {
 	if err := ctrl.NewControllerManagedBy(mgr).
 		Named("dgdr-run-publisher").
 		For(&v1beta2.DynamoGraphDeploymentRun{}).
-		Owns(&v1beta2.DynamoGraphDeploymentCandidate{}, builder.WithPredicates(ignoreCandidateUpdates)).
+		Owns(&v1beta2.DynamoGraphDeploymentCandidate{}).
 		Watches(&corev1.Pod{}, podToRun).
 		WatchesRawSource(source.Channel(events, &handler.EnqueueRequestForObject{})).
 		WithOptions(controller.Options{MaxConcurrentReconciles: 1}).
diff --git a/deploy/operator/internal/dgdrrunpublisher/outcome.go b/deploy/operator/internal/dgdrrunpublisher/outcome.go
index aea5b707ebee..dfcd1a9b6591 100644
--- a/deploy/operator/internal/dgdrrunpublisher/outcome.go
+++ b/deploy/operator/internal/dgdrrunpublisher/outcome.go
@@ -20,6 +20,7 @@ package dgdrrunpublisher
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 )
 
@@ -35,9 +36,9 @@ const (
 // terminal snapshot, which violates the producer protocol.
 var ErrMissingTerminalSnapshot = fmt.Errorf("%w: sweeper exited without publishing a terminal snapshot", ErrProtocolViolation)
 
-// ErrRunFailed means the Sweeper published a terminal Failed snapshot. The snapshot is
-// reconciled first; the error keeps a producer that reports failure but exits 0 from
-// letting the Job complete successfully.
+// ErrRunFailed means the Sweeper reported failure or produced no usable recommendation.
+// The snapshot is reconciled first; the error keeps that outcome from letting the Job
+// complete successfully.
 var ErrRunFailed = errors.New("sweeper reported a failed run")
 
 // ExitCode maps the result of Run to the process exit code.
@@ -60,12 +61,35 @@ type SweeperState struct {
 	ExitCode int32
 }
 
-// terminalResult is the outcome once a terminal snapshot has been reconciled.
-func terminalResult(phase string) error {
-	if phase == PhaseFailed {
+// terminalResult is the outcome once a terminal snapshot has been reconciled. A
+// successful run must have produced at least one materialized recommendation.
+func terminalResult(desired DesiredState) error {
+	if desired.Phase == PhaseSucceeded && len(desired.Candidates) > 0 {
+		return nil
+	}
+
+	// Preserve the producer's diagnostics in the process result consumed by the command.
+	details := make([]string, 0, 3)
+	if desired.Message != "" {
+		details = append(details, desired.Message)
+	}
+	if desired.Error != "" {
+		details = append(details, desired.Error)
+	}
+	if desired.Failed > 0 {
+		failure := "failures"
+		if desired.Failed == 1 {
+			failure = "failure"
+		}
+		details = append(details, fmt.Sprintf("%d candidate materialization %s", desired.Failed, failure))
+	}
+	if desired.Phase == PhaseSucceeded && len(desired.Candidates) == 0 {
+		details = append(details, "no candidate was materialized")
+	}
+	if len(details) == 0 {
 		return ErrRunFailed
 	}
-	return nil
+	return fmt.Errorf("%w: %s", ErrRunFailed, strings.Join(details, "; "))
 }
 
 // exitResult is the outcome when the Sweeper exited and the last snapshot it wrote (if
diff --git a/deploy/operator/internal/dgdrrunpublisher/outcome_test.go b/deploy/operator/internal/dgdrrunpublisher/outcome_test.go
index 7aa0d89a4468..9462ea9fd49a 100644
--- a/deploy/operator/internal/dgdrrunpublisher/outcome_test.go
+++ b/deploy/operator/internal/dgdrrunpublisher/outcome_test.go
@@ -20,7 +20,10 @@ package dgdrrunpublisher
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"testing"
+
+	"github.com/ai-dynamo/dynamo/deploy/operator/internal/dgdcreconcile"
 )
 
 func TestExitCodes(t *testing.T) {
@@ -30,6 +33,7 @@ func TestExitCodes(t *testing.T) {
 	}{
 		{nil, ExitAcknowledged},
 		{ErrRunFailed, ExitRunFailed},
+		{fmt.Errorf("%w: details", ErrRunFailed), ExitRunFailed},
 		{ErrMissingTerminalSnapshot, ExitProtocolViolation},
 		{violation("bad"), ExitProtocolViolation},
 		{fmt.Errorf("reconciling last snapshot: %w", violation("bad")), ExitProtocolViolation},
@@ -43,11 +47,16 @@ func TestExitCodes(t *testing.T) {
 }
 
 func TestTerminalResult(t *testing.T) {
-	if err := terminalResult(PhaseSucceeded); err != nil {
+	if err := terminalResult(DesiredState{Phase: PhaseSucceeded, Candidates: []dgdcreconcile.DesiredCandidate{{ID: "a"}}}); err != nil {
 		t.Fatalf("succeeded: %v", err)
 	}
-	if err := terminalResult(PhaseFailed); !errors.Is(err, ErrRunFailed) {
-		t.Fatalf("failed: %v", err)
+	failed := terminalResult(DesiredState{Phase: PhaseFailed, Message: "search failed", Error: "backend unavailable"})
+	if !errors.Is(failed, ErrRunFailed) || !strings.Contains(failed.Error(), "search failed") || !strings.Contains(failed.Error(), "backend unavailable") {
+		t.Fatalf("failed: %v", failed)
+	}
+	empty := terminalResult(DesiredState{Phase: PhaseSucceeded, Failed: 2})
+	if !errors.Is(empty, ErrRunFailed) || !strings.Contains(empty.Error(), "2 candidate materialization failures") || !strings.Contains(empty.Error(), "no candidate was materialized") {
+		t.Fatalf("empty success: %v", empty)
 	}
 }
 
diff --git a/deploy/operator/internal/dgdrrunpublisher/plan.go b/deploy/operator/internal/dgdrrunpublisher/plan.go
index 3f1efa3601f4..706d4db67c89 100644
--- a/deploy/operator/internal/dgdrrunpublisher/plan.go
+++ b/deploy/operator/internal/dgdrrunpublisher/plan.go
@@ -60,8 +60,12 @@ type DesiredState struct {
 	Status   RunStatus
 	Terminal bool
 	Phase    string
-	// Failed counts the candidates whose materialization failed. They are not created
-	// and the Run status has no field for them, so they are only logged.
+	// Message and Error are the producer's terminal diagnostics. They pass through the
+	// process result into the container termination message.
+	Message string
+	Error   string
+	// Failed counts the candidates whose materialization failed. They are not created;
+	// the count is logged and prevents an empty terminal projection from succeeding.
 	Failed int
 }
 
@@ -91,6 +95,8 @@ func Desire(runName, runUID string, snap *Snapshot) DesiredState {
 		Names:    make(map[string]string, len(snap.Candidates)),
 		Terminal: snap.Run.Terminal,
 		Phase:    snap.Run.Phase,
+		Message:  snap.Run.Message,
+		Error:    snap.Run.Error,
 	}
 	failed := 0
 	var names []string
diff --git a/deploy/operator/internal/dgdrrunpublisher/plan_test.go b/deploy/operator/internal/dgdrrunpublisher/plan_test.go
index c45ba1c6ea78..12268168a6a9 100644
--- a/deploy/operator/internal/dgdrrunpublisher/plan_test.go
+++ b/deploy/operator/internal/dgdrrunpublisher/plan_test.go
@@ -57,7 +57,7 @@ func TestSwapKeepsCandidateNamesAndOnlyReordersReferences(t *testing.T) {
 }
 
 func TestFailedMaterializationIsNotDesiredButCounted(t *testing.T) {
-	state := desire(t, snapshotJSON(t, PhaseRunning, cand{id: "a"}, cand{id: "x", failed: true}))
+	state := desire(t, withRunDiagnostics(t, snapshotJSON(t, PhaseRunning, cand{id: "a"}, cand{id: "x", failed: true}), "search warning", "render error"))
 	if len(state.Candidates) != 1 || state.Candidates[0].ID != "a" {
 		t.Fatalf("candidates = %+v", state.Candidates)
 	}
@@ -70,6 +70,9 @@ func TestFailedMaterializationIsNotDesiredButCounted(t *testing.T) {
 	if state.Status.Published != 1 {
 		t.Fatalf("published = %d, want 1: failed candidates are not published", state.Status.Published)
 	}
+	if state.Message != "search warning" || state.Error != "render error" {
+		t.Fatalf("terminal diagnostics = %q / %q", state.Message, state.Error)
+	}
 }
 
 func TestPublishedCountsTheReferencedCandidates(t *testing.T) {
diff --git a/deploy/operator/internal/dgdrrunpublisher/reconciler.go b/deploy/operator/internal/dgdrrunpublisher/reconciler.go
index 85cd6ad4efa8..470e8e180e89 100644
--- a/deploy/operator/internal/dgdrrunpublisher/reconciler.go
+++ b/deploy/operator/internal/dgdrrunpublisher/reconciler.go
@@ -29,6 +29,7 @@ import (
 	"github.com/ai-dynamo/dynamo/deploy/operator/internal/dgdcreconcile"
 	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/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -160,7 +161,7 @@ func (r *Reconciler) reconcile(ctx context.Context) error {
 	}
 	switch {
 	case desired.Terminal:
-		r.Completion.Finish(terminalResult(desired.Phase))
+		r.Completion.Finish(terminalResult(desired))
 	case exited:
 		r.Completion.Finish(exitResult(state))
 	}
@@ -202,13 +203,38 @@ func (r *Reconciler) apply(ctx context.Context, run *v1beta2.DynamoGraphDeployme
 		return fmt.Errorf("listing candidates: %w", err)
 	}
 	current := make([]dgdcreconcile.CurrentDGDC, 0, len(list.Items))
+	ownedObsolete := make([]string, 0, len(list.Items))
 	uids := make(map[string]types.UID, len(list.Items))
 	versions := make(map[string]string, len(list.Items)) // cached resourceVersion per candidate
+	expectedIDs := make(map[string]string, len(desired.Names))
+	for id, name := range desired.Names {
+		expectedIDs[name] = id
+	}
 	for i := range list.Items {
 		item := &list.Items[i]
+
+		// A label match scopes the cache but does not establish ownership. Foreign
+		// candidates must never satisfy desired state or become deletion targets.
+		owner := metav1.GetControllerOf(item)
+		if owner == nil || owner.UID != run.UID {
+			continue
+		}
+
+		// The deterministic name is the authoritative identity. An owned object under
+		// another name is obsolete; an expected object with a stale identity label is
+		// reapplied in place instead of deleted.
+		expectedID, isExpected := expectedIDs[item.Name]
+		if !isExpected {
+			ownedObsolete = append(ownedObsolete, item.Name)
+			uids[item.Name] = item.UID
+			continue
+		}
+		if item.Labels[LabelCandidateID] != expectedID {
+			continue
+		}
 		current = append(current, dgdcreconcile.CurrentDGDC{
 			Name:       item.Name,
-			ID:         item.Labels[LabelCandidateID],
+			ID:         expectedID,
 			Incomplete: item.Status.Metrics == nil,
 		})
 		uids[item.Name] = item.UID
@@ -218,6 +244,7 @@ func (r *Reconciler) apply(ctx context.Context, run *v1beta2.DynamoGraphDeployme
 	if err != nil {
 		return err
 	}
+	actions.Deletes = append(actions.Deletes, ownedObsolete...)
 	// populated names the candidates whose status the API server already has; they are
 	// known from the live read that follows each apply, not from the cache.
 	populated := map[string]struct{}{}
diff --git a/deploy/operator/internal/dgdrrunpublisher/reconciler_test.go b/deploy/operator/internal/dgdrrunpublisher/reconciler_test.go
index 27ff47cc988d..cb79429b82bc 100644
--- a/deploy/operator/internal/dgdrrunpublisher/reconciler_test.go
+++ b/deploy/operator/internal/dgdrrunpublisher/reconciler_test.go
@@ -90,7 +90,7 @@ func (w *memWriter) record(op string, err error) error {
 	return err
 }
 
-func (w *memWriter) CreateCandidate(ctx context.Context, _ *v1beta2.DynamoGraphDeploymentRun, name string, candidate dgdcreconcile.DesiredCandidate) (StoredCandidate, error) {
+func (w *memWriter) CreateCandidate(ctx context.Context, run *v1beta2.DynamoGraphDeploymentRun, name string, candidate dgdcreconcile.DesiredCandidate) (StoredCandidate, error) {
 	if err := w.record("create:"+name, w.fail("create")); err != nil {
 		return StoredCandidate{}, err
 	}
@@ -102,10 +102,11 @@ func (w *memWriter) CreateCandidate(ctx context.Context, _ *v1beta2.DynamoGraphD
 		return stored, nil
 	}
 	object := &v1beta2.DynamoGraphDeploymentCandidate{ObjectMeta: metav1.ObjectMeta{
-		Namespace: testNamespace,
-		Name:      name,
-		UID:       types.UID("uid-" + name),
-		Labels:    map[string]string{LabelRunUID: testRunUID, LabelCandidateID: candidate.ID},
+		Namespace:       testNamespace,
+		Name:            name,
+		UID:             types.UID("uid-" + name),
+		Labels:          map[string]string{LabelRunUID: testRunUID, LabelCandidateID: candidate.ID},
+		OwnerReferences: []metav1.OwnerReference{*metav1.NewControllerRef(run, v1beta2.DynamoGraphDeploymentRunGVK)},
 	}}
 	return stored, client.IgnoreAlreadyExists(w.c.Create(ctx, object))
 }
@@ -261,6 +262,24 @@ func (f *fixture) lastStatus() RunStatus {
 	return f.w.status[len(f.w.status)-1]
 }
 
+func (f *fixture) currentRun() *v1beta2.DynamoGraphDeploymentRun {
+	f.t.Helper()
+	var run v1beta2.DynamoGraphDeploymentRun
+	if err := f.c.Get(context.Background(), types.NamespacedName{Namespace: testNamespace, Name: testRunName}, &run); err != nil {
+		f.t.Fatal(err)
+	}
+	return &run
+}
+
+func candidateForRun(name, id string, run *v1beta2.DynamoGraphDeploymentRun) *v1beta2.DynamoGraphDeploymentCandidate {
+	return &v1beta2.DynamoGraphDeploymentCandidate{ObjectMeta: metav1.ObjectMeta{
+		Namespace:       run.Namespace,
+		Name:            name,
+		Labels:          map[string]string{LabelRunUID: string(run.UID), LabelCandidateID: id},
+		OwnerReferences: []metav1.OwnerReference{*metav1.NewControllerRef(run, v1beta2.DynamoGraphDeploymentRunGVK)},
+	}}
+}
+
 // withProgress rewrites a snapshot's progress counters.
 func withProgress(t *testing.T, data []byte, round, evaluated int) []byte {
 	t.Helper()
@@ -429,11 +448,7 @@ func TestFailedMaterializationIsNotCreatedButCounted(t *testing.T) {
 
 func TestCandidateWithoutStatusIsCompletedNotRecreated(t *testing.T) {
 	f := newFixture(t)
-	existing := &v1beta2.DynamoGraphDeploymentCandidate{ObjectMeta: metav1.ObjectMeta{
-		Namespace: testNamespace,
-		Name:      cn("a"),
-		Labels:    map[string]string{LabelRunUID: testRunUID, LabelCandidateID: "a"},
-	}}
+	existing := candidateForRun(cn("a"), "a", f.currentRun())
 	if err := f.c.Create(context.Background(), existing); err != nil {
 		t.Fatal(err)
 	}
@@ -475,9 +490,10 @@ func TestTerminalSucceededSnapshotIsReconciledThenAcknowledged(t *testing.T) {
 
 func TestTerminalFailedSnapshotIsReconciledThenFails(t *testing.T) {
 	f := newFixture(t)
-	f.publish(snapshotJSON(t, PhaseFailed, cand{id: "a"}))
+	f.publish(withRunDiagnostics(t, snapshotJSON(t, PhaseFailed, cand{id: "a"}), "search failed", "backend unavailable"))
 	f.mustReconcile()
-	if err := f.r.Completion.Err(); !errors.Is(err, ErrRunFailed) || ExitCode(err) != ExitRunFailed {
+	if err := f.r.Completion.Err(); !errors.Is(err, ErrRunFailed) || ExitCode(err) != ExitRunFailed ||
+		!strings.Contains(err.Error(), "search failed") || !strings.Contains(err.Error(), "backend unavailable") {
 		t.Fatalf("err = %v", err)
 	}
 	if len(f.w.status) != 1 {
@@ -485,6 +501,20 @@ func TestTerminalFailedSnapshotIsReconciledThenFails(t *testing.T) {
 	}
 }
 
+func TestTerminalSucceededSnapshotWithoutAMaterializedCandidateFails(t *testing.T) {
+	f := newFixture(t)
+	f.publish(snapshotJSON(t, PhaseSucceeded, cand{id: "x", failed: true}))
+	f.mustReconcile()
+
+	err := f.r.Completion.Err()
+	if !errors.Is(err, ErrRunFailed) || ExitCode(err) != ExitRunFailed || !strings.Contains(err.Error(), "1 candidate materialization failure") {
+		t.Fatalf("err = %v", err)
+	}
+	if ops := f.takeOps(); len(ops) != 0 {
+		t.Fatalf("a failed materialization must not produce cluster writes, got %v", ops)
+	}
+}
+
 func TestTerminalSnapshotReconcileFailureIsRetriedUntilItSucceeds(t *testing.T) {
 	f := newFixture(t)
 	f.w.failOn["status"] = errors.New("api unavailable")
@@ -594,23 +624,56 @@ func TestFinalReconcileFailureEndsTheProcessAfterBoundedRetries(t *testing.T) {
 	}
 }
 
-func TestMalformedProjectionFailsImmediately(t *testing.T) {
+func TestForeignLabeledCandidateDoesNotSatisfyOrGetDeleted(t *testing.T) {
 	f := newFixture(t)
+	foreignRun := &v1beta2.DynamoGraphDeploymentRun{ObjectMeta: metav1.ObjectMeta{
+		Namespace: testNamespace,
+		Name:      "foreign-run",
+		UID:       "foreign-run-uid",
+	}}
+	foreign := candidateForRun("foreign-candidate", "a", foreignRun)
+	foreign.Labels[LabelRunUID] = testRunUID
+	if err := f.c.Create(context.Background(), foreign); err != nil {
+		t.Fatal(err)
+	}
+
 	f.publish(snapshotJSON(t, PhaseRunning, cand{id: "a"}))
-	// Two live candidates claiming the same id make the diff input malformed.
-	for _, name := range []string{"x-1", "x-2"} {
-		object := &v1beta2.DynamoGraphDeploymentCandidate{ObjectMeta: metav1.ObjectMeta{
-			Namespace: testNamespace, Name: name,
-			Labels: map[string]string{LabelRunUID: testRunUID, LabelCandidateID: "dup"},
-		}}
-		if err := f.c.Create(context.Background(), object); err != nil {
-			t.Fatal(err)
-		}
+	f.mustReconcile()
+	if got, want := f.takeOps(), []string{"create:" + cn("a"), "populate:" + cn("a"), "status"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("ops = %v, want %v", got, want)
+	}
+	if err := f.c.Get(context.Background(), client.ObjectKeyFromObject(foreign), &v1beta2.DynamoGraphDeploymentCandidate{}); err != nil {
+		t.Fatalf("foreign candidate must remain untouched: %v", err)
+	}
+}
+
+func TestOwnedCandidateWithWrongNameIsReplacedBeforeDeletion(t *testing.T) {
+	f := newFixture(t)
+	wrong := candidateForRun("wrong-name", "a", f.currentRun())
+	if err := f.c.Create(context.Background(), wrong); err != nil {
+		t.Fatal(err)
 	}
+
+	f.publish(snapshotJSON(t, PhaseRunning, cand{id: "a"}))
 	f.mustReconcile()
-	var inputErr *dgdcreconcile.DiffInputError
-	if err := f.r.Completion.Err(); !errors.As(err, &inputErr) || ExitCode(err) != ExitReconcileFailed {
-		t.Fatalf("err = %v", err)
+	want := []string{"create:" + cn("a"), "populate:" + cn("a"), "status", "delete:wrong-name"}
+	if got := f.takeOps(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("ops = %v, want %v", got, want)
+	}
+}
+
+func TestOwnedCandidateWithWrongIdentityLabelIsReappliedNotDeleted(t *testing.T) {
+	f := newFixture(t)
+	wrong := candidateForRun(cn("a"), "wrong-id", f.currentRun())
+	if err := f.c.Create(context.Background(), wrong); err != nil {
+		t.Fatal(err)
+	}
+
+	f.publish(snapshotJSON(t, PhaseRunning, cand{id: "a"}))
+	f.mustReconcile()
+	want := []string{"create:" + cn("a"), "populate:" + cn("a"), "status"}
+	if got := f.takeOps(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("ops = %v, want %v", got, want)
 	}
 }
 
@@ -862,11 +925,7 @@ func TestNewCandidateStatusUsesTheVersionReturnedByTheCreate(t *testing.T) {
 
 func TestExistingCandidateStatusUsesItsCachedVersion(t *testing.T) {
 	f := newFixture(t)
-	existing := &v1beta2.DynamoGraphDeploymentCandidate{ObjectMeta: metav1.ObjectMeta{
-		Namespace: testNamespace,
-		Name:      cn("a"),
-		Labels:    map[string]string{LabelRunUID: testRunUID, LabelCandidateID: "a"},
-	}}
+	existing := candidateForRun(cn("a"), "a", f.currentRun())
 	if err := f.c.Create(context.Background(), existing); err != nil {
 		t.Fatal(err)
 	}
diff --git a/deploy/operator/internal/dgdrrunpublisher/snapshot_test.go b/deploy/operator/internal/dgdrrunpublisher/snapshot_test.go
index 3be65aa92092..e0287a5c36d3 100644
--- a/deploy/operator/internal/dgdrrunpublisher/snapshot_test.go
+++ b/deploy/operator/internal/dgdrrunpublisher/snapshot_test.go
@@ -95,6 +95,22 @@ func withTimestamp(t *testing.T, data []byte, timestamp string) []byte {
 	return out
 }
 
+func withRunDiagnostics(t *testing.T, data []byte, message, runError string) []byte {
+	t.Helper()
+	var doc map[string]any
+	if err := json.Unmarshal(data, &doc); err != nil {
+		t.Fatal(err)
+	}
+	run := doc["run"].(map[string]any)
+	run["message"] = message
+	run["error"] = runError
+	out, err := json.Marshal(doc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return out
+}
+
 func TestParseSnapshotValidation(t *testing.T) {
 	bad := map[string]string{
 		"schema":        `{"schemaVersion":2,"run":{"phase":"Running","terminal":false}}`,
diff --git a/deploy/operator/internal/dgdrrunpublisher/writer.go b/deploy/operator/internal/dgdrrunpublisher/writer.go
index 8a02f1c61934..1eb242f925f2 100644
--- a/deploy/operator/internal/dgdrrunpublisher/writer.go
+++ b/deploy/operator/internal/dgdrrunpublisher/writer.go
@@ -20,10 +20,12 @@ package dgdrrunpublisher
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"time"
 
 	v1beta2 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1beta2"
 	"github.com/ai-dynamo/dynamo/deploy/operator/internal/dgdcreconcile"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -87,6 +89,18 @@ func (w *clientWriter) apply(ctx context.Context, object *unstructured.Unstructu
 // tells the caller whether the status is populated already: the cache cannot, because it
 // is exactly what may be stale.
 func (w *clientWriter) CreateCandidate(ctx context.Context, run *v1beta2.DynamoGraphDeploymentRun, name string, candidate dgdcreconcile.DesiredCandidate) (StoredCandidate, error) {
+	// A cache miss may still find an object at the deterministic name through the API
+	// reader. Never let server-side apply adopt an object controlled by another owner.
+	var existing v1beta2.DynamoGraphDeploymentCandidate
+	if err := w.Reader.Get(ctx, types.NamespacedName{Namespace: w.Namespace, Name: name}, &existing); err == nil {
+		owner := metav1.GetControllerOf(&existing)
+		if owner == nil || owner.UID != run.UID {
+			return StoredCandidate{}, fmt.Errorf("candidate %s already exists without controller ownership by run %s", name, run.Name)
+		}
+	} else if !apierrors.IsNotFound(err) {
+		return StoredCandidate{}, fmt.Errorf("checking candidate %s ownership: %w", name, err)
+	}
+
 	object, err := buildCandidateApply(w.Client.Scheme(), w.Namespace, w.RunUID, run, name, candidate)
 	if err != nil {
 		return StoredCandidate{}, err
diff --git a/deploy/operator/internal/dgdrrunpublisher/writer_envtest_test.go b/deploy/operator/internal/dgdrrunpublisher/writer_envtest_test.go
index 77552d82ebb7..6d82f892d216 100644
--- a/deploy/operator/internal/dgdrrunpublisher/writer_envtest_test.go
+++ b/deploy/operator/internal/dgdrrunpublisher/writer_envtest_test.go
@@ -21,6 +21,8 @@ package dgdrrunpublisher
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"strings"
 	"testing"
 	"time"
@@ -28,9 +30,13 @@ import (
 	v1beta2 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1beta2"
 	"github.com/ai-dynamo/dynamo/deploy/operator/internal/dgdcreconcile"
 	"github.com/ai-dynamo/dynamo/deploy/operator/internal/testing/operatorenv"
+	"github.com/stretchr/testify/require"
+	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/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -44,6 +50,7 @@ type writerFixture struct {
 	t      *testing.T
 	ctx    context.Context
 	client client.Client
+	config *rest.Config
 	run    *v1beta2.DynamoGraphDeploymentRun
 	writer *clientWriter
 }
@@ -70,6 +77,7 @@ func newWriterFixture(t *testing.T) *writerFixture {
 		t:      t,
 		ctx:    ctx,
 		client: c,
+		config: env.RESTConfig(),
 		run:    run,
 		writer: &clientWriter{
 			Client:    c,
@@ -156,6 +164,45 @@ func TestEnvtestCandidateApplyIsAcceptedAndIdempotent(t *testing.T) {
 	}
 }
 
+func TestEnvtestCandidateApplyDoesNotAdoptAForeignObject(t *testing.T) {
+	f := newWriterFixture(t)
+	name := f.name("a")
+	candidate := envtestCandidate()
+	spec, _, err := decodeManifest(candidate)
+	if err != nil {
+		t.Fatal(err)
+	}
+	parameters, err := json.Marshal(candidate.Parameters)
+	if err != nil {
+		t.Fatal(err)
+	}
+	foreign := &v1beta2.DynamoGraphDeploymentCandidate{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: f.run.Namespace,
+			Name:      name,
+			Labels:    map[string]string{"owner": "foreign"},
+		},
+		Spec: v1beta2.DynamoGraphDeploymentCandidateSpec{
+			DynamoGraphDeploymentSpec: *spec,
+			Parameters:                runtime.RawExtension{Raw: parameters},
+		},
+	}
+	if err := f.client.Create(f.ctx, foreign); err != nil {
+		t.Fatalf("create foreign candidate: %v", err)
+	}
+
+	if _, err := f.writer.CreateCandidate(f.ctx, f.run, name, candidate); err == nil {
+		t.Fatal("applying a candidate over a foreign object succeeded")
+	}
+	got := f.candidate(name)
+	if owner := metav1.GetControllerOf(got); owner != nil {
+		t.Fatalf("foreign candidate was adopted by %+v", owner)
+	}
+	if got.Labels["owner"] != "foreign" || got.Labels[LabelRunUID] != "" {
+		t.Fatalf("foreign candidate was modified: labels = %v", got.Labels)
+	}
+}
+
 func TestEnvtestCandidateStatusIsWrittenOnceAndGuardedByResourceVersion(t *testing.T) {
 	f := newWriterFixture(t)
 	name := f.name("a")
@@ -263,3 +310,72 @@ func TestEnvtestDeleteCandidateHonorsTheUIDPrecondition(t *testing.T) {
 		t.Fatalf("deleting a missing candidate must not fail: %v", err)
 	}
 }
+
+func TestEnvtestCandidateUpdateTriggersStatusRepair(t *testing.T) {
+	f := newWriterFixture(t)
+	dir := t.TempDir()
+	podName := "publisher-pod"
+
+	t.Log("Create the publisher Pod and a running snapshot")
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: f.run.Namespace, Name: podName},
+		Spec:       corev1.PodSpec{Containers: []corev1.Container{{Name: "sweeper", Image: "example.invalid/sweeper"}}},
+	}
+	if err := f.client.Create(f.ctx, pod); err != nil {
+		t.Fatalf("create pod: %v", err)
+	}
+	writeSnapshot(t, dir, snapshotJSON(t, PhaseRunning, cand{id: "a", metrics: map[string]any{"throughput": 12.5}}))
+
+	t.Log("Start the real manager and wait for its initial publication")
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() {
+		done <- Run(ctx, f.config, Config{
+			Namespace:        f.run.Namespace,
+			RunName:          f.run.Name,
+			RunUID:           string(f.run.UID),
+			PodName:          podName,
+			SweeperContainer: "sweeper",
+			SnapshotDir:      dir,
+			MinPollInterval:  10 * time.Millisecond,
+			MaxPollInterval:  20 * time.Millisecond,
+		})
+	}()
+	t.Cleanup(func() {
+		cancel()
+		select {
+		case err := <-done:
+			if err != nil && !errors.Is(err, context.Canceled) {
+				t.Errorf("publisher stopped: %v", err)
+			}
+		case <-time.After(5 * time.Second):
+			t.Error("publisher did not stop")
+		}
+	})
+
+	name := f.name("a")
+	require.Eventually(t, func() bool {
+		var candidate v1beta2.DynamoGraphDeploymentCandidate
+		if err := f.client.Get(f.ctx, types.NamespacedName{Namespace: f.run.Namespace, Name: name}, &candidate); err != nil {
+			return false
+		}
+		return candidate.Status.Metrics != nil
+	}, 10*time.Second, 25*time.Millisecond, "publisher did not create and populate the candidate")
+
+	t.Log("Clear the Candidate status and let its update event trigger repair")
+	var candidate v1beta2.DynamoGraphDeploymentCandidate
+	if err := f.client.Get(f.ctx, types.NamespacedName{Namespace: f.run.Namespace, Name: name}, &candidate); err != nil {
+		t.Fatalf("get candidate: %v", err)
+	}
+	candidate.Status.Metrics = nil
+	if err := f.client.Status().Update(f.ctx, &candidate); err != nil {
+		t.Fatalf("clear candidate status: %v", err)
+	}
+	require.Eventually(t, func() bool {
+		var repaired v1beta2.DynamoGraphDeploymentCandidate
+		if err := f.client.Get(f.ctx, types.NamespacedName{Namespace: f.run.Namespace, Name: name}, &repaired); err != nil {
+			return false
+		}
+		return repaired.Status.Metrics != nil
+	}, 10*time.Second, 25*time.Millisecond, "candidate status update did not trigger repair")
+}

From f25d7c5c28bcf5102dfc1fa5a5124243bcba5c69 Mon Sep 17 00:00:00 2001
From: "Dr. Stefan Schimanski" <sschimanski@nvidia.com>
Date: Fri, 23 Oct 2026 18:28:49 +0200
Subject: [PATCH 2/2] fixup! operator/dgdr: harden run publisher reconciliation

Signed-off-by: Dr. Stefan Schimanski <sschimanski@nvidia.com>
---
 .../operator/internal/dgdrrunpublisher/doc.go |   7 +-
 .../internal/dgdrrunpublisher/manager.go      |  41 +--
 .../internal/dgdrrunpublisher/reconciler.go   |   8 +-
 .../dgdrrunpublisher/reconciler_test.go       |  37 ++-
 .../internal/dgdrrunpublisher/writer.go       |  55 ++--
 .../dgdrrunpublisher/writer_envtest_test.go   | 270 +++++++++++++++---
 6 files changed, 325 insertions(+), 93 deletions(-)

diff --git a/deploy/operator/internal/dgdrrunpublisher/doc.go b/deploy/operator/internal/dgdrrunpublisher/doc.go
index 308a5b2ec289..f0add7122a54 100644
--- a/deploy/operator/internal/dgdrrunpublisher/doc.go
+++ b/deploy/operator/internal/dgdrrunpublisher/doc.go
@@ -39,9 +39,10 @@
 // constrains its status writes to the resourceVersion it observed (a stale decision
 // conflicts and is retried through the workqueue); see Reconciler.
 //
-// The manager's cache covers only this Run, its candidates (by the run-UID label) and the
-// publisher's own Pod, so there is no API-server polling and no spec comparison:
-// candidate specs are immutable once created.
+// The manager's cache covers this Run and Pod by name and Candidates in their namespace.
+// Candidate membership is determined from controller ownership rather than mutable
+// labels, so there is no API-server polling and no spec comparison: candidate specs are
+// immutable once created.
 //
 // The sidecar is a regular container of the same Job pod as the Sweeper. The Job only
 // completes once the publisher has exited, so a successful exit acknowledges that the
diff --git a/deploy/operator/internal/dgdrrunpublisher/manager.go b/deploy/operator/internal/dgdrrunpublisher/manager.go
index a99fc47724f9..e7f33b047ceb 100644
--- a/deploy/operator/internal/dgdrrunpublisher/manager.go
+++ b/deploy/operator/internal/dgdrrunpublisher/manager.go
@@ -28,7 +28,6 @@ import (
 	v1beta2 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1beta2"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/fields"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -50,7 +49,7 @@ type Config struct {
 	Namespace string
 	RunName   string
 	// RunUID is the UID of the run this pod serves. The run controller knows it when it
-	// builds the Job, which is what lets the cache select this run's candidates by label.
+	// builds the Job; the publisher uses it to establish Candidate ownership.
 	RunUID           string
 	PodName          string
 	SweeperContainer string
@@ -82,9 +81,9 @@ func NewScheme() (*runtime.Scheme, error) {
 	return scheme, nil
 }
 
-// cacheOptions limits the manager's informers to this run, its candidates and its own
-// pod: one namespace, plus a field selector on the names of the run and the pod and a
-// label selector on the run UID for the candidates.
+// cacheOptions limits the manager's informers to one namespace and selects this Run and
+// Pod by name. Candidates remain namespace-scoped because their labels are mutable;
+// reconciliation establishes membership from their controller owner reference.
 func cacheOptions(c Config) cache.Options {
 	return cache.Options{
 		DefaultNamespaces: map[string]cache.Config{c.Namespace: {}},
@@ -92,9 +91,7 @@ func cacheOptions(c Config) cache.Options {
 			&v1beta2.DynamoGraphDeploymentRun{}: {
 				Field: fields.OneTermEqualSelector("metadata.name", c.RunName),
 			},
-			&v1beta2.DynamoGraphDeploymentCandidate{}: {
-				Label: labels.SelectorFromSet(labels.Set{LabelRunUID: c.RunUID}),
-			},
+			&v1beta2.DynamoGraphDeploymentCandidate{}: {},
 			&corev1.Pod{}: {
 				Field: fields.OneTermEqualSelector("metadata.name", c.PodName),
 			},
@@ -164,17 +161,7 @@ func Run(ctx context.Context, restConfig *rest.Config, cfg Config) error {
 		Completion:       completion,
 	}
 
-	podToRun := handler.EnqueueRequestsFromMapFunc(func(context.Context, client.Object) []reconcile.Request {
-		return []reconcile.Request{{NamespacedName: key}}
-	})
-	if err := ctrl.NewControllerManagedBy(mgr).
-		Named("dgdr-run-publisher").
-		For(&v1beta2.DynamoGraphDeploymentRun{}).
-		Owns(&v1beta2.DynamoGraphDeploymentCandidate{}).
-		Watches(&corev1.Pod{}, podToRun).
-		WatchesRawSource(source.Channel(events, &handler.EnqueueRequestForObject{})).
-		WithOptions(controller.Options{MaxConcurrentReconciles: 1}).
-		Complete(reconciler); err != nil {
+	if err := setupController(mgr, reconciler, key, events); err != nil {
 		return fmt.Errorf("building controller: %w", err)
 	}
 	if err := mgr.Add(manager.RunnableFunc(observer.Run)); err != nil {
@@ -204,6 +191,22 @@ func Run(ctx context.Context, restConfig *rest.Config, cfg Config) error {
 	return errors.New("manager stopped before the run finished")
 }
 
+// setupController centralizes the production event wiring: every Kubernetes and
+// local-input event reaches the one Run key.
+func setupController(mgr manager.Manager, reconciler reconcile.Reconciler, key types.NamespacedName, events <-chan event.GenericEvent) error {
+	podToRun := handler.EnqueueRequestsFromMapFunc(func(context.Context, client.Object) []reconcile.Request {
+		return []reconcile.Request{{NamespacedName: key}}
+	})
+	return ctrl.NewControllerManagedBy(mgr).
+		Named("dgdr-run-publisher").
+		For(&v1beta2.DynamoGraphDeploymentRun{}).
+		Owns(&v1beta2.DynamoGraphDeploymentCandidate{}).
+		Watches(&corev1.Pod{}, podToRun).
+		WatchesRawSource(source.Channel(events, &handler.EnqueueRequestForObject{})).
+		WithOptions(controller.Options{MaxConcurrentReconciles: 1}).
+		Complete(reconciler)
+}
+
 // SnapshotPath is where the Sweeper writes the snapshot inside the shared directory.
 func SnapshotPath(dir string) string {
 	return filepath.Join(dir, SnapshotFileName)
diff --git a/deploy/operator/internal/dgdrrunpublisher/reconciler.go b/deploy/operator/internal/dgdrrunpublisher/reconciler.go
index 470e8e180e89..9e51272b60d9 100644
--- a/deploy/operator/internal/dgdrrunpublisher/reconciler.go
+++ b/deploy/operator/internal/dgdrrunpublisher/reconciler.go
@@ -40,8 +40,8 @@ import (
 const (
 	// LabelRunUID ties a candidate to one incarnation of its run. The run name can be
 	// reused after a delete and recreate; the UID cannot, so candidates left over from a
-	// previous run of the same name are never mistaken for current ones. The manager's
-	// cache selects candidates by this label.
+	// previous run of the same name are never mistaken for current ones. It is a repairable
+	// index; the controller owner UID remains the authoritative ownership relation.
 	LabelRunUID = "nvidia.com/dgdr-run-uid"
 	// LabelCandidateID records the stable candidate id (the identity read back on list).
 	LabelCandidateID = "nvidia.com/dgdr-candidate-id"
@@ -199,7 +199,7 @@ func (r *Reconciler) apply(ctx context.Context, run *v1beta2.DynamoGraphDeployme
 		ctrl.LoggerFrom(ctx).V(1).Info("candidates failed materialization and are not published", "count", desired.Failed)
 	}
 	var list v1beta2.DynamoGraphDeploymentCandidateList
-	if err := r.Client.List(ctx, &list, client.InNamespace(r.Namespace), client.MatchingLabels{LabelRunUID: r.RunUID}); err != nil {
+	if err := r.Client.List(ctx, &list, client.InNamespace(r.Namespace)); err != nil {
 		return fmt.Errorf("listing candidates: %w", err)
 	}
 	current := make([]dgdcreconcile.CurrentDGDC, 0, len(list.Items))
@@ -229,7 +229,7 @@ func (r *Reconciler) apply(ctx context.Context, run *v1beta2.DynamoGraphDeployme
 			uids[item.Name] = item.UID
 			continue
 		}
-		if item.Labels[LabelCandidateID] != expectedID {
+		if item.Labels[LabelRunUID] != r.RunUID || item.Labels[LabelCandidateID] != expectedID {
 			continue
 		}
 		current = append(current, dgdcreconcile.CurrentDGDC{
diff --git a/deploy/operator/internal/dgdrrunpublisher/reconciler_test.go b/deploy/operator/internal/dgdrrunpublisher/reconciler_test.go
index cb79429b82bc..58a35b02e880 100644
--- a/deploy/operator/internal/dgdrrunpublisher/reconciler_test.go
+++ b/deploy/operator/internal/dgdrrunpublisher/reconciler_test.go
@@ -662,6 +662,21 @@ func TestOwnedCandidateWithWrongNameIsReplacedBeforeDeletion(t *testing.T) {
 	}
 }
 
+func TestOwnedObsoleteCandidateWithoutRunLabelIsDeleted(t *testing.T) {
+	f := newFixture(t)
+	obsolete := candidateForRun(cn("obsolete"), "obsolete", f.currentRun())
+	delete(obsolete.Labels, LabelRunUID)
+	if err := f.c.Create(context.Background(), obsolete); err != nil {
+		t.Fatal(err)
+	}
+
+	f.publish(snapshotJSON(t, PhaseRunning))
+	f.mustReconcile()
+	if got, want := f.takeOps(), []string{"delete:" + obsolete.Name}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("ops = %v, want %v", got, want)
+	}
+}
+
 func TestOwnedCandidateWithWrongIdentityLabelIsReappliedNotDeleted(t *testing.T) {
 	f := newFixture(t)
 	wrong := candidateForRun(cn("a"), "wrong-id", f.currentRun())
@@ -677,6 +692,22 @@ func TestOwnedCandidateWithWrongIdentityLabelIsReappliedNotDeleted(t *testing.T)
 	}
 }
 
+func TestOwnedCandidateWithMissingRunLabelIsReappliedNotDeleted(t *testing.T) {
+	f := newFixture(t)
+	drifted := candidateForRun(cn("a"), "a", f.currentRun())
+	delete(drifted.Labels, LabelRunUID)
+	if err := f.c.Create(context.Background(), drifted); err != nil {
+		t.Fatal(err)
+	}
+
+	f.publish(snapshotJSON(t, PhaseRunning, cand{id: "a"}))
+	f.mustReconcile()
+	want := []string{"create:" + cn("a"), "populate:" + cn("a"), "status"}
+	if got := f.takeOps(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("ops = %v, want %v", got, want)
+	}
+}
+
 func TestRunRecreatedUnderTheSameNameIsNotTouched(t *testing.T) {
 	f := newFixture(t)
 	var run v1beta2.DynamoGraphDeploymentRun
@@ -754,7 +785,7 @@ func TestObservedStatusMirrorsTheRunStatus(t *testing.T) {
 	}
 }
 
-func TestCacheOptionsCoverOnlyThisRunItsCandidatesAndItsPod(t *testing.T) {
+func TestCacheOptionsCoverThisRunPodAndNamespaceCandidates(t *testing.T) {
 	cfg := Config{Namespace: "ns", RunName: "run", RunUID: "uid-1", PodName: "pod-1"}
 	opts := cacheOptions(cfg)
 	if _, ok := opts.DefaultNamespaces["ns"]; !ok || len(opts.DefaultNamespaces) != 1 {
@@ -770,8 +801,8 @@ func TestCacheOptionsCoverOnlyThisRunItsCandidatesAndItsPod(t *testing.T) {
 				t.Errorf("run selector = %q", selector.Field)
 			}
 		case *v1beta2.DynamoGraphDeploymentCandidate:
-			if selector.Label.String() != LabelRunUID+"=uid-1" {
-				t.Errorf("candidate selector = %q", selector.Label)
+			if selector.Field != nil || selector.Label != nil {
+				t.Errorf("candidates must remain visible after mutable label drift: %+v", selector)
 			}
 		case *corev1.Pod:
 			if selector.Field.String() != "metadata.name=pod-1" {
diff --git a/deploy/operator/internal/dgdrrunpublisher/writer.go b/deploy/operator/internal/dgdrrunpublisher/writer.go
index 1eb242f925f2..e3b1030f59c6 100644
--- a/deploy/operator/internal/dgdrrunpublisher/writer.go
+++ b/deploy/operator/internal/dgdrrunpublisher/writer.go
@@ -47,9 +47,9 @@ type StoredCandidate struct {
 // writer is every write the reconciler makes to the API server. It is a seam so the
 // reconciler's behavior can be tested without server-side apply support in a fake client.
 type writer interface {
-	// CreateCandidate applies the candidate and returns the API server's own copy of it,
-	// read after the apply. It must be idempotent: applying a candidate that already
-	// exists changes nothing.
+	// CreateCandidate atomically creates a missing candidate or reapplies one already
+	// controlled by the Run, then returns the API server's own copy. It must never adopt
+	// an object that is not controlled by the Run.
 	CreateCandidate(ctx context.Context, run *v1beta2.DynamoGraphDeploymentRun, name string, candidate dgdcreconcile.DesiredCandidate) (StoredCandidate, error)
 	// PopulateStatus records the candidate's evaluation metrics. The
 	// write is constrained to resourceVersion, so a stale view of the candidate fails
@@ -83,28 +83,40 @@ func (w *clientWriter) apply(ctx context.Context, object *unstructured.Unstructu
 	return w.Client.Apply(ctx, configuration, client.FieldOwner(fieldOwner), client.ForceOwnership)
 }
 
-// CreateCandidate applies the candidate and reads it back from the API server. It is
-// only called for a candidate missing from the cache, and re-applying the same content
-// to a candidate that exists is a no-op, so no spec comparison is needed. The read-back
-// tells the caller whether the status is populated already: the cache cannot, because it
-// is exactly what may be stale.
+// CreateCandidate creates a candidate atomically, or reapplies it after a live ownership
+// check when the cache missed an existing object. The apply is constrained to the UID and
+// resourceVersion returned by Create or Get, so a concurrent delete/recreate or ownership
+// change conflicts instead of being adopted. The final live read tells the caller whether
+// status is populated already: the cache cannot, because it is exactly what may be stale.
 func (w *clientWriter) CreateCandidate(ctx context.Context, run *v1beta2.DynamoGraphDeploymentRun, name string, candidate dgdcreconcile.DesiredCandidate) (StoredCandidate, error) {
-	// A cache miss may still find an object at the deterministic name through the API
-	// reader. Never let server-side apply adopt an object controlled by another owner.
-	var existing v1beta2.DynamoGraphDeploymentCandidate
-	if err := w.Reader.Get(ctx, types.NamespacedName{Namespace: w.Namespace, Name: name}, &existing); err == nil {
+	object, err := buildCandidateApply(w.Client.Scheme(), w.Namespace, w.RunUID, run, name, candidate)
+	if err != nil {
+		return StoredCandidate{}, err
+	}
+
+	// POST provides the missing-object precondition that server-side apply lacks. If the
+	// name is already occupied, classify the live object before applying anything to it.
+	var observedUID types.UID
+	created := object.DeepCopy()
+	if err := w.Client.Create(ctx, created, client.FieldOwner(fieldOwner)); err == nil {
+		object.SetResourceVersion(created.GetResourceVersion())
+		observedUID = created.GetUID()
+	} else if apierrors.IsAlreadyExists(err) {
+		var existing v1beta2.DynamoGraphDeploymentCandidate
+		if err := w.Reader.Get(ctx, types.NamespacedName{Namespace: w.Namespace, Name: name}, &existing); err != nil {
+			return StoredCandidate{}, fmt.Errorf("reading existing candidate %s: %w", name, err)
+		}
 		owner := metav1.GetControllerOf(&existing)
 		if owner == nil || owner.UID != run.UID {
 			return StoredCandidate{}, fmt.Errorf("candidate %s already exists without controller ownership by run %s", name, run.Name)
 		}
-	} else if !apierrors.IsNotFound(err) {
-		return StoredCandidate{}, fmt.Errorf("checking candidate %s ownership: %w", name, err)
+		object.SetResourceVersion(existing.ResourceVersion)
+		observedUID = existing.UID
+	} else {
+		return StoredCandidate{}, fmt.Errorf("creating candidate %s: %w", name, err)
 	}
+	object.SetUID(observedUID)
 
-	object, err := buildCandidateApply(w.Client.Scheme(), w.Namespace, w.RunUID, run, name, candidate)
-	if err != nil {
-		return StoredCandidate{}, err
-	}
 	if err := w.apply(ctx, object, false); err != nil {
 		return StoredCandidate{}, err
 	}
@@ -112,6 +124,13 @@ func (w *clientWriter) CreateCandidate(ctx context.Context, run *v1beta2.DynamoG
 	if err := w.Reader.Get(ctx, types.NamespacedName{Namespace: w.Namespace, Name: name}, &stored); err != nil {
 		return StoredCandidate{}, err
 	}
+	owner := metav1.GetControllerOf(&stored)
+	if stored.UID != observedUID {
+		return StoredCandidate{}, fmt.Errorf("candidate %s changed uid from %s to %s during apply", name, observedUID, stored.UID)
+	}
+	if owner == nil || owner.UID != run.UID {
+		return StoredCandidate{}, fmt.Errorf("candidate %s is no longer controlled by run %s after apply", name, run.Name)
+	}
 	return StoredCandidate{
 		ResourceVersion: stored.ResourceVersion,
 		Populated:       stored.Status.Metrics != nil,
diff --git a/deploy/operator/internal/dgdrrunpublisher/writer_envtest_test.go b/deploy/operator/internal/dgdrrunpublisher/writer_envtest_test.go
index 6d82f892d216..6437e8c363cb 100644
--- a/deploy/operator/internal/dgdrrunpublisher/writer_envtest_test.go
+++ b/deploy/operator/internal/dgdrrunpublisher/writer_envtest_test.go
@@ -24,6 +24,7 @@ import (
 	"encoding/json"
 	"errors"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -39,6 +40,8 @@ import (
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 )
 
 // These tests send the publisher's real server-side-apply payloads to an API server that
@@ -120,6 +123,82 @@ func envtestCandidate() dgdcreconcile.DesiredCandidate {
 	}
 }
 
+func foreignCandidate(t *testing.T, namespace, name string, candidate dgdcreconcile.DesiredCandidate) *v1beta2.DynamoGraphDeploymentCandidate {
+	t.Helper()
+	spec, _, err := decodeManifest(candidate)
+	if err != nil {
+		t.Fatal(err)
+	}
+	parameters, err := json.Marshal(candidate.Parameters)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return &v1beta2.DynamoGraphDeploymentCandidate{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+			Labels:    map[string]string{"owner": "foreign"},
+		},
+		Spec: v1beta2.DynamoGraphDeploymentCandidateSpec{
+			DynamoGraphDeploymentSpec: *spec,
+			Parameters:                runtime.RawExtension{Raw: parameters},
+		},
+	}
+}
+
+type writeInterceptClient struct {
+	client.Client
+	beforeCreate func()
+	beforeApply  func()
+}
+
+func (c *writeInterceptClient) Create(ctx context.Context, object client.Object, options ...client.CreateOption) error {
+	if c.beforeCreate != nil {
+		hook := c.beforeCreate
+		c.beforeCreate = nil
+		hook()
+	}
+	return c.Client.Create(ctx, object, options...)
+}
+
+func (c *writeInterceptClient) Apply(ctx context.Context, configuration runtime.ApplyConfiguration, options ...client.ApplyOption) error {
+	if c.beforeApply != nil {
+		hook := c.beforeApply
+		c.beforeApply = nil
+		hook()
+	}
+	return c.Client.Apply(ctx, configuration, options...)
+}
+
+type reconcileRecorder struct {
+	delegate *Reconciler
+	count    atomic.Int64
+}
+
+func (r *reconcileRecorder) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.Result, error) {
+	result, err := r.delegate.Reconcile(ctx, request)
+	r.count.Add(1)
+	return result, err
+}
+
+func waitForReconcileQuiet(t *testing.T, recorder *reconcileRecorder, quiet, timeout time.Duration) {
+	t.Helper()
+	var last int64 = -1
+	var stableSince time.Time
+	require.Eventually(t, func() bool {
+		current := recorder.count.Load()
+		if current == 0 {
+			return false
+		}
+		if current != last {
+			last = current
+			stableSince = time.Now()
+			return false
+		}
+		return time.Since(stableSince) >= quiet
+	}, timeout, 10*time.Millisecond, "reconciliation count did not become stable")
+}
+
 func hasApplyManager(fields []metav1.ManagedFieldsEntry, subresource string) bool {
 	for _, entry := range fields {
 		if entry.Manager == fieldOwner && entry.Operation == metav1.ManagedFieldsOperationApply && entry.Subresource == subresource {
@@ -162,31 +241,25 @@ func TestEnvtestCandidateApplyIsAcceptedAndIdempotent(t *testing.T) {
 	if again.ResourceVersion != stored.ResourceVersion {
 		t.Errorf("re-applying the same candidate changed resourceVersion %s -> %s", stored.ResourceVersion, again.ResourceVersion)
 	}
+
+	drifted := f.candidate(name)
+	delete(drifted.Labels, LabelRunUID)
+	if err := f.client.Update(f.ctx, drifted); err != nil {
+		t.Fatalf("remove run label: %v", err)
+	}
+	if _, err := f.writer.CreateCandidate(f.ctx, f.run, name, envtestCandidate()); err != nil {
+		t.Fatalf("repair candidate labels: %v", err)
+	}
+	if got := f.candidate(name); got.Labels[LabelRunUID] != string(f.run.UID) {
+		t.Fatalf("run label was not repaired: %v", got.Labels)
+	}
 }
 
 func TestEnvtestCandidateApplyDoesNotAdoptAForeignObject(t *testing.T) {
 	f := newWriterFixture(t)
 	name := f.name("a")
 	candidate := envtestCandidate()
-	spec, _, err := decodeManifest(candidate)
-	if err != nil {
-		t.Fatal(err)
-	}
-	parameters, err := json.Marshal(candidate.Parameters)
-	if err != nil {
-		t.Fatal(err)
-	}
-	foreign := &v1beta2.DynamoGraphDeploymentCandidate{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: f.run.Namespace,
-			Name:      name,
-			Labels:    map[string]string{"owner": "foreign"},
-		},
-		Spec: v1beta2.DynamoGraphDeploymentCandidateSpec{
-			DynamoGraphDeploymentSpec: *spec,
-			Parameters:                runtime.RawExtension{Raw: parameters},
-		},
-	}
+	foreign := foreignCandidate(t, f.run.Namespace, name, candidate)
 	if err := f.client.Create(f.ctx, foreign); err != nil {
 		t.Fatalf("create foreign candidate: %v", err)
 	}
@@ -203,6 +276,69 @@ func TestEnvtestCandidateApplyDoesNotAdoptAForeignObject(t *testing.T) {
 	}
 }
 
+func TestEnvtestCandidateCreateRaceDoesNotAdoptForeignObject(t *testing.T) {
+	f := newWriterFixture(t)
+	name := f.name("a")
+	candidate := envtestCandidate()
+	foreign := foreignCandidate(t, f.run.Namespace, name, candidate)
+	intercept := &writeInterceptClient{Client: f.client}
+	intercept.beforeCreate = func() {
+		if err := f.client.Create(f.ctx, foreign); err != nil {
+			t.Fatalf("create racing foreign candidate: %v", err)
+		}
+	}
+	writer := *f.writer
+	writer.Client = intercept
+
+	if _, err := writer.CreateCandidate(f.ctx, f.run, name, candidate); err == nil {
+		t.Fatal("candidate creation adopted the foreign object that won the name race")
+	}
+	got := f.candidate(name)
+	if owner := metav1.GetControllerOf(got); owner != nil {
+		t.Fatalf("racing foreign candidate was adopted by %+v", owner)
+	}
+	if got.Labels["owner"] != "foreign" || got.Labels[LabelRunUID] != "" {
+		t.Fatalf("racing foreign candidate was modified: labels = %v", got.Labels)
+	}
+}
+
+func TestEnvtestCandidateApplyRejectsDeleteRecreateRace(t *testing.T) {
+	f := newWriterFixture(t)
+	name := f.name("a")
+	candidate := envtestCandidate()
+	if _, err := f.writer.CreateCandidate(f.ctx, f.run, name, candidate); err != nil {
+		t.Fatalf("create owned candidate: %v", err)
+	}
+	ownedUID := f.candidate(name).UID
+	foreign := foreignCandidate(t, f.run.Namespace, name, candidate)
+	intercept := &writeInterceptClient{Client: f.client}
+	intercept.beforeApply = func() {
+		current := f.candidate(name)
+		if err := f.client.Delete(f.ctx, current); err != nil {
+			t.Fatalf("delete owned candidate during apply race: %v", err)
+		}
+		if err := f.client.Create(f.ctx, foreign); err != nil {
+			t.Fatalf("recreate foreign candidate during apply race: %v", err)
+		}
+	}
+	writer := *f.writer
+	writer.Client = intercept
+
+	if _, err := writer.CreateCandidate(f.ctx, f.run, name, candidate); err == nil {
+		t.Fatal("UID- and resourceVersion-constrained apply accepted a replacement object")
+	}
+	got := f.candidate(name)
+	if got.UID == ownedUID {
+		t.Fatalf("candidate UID did not change across delete/recreate: %s", got.UID)
+	}
+	if owner := metav1.GetControllerOf(got); owner != nil {
+		t.Fatalf("replacement foreign candidate was adopted by %+v", owner)
+	}
+	if got.Labels["owner"] != "foreign" || got.Labels[LabelRunUID] != "" {
+		t.Fatalf("replacement foreign candidate was modified: labels = %v", got.Labels)
+	}
+}
+
 func TestEnvtestCandidateStatusIsWrittenOnceAndGuardedByResourceVersion(t *testing.T) {
 	f := newWriterFixture(t)
 	name := f.name("a")
@@ -313,10 +449,15 @@ func TestEnvtestDeleteCandidateHonorsTheUIDPrecondition(t *testing.T) {
 
 func TestEnvtestCandidateUpdateTriggersStatusRepair(t *testing.T) {
 	f := newWriterFixture(t)
-	dir := t.TempDir()
 	podName := "publisher-pod"
+	candidate := envtestCandidate()
+	snapshot, err := ParseSnapshot(snapshotJSON(t, PhaseRunning, cand{id: "a", metrics: candidate.Metrics}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	desired := Desire(f.run.Name, string(f.run.UID), snapshot)
 
-	t.Log("Create the publisher Pod and a running snapshot")
+	t.Log("Arrange a fully converged Run and Candidate before starting the manager")
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{Namespace: f.run.Namespace, Name: podName},
 		Spec:       corev1.PodSpec{Containers: []corev1.Container{{Name: "sweeper", Image: "example.invalid/sweeper"}}},
@@ -324,22 +465,63 @@ func TestEnvtestCandidateUpdateTriggersStatusRepair(t *testing.T) {
 	if err := f.client.Create(f.ctx, pod); err != nil {
 		t.Fatalf("create pod: %v", err)
 	}
-	writeSnapshot(t, dir, snapshotJSON(t, PhaseRunning, cand{id: "a", metrics: map[string]any{"throughput": 12.5}}))
+	name := f.name("a")
+	stored, err := f.writer.CreateCandidate(f.ctx, f.run, name, candidate)
+	if err != nil {
+		t.Fatalf("create candidate: %v", err)
+	}
+	if err := f.writer.PopulateStatus(f.ctx, name, stored.ResourceVersion, candidate); err != nil {
+		t.Fatalf("populate candidate: %v", err)
+	}
+	if err := f.writer.UpdateRunStatus(f.ctx, f.currentRun().ResourceVersion, desired.Status); err != nil {
+		t.Fatalf("converge run status: %v", err)
+	}
+
+	t.Log("Start the production controller watches around an instrumented Reconciler")
+	scheme, err := NewScheme()
+	if err != nil {
+		t.Fatal(err)
+	}
+	manager, err := ctrl.NewManager(f.config, ctrl.Options{
+		Scheme:         scheme,
+		Cache:          cacheOptions(Config{Namespace: f.run.Namespace, RunName: f.run.Name, RunUID: string(f.run.UID), PodName: podName}),
+		Metrics:        metricsserver.Options{BindAddress: "0"},
+		LeaderElection: false,
+	})
+	if err != nil {
+		t.Fatalf("create manager: %v", err)
+	}
+	store := &InputStore{}
+	store.accept(snapshot)
+	completion := NewCompletion()
+	reconciler := &Reconciler{
+		Client: manager.GetClient(),
+		Writer: &clientWriter{
+			Client:    manager.GetClient(),
+			Reader:    manager.GetAPIReader(),
+			Namespace: f.run.Namespace,
+			RunName:   f.run.Name,
+			RunUID:    string(f.run.UID),
+		},
+		Namespace:        f.run.Namespace,
+		RunName:          f.run.Name,
+		RunUID:           string(f.run.UID),
+		PodName:          podName,
+		SweeperContainer: "sweeper",
+		Store:            store,
+		Completion:       completion,
+	}
+	recorder := &reconcileRecorder{delegate: reconciler}
+	events := make(chan event.GenericEvent, 1)
+	key := types.NamespacedName{Namespace: f.run.Namespace, Name: f.run.Name}
+	if err := setupController(manager, recorder, key, events); err != nil {
+		t.Fatalf("setup controller: %v", err)
+	}
 
-	t.Log("Start the real manager and wait for its initial publication")
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan error, 1)
 	go func() {
-		done <- Run(ctx, f.config, Config{
-			Namespace:        f.run.Namespace,
-			RunName:          f.run.Name,
-			RunUID:           string(f.run.UID),
-			PodName:          podName,
-			SweeperContainer: "sweeper",
-			SnapshotDir:      dir,
-			MinPollInterval:  10 * time.Millisecond,
-			MaxPollInterval:  20 * time.Millisecond,
-		})
+		done <- manager.Start(ctx)
 	}()
 	t.Cleanup(func() {
 		cancel()
@@ -353,24 +535,20 @@ func TestEnvtestCandidateUpdateTriggersStatusRepair(t *testing.T) {
 		}
 	})
 
-	name := f.name("a")
-	require.Eventually(t, func() bool {
-		var candidate v1beta2.DynamoGraphDeploymentCandidate
-		if err := f.client.Get(f.ctx, types.NamespacedName{Namespace: f.run.Namespace, Name: name}, &candidate); err != nil {
-			return false
-		}
-		return candidate.Status.Metrics != nil
-	}, 10*time.Second, 25*time.Millisecond, "publisher did not create and populate the candidate")
+	waitForReconcileQuiet(t, recorder, 250*time.Millisecond, 10*time.Second)
+	beforeUpdate := recorder.count.Load()
 
-	t.Log("Clear the Candidate status and let its update event trigger repair")
-	var candidate v1beta2.DynamoGraphDeploymentCandidate
-	if err := f.client.Get(f.ctx, types.NamespacedName{Namespace: f.run.Namespace, Name: name}, &candidate); err != nil {
+	t.Log("Clear only Candidate status after the controller is quiet")
+	var current v1beta2.DynamoGraphDeploymentCandidate
+	if err := f.client.Get(f.ctx, types.NamespacedName{Namespace: f.run.Namespace, Name: name}, &current); err != nil {
 		t.Fatalf("get candidate: %v", err)
 	}
-	candidate.Status.Metrics = nil
-	if err := f.client.Status().Update(f.ctx, &candidate); err != nil {
+	current.Status.Metrics = nil
+	if err := f.client.Status().Update(f.ctx, &current); err != nil {
 		t.Fatalf("clear candidate status: %v", err)
 	}
+	require.Eventually(t, func() bool { return recorder.count.Load() > beforeUpdate }, 10*time.Second, 25*time.Millisecond,
+		"Candidate update did not trigger a new reconciliation")
 	require.Eventually(t, func() bool {
 		var repaired v1beta2.DynamoGraphDeploymentCandidate
 		if err := f.client.Get(f.ctx, types.NamespacedName{Namespace: f.run.Namespace, Name: name}, &repaired); err != nil {