fix(operator): harden DGDR run publication

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)
+}

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)])
+	}
+}

deploy/operator/internal/dgdrrunpublisher/doc.go

@@ -37,17 +37,18 @@
 // 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:
-// 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
 // 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):
 //

deploy/operator/internal/dgdrrunpublisher/manager.go

@@ -28,21 +28,18 @@ 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"
 	"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"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"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"
 )
@@ -52,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
@@ -84,35 +81,24 @@ 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: {}},
 		ByObject: map[client.Object]cache.ByObject{
 			&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),
 			},
 		},
 	}
 }
 
-// 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.
@@ -175,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{}, builder.WithPredicates(ignoreCandidateUpdates)).
-		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 {
@@ -215,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)

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

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)
 	}
 }
 

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

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) {