fix(queue): validate CPU topology flavor compatibility

charts/tau-core-controller/kustomize/rbac.yaml

@@ -40,6 +40,9 @@ rules:
   - apiGroups: ["kueue.x-k8s.io"]
     resources: ["clusterqueues"]
     verbs: ["get"]
+  - apiGroups: ["kueue.x-k8s.io"]
+    resources: ["resourceflavors"]
+    verbs: ["get"]
   - apiGroups: ["tau.azure.com"]
     resources: ["clusters"]
     resourceNames: ["cluster"]

charts/tau-core-controller/templates/rbac.yaml

@@ -38,6 +38,9 @@ rules:
   - apiGroups: ["kueue.x-k8s.io"]
     resources: ["clusterqueues"]
     verbs: ["get"]
+  - apiGroups: ["kueue.x-k8s.io"]
+    resources: ["resourceflavors"]
+    verbs: ["get"]
   - apiGroups: ["tau.azure.com"]
     resources: ["clusters"]
     resourceNames: ["cluster"]

charts/tau-core-controller/tests/rbac_test.yaml

@@ -40,6 +40,20 @@ tests:
             resources: ["rayjobs", "rayservices"]
             verbs: ["create", "get", "list", "watch", "delete", "patch", "update"]
 
+  - it: grants workspace researchers ResourceFlavor reads for queue preflight
+    documentSelector:
+      path: metadata.name
+      value: tau-clusterqueue-reader-v1
+    asserts:
+      - isKind:
+          of: ClusterRole
+      - contains:
+          path: rules
+          content:
+            apiGroups: ["kueue.x-k8s.io"]
+            resources: ["resourceflavors"]
+            verbs: ["get"]
+
   - it: grants the controller create-only mutation access to Kueue Topologies
     asserts:
       - contains:

cli/internal/cli/topology_flags.go

