diff --git a/lib/galaxy/job_execution/output_collect.py b/lib/galaxy/job_execution/output_collect.py
index 921828323463..e251c8655e8c 100644
--- a/lib/galaxy/job_execution/output_collect.py
+++ b/lib/galaxy/job_execution/output_collect.py
@@ -727,11 +727,19 @@ def default_exit_code_file(files_dir, id_tag):
     return os.path.join(files_dir, f"galaxy_{id_tag}.ec")
 
 
-def collect_extra_files(object_store: ObjectStore, dataset: "DatasetInstance", job_working_directory: str) -> None:
+def collect_extra_files(
+    object_store: ObjectStore,
+    dataset: "DatasetInstance",
+    job_working_directory: str,
+    outputs_to_working_directory: bool = False,
+):
     # TODO: should this use compute_environment to determine the extra files path ?
     assert dataset.dataset
-    file_name = dataset.dataset.extra_files_path_name_from(object_store)
-    assert file_name
+    real_file_name = file_name = dataset.dataset.extra_files_path_name_from(object_store)
+    if outputs_to_working_directory:
+        # OutputsToWorkingDirectoryPathRewriter always rewrites extra files to uuid path,
+        # so we have to collect from that path even if the real extra files path is dataset_N_files
+        file_name = f"dataset_{dataset.dataset.uuid}_files"
     output_location = "outputs"
     temp_file_path = os.path.join(job_working_directory, output_location, file_name)
     if not os.path.exists(temp_file_path):
@@ -746,7 +754,12 @@ def collect_extra_files(object_store: ObjectStore, dataset: "DatasetInstance", j
         # automatically creates them.  However, empty directories will
         # not be created in the object store at all, which might be a
         # problem.
-        persist_extra_files(object_store=object_store, src_extra_files_path=temp_file_path, primary_data=dataset)
+        persist_extra_files(
+            object_store=object_store,
+            src_extra_files_path=temp_file_path,
+            primary_data=dataset,
+            extra_files_path_name=real_file_name,
+        )
     except Exception as e:
         log.debug("Error in collect_associated_files: %s", unicodify(e))
 
diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index aae0f8486de6..91a0ed792be1 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -1746,7 +1746,7 @@ def _finish_dataset(self, output_name, dataset, job, context, final_job_state, r
             dataset.dataset.uuid = context["uuid"]
         self.__update_output(job, dataset)
         if not purged:
-            collect_extra_files(self.object_store, dataset, self.working_directory)
+            collect_extra_files(self.object_store, dataset, self.working_directory, self.outputs_to_working_directory)
         if job.states.ERROR == final_job_state:
             dataset.blurb = "error"
             if not implicit_collection_jobs:
diff --git a/lib/galaxy/objectstore/__init__.py b/lib/galaxy/objectstore/__init__.py
index 4560573fa9eb..31bc280244b1 100644
--- a/lib/galaxy/objectstore/__init__.py
+++ b/lib/galaxy/objectstore/__init__.py
@@ -1587,10 +1587,16 @@ def set_dataset_object_store_id(self, dataset, require_shareable=True):
         self.object_store_id = dataset.object_store_id  # these will be the same thing after the first output
 
 
-def persist_extra_files(object_store: ObjectStore, src_extra_files_path: str, primary_data: "DatasetInstance") -> None:
+def persist_extra_files(
+    object_store: ObjectStore,
+    src_extra_files_path: str,
+    primary_data: "DatasetInstance",
+    extra_files_path_name: Optional[str] = None,
+) -> None:
     if os.path.exists(src_extra_files_path):
         assert primary_data.dataset
-        extra_files_path_name = primary_data.dataset.extra_files_path_name_from(object_store)
+        if not extra_files_path_name:
+            extra_files_path_name = primary_data.dataset.extra_files_path_name_from(object_store)
         assert extra_files_path_name
         for root, _dirs, files in safe_walk(src_extra_files_path):
             extra_dir = os.path.join(extra_files_path_name, os.path.relpath(root, src_extra_files_path))
diff --git a/test/integration/test_change_datatype_with_store_by_id.py b/test/integration/test_legacy_store_by.py
similarity index 64%
rename from test/integration/test_change_datatype_with_store_by_id.py
rename to test/integration/test_legacy_store_by.py
index a7c2c0afd3af..aede8c311c9e 100644
--- a/test/integration/test_change_datatype_with_store_by_id.py
+++ b/test/integration/test_legacy_store_by.py
@@ -21,3 +21,21 @@ def setUp(self):
         super().setUp()
         self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
         self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
+
+
+class StoreByIdTestCase(integration_util.IntegrationInstance):
+    framework_tool_and_types = True
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["object_store_store_by"] = "id"
+        config["outputs_to_working_directory"] = True
+
+
+instance = integration_util.integration_module_instance(StoreByIdTestCase)
+
+test_tools = integration_util.integration_tool_runner(
+    [
+        "composite_output_tests",
+    ]
+)