[scheduler] Fix regression with layers containing multiple tags (#2258)

Layers can be setup with multiple tags using both '|' and ' | ' as
separators, but the new scheduler only takes ' | '. This change fixes
that.

## LLM usage disclosure
Claude Opus was used to apply the query changes to accept both
separators

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Bug Fixes**
* Fixed tag matching to properly handle tags with inconsistent spacing
around delimiters, improving query accuracy and flexibility.
* Enhanced tag parsing to normalize whitespace and filter out empty
values.

* **Tests**
* Added comprehensive smoke tests for tag matching with various
delimiter spacing formats.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: DiegoTavares

rust/crates/scheduler/src/dao/job_dao.rs

@@ -97,7 +97,7 @@ filtered_jobs AS (
         AND (fr.int_max_cores = -1 OR fr.int_cores + l.int_cores_min < fr.int_max_cores)
         AND (fr.int_max_gpus = -1 OR fr.int_gpus + l.int_gpus_min < fr.int_max_gpus)
         -- Match tags: jobs with at least one layer that contains the queried tag
-        AND string_to_array($3, ' | ') && string_to_array(l.str_tags, ' | ')
+        AND string_to_array(REPLACE($3, ' ', ''), '|') && string_to_array(REPLACE(l.str_tags, ' ', ''), '|')
         AND LOWER(j.pk_facility) = LOWER($4)
 )
 SELECT DISTINCT

rust/crates/scheduler/src/dao/layer_dao.rs

@@ -139,7 +139,12 @@ impl DispatchLayer {
                 .try_into()
                 .expect("gpus_min should fit on a i32"),
             gpu_mem_min: ByteSize::kb(layer.int_gpu_mem_min as u64),
-            tags: layer.str_tags.split(" | ").map(|t| t.to_string()).collect(),
+            tags: layer
+                .str_tags
+                .split('|')
+                .map(|t| t.trim().to_string())
+                .filter(|t| !t.is_empty())
+                .collect(),
             frames: frames.into_iter().map(|f| f.into()).collect(),
         }
     }
@@ -204,7 +209,7 @@ WITH dispatch_frames AS (
         INNER JOIN layer_stat ls on l.pk_layer = ls.pk_layer
     WHERE j.pk_job = $1
         AND ls.int_waiting_count > 0
-        AND string_to_array($2, ' | ') && string_to_array(l.str_tags, ' | ')
+        AND string_to_array(REPLACE($2, ' ', ''), '|') && string_to_array(REPLACE(l.str_tags, ' ', ''), '|')
         AND f.str_state = 'WAITING'
 ),
 limited_frames AS (
@@ -277,7 +282,7 @@ FROM job j
     LEFT JOIN limited_frames lf ON l.pk_layer = lf.pk_layer
 WHERE j.pk_job = $1
     AND ls.int_waiting_count > 0
-    AND string_to_array($2, ' | ') && string_to_array(l.str_tags, ' | ')
+    AND string_to_array(REPLACE($2, ' ', ''), '|') && string_to_array(REPLACE(l.str_tags, ' ', ''), '|')
 ORDER BY
     l.int_dispatch_order,
     lf.int_dispatch_order,

rust/crates/scheduler/tests/smoke_tests.rs

@@ -1358,6 +1358,127 @@ mod scheduler_smoke_test {
         assert!(total_frames > 0, "Expected at least 1 frame to be dispatched");
     }
 
+    // ============================================================
+    // Tag delimiter tolerance: str_tags may be stored as 'a|b' or 'a | b'
+    // ============================================================
+
+    #[tokio::test]
+    #[traced_test]
+    #[serial]
+    async fn test_query_layers_matches_tags_without_spaces() {
+        // Regression: layers stored with 'tag1|tag2' (no spaces) were dropped
+        // by string_to_array(..., ' | ') and never dispatched.
+        let (pool, show_id, facility_id, dept_id, folder_id, _alloc_id, suffix) =
+            setup_resource_limit_test("tag_nospace").await.expect("setup failed");
+
+        let tag_a = format!("tag_a_{}", suffix);
+        let tag_b = format!("tag_b_{}", suffix);
+        let stored_tags = format!("{}|{}", tag_a, tag_b);
+
+        let job = create_job_scenario(
+            &pool,
+            show_id,
+            facility_id,
+            dept_id,
+            folder_id,
+            &format!("tag_nospace_job_{}", suffix),
+            vec![(
+                &format!("tag_nospace_layer_{}", suffix),
+                &stored_tags,
+                1,
+                1024 * 1024,
+                0,
+                0,
+            )],
+            3,
+        )
+        .await
+        .expect("failed to create job");
+
+        let layer_dao = scheduler::dao::LayerDao::new()
+            .await
+            .expect("failed to create LayerDao");
+
+        let layers = layer_dao
+            .query_layers(job.id, vec![tag_a.clone()])
+            .await
+            .expect("query_layers failed");
+
+        let total_frames = count_dispatch_frames(&layers);
+        assert!(
+            total_frames > 0,
+            "Expected frames for no-space tag format '{}' when querying '{}', got 0",
+            stored_tags,
+            tag_a,
+        );
+    }
+
+    #[tokio::test]
+    #[traced_test]
+    #[serial]
+    async fn test_query_layers_matches_mixed_tag_delimiters() {
+        // Some rows end up with inconsistent delimiters ('a|b | c'); the query
+        // should still match any individual tag.
+        let (pool, show_id, facility_id, dept_id, folder_id, _alloc_id, suffix) =
+            setup_resource_limit_test("tag_mixed").await.expect("setup failed");
+
+        let tag_a = format!("tag_a_{}", suffix);
+        let tag_b = format!("tag_b_{}", suffix);
+        let tag_c = format!("tag_c_{}", suffix);
+        let stored_tags = format!("{}|{} | {}", tag_a, tag_b, tag_c);
+
+        let job = create_job_scenario(
+            &pool,
+            show_id,
+            facility_id,
+            dept_id,
+            folder_id,
+            &format!("tag_mixed_job_{}", suffix),
+            vec![(
+                &format!("tag_mixed_layer_{}", suffix),
+                &stored_tags,
+                1,
+                1024 * 1024,
+                0,
+                0,
+            )],
+            3,
+        )
+        .await
+        .expect("failed to create job");
+
+        let layer_dao = scheduler::dao::LayerDao::new()
+            .await
+            .expect("failed to create LayerDao");
+
+        // Query for the middle tag which is adjacent to both delimiter styles.
+        let layers = layer_dao
+            .query_layers(job.id, vec![tag_b.clone()])
+            .await
+            .expect("query_layers failed");
+
+        let total_frames = count_dispatch_frames(&layers);
+        assert!(
+            total_frames > 0,
+            "Expected frames for mixed-delimiter tags '{}' when querying '{}', got 0",
+            stored_tags,
+            tag_b,
+        );
+
+        // Returned DispatchLayer.tags should be normalized (trimmed, no empties).
+        let returned_tags = &layers[0].tags;
+        assert!(
+            returned_tags.contains(&tag_a)
+                && returned_tags.contains(&tag_b)
+                && returned_tags.contains(&tag_c),
+            "DispatchLayer.tags should be normalized to [{}, {}, {}], got {:?}",
+            tag_a,
+            tag_b,
+            tag_c,
+            returned_tags,
+        );
+    }
+
     // #[tokio::test]
     // #[traced_test]
     // async fn test_full_service_cluster_discovery() {