refactor(operator): separate managed Grove components

deploy/operator/internal/AGENTS.md

@@ -39,6 +39,16 @@ reconciliation, rollout, restart, readiness, status, watches, and provider integ
 - The program request contains only the explicitly mutable DGD. Clients, recorders,
   configuration, and other dependencies live on the concrete program or collaborator
   that uses them.
+- A `DynamoGraphDeployment` value always represents the complete API object. Never
+  construct or pass a copy with `spec.components` filtered.
+- When a composite program divides workload ownership between managed and delegated
+  paths, set the ownership predicate once at the program composition root and pass it
+  with the complete DGD to nested reconcilers that select or observe provider-owned
+  workloads. Those reconcilers must not reclassify component ownership or accept a
+  redundant component slice they can derive from the request. Graph-wide reconcilers
+  consume the complete DGD directly when workload ownership is irrelevant.
+  A renderer may copy and transform one component locally while emitting provider
+  resources; it must not expose a parallel component collection as another graph view.
 - The program result contains `ctrl.Result`, the complete desired DGD status, and queued
   status-transition events.
 - Do not add speculative `Facts`, `Inputs`, `ComputedState`, `WorkloadModel`, render

deploy/operator/internal/controller/dgd_grove_lpx_restart.go

@@ -37,36 +37,39 @@ func (r *lpxRestartProgressResolver) Resolve(
 // resolveCompositeGroveRestartProgress composes child-owned LPX and ordinary Grove observations.
 func resolveCompositeGroveRestartProgress(
 	ctx context.Context,
-	source *v1beta1.DynamoGraphDeployment,
-	ordinaryDGD *v1beta1.DynamoGraphDeployment,
+	req groveReconcileRequest,
 	inProgress []string,
 	ordinaryResolver *groveRestartProgressResolver,
 	lpxResolver *lpxRestartProgressResolver,
 ) []string {
 	ordinary := make([]string, 0, len(inProgress))
-	lpxComponents := make([]string, 0, len(inProgress))
+	delegated := make([]string, 0, len(inProgress))
 	pending := make(map[string]bool, len(inProgress))
+	delegatedNames := make(map[string]struct{})
+	for _, component := range req.DelegatedComponents() {
+		delegatedNames[component.ComponentName] = struct{}{}
+	}
 
 	for _, name := range inProgress {
-		component := source.GetComponentByName(name)
+		component := req.DGD.GetComponentByName(name)
 		if component == nil {
 			continue
 		}
-		if component.IsLPX() {
-			lpxComponents = append(lpxComponents, name)
+		if _, found := delegatedNames[name]; found {
+			delegated = append(delegated, name)
 		} else {
 			ordinary = append(ordinary, name)
 		}
 	}
 
 	// Observe the shared LPX child before ordinary Grove restart progress.
-	if len(lpxComponents) > 0 {
-		for _, name := range lpxResolver.Resolve(ctx, source, lpxComponents) {
+	if len(delegated) > 0 {
+		for _, name := range lpxResolver.Resolve(ctx, req.DGD, delegated) {
 			pending[name] = true
 		}
 	}
 	if len(ordinary) > 0 {
-		for _, name := range ordinaryResolver.Resolve(ctx, ordinaryDGD, ordinary) {
+		for _, name := range ordinaryResolver.Resolve(ctx, req, ordinary) {
 			pending[name] = true
 		}
 	}

deploy/operator/internal/controller/dgd_grove_program.go

@@ -20,7 +20,6 @@ package controller
 import (
 	"context"
 	"fmt"
-	"slices"
 	"time"
 
 	nvidiacomv1beta1 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1beta1"
@@ -43,6 +42,41 @@ type groveProgram struct {
 	lpx                *dgdLPXHandoff
 }
 
+// groveReconcileRequest keeps the complete DGD together with the component
+// ownership boundary selected by the outer Grove program.
+type groveReconcileRequest struct {
+	DGD *nvidiacomv1beta1.DynamoGraphDeployment
+
+	// IsDelegated reports whether another controller reconciles this component's
+	// workloads. A nil predicate delegates no components.
+	IsDelegated func(*nvidiacomv1beta1.DynamoComponentDeploymentSharedSpec) bool
+}
+
+func (r groveReconcileRequest) ManagedComponents() []nvidiacomv1beta1.DynamoComponentDeploymentSharedSpec {
+	components := make([]nvidiacomv1beta1.DynamoComponentDeploymentSharedSpec, 0, len(r.DGD.Spec.Components))
+	for i := range r.DGD.Spec.Components {
+		component := &r.DGD.Spec.Components[i]
+		if r.IsDelegated == nil || !r.IsDelegated(component) {
+			components = append(components, *component)
+		}
+	}
+	return components
+}
+
+func (r groveReconcileRequest) DelegatedComponents() []nvidiacomv1beta1.DynamoComponentDeploymentSharedSpec {
+	if r.IsDelegated == nil {
+		return nil
+	}
+	components := make([]nvidiacomv1beta1.DynamoComponentDeploymentSharedSpec, 0, len(r.DGD.Spec.Components))
+	for i := range r.DGD.Spec.Components {
+		component := &r.DGD.Spec.Components[i]
+		if r.IsDelegated(component) {
+			components = append(components, *component)
+		}
+	}
+	return components
+}
+
 // newGroveProgram wires the Grove pathway at the DGD composition root.
 func (r *DynamoGraphDeploymentReconciler) newGroveProgram() *groveProgram {
 	rollout := newDGDWorkerRolloutReconciler(r.Client, r.Recorder)
@@ -81,8 +115,12 @@ func (r *DynamoGraphDeploymentReconciler) newGroveProgram() *groveProgram {
 // are persisted through req.DGD; status accumulates in the returned result.
 func (p *groveProgram) Reconcile(
 	ctx context.Context,
-	req workloadProgramRequest,
+	programReq workloadProgramRequest,
 ) (programResult workloadProgramResult, retErr error) {
+	req := groveReconcileRequest{
+		DGD:         programReq.DGD,
+		IsDelegated: (*nvidiacomv1beta1.DynamoComponentDeploymentSharedSpec).ManagedByExternalController,
+	}
 	programResult = newWorkloadProgramResult(req.DGD)
 	clearComponentGPUShapes(programResult.Status.Components)
 
@@ -95,8 +133,6 @@ func (p *groveProgram) Reconcile(
 		programResult.Fail(req.DGD.Generation, reasonSelectedWorkloadProviderUnavailable, err)
 		return programResult, reconcile.TerminalError(err)
 	}
-	var ordinaryDGD *nvidiacomv1beta1.DynamoGraphDeployment
-
 	defer func() {
 		if retErr != nil {
 			reason := reasonFailedToReconcileResources
@@ -105,10 +141,7 @@ func (p *groveProgram) Reconcile(
 			}
 			programResult.Fail(req.DGD.Generation, reason, retErr)
 		}
-		if ordinaryDGD == nil {
-			ordinaryDGD = projectWithoutExternallyManagedComponents(req.DGD)
-		}
-		p.topology.Reconcile(ctx, ordinaryDGD, &programResult)
+		p.topology.Reconcile(ctx, req, &programResult)
 	}()
 	log.FromContext(ctx).Info(
 		"Reconciling Grove resources",
@@ -126,18 +159,15 @@ func (p *groveProgram) Reconcile(
 	if err != nil {
 		return programResult, err
 	}
-	ordinaryDGD = projectWithoutExternallyManagedComponents(req.DGD)
-
 	previousRestart := programResult.Status.Restart
 	restart := p.restart.Resolve(
 		ctx,
 		req.DGD,
 		&programResult.Status,
-		func(ctx context.Context, source *nvidiacomv1beta1.DynamoGraphDeployment, inProgress []string) []string {
+		func(ctx context.Context, _ *nvidiacomv1beta1.DynamoGraphDeployment, inProgress []string) []string {
 			return resolveCompositeGroveRestartProgress(
 				ctx,
-				source,
-				ordinaryDGD,
+				req,
 				inProgress,
 				p.restartProgress,
 				p.lpxRestartProgress,
@@ -149,8 +179,7 @@ func (p *groveProgram) Reconcile(
 
 	result, err := p.workloads.Reconcile(
 		ctx,
-		req.DGD,
-		ordinaryDGD,
+		req,
 		restart.State,
 		checkpoints.Infos,
 	)
@@ -181,7 +210,7 @@ func (p *groveProgram) Reconcile(
 	result = applyCheckpointStartupReadiness(result, checkpoints.Infos)
 
 	if result.State != nvidiacomv1beta1.DGDStatePending || result.Reason != reasonWaitingForCheckpoint {
-		if err := p.scalingAdapters.Reconcile(ctx, ordinaryDGD); err != nil {
+		if err := p.scalingAdapters.Reconcile(ctx, req.DGD); err != nil {
 			log.FromContext(ctx).Error(err, "Failed to reconcile scaling adapters")
 			return programResult, fmt.Errorf("failed to reconcile scaling adapters: %w", err)
 		}
@@ -190,16 +219,3 @@ func (p *groveProgram) Reconcile(
 	programResult.applyReconcileResult(req.DGD.Generation, result)
 	return programResult, nil
 }
-
-// projectWithoutExternallyManagedComponents copies source without externally managed components.
-// The source must be non-nil; shared nested data must not be mutated by callers.
-func projectWithoutExternallyManagedComponents(source *nvidiacomv1beta1.DynamoGraphDeployment) *nvidiacomv1beta1.DynamoGraphDeployment {
-	projected := *source
-	projected.Spec.Components = slices.DeleteFunc(
-		slices.Clone(projected.Spec.Components),
-		func(component nvidiacomv1beta1.DynamoComponentDeploymentSharedSpec) bool {
-			return component.ManagedByExternalController()
-		},
-	)
-	return &projected
-}

deploy/operator/internal/controller/dgd_grove_restart_progress_resolver.go

@@ -20,7 +20,6 @@ package controller
 import (
 	"context"
 
-	nvidiacomv1beta1 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1beta1"
 	"github.com/ai-dynamo/dynamo/deploy/operator/internal/dynamo"
 	grovev1alpha1 "github.com/ai-dynamo/grove/operator/api/core/v1alpha1"
 	"k8s.io/apimachinery/pkg/types"
@@ -40,26 +39,26 @@ func newGroveRestartProgressResolver(reader client.Reader) *groveRestartProgress
 
 func (r *groveRestartProgressResolver) Resolve(
 	ctx context.Context,
-	dgd *nvidiacomv1beta1.DynamoGraphDeployment,
+	req groveReconcileRequest,
 	inProgress []string,
 ) []string {
 	logger := log.FromContext(ctx)
 
 	pcs := &grovev1alpha1.PodCliqueSet{}
-	pcsName := dynamo.PCSNameForDGD(dgd.Name, dgd.Spec.Components)
-	if err := r.reader.Get(ctx, types.NamespacedName{Name: pcsName, Namespace: dgd.Namespace}, pcs); err != nil {
+	pcsName := dynamo.PCSNameForDGD(req.DGD, req.IsDelegated)
+	if err := r.reader.Get(ctx, types.NamespacedName{Name: pcsName, Namespace: req.DGD.Namespace}, pcs); err != nil {
 		logger.Error(err, "failed to get PodCliqueSet")
 		return inProgress
 	}
 
 	if pcs.Status.ObservedGeneration == nil {
-		logger.Info("PodCliqueSet observedGeneration is nil", "name", dgd.Name)
+		logger.Info("PodCliqueSet observedGeneration is nil", "name", req.DGD.Name)
 		return inProgress
 	}
 	if *pcs.Status.ObservedGeneration < pcs.Generation {
 		logger.Info(
 			"PodCliqueSet not yet reconciled",
-			"name", dgd.Name,
+			"name", req.DGD.Name,
 			"generation", pcs.Generation,
 			"observedGeneration", *pcs.Status.ObservedGeneration,
 		)
@@ -68,12 +67,12 @@ func (r *groveRestartProgressResolver) Resolve(
 
 	updatedInProgress := make([]string, 0, len(inProgress))
 	for _, componentName := range inProgress {
-		component := dgd.GetComponentByName(componentName)
+		component := req.DGD.GetComponentByName(componentName)
 		if component == nil {
 			logger.V(1).Info("component not found in DGD", "componentName", componentName)
 			continue
 		}
-		resourceName := dynamo.GroveComponentResourceName(dgd, componentName)
+		resourceName := dynamo.GroveComponentResourceName(pcsName, componentName)
 
 		var (
 			isReady bool
@@ -87,15 +86,15 @@ func (r *groveRestartProgressResolver) Resolve(
 				ctx,
 				r.reader,
 				resourceName,
-				dgd.Namespace,
+				req.DGD.Namespace,
 				logger,
 			)
 		} else {
 			isReady, reason, _, _, _ = dynamo.CheckPodCliqueReady(
 				ctx,
 				r.reader,
 				resourceName,
-				dgd.Namespace,
+				req.DGD.Namespace,
 				logger,
 			)
 		}

deploy/operator/internal/controller/dgd_grove_scaler.go

@@ -22,7 +22,6 @@ import (
 	"fmt"
 
 	nvidiacomv1alpha1 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1alpha1"
-	nvidiacomv1beta1 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1beta1"
 	"github.com/ai-dynamo/dynamo/deploy/operator/internal/checkpoint"
 	"github.com/ai-dynamo/dynamo/deploy/operator/internal/consts"
 	commoncontroller "github.com/ai-dynamo/dynamo/deploy/operator/internal/controller_common"
@@ -45,14 +44,16 @@ func newGroveScaler(kubeClient client.Client) *groveScaler {
 // asynchronously from the PodCliqueSet.
 func (s *groveScaler) Reconcile(
 	ctx context.Context,
-	dgd *nvidiacomv1beta1.DynamoGraphDeployment,
+	req groveReconcileRequest,
 	checkpointInfos map[string]*checkpoint.CheckpointInfo,
 ) error {
 	logger := log.FromContext(ctx)
 	logger.V(1).Info("Reconciling Grove scaling operations")
+	managedComponents := req.ManagedComponents()
+	pcsName := dynamo.PCSNameForDGD(req.DGD, req.IsDelegated)
 
-	for i := range dgd.Spec.Components {
-		component := &dgd.Spec.Components[i]
+	for i := range managedComponents {
+		component := &managedComponents[i]
 		componentName := component.ComponentName
 		info := checkpointInfos[componentName]
 		gated := info != nil &&
@@ -71,7 +72,7 @@ func (s *groveScaler) Reconcile(
 		}
 
 		usesPCSG := component.UsesPCSG()
-		resourceName := dynamo.GroveComponentResourceName(dgd, componentName)
+		resourceName := dynamo.GroveComponentResourceName(pcsName, componentName)
 		resourceKind := "PodClique"
 		gvr := consts.PodCliqueGVR
 		if usesPCSG {
@@ -82,7 +83,7 @@ func (s *groveScaler) Reconcile(
 			ctx,
 			gvr,
 			resourceName,
-			dgd.Namespace,
+			req.DGD.Namespace,
 			replicas,
 		); err != nil {
 			logger.Error(

deploy/operator/internal/controller/dgd_grove_scaler_clustertest_test.go

@@ -50,7 +50,7 @@ func TestClusterGroveScalerUsesScaleSubresource(t *testing.T) {
 	require.NoError(t, env.Client().Create(ctx, podClique))
 
 	t.Log("Scale the PodClique through the production controller-runtime subresource path")
-	require.NoError(t, newGroveScaler(env.Client()).Reconcile(ctx, dgd, nil))
+	require.NoError(t, newGroveScaler(env.Client()).Reconcile(ctx, groveReconcileRequest{DGD: dgd}, nil))
 
 	t.Log("Read the PodClique back and verify the API server applied the scale update")
 	require.NoError(t, env.Client().Get(ctx, client.ObjectKeyFromObject(podClique), podClique))

deploy/operator/internal/controller/dgd_grove_scaler_test.go

@@ -73,7 +73,7 @@ func TestGroveScaler_ReconcileTargetsExpectedGroveChildren(t *testing.T) {
 
 	err := newGroveScaler(kubeClient).Reconcile(
 		t.Context(),
-		dgd,
+		groveReconcileRequest{DGD: dgd},
 		map[string]*checkpoint.CheckpointInfo{
 			"gated": {
 				Enabled:       true,
@@ -172,7 +172,7 @@ func TestGroveScaler_ReconcileHandlesScaleReadErrors(t *testing.T) {
 			WithScheme(newDynamoGraphDeploymentControllerTestScheme(t)).
 			WithRESTMapper(groveScaleRESTMapper()).
 			Build()
-		require.NoError(t, newGroveScaler(kubeClient).Reconcile(t.Context(), dgd, nil))
+		require.NoError(t, newGroveScaler(kubeClient).Reconcile(t.Context(), groveReconcileRequest{DGD: dgd}, nil))
 	})
 
 	t.Run("other errors are propagated", func(t *testing.T) {
@@ -185,7 +185,7 @@ func TestGroveScaler_ReconcileHandlesScaleReadErrors(t *testing.T) {
 				},
 			}).
 			Build()
-		err := newGroveScaler(kubeClient).Reconcile(t.Context(), dgd, nil)
+		err := newGroveScaler(kubeClient).Reconcile(t.Context(), groveReconcileRequest{DGD: dgd}, nil)
 		require.ErrorContains(t, err, "scale read failed")
 	})
 }