feat: add end-to-end priority queueing

charts/taugrid/tests/baseline_queue_test.yaml

@@ -54,6 +54,14 @@ tests:
           path: metadata.name
           value: jobqueue
         documentIndex: 3
+      - equal:
+          path: spec.queueingStrategy
+          value: BestEffortFIFO
+        documentIndex: 3
+      - equal:
+          path: spec.preemption.withinClusterQueue
+          value: LowerPriority
+        documentIndex: 3
       - equal:
           path: spec.namespaceSelector.matchExpressions[0].key
           value: tau.azure.com/workspace
@@ -89,6 +97,54 @@ tests:
             - name: nvidia.com/gpu
               nominalQuota: "1000"
         documentIndex: 3
+      - equal:
+          path: kind
+          value: PriorityClass
+        documentIndex: 4
+      - equal:
+          path: metadata.name
+          value: taugrid-default
+        documentIndex: 4
+      - equal:
+          path: value
+          value: 1000
+        documentIndex: 4
+      - equal:
+          path: preemptionPolicy
+          value: PreemptLowerPriority
+        documentIndex: 4
+      - equal:
+          path: metadata.name
+          value: taugrid-priority
+        documentIndex: 5
+      - equal:
+          path: value
+          value: 1200
+        documentIndex: 5
+      - equal:
+          path: preemptionPolicy
+          value: PreemptLowerPriority
+        documentIndex: 5
+      - equal:
+          path: kind
+          value: WorkloadPriorityClass
+        documentIndex: 6
+      - equal:
+          path: metadata.name
+          value: taugrid-default
+        documentIndex: 6
+      - equal:
+          path: value
+          value: 1000
+        documentIndex: 6
+      - equal:
+          path: metadata.name
+          value: taugrid-priority
+        documentIndex: 7
+      - equal:
+          path: value
+          value: 1200
+        documentIndex: 7
 
   - it: disables every baseline queue object together
     set:

cli/internal/cli/run_workload_profile.go

