fix(scheduler): don't leak pipeline-only fit errors onto live jobs

.changes/unreleased/Fixed-20260722-035021.yaml

@@ -0,0 +1,3 @@
+kind: Fixed
+body: |-
+  Pipeline-only allocation no longer leaks simulation fit errors onto pending jobs, preserving authoritative scheduling diagnostics. [#1948](https://github.com/kai-scheduler/KAI-Scheduler/issues/1948)

pkg/scheduler/actions/common/allocate.go

@@ -140,7 +140,10 @@ func allocateTasksOnNodeSet(ssn *framework.Session, stmt *framework.Statement, n
 	for index, task := range tasksToAllocate {
 		success := allocateTask(ssn, stmt, nodes, task, isPipelineOnly)
 		if !success {
-			handleFailedTaskAllocation(job, task, index)
+			// Skip job fit errors during pipeline-only simulations; see allocateTask (#1948).
+			if !isPipelineOnly {
+				handleFailedTaskAllocation(job, task, index)
+			}
 			return false
 		}
 	}
@@ -159,9 +162,15 @@ func allocateTask(ssn *framework.Session, stmt *framework.Statement, nodes []*no
 		log.InfraLogger.V(6).Infof("pre-predicates failed on task %s/%s. Error: %v",
 			task.Namespace, task.Name, err)
 
-		fitErrors := common_info.NewFitErrors()
-		fitErrors.SetError(err.Error())
-		job.AddTaskFitErrors(task, fitErrors)
+		// Pipeline-only allocation runs inside solver simulations that are rolled back.
+		// Statement.Rollback does not restore fit errors, so recording them here would leak
+		// simulation-only diagnostics onto the live job and mask the authoritative Allocate
+		// reason (#1948). Mirror FittingNode's writeFittingDelta=!isPipelineOnly behavior.
+		if !isPipelineOnly {
+			fitErrors := common_info.NewFitErrors()
+			fitErrors.SetError(err.Error())
+			job.AddTaskFitErrors(task, fitErrors)
+		}
 		return false
 	}
 

pkg/scheduler/actions/integration_tests/reclaim/reclaim_fit_errors_test.go

@@ -0,0 +1,160 @@
+// Copyright 2026 NVIDIA CORPORATION
+// SPDX-License-Identifier: Apache-2.0
+
+package reclaim
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+	"gopkg.in/h2non/gock.v1"
+
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/common_info"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/node_info"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/pod_info"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/pod_status"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/podgroup_info"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/constants"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/test_utils"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/test_utils/jobs_fake"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/test_utils/nodes_fake"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/test_utils/tasks_fake"
+)
+
+func TestSchedulingCyclePreservesAllocateFitErrors(t *testing.T) {
+	defer gock.Off()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	ssn := test_utils.BuildSession(buildFailedGangReclaimTopology(), ctrl)
+	ssn.AddNodeOrderFn(func(_ *pod_info.PodInfo, node *node_info.NodeInfo) (float64, error) {
+		if node.Name == "node0" {
+			return 1000, nil
+		}
+		return 0, nil
+	})
+	failedPrePredicateCalls := 0
+	ssn.AddPrePredicateFn(func(
+		task *pod_info.PodInfo,
+		_ *podgroup_info.PodGroupInfo,
+	) error {
+		if task.Job != common_info.PodGroupID("pending-gang") || task.Name != "pending-gang-0" {
+			return nil
+		}
+		releasingTasks := 0
+		for _, job := range ssn.ClusterInfo.PodGroupInfos {
+			releasingTasks += len(job.PodStatusIndex[pod_status.Releasing])
+		}
+		if releasingTasks < 2 {
+			return nil
+		}
+		failedPrePredicateCalls++
+		return errors.New("simulated pre-predicate failure")
+	})
+	onJobSolutionStartCalls := 0
+	ssn.AddOnJobSolutionStartFn(func() {
+		onJobSolutionStartCalls++
+	})
+	actions := manySingleGPUJobsSchedulingCycleActions()
+	actions[0].Execute(ssn)
+
+	job := ssn.ClusterInfo.PodGroupInfos[common_info.PodGroupID("pending-gang")]
+	require.NotNil(t, job)
+	require.NotEmpty(t, job.TasksFitErrors)
+	require.NotEmpty(t, job.JobFitErrors)
+	require.NotContains(t, job.TasksFitErrors, common_info.PodID("pending-gang-0"))
+	require.Zero(t, failedPrePredicateCalls)
+
+	allocateTaskErrors := taskFitErrorMessages(job)
+	allocateJobErrors := jobFitErrorMessages(job)
+	firstTaskFitErrorMutation := ""
+	firstJobFitErrorMutation := ""
+	for _, action := range actions[1:] {
+		action.Execute(ssn)
+		currentJob := ssn.ClusterInfo.PodGroupInfos[common_info.PodGroupID("pending-gang")]
+		if firstTaskFitErrorMutation == "" &&
+			!assert.ObjectsAreEqual(allocateTaskErrors, taskFitErrorMessages(currentJob)) {
+			firstTaskFitErrorMutation = string(action.Name())
+		}
+		if firstJobFitErrorMutation == "" &&
+			!assert.ObjectsAreEqual(allocateJobErrors, jobFitErrorMessages(currentJob)) {
+			firstJobFitErrorMutation = string(action.Name())
+		}
+	}
+
+	job = ssn.ClusterInfo.PodGroupInfos[common_info.PodGroupID("pending-gang")]
+	assert.Equalf(t, allocateTaskErrors, taskFitErrorMessages(job),
+		"task fit errors produced by Allocate were first changed by %s", firstTaskFitErrorMutation)
+	assert.Equalf(t, allocateJobErrors, jobFitErrorMessages(job),
+		"job fit errors produced by Allocate were first changed by %s", firstJobFitErrorMutation)
+	require.Len(t, job.PodStatusIndex[pod_status.Pending], 2)
+	require.NotZero(t, onJobSolutionStartCalls)
+	require.NotZero(t, failedPrePredicateCalls)
+}
+
+func buildFailedGangReclaimTopology() test_utils.TestTopologyBasic {
+	return test_utils.TestTopologyBasic{
+		Name: "failed gang reclaim preserves allocate fit errors",
+		Nodes: map[string]nodes_fake.TestNodeBasic{
+			"node0": {GPUs: 1},
+			"node1": {GPUs: 1},
+		},
+		Jobs: []*jobs_fake.TestJobBasic{
+			{
+				Name:                "running-victim",
+				RequiredGPUsPerTask: 1,
+				Priority:            constants.PriorityTrainNumber,
+				QueueName:           "running-queue",
+				Tasks: []*tasks_fake.TestTaskBasic{{
+					NodeName: "node0",
+					State:    pod_status.Releasing,
+				}},
+			},
+			{
+				Name:                "running-victim-2",
+				RequiredGPUsPerTask: 1,
+				Priority:            constants.PriorityTrainNumber,
+				QueueName:           "running-queue",
+				Tasks: []*tasks_fake.TestTaskBasic{{
+					NodeName: "node1",
+					State:    pod_status.Running,
+				}},
+			},
+			{
+				Name:                "pending-gang",
+				RequiredGPUsPerTask: 1,
+				Priority:            constants.PriorityTrainNumber,
+				QueueName:           "pending-queue",
+				Tasks: []*tasks_fake.TestTaskBasic{
+					{State: pod_status.Pending},
+					{State: pod_status.Pending},
+				},
+			},
+		},
+		Queues: []test_utils.TestQueueBasic{
+			{Name: "running-queue", DeservedGPUs: 0, GPUOverQuotaWeight: 0},
+			{Name: "pending-queue", DeservedGPUs: 2, GPUOverQuotaWeight: 0},
+		},
+		Mocks: &test_utils.TestMock{CacheRequirements: &test_utils.CacheMocking{}},
+	}
+}
+
+func taskFitErrorMessages(job *podgroup_info.PodGroupInfo) map[common_info.PodID]string {
+	messages := make(map[common_info.PodID]string, len(job.TasksFitErrors))
+	for taskID, fitErrors := range job.TasksFitErrors {
+		messages[taskID] = fitErrors.Error()
+	}
+	return messages
+}
+
+func jobFitErrorMessages(job *podgroup_info.PodGroupInfo) []string {
+	messages := make([]string, 0, len(job.JobFitErrors))
+	for _, fitError := range job.JobFitErrors {
+		messages = append(messages, fitError.DetailedMessage())
+	}
+	return messages
+}