diff --git a/internal/push/preflight.go b/internal/push/preflight.go
index a2810eaf..5f7fffbf 100644
--- a/internal/push/preflight.go
+++ b/internal/push/preflight.go
@@ -247,8 +247,11 @@ func ValidateImages(images []string, expectedW, expectedH int) error {
 // direction for image_classification) — the caller may surface them as a
 // note.
 //
-// filenameColumn is the CSV's first column (the ingestor reads filenames
-// positionally from the id column of labels.csv).
+// The image filename is read from the column NAMED "filename", not positionally:
+// the ingestor does record.get("filename") over the header-keyed record
+// (file_transfer.py / record_processor.py), so a `label,filename` header (filename
+// not first) resolves to that column — reading rec[0] would treat the LABEL value
+// as the filename and false-reject a layout the cluster ingests cleanly.
 func CrossCheckLabels(csvPath string, images []string, extension string) (missing []string, orphans []string, err error) {
 	f, err := os.Open(csvPath)
 	if err != nil {
@@ -268,12 +271,14 @@ func CrossCheckLabels(csvPath string, images []string, extension string) (missin
 	}
 	referenced := make(map[string]bool)
 
-	if _, err := r.Read(); err != nil { // header
+	header, err := r.Read()
+	if err != nil {
 		if errors.Is(err, io.EOF) {
 			return nil, nil, nil // emptiness is CheckHasDataRows' diagnostic
 		}
 		return nil, nil, fmt.Errorf("reading %s: %w", filepath.Base(csvPath), err)
 	}
+	fnIdx := imageFileColIndex(header)
 	for {
 		rec, err := r.Read()
 		if errors.Is(err, io.EOF) {
@@ -282,10 +287,10 @@ func CrossCheckLabels(csvPath string, images []string, extension string) (missin
 		if err != nil {
 			return nil, nil, fmt.Errorf("reading %s: %w", filepath.Base(csvPath), err)
 		}
-		if len(rec) == 0 {
-			continue
+		if fnIdx >= len(rec) {
+			continue // short/ragged row — no filename cell to check
 		}
-		name := strings.TrimSpace(rec[0])
+		name := strings.TrimSpace(rec[fnIdx])
 		if name == "" {
 			continue
 		}
@@ -311,6 +316,34 @@ func CrossCheckLabels(csvPath string, images []string, extension string) (missin
 	return missing, orphans, nil
 }
 
+// imageFileColIndex returns the header index of the column the ingestor reads each
+// image's file key from: "filename" if present, else "data_id" — the ingestor's own
+// precedence (image_paths.prepare_classification_pytorch_image_df / image_loader),
+// position-independent for both. A label,data_id CSV (no filename column) is ingested
+// cleanly by the cluster, so matching only "filename" and falling back to index 0
+// would read the label column as filenames and false-reject it (exit 3).
+//
+// Each name is matched exactly first, then case-insensitively with surrounding
+// whitespace stripped (the ingestor's _match_column rule). "filename" wins over
+// "data_id" when both resolve. Falls back to 0 only when NEITHER column exists — a
+// labels.csv the ingestor rejects at validate_data regardless, not this check's job
+// to diagnose.
+func imageFileColIndex(header []string) int {
+	for _, want := range []string{"filename", "data_id"} {
+		for i, h := range header {
+			if strings.TrimSpace(h) == want {
+				return i
+			}
+		}
+		for i, h := range header {
+			if strings.EqualFold(strings.TrimSpace(h), want) {
+				return i
+			}
+		}
+	}
+	return 0
+}
+
 // CheckAnnotationPairing previews the ingestor's FilePairingValidator
 // (file_pairing_validator.py) for object_detection: every image must have
 // an annotation with the same filename stem and vice versa — a mismatch in
diff --git a/internal/push/preflight_test.go b/internal/push/preflight_test.go
index ed546f4e..6c3381a8 100644
--- a/internal/push/preflight_test.go
+++ b/internal/push/preflight_test.go
@@ -163,8 +163,10 @@ func TestCrossCheckLabels(t *testing.T) {
 	// One row exact, one extensionless (the ingestor appends the dataset
 	// extension — the check must mirror that), one missing.
 	csvPath := filepath.Join(dir, "labels.csv")
+	// Realistic image_classification labels.csv: the ingestor reads the image name
+	// from the column NAMED "filename" (record.get("filename")).
 	if err := os.WriteFile(csvPath,
-		[]byte("image_id,label\na.jpg,cat\nb,dog\nghost.jpg,cat\n"), 0o644); err != nil {
+		[]byte("filename,label\na.jpg,cat\nb,dog\nghost.jpg,cat\n"), 0o644); err != nil {
 		t.Fatal(err)
 	}
 	images := []string{filepath.Join(imgs, "a.jpg"), filepath.Join(imgs, "b.jpg"), filepath.Join(imgs, "extra.jpg")}
@@ -180,6 +182,95 @@ func TestCrossCheckLabels(t *testing.T) {
 	}
 }
 
+// The image filename is resolved by the "filename" COLUMN NAME, not positionally —
+// so a `label,filename` header (filename not first, a layout the ingestor accepts
+// via record.get("filename")) must not false-reject. Reading rec[0] would treat the
+// label value ("cat"/"dog") as the filename and flag every row missing (exit 3).
+func TestCrossCheckLabels_FilenameColumnNotFirst(t *testing.T) {
+	dir := t.TempDir()
+	imgs := filepath.Join(dir, "images")
+	if err := os.MkdirAll(imgs, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	for _, n := range []string{"a.jpg", "b.jpg", "extra.jpg"} {
+		if err := os.WriteFile(filepath.Join(imgs, n), []byte("x"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	csvPath := filepath.Join(dir, "labels.csv")
+	// filename is the SECOND column, and mixed-case to exercise the ci match.
+	if err := os.WriteFile(csvPath,
+		[]byte("label,Filename\ncat,a.jpg\ndog,b\ncat,ghost.jpg\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	images := []string{filepath.Join(imgs, "a.jpg"), filepath.Join(imgs, "b.jpg"), filepath.Join(imgs, "extra.jpg")}
+	missing, orphans, err := CrossCheckLabels(csvPath, images, ".jpg")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(missing) != 1 || missing[0] != "ghost.jpg" {
+		t.Errorf("missing = %v, want [ghost.jpg] — the label values must NOT be read as filenames", missing)
+	}
+	if len(orphans) != 1 || orphans[0] != "extra.jpg" {
+		t.Errorf("orphans = %v, want [extra.jpg]", orphans)
+	}
+}
+
+// A `label,data_id` header (no filename column) is ingested cleanly by the cluster —
+// the ingestor resolves the image column as filename ELSE data_id. CrossCheckLabels
+// must resolve data_id, not fall back to index 0 (the label column), which would read
+// "cat"/"dog" as filenames and false-reject every row (exit 3). Regression for Asad's
+// #207 review.
+func TestCrossCheckLabels_DataIDColumn(t *testing.T) {
+	dir := t.TempDir()
+	imgs := filepath.Join(dir, "images")
+	if err := os.MkdirAll(imgs, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	for _, n := range []string{"a.jpg", "b.jpg", "extra.jpg"} {
+		if err := os.WriteFile(filepath.Join(imgs, n), []byte("x"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	csvPath := filepath.Join(dir, "labels.csv")
+	// data_id instead of filename, and not first — a layout the ingestor accepts.
+	if err := os.WriteFile(csvPath,
+		[]byte("label,data_id\ncat,a.jpg\ndog,b\ncat,ghost.jpg\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	images := []string{filepath.Join(imgs, "a.jpg"), filepath.Join(imgs, "b.jpg"), filepath.Join(imgs, "extra.jpg")}
+	missing, orphans, err := CrossCheckLabels(csvPath, images, ".jpg")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(missing) != 1 || missing[0] != "ghost.jpg" {
+		t.Errorf("missing = %v, want [ghost.jpg] — data_id must resolve; label values must NOT be read as filenames", missing)
+	}
+	if len(orphans) != 1 || orphans[0] != "extra.jpg" {
+		t.Errorf("orphans = %v, want [extra.jpg]", orphans)
+	}
+}
+
+func TestImageFileColIndex(t *testing.T) {
+	cases := []struct {
+		header []string
+		want   int
+	}{
+		{[]string{"filename", "label"}, 0},            // filename by name, first
+		{[]string{"label", "filename"}, 1},            // filename by name, not first (the original bug)
+		{[]string{"label", " Filename "}, 1},          // case + whitespace insensitive
+		{[]string{"label", "data_id"}, 1},             // no filename → data_id (the ingestor's fallback)
+		{[]string{"label", "Data_ID"}, 1},             // data_id, case-insensitive
+		{[]string{"data_id", "filename", "label"}, 1}, // filename wins over data_id when both present
+		{[]string{"image_id", "label"}, 0},            // neither → fallback to 0
+	}
+	for _, c := range cases {
+		if got := imageFileColIndex(c.header); got != c.want {
+			t.Errorf("imageFileColIndex(%v) = %d, want %d", c.header, got, c.want)
+		}
+	}
+}
+
 func TestCheckAnnotationPairing(t *testing.T) {
 	imgs := []string{"images/a.jpg", "images/b.jpg"}
 	anns := []string{"annotations/a.xml", "annotations/c.xml"}