@@ -260,6 +260,7 @@ type renderedQueueContract struct {
 	QueueName       string
 	GPUClass        string
 	GPUCount        int
+	CPURequest      bool
 	GPUResourceName string
 	NodeSelector    map[string]string
 	PodTolerations  [][]kueueapi.Toleration
@@ -306,6 +307,7 @@ func inspectRenderedQueue(ctx context.Context, r kubeRawRunner, namespace string
 		NodeSelector:            nodeSelector,
 		PodTolerations:          contract.PodTolerations,
 		GPUCount:                contract.GPUCount,
+		CPURequest:              contract.CPURequest,
 		GPUResourceName:         gpuResourceName,
 		TopologyRequest:         contract.TopologyRequest,
 		TopologyName:            policy.TopologyName,
@@ -366,6 +368,9 @@ func renderedQueueContractFromManifest(manifest []byte) (renderedQueueContract,
 		if renderedGPUObjectsRequestTopology(obj) {
 			out.TopologyRequest = true
 		}
+		if renderedObjectsRequestCPU(obj) {
+			out.CPURequest = true
+		}
 		resourceNames := map[string]struct{}{}
 		collectRenderedGPUResourceNames(obj, resourceNames)
 		for resourceName := range resourceNames {
@@ -429,7 +434,7 @@ func renderedGPUObjectsRequestTopology(value any) bool {
 
 func renderedPodTemplateRequestsTopology(value map[string]any) bool {
 	spec, ok := value["spec"].(map[string]any)
-	if !ok || collectRenderedDirectPodGPUCount(spec) == 0 {
+	if !ok || !renderedDirectPodRequestsCompute(spec) {
 		return false
 	}
 	metadata, _ := value["metadata"].(map[string]any)
@@ -446,6 +451,57 @@ func renderedPodTemplateRequestsTopology(value map[string]any) bool {
 	return false
 }
 
+func renderedDirectPodRequestsCompute(spec map[string]any) bool {
+	return collectRenderedDirectPodGPUCount(spec) > 0 ||
+		renderedResourceRequested(spec["containers"], "cpu") ||
+		renderedResourceRequested(spec["initContainers"], "cpu")
+}
+
+func renderedObjectsRequestCPU(value any) bool {
+	switch typed := value.(type) {
+	case map[string]any:
+		if spec, ok := typed["spec"].(map[string]any); ok &&
+			(renderedResourceRequested(spec["containers"], "cpu") ||
+				renderedResourceRequested(spec["initContainers"], "cpu")) {
+			return true
+		}
+		for _, child := range typed {
+			if renderedObjectsRequestCPU(child) {
+				return true
+			}
+		}
+	case []any:
+		for _, child := range typed {
+			if renderedObjectsRequestCPU(child) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func renderedResourceRequested(value any, resourceName string) bool {
+	switch typed := value.(type) {
+	case map[string]any:
+		for key, child := range typed {
+			if key == resourceName && strings.TrimSpace(fmt.Sprint(child)) != "" &&
+				strings.TrimSpace(fmt.Sprint(child)) != "0" {
+				return true
+			}
+			if renderedResourceRequested(child, resourceName) {
+				return true
+			}
+		}
+	case []any:
+		for _, child := range typed {
+			if renderedResourceRequested(child, resourceName) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func collectRenderedDirectPodGPUCount(spec map[string]any) int {
 	count := collectRenderedGPUCount(spec["containers"])
 	count = maxInt(count, collectRenderedGPUCount(spec["initContainers"]))

cli/internal/cli/topology_flags_test.go

@@ -279,6 +279,70 @@ spec:
 	}
 }
 
+func TestValidateRenderedQueueRejectsCPUTopologyOnTopologyFreeFlavor(t *testing.T) {
+	runner := &fakeRawRunner{
+		outputs: map[string]string{
+			fakeRawKey("-n", "workspace", "get", "localqueue.kueue.x-k8s.io", "cpu-queue", "-o", "json"): `{"metadata":{"name":"cpu-queue"},"spec":{"clusterQueue":"cpu-cq"}}`,
+			fakeRawKey("get", "clusterqueue.kueue.x-k8s.io", "cpu-cq", "-o", "json"):                     `{"metadata":{"name":"cpu-cq"},"spec":{"resourceGroups":[{"flavors":[{"name":"cpu","resources":[{"name":"cpu","nominalQuota":"100"}]}]}]}}`,
+			fakeRawKey("get", "resourceflavor.kueue.x-k8s.io", "cpu", "-o", "json"):                      `{"metadata":{"name":"cpu"},"spec":{}}`,
+		},
+		errors: map[string]error{},
+	}
+	manifest := []byte(`apiVersion: batch/v1
+kind: Job
+metadata:
+  labels:
+    kueue.x-k8s.io/queue-name: cpu-queue
+spec:
+  template:
+    metadata:
+      annotations:
+        kueue.x-k8s.io/podset-unconstrained-topology: "true"
+    spec:
+      containers:
+      - resources:
+          requests:
+            cpu: "1"
+`)
+
+	err := validateRenderedQueue(context.Background(), runner, "workspace", manifest, jobrender.Options{}, queueValidationPolicy{})
+	if err == nil || !strings.Contains(err.Error(), "CPU workload sets policy.topology") ||
+		!strings.Contains(err.Error(), "do not support TopologyAwareScheduling") {
+		t.Fatalf("expected actionable CPU topology mismatch, got %v", err)
+	}
+}
+
+func TestValidateRenderedQueueAcceptsCPUTopologyOnTASFlavor(t *testing.T) {
+	runner := &fakeRawRunner{
+		outputs: map[string]string{
+			fakeRawKey("-n", "workspace", "get", "localqueue.kueue.x-k8s.io", "cpu-queue", "-o", "json"): `{"metadata":{"name":"cpu-queue"},"spec":{"clusterQueue":"cpu-cq"}}`,
+			fakeRawKey("get", "clusterqueue.kueue.x-k8s.io", "cpu-cq", "-o", "json"):                     `{"metadata":{"name":"cpu-cq"},"spec":{"resourceGroups":[{"flavors":[{"name":"cpu-tas","resources":[{"name":"cpu","nominalQuota":"100"}]}]}]}}`,
+			fakeRawKey("get", "resourceflavor.kueue.x-k8s.io", "cpu-tas", "-o", "json"):                  `{"metadata":{"name":"cpu-tas"},"spec":{"topologyName":"default-node-topology"}}`,
+		},
+		errors: map[string]error{},
+	}
+	manifest := []byte(`apiVersion: batch/v1
+kind: Job
+metadata:
+  labels:
+    kueue.x-k8s.io/queue-name: cpu-queue
+spec:
+  template:
+    metadata:
+      annotations:
+        kueue.x-k8s.io/podset-unconstrained-topology: "true"
+    spec:
+      containers:
+      - resources:
+          requests:
+            cpu: "1"
+`)
+
+	if err := validateRenderedQueue(context.Background(), runner, "workspace", manifest, jobrender.Options{}, queueValidationPolicy{}); err != nil {
+		t.Fatalf("CPU topology-capable flavor rejected: %v", err)
+	}
+}
+
 func TestRenderedQueueContractIgnoresTopologyOutsideGPUPodTemplate(t *testing.T) {
 	manifest := []byte(`apiVersion: ray.io/v1
 kind: RayJob

cli/internal/queueresolve/validate.go

@@ -34,6 +34,7 @@ type ValidationOptions struct {
 	NodeSelector            map[string]string
 	PodTolerations          [][]kueueapi.Toleration
 	GPUCount                int
+	CPURequest              bool
 	GPUResourceName         string
 }
 
@@ -61,6 +62,7 @@ type validationTarget struct {
 	NodeSelector            map[string]string
 	PodTolerations          [][]kueueapi.Toleration
 	GPUCount                int
+	CPURequest              bool
 	GPUResourceName         string
 }
 
@@ -109,6 +111,9 @@ func ValidateSelection(ctx context.Context, r RawRunner, opts ValidationOptions)
 	if err := validateOptionalTopologyLabels("ClusterQueue", target.ClusterQueue, cq.Metadata.Labels, target); err != nil {
 		return report, err
 	}
+	if err := validateCPUQueueTopologyIntent(ctx, r, cq, target); err != nil {
+		return report, err
+	}
 
 	policyTopologyContract := target.CatalogTopologyContract
 	if policyTopologyContract {
@@ -291,6 +296,7 @@ func (o ValidationOptions) resolve() validationTarget {
 		NodeSelector:            o.NodeSelector,
 		PodTolerations:          o.PodTolerations,
 		GPUCount:                o.GPUCount,
+		CPURequest:              o.CPURequest,
 		GPUResourceName:         strings.TrimSpace(o.GPUResourceName),
 	}
 	if target.Namespace == "" {
@@ -622,6 +628,68 @@ func validateQueueTopologyIntent(ctx context.Context, r RawRunner, cq kueueapi.C
 		target.ClusterQueue, strings.Join(names, ", "))
 }
 
+func validateCPUQueueTopologyIntent(ctx context.Context, r RawRunner, cq kueueapi.ClusterQueue, target validationTarget) error {
+	if !target.CPURequest || target.GPUCount > 0 || !target.TopologyRequest {
+		return nil
+	}
+	flavorNames := resourceFlavorNames(cq, "cpu")
+	if target.ResourceFlavor != "" {
+		flavorNames = []string{target.ResourceFlavor}
+	}
+	if len(flavorNames) == 0 {
+		return nil
+	}
+	var topologyFlavors []string
+	var unreadable []string
+	for _, name := range flavorNames {
+		rf, err := getResourceFlavor(ctx, r, name)
+		if err != nil {
+			unreadable = append(unreadable, fmt.Sprintf("%s (%s)", name, err))
+			continue
+		}
+		if strings.TrimSpace(rf.Spec.TopologyName) != "" {
+			topologyFlavors = append(topologyFlavors, name)
+		}
+	}
+	if len(unreadable) > 0 {
+		return fmt.Errorf(
+			"cannot verify CPU policy.topology compatibility with ClusterQueue %q because ResourceFlavor capabilities could not be read: %s; grant read access to ResourceFlavors or ask the platform owner to inspect the queue",
+			target.ClusterQueue,
+			strings.Join(unreadable, ", "),
+		)
+	}
+	if len(topologyFlavors) > 0 {
+		return nil
+	}
+	return fmt.Errorf(
+		"CPU workload sets policy.topology, but CPU ResourceFlavors in ClusterQueue %q do not support TopologyAwareScheduling; use a topology-capable CPU flavor or omit independent/unconstrained topology for this queue",
+		target.ClusterQueue,
+	)
+}
+
+func resourceFlavorNames(cq kueueapi.ClusterQueue, resourceName string) []string {
+	seen := map[string]bool{}
+	var names []string
+	for _, group := range cq.Spec.ResourceGroups {
+		for _, flavor := range group.Flavors {
+			if seen[flavor.Name] {
+				continue
+			}
+			for _, resource := range flavor.Resources {
+				if resource.Name != resourceName ||
+					resource.NominalQuota.Int64()+resource.BorrowingLimit.Int64() <= 0 {
+					continue
+				}
+				seen[flavor.Name] = true
+				names = append(names, flavor.Name)
+				break
+			}
+		}
+	}
+	sort.Strings(names)
+	return names
+}
+
 func compatibleGPUFlavorNames(cq kueueapi.ClusterQueue, target validationTarget) []string {
 	seen := map[string]struct{}{}
 	var names []string

cli/internal/queueresolve/validate_test.go

@@ -6,6 +6,7 @@ package queueresolve
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"reflect"
 	"strings"
@@ -568,6 +569,35 @@ func TestValidateSelectionSkipsTopologyCapabilityForCPUOnlyConfig(t *testing.T)
 	}
 }
 
+func TestValidateSelectionExplainsCPUResourceFlavorReadDenial(t *testing.T) {
+	runner := &validationFakeRunner{
+		outputs: map[string]string{
+			validationKey("-n", "workspace", "get", "localqueue.kueue.x-k8s.io", "cpu-queue", "-o", "json"): localQueueObject("cpu-queue", "cpu-cq", nil),
+			validationKey("get", "clusterqueue.kueue.x-k8s.io", "cpu-cq", "-o", "json"): `{
+				"metadata":{"name":"cpu-cq"},
+				"spec":{"resourceGroups":[{"flavors":[
+					{"name":"cpu-tas","resources":[{"name":"cpu","nominalQuota":"100"}]}
+				]}]}
+			}`,
+		},
+		errors: map[string]error{
+			validationKey("get", "resourceflavor.kueue.x-k8s.io", "cpu-tas", "-o", "json"): errors.New("Error from server (Forbidden): resourceflavors.kueue.x-k8s.io is forbidden"),
+		},
+	}
+
+	_, err := ValidateSelection(context.Background(), runner, ValidationOptions{
+		Namespace:       "workspace",
+		QueueName:       "cpu-queue",
+		TopologyRequest: true,
+		CPURequest:      true,
+	})
+	if err == nil ||
+		!strings.Contains(err.Error(), "Forbidden") ||
+		!strings.Contains(err.Error(), "grant read access to ResourceFlavors") {
+		t.Fatalf("CPU ResourceFlavor read denial should be actionable, got %v", err)
+	}
+}
+
 func TestValidateSelectionAcceptsTopologyCapableWorkspaceQueue(t *testing.T) {
 	runner := &validationFakeRunner{
 		outputs: map[string]string{

controllers/tau-core/internal/controller/workspace_rbac_test.go

@@ -51,6 +51,21 @@ func TestKustomizeResearcherRoleGrantsRayServicePermissions(t *testing.T) {
 	t.Fatalf("researcher role must grant RayService lifecycle permissions: %#v", role.Rules)
 }
 
+func TestKustomizeClusterQueueReaderRoleGrantsResourceFlavorRead(t *testing.T) {
+	role := kustomizeClusterRole(t, clusterQueueReaderRoleName)
+	want := rbacv1.PolicyRule{
+		APIGroups: []string{"kueue.x-k8s.io"},
+		Resources: []string{"resourceflavors"},
+		Verbs:     []string{"get"},
+	}
+	for _, rule := range role.Rules {
+		if reflect.DeepEqual(rule, want) {
+			return
+		}
+	}
+	t.Fatalf("ClusterQueue reader role must grant ResourceFlavor reads for queue preflight: %#v", role.Rules)
+}
+
 func TestKustomizeControllerRoleMatchesTopologyAndGPUFlavorPermissions(t *testing.T) {
 	role := kustomizeClusterRole(t, "tau-core-controller")
 	for _, want := range []rbacv1.PolicyRule{

controllers/tau-core/scripts/kind-e2e.sh

@@ -872,8 +872,10 @@ kubectl auth can-i get configmaps -n "${TARGET_NAMESPACE}" --as=researcher@examp
 kubectl auth can-i get "workspaces.tau.azure.com/${WORKSPACE_NAME}" -n "${SYSTEM_NAMESPACE}" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes
 kubectl auth can-i create quotarequests.tau.azure.com -n "${SYSTEM_NAMESPACE}" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes
 kubectl auth can-i get "clusterqueues.kueue.x-k8s.io/${WORKSPACE_NAME}" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes
+kubectl auth can-i get resourceflavors.kueue.x-k8s.io --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes
 kubectl auth can-i get "clusters.tau.azure.com/cluster" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes
 [[ "$(kubectl auth can-i list clusterqueues.kueue.x-k8s.io --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]]
+[[ "$(kubectl auth can-i list resourceflavors.kueue.x-k8s.io --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]]
 [[ "$(kubectl auth can-i list clusters.tau.azure.com --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]]
 [[ "$(kubectl auth can-i get clusters.tau.azure.com/other --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]]
 [[ "$(kubectl auth can-i list workspaces.tau.azure.com -n "${SYSTEM_NAMESPACE}" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]]