fix: normalise worker type, fix matching

tests/unit/mazepa/test_task_router.py

@@ -11,8 +11,8 @@
 def test_constructor(mocker):
     queue_a = mocker.MagicMock()
     queue_b = mocker.MagicMock()
-    queue_a.name = "_type_a"
-    queue_b.name = "_type_b"
+    queue_a.name = "run-xxx_type-a_work"
+    queue_b.name = "run-xxx_type-b_work"
     meq = TaskRouter([queue_a, queue_b])
     assert queue_a.name in meq.name
     assert queue_b.name in meq.name
@@ -21,8 +21,8 @@ def test_constructor(mocker):
 def test_push_tasks(mocker):
     queue_a = mocker.MagicMock()
     queue_b = mocker.MagicMock()
-    queue_a.name = "_type_a"
-    queue_b.name = "_type_b"
+    queue_a.name = "run-xxx_type-a_work"
+    queue_b.name = "run-xxx_type-b_work"
     meq = TaskRouter([queue_a, queue_b])
     task_a = make_test_task(lambda: None, id_="dummy").with_worker_type("type_a")
     task_b = make_test_task(lambda: None, "dummy").with_worker_type("type_b")
@@ -35,9 +35,26 @@ def test_push_tasks(mocker):
 def test_push_tasks_exc(mocker):
     queue_a = mocker.MagicMock()
     queue_b = mocker.MagicMock()
-    queue_a.name = "_type_a"
-    queue_b.name = "_type_b"
+    queue_a.name = "run-xxx_type-a_work"
+    queue_b.name = "run-xxx_type-b_work"
     meq = TaskRouter([queue_a, queue_b])
     task_c = Task(lambda: None, "dummy", worker_type="type_c")
     with pytest.raises(RuntimeError):
         meq.push([task_c])
+
+
+def test_push_tasks_mem_vs_mem_agg(mocker):
+    """Test that mem and mem_agg worker types route to correct queues."""
+    queue_mem = mocker.MagicMock()
+    queue_mem_agg = mocker.MagicMock()
+    queue_mem.name = "run-xxx_mem_work"
+    queue_mem_agg.name = "run-xxx_mem-agg_work"
+    meq = TaskRouter([queue_mem, queue_mem_agg])
+
+    task_mem = make_test_task(lambda: None, id_="task1").with_worker_type("mem")
+    task_mem_agg = make_test_task(lambda: None, id_="task2").with_worker_type("mem_agg")
+
+    meq.push([task_mem, task_mem_agg])
+
+    queue_mem.push.assert_called_with([task_mem])
+    queue_mem_agg.push.assert_called_with([task_mem_agg])

zetta_utils/mazepa/task_router.py

@@ -11,11 +11,14 @@
 
 
 def _is_compatible_task(task: Task, queue_name: str) -> bool:
-    return (
-        task.worker_type is None
-        or queue_name.startswith("local_")
-        or f"_{task.worker_type}" in queue_name
-    )
+    if task.worker_type is None:
+        return True
+    if queue_name.startswith("local_"):
+        return True
+    # Normalize worker_type to match queue naming convention (underscores -> dashes)
+    normalized_type = task.worker_type.replace("_", "-")
+    # Check if worker type appears as a complete segment in the queue name
+    return f"_{normalized_type}_" in queue_name
 
 
 @typechecked