fix(dynamo): use Job conditions for failure/completion detection

Check authoritative Job conditions (JobComplete, JobFailed) before
falling back to counter-based detection. This correctly handles
activeDeadlineSeconds and podFailurePolicy failures where the Job
gets a Failed condition but Status.Failed may not exceed backoffLimit.

Also change fallback comparison from > to >= for robustness when
conditions haven't been set yet.

Signed-off-by: Suraj Deshmukh <suraj.deshmukh@microsoft.com>

Author: surajssd

providers/dynamo/download.go

@@ -153,17 +153,33 @@ func EnsureDownloadJob(ctx context.Context, c client.Client, md *kubeairunwayv1a
 		return false, nil // requeue → next reconcile creates fresh Job
 	}
 
-	// Job exists, check status
+	// Job exists — check conditions (authoritative) then counters (fallback).
+	for _, cond := range existing.Status.Conditions {
+		if cond.Status != corev1.ConditionTrue {
+			continue
+		}
+		switch cond.Type {
+		case batchv1.JobComplete:
+			logger.Info("Model download Job completed", "name", jobName)
+			return true, nil
+		case batchv1.JobFailed:
+			return false, fmt.Errorf("model download Job %s failed permanently: %s",
+				jobName, cond.Message)
+		}
+	}
+
+	// Fallback: counter-based detection for older clusters or edge cases
+	// where conditions haven't been set yet.
 	if existing.Status.Succeeded >= 1 {
-		logger.Info("Model download Job completed", "name", jobName)
+		logger.Info("Model download Job completed (counter)", "name", jobName)
 		return true, nil
 	}
 
 	backoffLimit := defaultBackoffLimit
 	if existing.Spec.BackoffLimit != nil {
 		backoffLimit = *existing.Spec.BackoffLimit
 	}
-	if existing.Status.Failed > backoffLimit {
+	if existing.Status.Failed >= backoffLimit {
 		return false, fmt.Errorf("model download Job %s failed permanently (failed=%d, backoffLimit=%d)",
 			jobName, existing.Status.Failed, backoffLimit)
 	}

providers/dynamo/download_test.go

@@ -321,6 +321,13 @@ func TestEnsureDownloadJobCompleted(t *testing.T) {
 		},
 		Status: batchv1.JobStatus{
 			Succeeded: 1,
+			Conditions: []batchv1.JobCondition{
+				{
+					Type:    batchv1.JobComplete,
+					Status:  corev1.ConditionTrue,
+					Message: "Job completed successfully",
+				},
+			},
 		},
 	}
 
@@ -402,6 +409,13 @@ func TestEnsureDownloadJobFailed(t *testing.T) {
 		},
 		Status: batchv1.JobStatus{
 			Failed: 4, // exceeds backoffLimit of 3
+			Conditions: []batchv1.JobCondition{
+				{
+					Type:    batchv1.JobFailed,
+					Status:  corev1.ConditionTrue,
+					Message: "Job has reached the specified backoff limit",
+				},
+			},
 		},
 	}
 
@@ -416,6 +430,98 @@ func TestEnsureDownloadJobFailed(t *testing.T) {
 	}
 }
 
+func TestEnsureDownloadJobFailedByConditionOnly(t *testing.T) {
+	scheme := newScheme()
+	_ = batchv1.AddToScheme(scheme)
+
+	md := newDownloadMD("my-model", "default")
+
+	// Job failed via activeDeadlineSeconds: JobFailed condition is set,
+	// but Failed count is below the backoff limit.
+	backoffLimit := int32(3)
+	existingJob := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "my-model-model-download",
+			Namespace: "default",
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: "kubeairunway.ai/v1alpha1",
+					Kind:       "ModelDeployment",
+					Name:       "my-model",
+					UID:        "test-uid",
+				},
+			},
+		},
+		Spec: batchv1.JobSpec{
+			BackoffLimit: &backoffLimit,
+		},
+		Status: batchv1.JobStatus{
+			Failed: 1, // below backoffLimit
+			Conditions: []batchv1.JobCondition{
+				{
+					Type:    batchv1.JobFailed,
+					Status:  corev1.ConditionTrue,
+					Message: "Job was active longer than specified deadline",
+				},
+			},
+		},
+	}
+
+	c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(existingJob).WithStatusSubresource(existingJob).Build()
+
+	_, err := EnsureDownloadJob(context.Background(), c, md, DefaultDownloadJobImage)
+	if err == nil {
+		t.Fatal("expected error for Job failed by condition (activeDeadlineSeconds)")
+	}
+	if !strings.Contains(err.Error(), "failed permanently") {
+		t.Errorf("expected permanent failure error, got: %v", err)
+	}
+	if !strings.Contains(err.Error(), "active longer than specified deadline") {
+		t.Errorf("expected condition message in error, got: %v", err)
+	}
+}
+
+func TestEnsureDownloadJobFailedAtBackoffLimit(t *testing.T) {
+	scheme := newScheme()
+	_ = batchv1.AddToScheme(scheme)
+
+	md := newDownloadMD("my-model", "default")
+
+	// Job with Failed == BackoffLimit but no condition set yet.
+	// The >= fallback should catch this.
+	backoffLimit := int32(3)
+	existingJob := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "my-model-model-download",
+			Namespace: "default",
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: "kubeairunway.ai/v1alpha1",
+					Kind:       "ModelDeployment",
+					Name:       "my-model",
+					UID:        "test-uid",
+				},
+			},
+		},
+		Spec: batchv1.JobSpec{
+			BackoffLimit: &backoffLimit,
+		},
+		Status: batchv1.JobStatus{
+			Failed: 3, // exactly at backoffLimit, no conditions
+		},
+	}
+
+	c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(existingJob).WithStatusSubresource(existingJob).Build()
+
+	_, err := EnsureDownloadJob(context.Background(), c, md, DefaultDownloadJobImage)
+	if err == nil {
+		t.Fatal("expected error when Failed == BackoffLimit (fallback detection)")
+	}
+	if !strings.Contains(err.Error(), "failed permanently") {
+		t.Errorf("expected permanent failure error, got: %v", err)
+	}
+}
+
 func TestDeleteManagedJobs(t *testing.T) {
 	scheme := newScheme()
 	_ = batchv1.AddToScheme(scheme)