@@ -94,6 +94,14 @@ func applySelectedWorkloadProfile(
 	selection profile.Selection,
 	renderProfile profile.Profile,
 ) (unresolvedRunOptions, error) {
+	if strings.TrimSpace(o.priorityTier) != "" &&
+		(strings.TrimSpace(o.workloadPriorityClass) != "" ||
+			strings.TrimSpace(o.podPriorityClass) != "" ||
+			o.disableDefaultPriorities) {
+		return o, fmt.Errorf(
+			"priority tier cannot be combined with explicit priority classes or disabled default priorities",
+		)
+	}
 	conflicts := []struct {
 		field    string
 		key      string
@@ -132,12 +140,6 @@ func applySelectedWorkloadProfile(
 			renderProfile.Topology.DisableDefaultPriorities,
 		)
 	}
-	if policyFieldWasExplicit(o, "priority_tier", strings.TrimSpace(o.priorityTier) != "") {
-		return o, fmt.Errorf(
-			"policy.priority_tier cannot be combined with authoritative priority classes from workload profile %q; remove policy.priority_tier",
-			selection.Profile.Name,
-		)
-	}
 	for _, unsupported := range []struct {
 		field string
 		key   string
@@ -203,10 +205,15 @@ func applySelectedWorkloadProfile(
 	o.queue = renderProfile.Queue
 	o.mode = renderProfile.Topology.Mode
 	o.topology = renderProfile.Topology.Placement
-	o.priorityTier = ""
-	o.workloadPriorityClass = renderProfile.Topology.WorkloadPriorityClassName
-	o.podPriorityClass = renderProfile.Topology.PodPriorityClassName
-	o.disableDefaultPriorities = renderProfile.Topology.DisableDefaultPriorities
+	if strings.TrimSpace(o.priorityTier) != "" {
+		o.workloadPriorityClass = ""
+		o.podPriorityClass = ""
+		o.disableDefaultPriorities = false
+	} else {
+		o.workloadPriorityClass = renderProfile.Topology.WorkloadPriorityClassName
+		o.podPriorityClass = renderProfile.Topology.PodPriorityClassName
+		o.disableDefaultPriorities = renderProfile.Topology.DisableDefaultPriorities
+	}
 	clusterQueue := ""
 	if o.selectedWorkloadProfile != nil {
 		clusterQueue = o.selectedWorkloadProfile.ClusterQueue

cli/internal/cli/run_workload_profile_test.go

@@ -241,6 +241,88 @@ func TestSelectedProfileRejectsAuthoritativeConflicts(t *testing.T) {
 	}
 }
 
+func TestSelectedProfileAllowsBuiltInPriorityTier(t *testing.T) {
+	provider := testRunProfileProvider(t, profile.ExecutionTargetSingleCluster, 2, 3)
+	options := defaultRunDispatchOptions()
+	options.engine = runconfig.EngineRayJob
+	options.profileName = "research-profile"
+	options.namespace = "alpha"
+	options.team = "research"
+	options.lane = "training"
+	options.priorityTier = "priority"
+	options.explicitPolicyFields = map[string]bool{"priority_tier": true}
+
+	selected, err := selectRunWorkloadProfile(context.Background(), options, provider)
+	if err != nil {
+		t.Fatalf("selectRunWorkloadProfile: %v", err)
+	}
+	if selected.priorityTier != "priority" {
+		t.Fatalf("priority tier = %q, want priority", selected.priorityTier)
+	}
+	if selected.workloadPriorityClass != "" || selected.podPriorityClass != "" {
+		t.Fatalf(
+			"explicit tier should defer class selection to topology: workload=%q pod=%q",
+			selected.workloadPriorityClass,
+			selected.podPriorityClass,
+		)
+	}
+}
+
+func TestSelectedProfilePriorityTierOverridesDisabledProfilePriorities(t *testing.T) {
+	resolved := testRunResolvedProfile(profile.ExecutionTargetSingleCluster, 2, 3, 11)
+	resolved.Priorities = profile.ProfilePriorities{DisableDefaultPriorities: true}
+	resolved.WorkloadPriorityClasses = nil
+	resolved.PodPriorityClasses = nil
+	provider := testRunProviderForResolved(t, 11, resolved)
+
+	options := defaultRunDispatchOptions()
+	options.engine = runconfig.EngineRayJob
+	options.profileName = "research-profile"
+	options.namespace = "alpha"
+	options.team = "research"
+	options.lane = "training"
+	options.priorityTier = "priority"
+	options.explicitPolicyFields = map[string]bool{"priority_tier": true}
+
+	selected, err := selectRunWorkloadProfile(context.Background(), options, provider)
+	if err != nil {
+		t.Fatalf("selectRunWorkloadProfile: %v", err)
+	}
+	if selected.disableDefaultPriorities {
+		t.Fatal("explicit tier should enable TauGrid-managed priority classes")
+	}
+}
+
+func TestSelectedProfileRejectsAmbiguousPriorityTier(t *testing.T) {
+	provider := testRunProfileProvider(t, profile.ExecutionTargetSingleCluster, 2, 3)
+	for _, test := range []struct {
+		name   string
+		mutate func(*unresolvedRunOptions)
+	}{
+		{"exact class", func(o *unresolvedRunOptions) { o.podPriorityClass = "tau-default" }},
+		{"disabled priorities", func(o *unresolvedRunOptions) {
+			o.disableDefaultPriorities, o.disablePrioritiesExplicit = true, true
+		}},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			options := defaultRunDispatchOptions()
+			options.engine = runconfig.EngineRayJob
+			options.profileName = "research-profile"
+			options.namespace = "alpha"
+			options.team = "research"
+			options.lane = "training"
+			options.priorityTier = "priority"
+			options.explicitPolicyFields = map[string]bool{"priority_tier": true}
+			test.mutate(&options)
+
+			_, err := selectRunWorkloadProfile(context.Background(), options, provider)
+			if err == nil || !strings.Contains(err.Error(), "cannot be combined") {
+				t.Fatalf("error = %v, want ambiguous priority rejection", err)
+			}
+		})
+	}
+}
+
 func TestSelectedProfilePreservesExplicitRuntimeAndStorageSettings(t *testing.T) {
 	options := defaultRunDispatchOptions()
 	options.engine = runconfig.EngineRayJob

cli/internal/cli/run_workspace.go

@@ -158,7 +158,7 @@ func workspaceHasDurableOutputMount(o unresolvedRunOptions) bool {
 func workspacePriorityTier(priority string) string {
 	switch priority {
 	case "normal":
-		return "default"
+		return ""
 	default:
 		return priority
 	}

cli/internal/cli/run_workspace_test.go

@@ -40,7 +40,7 @@ func TestApplyWorkspaceDefaultsFillsPolicyFields(t *testing.T) {
 		t.Fatalf("applyWorkspaceDefaults: %v", err)
 	}
 
-	if got.namespace != "sample" || got.queue != "sample" || got.priorityTier != "default" {
+	if got.namespace != "sample" || got.queue != "sample" || got.priorityTier != "" {
 		t.Fatalf("policy defaults = namespace %q queue %q priority %q", got.namespace, got.queue, got.priorityTier)
 	}
 	if got.serviceAccountName != "tau-workload" {

cli/internal/jobrender/render_test.go

@@ -1442,6 +1442,28 @@ func TestRender_ElasticUsesLowPriorityAndSharedQueue(t *testing.T) {
 	}
 }
 
+func TestRender_PriorityTierUsesTauGridManagedClasses(t *testing.T) {
+	out, err := Render(trainProfile(), Options{
+		Name:         "priority-job",
+		Namespace:    "tau",
+		Command:      []string{"true"},
+		QueueName:    "jobqueue",
+		PriorityTier: "priority",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	manifest := parseYAML(t, out)
+	labels := manifest["metadata"].(map[string]any)["labels"].(map[string]any)
+	if labels["kueue.x-k8s.io/priority-class"] != "taugrid-priority" {
+		t.Fatalf("workload priority = %v", labels["kueue.x-k8s.io/priority-class"])
+	}
+	pod := manifest["spec"].(map[string]any)["template"].(map[string]any)["spec"].(map[string]any)
+	if pod["priorityClassName"] != "taugrid-priority" {
+		t.Fatalf("pod priority = %v", pod["priorityClassName"])
+	}
+}
+
 func TestRender_NoImage_NoOverride_Errors(t *testing.T) {
 	p := trainProfile()
 	p.Runtime = profile.Runtime{}

cli/internal/rayjobrender/render_test.go

@@ -91,6 +91,38 @@ func TestRenderRestrictedSecurityCoversRayPods(t *testing.T) {
 	}
 }
 
+func TestRenderPriorityTierAppliesToRayJobAndEveryPod(t *testing.T) {
+	out, err := Render(Options{
+		Name:          "priority-ray",
+		Namespace:     "tau",
+		ScriptName:    "train.py",
+		Script:        []byte("print('ok')\n"),
+		Workers:       1,
+		GPUsPerWorker: 1,
+		TopologyOptions: topology.Options{
+			QueueName:    "jobqueue",
+			PriorityTier: "priority",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rayJob := decodeDocs(t, out)[0]
+	labels := rayJob["metadata"].(map[string]any)["labels"].(map[string]any)
+	if labels["kueue.x-k8s.io/priority-class"] != "taugrid-priority" {
+		t.Fatalf("workload priority = %v", labels["kueue.x-k8s.io/priority-class"])
+	}
+	cluster := rayJob["spec"].(map[string]any)["rayClusterSpec"].(map[string]any)
+	head := cluster["headGroupSpec"].(map[string]any)["template"].(map[string]any)["spec"].(map[string]any)
+	workers := cluster["workerGroupSpecs"].([]any)
+	worker := workers[0].(map[string]any)["template"].(map[string]any)["spec"].(map[string]any)
+	for name, pod := range map[string]map[string]any{"head": head, "worker": worker} {
+		if pod["priorityClassName"] != "taugrid-priority" {
+			t.Fatalf("%s pod priority = %v", name, pod["priorityClassName"])
+		}
+	}
+}
+
 func TestRenderRDMAAppliesToWorkersOnly(t *testing.T) {
 	out, err := Render(Options{
 		Name:          "rdma-ray",