Fix motif collapse dropping activities on duplicate matches

src/flowx/motifs/collapser.py

@@ -38,26 +38,43 @@ def collapse_motifs(
     # _rewire_dependencies can match Dependency.task_key (also sanitised); raw names would miss edges.
     motif_task_keys: dict[str, str] = {}
 
-    inserted_motifs: set[str] = set()
+    # Track how many times each motif_id has been inserted so multiple distinct matches of the same
+    # motif (e.g. two File Existence Validation groups in one pipeline) get unique task_keys instead of
+    # colliding. Keyed by the match's *contents* (the frozenset of its matched activity names, which is
+    # unique per match because activities are claimed exactly once) so we insert each group exactly once
+    # while still recording *every* member's task_key -> motif mapping for dependency rewiring. Content
+    # keying is deliberate: an object-identity key (id(detected)) would silently break if
+    # _find_motif_for_activity is ever changed to return a fresh object per call.
+    inserted_matches: dict[frozenset[str], str] = {}
+    motif_id_counts: dict[str, int] = {}
     for task in pipeline.tasks:
         if task.name in claimed_names:
             detected = _find_motif_for_activity(task.name, motifs)
             if detected is None:
                 new_tasks.append(task)
                 continue
 
-            motif_id = detected.definition.motif_id
-            if motif_id in inserted_motifs:
-                continue
-            inserted_motifs.add(motif_id)
-
-            motif_activity = _build_motif_activity(detected, tasks_by_name)
-            new_tasks.append(motif_activity)
-
-            for matched_name in detected.matched_activities:
-                matched_task = tasks_by_name.get(matched_name)
-                if matched_task is not None:
-                    motif_task_keys[matched_task.task_key] = motif_activity.task_key
+            match_id = frozenset(detected.matched_activities)
+            existing_key = inserted_matches.get(match_id)
+            if existing_key is None:
+                # First activity of this match: mint a unique task_key and insert the motif node once.
+                # The occurrence suffix (`""`, `_2`, `_3`, ...) is assigned in `pipeline.tasks` iteration
+                # order, so a motif's task_key is stable only as long as that order is deterministic
+                # (it is: `pipeline.tasks` is an ordered list built once by the translator). If detection
+                # or task ordering is ever parallelised/reordered, these keys would shift between runs.
+                motif_id = detected.definition.motif_id
+                occurrence = motif_id_counts.get(motif_id, 0)
+                motif_id_counts[motif_id] = occurrence + 1
+                suffix = "" if occurrence == 0 else f"_{occurrence + 1}"
+                motif_activity = _build_motif_activity(detected, tasks_by_name, task_key_suffix=suffix)
+                new_tasks.append(motif_activity)
+                inserted_matches[match_id] = motif_activity.task_key
+                existing_key = motif_activity.task_key
+
+                for matched_name in detected.matched_activities:
+                    matched_task = tasks_by_name.get(matched_name)
+                    if matched_task is not None:
+                        motif_task_keys[matched_task.task_key] = existing_key
         else:
             new_tasks.append(task)
 
@@ -94,11 +111,16 @@ def _find_motif_for_activity(
 def _build_motif_activity(
     motif: DetectedMotif,
     tasks_by_name: dict[str, Activity],
+    task_key_suffix: str = "",
 ) -> MotifActivity:
-    """Builds a MotifActivity from a detected motif and the original tasks."""
+    """Builds a MotifActivity from a detected motif and the original tasks.
+
+    ``task_key_suffix`` disambiguates multiple matches of the same motif in one
+    pipeline so their task_keys don't collide (which would drop a match).
+    """
     definition = motif.definition
 
-    task_key = f"motif_{definition.motif_id}"
+    task_key = f"motif_{definition.motif_id}{task_key_suffix}"
     display_name = definition.display_name
 
     original_activities = [tasks_by_name[name] for name in motif.matched_activities if name in tasks_by_name]

tests/unit/test_motifs.py

@@ -250,3 +250,65 @@ def test_collapse_handles_activity_name_distinct_from_task_key(self):
         assert downstream_keys == {motif.task_key}, (
             f"Downstream task should be rewired to point at the motif, got {downstream_keys}"
         )
+
+    def test_collapse_two_matches_of_same_motif_are_both_kept(self):
+        """Regression (#22): two matches of the same motif in one pipeline must both survive.
+
+        The collapser used to dedupe by ``motif_id`` and ``continue``-skip any
+        further match of an already-inserted motif, which dropped the second
+        group's activities entirely and left dangling ``depends_on`` edges
+        (invalid dependency graph at deploy). Each distinct match must now
+        collapse into its own node with a unique ``task_key``
+        (``motif_<id>`` and ``motif_<id>_2``) and no activity may be dropped.
+        """
+        from flowx.models.motifs import DetectedMotif
+
+        pipeline = Pipeline(
+            name="test",
+            tasks=[
+                _ir_activity("GetTableList1"),
+                _ir_activity("ForEachTable1", depends_on=["GetTableList1"]),
+                _ir_activity("GetTableList2"),
+                _ir_activity("ForEachTable2", depends_on=["GetTableList2"]),
+                _ir_activity("PostProcessing", depends_on=["ForEachTable2"]),
+            ],
+        )
+        detected1 = DetectedMotif(
+            definition=MOTIF_METADATA_DRIVEN_BULK_COPY,
+            matched_activities=["GetTableList1", "ForEachTable1"],
+            source_type_hint="database",
+        )
+        detected2 = DetectedMotif(
+            definition=MOTIF_METADATA_DRIVEN_BULK_COPY,
+            matched_activities=["GetTableList2", "ForEachTable2"],
+            source_type_hint="database",
+        )
+        result = collapse_motifs(pipeline, [detected1, detected2])
+
+        # Both matches collapse into their own node -- neither is dropped.
+        motif_tasks = [t for t in result.tasks if isinstance(t, MotifActivity)]
+        assert len(motif_tasks) == 2, "Both matches of the same motif must produce a node"
+
+        # Unique, deterministic task_keys assigned in pipeline.tasks order.
+        motif_keys = [t.task_key for t in motif_tasks]
+        assert motif_keys == ["motif_metadata_driven_bulk_copy", "motif_metadata_driven_bulk_copy_2"], (
+            f"Duplicate matches must get unique task_keys, got {motif_keys}"
+        )
+
+        # No claimed activity leaked through as a standalone task, and nothing was dropped:
+        # 2 collapsed motifs + the single unclaimed PostProcessing task.
+        remaining_names = [t.name for t in result.tasks]
+        assert "PostProcessing" in remaining_names
+        for claimed in ("GetTableList1", "ForEachTable1", "GetTableList2", "ForEachTable2"):
+            assert claimed not in remaining_names, f"{claimed} should be collapsed, not left standalone"
+        collapsed = set()
+        for m in motif_tasks:
+            collapsed.update(m.matched_activity_names)
+        assert collapsed == {"GetTableList1", "ForEachTable1", "GetTableList2", "ForEachTable2"}
+
+        # The downstream dep on the second group is rewired to that group's motif node (no dangling edge).
+        post = next(t for t in result.tasks if t.name == "PostProcessing")
+        post_keys = {dep.task_key for dep in post.depends_on or []}
+        assert post_keys == {"motif_metadata_driven_bulk_copy_2"}, (
+            f"Downstream task must be rewired to the second motif, got {post_keys}"
+        )