Fix extra files collection for outputs in `store_by="id"` object stores

lib/galaxy/job_execution/output_collect.py

@@ -718,9 +718,13 @@ 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, dataset, job_working_directory):
+def collect_extra_files(object_store, dataset, job_working_directory, outputs_to_working_directory=False):
     # TODO: should this use compute_environment to determine the extra files path ?
-    file_name = dataset.dataset.extra_files_path_name_from(object_store)
+    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):
@@ -739,7 +743,7 @@ def collect_extra_files(object_store, dataset, job_working_directory):
             for f in files:
                 object_store.update_from_file(
                     dataset.dataset,
-                    extra_dir=os.path.normpath(os.path.join(file_name, os.path.relpath(root, temp_file_path))),
+                    extra_dir=os.path.normpath(os.path.join(real_file_name, os.path.relpath(root, temp_file_path))),
                     alt_name=f,
                     file_name=os.path.join(root, f),
                     create=True,

lib/galaxy/jobs/__init__.py

@@ -1652,7 +1652,7 @@ class MinimalJobWrapper(HasResourceParameters):
             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:

test/integration/test_legacy_store_by.py

@@ -0,0 +1,21 @@
+from galaxy_test.driver import integration_util
+
+
+class StoreByIdTestCase(integration_util.IntegrationInstance):
+    """Describe a Galaxy test instance with embedded pulsar configured."""
+
+    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",
+    ]
+)