diff --git a/lib/galaxy_test/api/test_workflows.py b/lib/galaxy_test/api/test_workflows.py
index 60f6889cfa63..88d4112966fd 100644
--- a/lib/galaxy_test/api/test_workflows.py
+++ b/lib/galaxy_test/api/test_workflows.py
@@ -2800,44 +2800,6 @@ def run_test(workflow_text):
 
         run_test(NESTED_WORKFLOW_AUTO_LABELS_MODERN_SYNTAX)
 
-    @skip_without_tool("collection_paired_test")
-    def test_workflow_flatten_with_mapped_over_execution(self):
-        with self.dataset_populator.test_history() as history_id:
-            self._run_jobs(
-                r"""
-class: GalaxyWorkflow
-inputs:
-  input_fastqs: collection
-steps:
-  split_up:
-    tool_id: collection_split_on_column
-    in:
-      input1: input_fastqs
-  flatten:
-    tool_id: '__FLATTEN__'
-    in:
-      input: split_up/split_output
-    join_identifier: '-'
-test_data:
-  input_fastqs:
-    collection_type: list
-    elements:
-      - identifier: samp1
-        content: "0\n1"
-""",
-                history_id=history_id,
-            )
-            history = self._get(f"histories/{history_id}/contents").json()
-            flattened_collection = history[-1]
-            assert flattened_collection["history_content_type"] == "dataset_collection"
-            assert flattened_collection["collection_type"] == "list"
-            assert flattened_collection["element_count"] == 2
-            nested_collection = self.dataset_populator.get_history_collection_details(history_id, hid=3)
-            assert nested_collection["collection_type"] == "list:list"
-            assert nested_collection["element_count"] == 1
-            assert nested_collection["elements"][0]["object"]["populated"]
-            assert nested_collection["elements"][0]["object"]["element_count"] == 2
-
     @skip_without_tool("cat")
     def test_workflow_invocation_report_1(self):
         test_data = """
diff --git a/lib/galaxy_test/workflow/flatten_collection_over_execution.gxwf-tests.yml b/lib/galaxy_test/workflow/flatten_collection_over_execution.gxwf-tests.yml
new file mode 100644
index 000000000000..5c54212815e9
--- /dev/null
+++ b/lib/galaxy_test/workflow/flatten_collection_over_execution.gxwf-tests.yml
@@ -0,0 +1,19 @@
+- doc: | 
+    Test to verify collection flatten collection operation mid workflow.
+  job:
+    input_fastqs:
+      collection_type: list
+      elements:
+        - identifier: samp1
+          content: "0 mycoolline\n1 mysecondline\n"
+  outputs:
+    out:
+      elements:
+        'samp1-0':
+          asserts:
+          - that: has_text
+            text: "mycoolline"
+        'samp1-1':
+          asserts:
+          - that: has_text
+            text: "mysecondline"
diff --git a/lib/galaxy_test/workflow/flatten_collection_over_execution.gxwf.yml b/lib/galaxy_test/workflow/flatten_collection_over_execution.gxwf.yml
new file mode 100644
index 000000000000..172eaa7d4b26
--- /dev/null
+++ b/lib/galaxy_test/workflow/flatten_collection_over_execution.gxwf.yml
@@ -0,0 +1,17 @@
+class: GalaxyWorkflow
+inputs:
+  input_fastqs: collection
+outputs:
+  out:
+    outputSource: flatten/output
+steps:
+  split_up:
+    tool_id: collection_split_on_column
+    in:
+      input1: input_fastqs
+  flatten:
+    tool_id: '__FLATTEN__'
+    state:
+      join_identifier: '-'
+    in:
+      input: split_up/split_output