diff --git a/lib/galaxy/job_execution/output_collect.py b/lib/galaxy/job_execution/output_collect.py
index aae1336f6484..8e37864767d0 100644
--- a/lib/galaxy/job_execution/output_collect.py
+++ b/lib/galaxy/job_execution/output_collect.py
@@ -521,7 +521,8 @@ def collect_primary_datasets(job_context: Union[JobContext, SessionlessJobContex
                 outdata.designation = designation
                 outdata.dataset.external_filename = None  # resets filename_override
                 # Move data from temp location to dataset location
-                job_context.object_store.update_from_file(outdata.dataset, file_name=filename, create=True)
+                if not outdata.dataset.purged:
+                    job_context.object_store.update_from_file(outdata.dataset, file_name=filename, create=True)
                 primary_output_assigned = True
                 continue
             if name not in primary_datasets:
@@ -554,6 +555,7 @@ def collect_primary_datasets(job_context: Union[JobContext, SessionlessJobContex
                 dataset_attributes=new_primary_datasets_attributes,
                 creating_job_id=job_context.get_job_id() if job_context else None,
                 storage_callbacks=storage_callbacks,
+                purged=outdata.dataset.purged,
             )
             # Associate new dataset with job
             job_context.add_output_dataset_association(f"__new_primary_file_{name}|{designation}__", primary_data)
@@ -563,7 +565,13 @@ def collect_primary_datasets(job_context: Union[JobContext, SessionlessJobContex
         if primary_output_assigned:
             outdata.name = new_outdata_name
             outdata.init_meta()
-            outdata.set_meta()
+            if not outdata.dataset.purged:
+                try:
+                    outdata.set_meta()
+                except Exception:
+                    # We don't want to fail here on a single "bad" discovered dataset
+                    log.debug("set meta failed for %s", outdata, exc_info=True)
+                    outdata.state = HistoryDatasetAssociation.states.FAILED_METADATA
             outdata.set_peek()
             outdata.discovered = True
             sa_session = job_context.sa_session
diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index fe3d4995fb4e..e6fd99f2340e 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -2000,10 +2000,14 @@ def fail(message=job.info, exception=None):
         quota_source_info = None
         # Once datasets are collected, set the total dataset size (includes extra files)
         for dataset_assoc in job.output_datasets:
-            if not dataset_assoc.dataset.dataset.purged:
+            dataset = dataset_assoc.dataset.dataset
+            if not dataset.purged:
                 # assume all datasets in a job get written to the same objectstore
-                quota_source_info = dataset_assoc.dataset.dataset.quota_source_info
-                collected_bytes += dataset_assoc.dataset.set_total_size()
+                quota_source_info = dataset.quota_source_info
+                collected_bytes += dataset.set_total_size()
+            else:
+                # Purge, in case job wrote directly to object store
+                dataset.full_delete()
 
         user = job.user
         if user and collected_bytes > 0 and quota_source_info is not None and quota_source_info.use:
diff --git a/lib/galaxy/jobs/command_factory.py b/lib/galaxy/jobs/command_factory.py
index 1c2a67da86ad..bda5c81ce845 100644
--- a/lib/galaxy/jobs/command_factory.py
+++ b/lib/galaxy/jobs/command_factory.py
@@ -293,7 +293,11 @@ def __copy_if_exists_command(work_dir_output):
     source_file, destination = work_dir_output
     if "?" in source_file or "*" in source_file:
         source_file = source_file.replace("*", '"*"').replace("?", '"?"')
-    return f'\nif [ -f "{source_file}" ] ; then cp "{source_file}" "{destination}" ; fi'
+    # Check if source and destination exist.
+    # Users can purge outputs before the job completes,
+    # in that case we don't want to copy the output to a purged path.
+    # Static, non work_dir_output files are handled in job_finish code.
+    return f'\nif [ -f "{source_file}" -a -f "{destination}" ] ; then cp "{source_file}" "{destination}" ; fi'
 
 
 class CommandsBuilder:
diff --git a/lib/galaxy/jobs/runners/__init__.py b/lib/galaxy/jobs/runners/__init__.py
index 58485fc955ec..db1e1975f5e3 100644
--- a/lib/galaxy/jobs/runners/__init__.py
+++ b/lib/galaxy/jobs/runners/__init__.py
@@ -380,6 +380,13 @@ def get_work_dir_outputs(
         job_tool = job_wrapper.tool
         for joda, dataset in self._walk_dataset_outputs(job):
             if joda and job_tool:
+                if dataset.dataset.purged:
+                    log.info(
+                        "Output dataset %s for job %s purged before job completed, skipping output collection.",
+                        joda.name,
+                        job.id,
+                    )
+                    continue
                 hda_tool_output = job_tool.find_output_def(joda.name)
                 if hda_tool_output and hda_tool_output.from_work_dir:
                     # Copy from working dir to HDA.
@@ -618,10 +625,23 @@ def _finish_or_resubmit_job(self, job_state: "JobState", job_stdout, job_stderr,
 
             tool_stdout_path = os.path.join(outputs_directory, "tool_stdout")
             tool_stderr_path = os.path.join(outputs_directory, "tool_stderr")
-            with open(tool_stdout_path, "rb") as stdout_file:
-                tool_stdout = self._job_io_for_db(stdout_file)
-            with open(tool_stderr_path, "rb") as stderr_file:
-                tool_stderr = self._job_io_for_db(stderr_file)
+            try:
+                with open(tool_stdout_path, "rb") as stdout_file:
+                    tool_stdout = self._job_io_for_db(stdout_file)
+                with open(tool_stderr_path, "rb") as stderr_file:
+                    tool_stderr = self._job_io_for_db(stderr_file)
+            except FileNotFoundError:
+                if job.state in (model.Job.states.DELETING, model.Job.states.DELETED):
+                    # We killed the job, so we may not even have the tool stdout / tool stderr
+                    tool_stdout = ""
+                    tool_stderr = "Job cancelled"
+                else:
+                    # Should we instead just move on ?
+                    # In the end the only consequence here is that we won't be able to determine
+                    # if the job failed for known tool reasons (check_tool_output).
+                    # OTOH I don't know if this can even be reached
+                    # Deal with it if we ever get reports about this.
+                    raise
 
             check_output_detected_state = job_wrapper.check_tool_output(
                 tool_stdout,
diff --git a/lib/galaxy/metadata/__init__.py b/lib/galaxy/metadata/__init__.py
index 975d7ccf79f4..1f44e272ab8a 100644
--- a/lib/galaxy/metadata/__init__.py
+++ b/lib/galaxy/metadata/__init__.py
@@ -118,7 +118,7 @@ def _metadata_results_from_file(self, dataset, filename_results_code):
             rstring = f"Metadata results could not be read from '{filename_results_code}'"
 
         if not rval:
-            log.debug(f"setting metadata externally failed for {dataset.__class__.__name__} {dataset.id}: {rstring}")
+            log.warning(f"setting metadata externally failed for {dataset.__class__.__name__} {dataset.id}: {rstring}")
         return rval
 
 
diff --git a/lib/galaxy/metadata/set_metadata.py b/lib/galaxy/metadata/set_metadata.py
index 8f1f9ea86be9..38f3f7917729 100644
--- a/lib/galaxy/metadata/set_metadata.py
+++ b/lib/galaxy/metadata/set_metadata.py
@@ -96,7 +96,7 @@ def push_if_necessary(object_store: ObjectStore, dataset: DatasetInstance, exter
     # or a remote object store from its cache path.
     # empty files could happen when outputs are discovered from working dir,
     # empty file check needed for e.g. test/integration/test_extended_metadata_outputs_to_working_directory.py::test_tools[multi_output_assign_primary]
-    if os.path.getsize(external_filename):
+    if not dataset.dataset.purged and os.path.getsize(external_filename):
         object_store.update_from_file(dataset.dataset, file_name=external_filename, create=True)
 
 
@@ -426,6 +426,10 @@ def set_meta(new_dataset_instance, file_dict):
                         # as opposed to perhaps a storage issue.
                         with open(external_filename, "wb"):
                             pass
+                    elif not os.path.exists(dataset_filename_override):
+                        # purged output ?
+                        dataset.purged = True
+                        dataset.dataset.purged = True
                     else:
                         raise Exception(f"Output file '{external_filename}' not found")
 
@@ -477,7 +481,7 @@ def set_meta(new_dataset_instance, file_dict):
                     object_store_update_actions.append(partial(reset_external_filename, dataset))
                 object_store_update_actions.append(partial(dataset.set_total_size))
                 object_store_update_actions.append(partial(export_store.add_dataset, dataset))
-                if dataset_instance_id not in unnamed_id_to_path:
+                if dataset_instance_id not in unnamed_id_to_path and not dataset.dataset.purged:
                     object_store_update_actions.append(partial(collect_extra_files, object_store, dataset, "."))
                     dataset_state = "deferred" if (is_deferred and final_job_state == "ok") else final_job_state
                     if not dataset.state == dataset.states.ERROR:
@@ -485,7 +489,8 @@ def set_meta(new_dataset_instance, file_dict):
                         dataset.state = dataset.dataset.state = dataset_state
                     # We're going to run through set_metadata in collect_dynamic_outputs with more contextual metadata,
                     # so only run set_meta for fixed outputs
-                    set_meta(dataset, file_dict)
+                    if not dataset.dataset.purged:
+                        set_meta(dataset, file_dict)
                 # TODO: merge expression_context into tool_provided_metadata so we don't have to special case this (here and in _finish_dataset)
                 meta = tool_provided_metadata.get_dataset_meta(output_name, dataset.dataset.id, dataset.dataset.uuid)
                 if meta:
@@ -512,7 +517,7 @@ def set_meta(new_dataset_instance, file_dict):
                         context_value = context[context_key]
                         setattr(dataset, context_key, context_value)
             else:
-                if dataset_instance_id not in unnamed_id_to_path:
+                if dataset_instance_id not in unnamed_id_to_path and not dataset.dataset.purged:
                     # We're going to run through set_metadata in collect_dynamic_outputs with more contextual metadata,
                     # so only run set_meta for fixed outputs
                     set_meta(dataset, file_dict)
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 97ff0f6b0fbf..84a28aaa396a 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -4211,6 +4211,8 @@ def full_delete(self):
         # TODO: purge metadata files
         self.deleted = True
         self.purged = True
+        self.file_size = 0
+        self.total_size = 0
 
     def get_access_roles(self, security_agent):
         roles = []
@@ -9485,13 +9487,14 @@ def dataset(self) -> Optional[Dataset]:
     def update_from_file(self, file_name):
         if not self.dataset:
             raise Exception("Attempted to write MetadataFile, but no DatasetAssociation set")
-        self.dataset.object_store.update_from_file(
-            self,
-            file_name=file_name,
-            extra_dir="_metadata_files",
-            extra_dir_at_root=True,
-            alt_name=os.path.basename(self.get_file_name()),
-        )
+        if not self.dataset.purged:
+            self.dataset.object_store.update_from_file(
+                self,
+                file_name=file_name,
+                extra_dir="_metadata_files",
+                extra_dir_at_root=True,
+                alt_name=os.path.basename(self.get_file_name()),
+            )
 
     def get_file_name(self, sync_cache=True):
         # Ensure the directory structure and the metadata file object exist
diff --git a/lib/galaxy/model/store/__init__.py b/lib/galaxy/model/store/__init__.py
index 1b78f52ebcc8..4880b83e7b96 100644
--- a/lib/galaxy/model/store/__init__.py
+++ b/lib/galaxy/model/store/__init__.py
@@ -183,8 +183,6 @@ class ImportDiscardedDataType(Enum):
 
 class DatasetAttributeImportModel(BaseModel):
     state: Optional[DatasetStateField] = None
-    deleted: Optional[bool] = None
-    purged: Optional[bool] = None
     external_filename: Optional[str] = None
     _extra_files_path: Optional[str] = None
     file_size: Optional[int] = None
@@ -470,6 +468,8 @@ def handle_dataset_object_edit(dataset_instance, dataset_attrs):
                 )
                 for attribute, value in dataset_attributes.items():
                     setattr(dataset_instance.dataset, attribute, value)
+                if dataset_instance.dataset.purged:
+                    dataset_instance.dataset.full_delete()
                 self._attach_dataset_hashes(dataset_attrs["dataset"], dataset_instance)
                 self._attach_dataset_sources(dataset_attrs["dataset"], dataset_instance)
                 if "id" in dataset_attrs["dataset"] and self.import_options.allow_edit:
@@ -654,17 +654,18 @@ def handle_dataset_object_edit(dataset_instance, dataset_attrs):
                         dataset_instance.state = dataset_state
                         if not self.object_store:
                             raise Exception(f"self.object_store is missing from {self}.")
-                        self.object_store.update_from_file(
-                            dataset_instance.dataset, file_name=temp_dataset_file_name, create=True
-                        )
+                        if not dataset_instance.dataset.purged:
+                            self.object_store.update_from_file(
+                                dataset_instance.dataset, file_name=temp_dataset_file_name, create=True
+                            )
 
-                        # Import additional files if present. Histories exported previously might not have this attribute set.
-                        dataset_extra_files_path = dataset_attrs.get("extra_files_path", None)
-                        if dataset_extra_files_path:
-                            assert file_source_root
-                            dataset_extra_files_path = os.path.join(file_source_root, dataset_extra_files_path)
-                            persist_extra_files(self.object_store, dataset_extra_files_path, dataset_instance)
-                        # Don't trust serialized file size
+                            # Import additional files if present. Histories exported previously might not have this attribute set.
+                            dataset_extra_files_path = dataset_attrs.get("extra_files_path", None)
+                            if dataset_extra_files_path:
+                                assert file_source_root
+                                dataset_extra_files_path = os.path.join(file_source_root, dataset_extra_files_path)
+                                persist_extra_files(self.object_store, dataset_extra_files_path, dataset_instance)
+                            # Don't trust serialized file size
                         dataset_instance.dataset.file_size = None
                         dataset_instance.dataset.set_total_size()  # update the filesize record in the database
 
diff --git a/lib/galaxy/model/store/discover.py b/lib/galaxy/model/store/discover.py
index 9329bda9ee40..679bd2aa38f7 100644
--- a/lib/galaxy/model/store/discover.py
+++ b/lib/galaxy/model/store/discover.py
@@ -91,6 +91,7 @@ def create_dataset(
         creating_job_id=None,
         output_name=None,
         storage_callbacks=None,
+        purged=False,
     ):
         tag_list = tag_list or []
         sources = sources or []
@@ -190,7 +191,11 @@ def create_dataset(
 
         if info is not None:
             primary_data.info = info
-        if filename:
+
+        if purged:
+            primary_data.dataset.purged = True
+            primary_data.purged = True
+        if filename and not purged:
             if storage_callbacks is None:
                 self.finalize_storage(
                     primary_data=primary_data,
@@ -214,6 +219,11 @@ def create_dataset(
         return primary_data
 
     def finalize_storage(self, primary_data, dataset_attributes, extra_files, filename, link_data, output_name):
+        if primary_data.dataset.purged:
+            # metadata won't be set, maybe we should do that, then purge ?
+            primary_data.dataset.file_size = 0
+            primary_data.dataset.total_size = 0
+            return
         # Move data from temp location to dataset location
         if not link_data:
             dataset = primary_data.dataset
diff --git a/lib/galaxy/objectstore/__init__.py b/lib/galaxy/objectstore/__init__.py
index f6bb0f4b07ae..49b559525ea9 100644
--- a/lib/galaxy/objectstore/__init__.py
+++ b/lib/galaxy/objectstore/__init__.py
@@ -1670,7 +1670,7 @@ def persist_extra_files(
     primary_data: "DatasetInstance",
     extra_files_path_name: Optional[str] = None,
 ) -> None:
-    if os.path.exists(src_extra_files_path):
+    if not primary_data.dataset.purged and os.path.exists(src_extra_files_path):
         assert primary_data.dataset
         if not extra_files_path_name:
             extra_files_path_name = primary_data.dataset.extra_files_path_name_from(object_store)
diff --git a/lib/galaxy/webapps/galaxy/api/job_files.py b/lib/galaxy/webapps/galaxy/api/job_files.py
index 74c25e07ea35..3bae6d62d137 100644
--- a/lib/galaxy/webapps/galaxy/api/job_files.py
+++ b/lib/galaxy/webapps/galaxy/api/job_files.py
@@ -99,6 +99,8 @@ def create(self, trans, job_id, payload, **kwargs):
         """
         job = self.__authorize_job_access(trans, job_id, **payload)
         path = payload.get("path")
+        if not path:
+            raise exceptions.RequestParameterInvalidException("'path' parameter not provided or empty.")
         self.__check_job_can_write_to_path(trans, job, path)
 
         # Is this writing an unneeded file? Should this just copy in Python?
diff --git a/test/functional/tools/all_output_types.xml b/test/functional/tools/all_output_types.xml
new file mode 100644
index 000000000000..58cd10e8a4a9
--- /dev/null
+++ b/test/functional/tools/all_output_types.xml
@@ -0,0 +1,90 @@
+<tool id="all_output_types" name="all_output_types" version="1.0.0" profile="24.0">
+    <command><![CDATA[
+        sleep $sleep_param &&
+        echo hi > output.txt &&
+        echo hi > '$static_output' &&
+        echo hi > '$static_output_2' &&
+        cp '$c1' galaxy.json
+        ]]>
+    </command>
+    <configfiles>
+        <configfile name="c1">{"output_tool_supplied_metadata": {
+    "name": "my dynamic name",
+    "ext": "txt",
+    "info": "my dynamic info"
+  }}
+        </configfile>
+    </configfiles>
+    <inputs>
+        <param name="sleep_param" type="integer" value="0" />
+    </inputs>
+    <outputs>
+        <data name="static_output" format="txt" />
+        <data name="static_output_2" format="txt" />
+        <data name="output_workdir" from_work_dir="output.txt" format="txt" />
+        <data name="output_tool_supplied_metadata" from_work_dir="output.txt" format="auto" />
+        <data format="txt" name="discovered_output">
+            <discover_datasets pattern="(?P&lt;designation&gt;.+)\.txt" ext="txt" visible="true"/>
+        </data>
+        <data format="txt" name="discovered_output_replaced">
+            <discover_datasets pattern="(?P&lt;designation&gt;.+)\.txt" ext="txt" visible="true" assign_primary_output="true" />
+        </data>
+        <collection type="paired" name="static_pair" format="txt">
+            <data name="forward" from_work_dir="output.txt"></data>
+            <data name="reverse" from_work_dir="output.txt"></data>
+        </collection>
+        <collection type="list" name="discovered_list" format="txt">
+            <discover_datasets pattern="(?P&lt;designation&gt;.+)\.txt" ext="txt" visible="true" />
+        </collection>
+    </outputs>
+    <tests>
+        <test>
+            <output name="static_output">
+                <assert_contents>
+                    <has_text text="hi"/>
+                </assert_contents>
+            </output>
+            <output name="output_workdir">
+                <assert_contents>
+                    <has_text text="hi"/>
+                </assert_contents>
+            </output>
+            <output name="output_tool_supplied_metadata">
+                <assert_contents>
+                    <has_text text="hi"/>
+                </assert_contents>
+            </output>
+            <output name="discovered_output">
+                <discovered_dataset designation="output" ftype="txt">
+                    <assert_contents>
+                        <has_text text="hi"/>
+                    </assert_contents>
+                  </discovered_dataset>
+            </output>
+            <output name="discovered_output_replaced" count="1">
+                <assert_contents>
+                    <has_text text="hi"/>
+                </assert_contents>
+            </output>
+            <output_collection name="static_pair" type="paired">
+                <element name="forward" ftype="txt">
+                    <assert_contents>
+                        <has_text text="hi"></has_text>
+                    </assert_contents>
+                </element>
+                <element name="reverse" ftype="txt">
+                    <assert_contents>
+                        <has_text text="hi"></has_text>
+                    </assert_contents>
+                </element>
+            </output_collection>
+            <output_collection name="discovered_list">
+                <element name="output" ftype="txt">
+                    <assert_contents>
+                        <has_text text="hi"></has_text>
+                    </assert_contents>
+                </element>
+            </output_collection>
+        </test>
+    </tests>
+</tool>
diff --git a/test/functional/tools/sample_tool_conf.xml b/test/functional/tools/sample_tool_conf.xml
index 8775d130e520..2c074381db5e 100644
--- a/test/functional/tools/sample_tool_conf.xml
+++ b/test/functional/tools/sample_tool_conf.xml
@@ -8,10 +8,10 @@
     <tool file="param_text_option.xml" />
     <tool file="column_param.xml" />
   </section>
-  <tool file="image_diff.xml"/>
-  <tool file="output_format_input.xml"/>
-  <tool file="ucsc_tablebrowser.xml"/>
-  <tool file="test_data_source.xml"/>
+  <tool file="image_diff.xml" />
+  <tool file="output_format_input.xml" />
+  <tool file="ucsc_tablebrowser.xml" />
+  <tool file="test_data_source.xml" />
   <tool file="simple_constructs.xml" />
   <tool file="color_param.xml" />
   <tool file="inheritance_simple.xml" />
@@ -89,7 +89,7 @@
   <tool file="sim_size_delta.xml" />
   <tool file="composite_shapefile.xml" />
   <tool file="is_valid_xml.xml" />
-  <tool file="expression_tools/parse_values_from_file.xml"/>
+  <tool file="expression_tools/parse_values_from_file.xml" />
   <tool file="expression_tools/pick_value.xml" />
   <!--
   TODO: Figure out why this transiently fails on Jenkins.
@@ -137,7 +137,7 @@
   <tool file="column_param_configfile.xml" />
   <tool file="column_param_list.xml" />
   <tool file="column_multi_param.xml" />
-  <tool file="select_optional.xml"/>
+  <tool file="select_optional.xml" />
   <tool file="hidden_param.xml" />
   <tool file="special_params.xml" />
   <tool file="section.xml" />
@@ -150,10 +150,10 @@
   <tool file="validation_default.xml" />
   <tool file="validation_sanitizer.xml" />
   <tool file="validation_repeat.xml" />
-  <tool file="validation_metadata_in_range.xml"/>
-  <tool file="validation_metadata_in_datatable.xml"/>
-  <tool file="validation_dataset_metadata_in_file.xml"/>
-  <tool file="validation_value_in_datatable.xml"/>
+  <tool file="validation_metadata_in_range.xml" />
+  <tool file="validation_metadata_in_datatable.xml" />
+  <tool file="validation_dataset_metadata_in_file.xml" />
+  <tool file="validation_value_in_datatable.xml" />
   <tool file="empty_output.xml" />
   <tool file="validation_empty_dataset.xml" />
   <tool file="implicit_conversion.xml" />
@@ -170,7 +170,7 @@
   <tool file="identifier_multiple_in_conditional.xml" />
   <tool file="identifier_multiple_in_repeat.xml" />
   <tool file="identifier_collection.xml" />
-  <tool file="identifier_all_collection_types.xml"/>
+  <tool file="identifier_all_collection_types.xml" />
   <tool file="identifier_in_actions.xml" />
   <tool file="fail_identifier.xml" />
   <tool file="fail_writing_work_dir_file.xml" />
@@ -224,7 +224,7 @@
   <tool file="expect_num_outputs.xml" />
   <tool file="text_repeat.xml" />
   <tool file="integer_default.xml" />
-
+  <tool file="all_output_types.xml" />
   <tool file="multiple_versions_v01.xml" />
   <tool file="multiple_versions_v01galaxy6.xml" />
   <tool file="multiple_versions_v02.xml" />
diff --git a/test/integration/objectstore/_purged_handling.py b/test/integration/objectstore/_purged_handling.py
new file mode 100644
index 000000000000..1437f6742ae5
--- /dev/null
+++ b/test/integration/objectstore/_purged_handling.py
@@ -0,0 +1,71 @@
+import os
+import time
+
+from galaxy_test.base.populators import DatasetPopulator
+
+
+def purge_while_job_running(dataset_populator: DatasetPopulator, extra_sleep=0):
+    with dataset_populator.test_history() as history_id:
+        response = dataset_populator.run_tool(
+            "all_output_types",
+            inputs={
+                "sleep_param": 5,
+            },
+            history_id=history_id,
+        )
+        job = dataset_populator.get_job_details(response["jobs"][0]["id"], full=True).json()
+        # Make sure job runs (and thus command is templated before purging output)
+        dataset_populator.wait_for_job(job["id"], ok_states=["running"])
+        time.sleep(extra_sleep)
+        hda_ids = []
+        hda_filenames = []
+        for output_name, output in job["outputs"].items():
+            if output_name == "static_output_2":
+                # We need to keep one output so the job doesn't get cancelled
+                continue
+            details = dataset_populator.get_history_dataset_details(
+                history_id=history_id, content_id=output["id"], wait=False
+            )
+            if not output_name.startswith("discovered_output"):
+                # We're not precreating discovered outputs on disk
+                hda_filenames.append(details["file_name"])
+            dataset_populator.delete_dataset(
+                history_id=history_id, content_id=output["id"], purge=True, wait_for_purge=True
+            )
+            hda_ids.append(output["id"])
+        for output_collection in job["output_collections"].values():
+            hdca = dataset_populator.get_history_collection_details(
+                history_id=history_id, content_id=output_collection["id"]
+            )
+            for element in hdca["elements"]:
+                # Technically the static pair elements are already included in job["outputs"],
+                # but no harm purging them again here, in case we ever change that logic.
+                hda_id = element["object"]["id"]
+                dataset_populator.delete_dataset(
+                    history_id=history_id, content_id=hda_id, purge=True, wait_for_purge=True
+                )
+                hda_ids.append(hda_id)
+        dataset_populator.wait_for_job(job["id"], assert_ok=True)
+        # Now make sure we can't find the datasets on disk
+        # We're not covering discovered datasets here, those can't be purged while the job is running.
+        for hda_id in hda_ids:
+            exception = None
+            try:
+                dataset_populator.get_history_dataset_content(history_id=history_id, dataset_id=hda_id)
+            except AssertionError as e:
+                exception = e
+            assert exception and "The dataset you are attempting to view has been purged" in str(exception), str(
+                exception
+            )
+            output_details = dataset_populator.get_history_dataset_details(
+                history_id=history_id, content_id=hda_id, wait=False
+            )
+            # Make sure that we don't revert state while finishing job
+            assert output_details["purged"], f"expected output '{output_name}' to be purged, but it is not purged."
+            assert not output_details.get("file_name")
+            assert (
+                output_details["file_size"] == 0
+            ), f"expected file_size for '{output_name}' to be 0, but it is {output_details['file_size']}."
+        for file_name in hda_filenames:
+            # Make sure job didn't push to object store
+            assert not os.path.exists(file_name), f"Expected {file_name} to be purged."
diff --git a/test/integration/test_extended_metadata.py b/test/integration/test_extended_metadata.py
index ea433f5c2fab..d267ed01c418 100644
--- a/test/integration/test_extended_metadata.py
+++ b/test/integration/test_extended_metadata.py
@@ -5,6 +5,7 @@
     LibraryPopulator,
 )
 from galaxy_test.driver import integration_util
+from .objectstore._purged_handling import purge_while_job_running
 
 TEST_TOOL_IDS = [
     "from_work_dir_glob",
@@ -47,6 +48,7 @@
 
 class TestExtendedMetadataIntegration(integration_util.IntegrationTestCase):
     dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
 
     def setUp(self):
         super().setUp()
@@ -95,6 +97,11 @@ def test_fetch_data_library(self):
         assert dataset["file_ext"] == "bed", dataset
         assert dataset["created_from_basename"] == "4.bed"
 
+    def test_purge_while_job_running(self):
+        # pass extra_sleep, since templating the command line will fail if the output
+        # is deleted before remote_tool_eval runs.
+        purge_while_job_running(self.dataset_populator, extra_sleep=4)
+
 
 class TestExtendedMetadataDeferredIntegration(integration_util.IntegrationTestCase):
     dataset_populator: DatasetPopulator
diff --git a/test/integration/test_pulsar_embedded_mq.py b/test/integration/test_pulsar_embedded_mq.py
index e0b9e156992b..19022a9b4b2f 100644
--- a/test/integration/test_pulsar_embedded_mq.py
+++ b/test/integration/test_pulsar_embedded_mq.py
@@ -7,7 +7,9 @@
 import pytest
 
 from galaxy.util import safe_makedirs
+from galaxy_test.base.populators import DatasetPopulator
 from galaxy_test.driver import integration_util
+from .objectstore._purged_handling import purge_while_job_running
 
 SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
 EMBEDDED_PULSAR_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "embedded_pulsar_mq_job_conf.yml")
@@ -46,6 +48,53 @@
 """
 
 
+def _handle_galaxy_config_kwds(cls, config):
+    amqp_url = os.environ.get("GALAXY_TEST_AMQP_URL", None)
+    if amqp_url is None:
+        pytest.skip("External AMQP URL not configured for test")
+
+    jobs_directory = os.path.join(cls._test_driver.mkdtemp(), "pulsar_staging")
+    safe_makedirs(jobs_directory)
+    job_conf_template = string.Template(JOB_CONF_TEMPLATE)
+    job_conf_str = job_conf_template.substitute(
+        amqp_url=AMQP_URL, jobs_directory=jobs_directory, galaxy_home=os.path.join(SCRIPT_DIRECTORY, os.pardir)
+    )
+    with tempfile.NamedTemporaryFile(suffix="_mq_job_conf.yml", mode="w", delete=False) as job_conf:
+        job_conf.write(job_conf_str)
+    config["job_config_file"] = job_conf.name
+    infrastructure_url = "http://localhost:$GALAXY_WEB_PORT"
+    config["galaxy_infrastructure_url"] = infrastructure_url
+
+
+class TestEmbeddedMessageQueuePulsarPurge(integration_util.IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        _handle_galaxy_config_kwds(cls, config)
+
+    def test_purge_while_job_running(self):
+        purge_while_job_running(self.dataset_populator)
+
+
+class TestEmbeddedMessageQueuePulsarExtendedMetadataPurge(TestEmbeddedMessageQueuePulsarPurge):
+    """Describe a Galaxy test instance with embedded pulsar and extended metadata configured.
+
+    $ Setup RabbitMQ (e.g. https://www.rabbitmq.com/install-homebrew.html)
+    $ GALAXY_TEST_AMQP_URL='amqp://guest:guest@localhost:5672//' pytest -s test/integration/test_pulsar_embedded_mq.py
+    """
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["metadata_strategy"] = "extended"
+        _handle_galaxy_config_kwds(cls, config)
+
+
 class EmbeddedMessageQueuePulsarIntegrationInstance(integration_util.IntegrationInstance):
     """Describe a Galaxy test instance with embedded pulsar configured.
 
@@ -53,27 +102,19 @@ class EmbeddedMessageQueuePulsarIntegrationInstance(integration_util.Integration
     $ GALAXY_TEST_AMQP_URL='amqp://guest:guest@localhost:5672//' pytest -s test/integration/test_pulsar_embedded_mq.py
     """
 
+    dataset_populator: DatasetPopulator
     framework_tool_and_types = True
 
     @classmethod
     def handle_galaxy_config_kwds(cls, config):
-        amqp_url = os.environ.get("GALAXY_TEST_AMQP_URL", None)
-        if amqp_url is None:
-            pytest.skip("External AMQP URL not configured for test")
-
-        jobs_directory = os.path.join(cls._test_driver.mkdtemp(), "pulsar_staging")
-        safe_makedirs(jobs_directory)
-        job_conf_template = string.Template(JOB_CONF_TEMPLATE)
-        job_conf_str = job_conf_template.substitute(
-            amqp_url=AMQP_URL, jobs_directory=jobs_directory, galaxy_home=os.path.join(SCRIPT_DIRECTORY, os.pardir)
-        )
-        with tempfile.NamedTemporaryFile(suffix="_mq_job_conf.yml", mode="w", delete=False) as job_conf:
-            job_conf.write(job_conf_str)
-        config["job_config_file"] = job_conf.name
-        infrastructure_url = "http://localhost:$GALAXY_WEB_PORT"
-        config["galaxy_infrastructure_url"] = infrastructure_url
+        _handle_galaxy_config_kwds(cls, config)
+
+    def test_purge_while_job_running(self):
+        purge_while_job_running(self.dataset_populator)
 
 
 instance = integration_util.integration_module_instance(EmbeddedMessageQueuePulsarIntegrationInstance)
 
-test_tools = integration_util.integration_tool_runner(["simple_constructs", "composite_output_tests"])
+test_tools = integration_util.integration_tool_runner(
+    ["simple_constructs", "composite_output_tests", "all_output_types"]
+)
diff --git a/test/unit/app/jobs/test_command_factory.py b/test/unit/app/jobs/test_command_factory.py
index 48c097a37139..e185904e94e6 100644
--- a/test/unit/app/jobs/test_command_factory.py
+++ b/test/unit/app/jobs/test_command_factory.py
@@ -21,7 +21,7 @@
 TEST_FILES_PATH = "file_path"
 TEE_REDIRECT = '> "$__out" 2> "$__err"'
 RETURN_CODE_CAPTURE = "; return_code=$?; echo $return_code > galaxy_1.ec"
-CP_WORK_DIR_OUTPUTS = '; \nif [ -f "foo" ] ; then cp "foo" "bar" ; fi'
+CP_WORK_DIR_OUTPUTS = '; \nif [ -f "foo" -a -f "bar" ] ; then cp "foo" "bar" ; fi'
 
 
 class TestCommandFactory(TestCase):
@@ -105,7 +105,7 @@ def test_workdir_outputs_with_glob(self):
         self.workdir_outputs = [("foo*bar", "foo_x_bar")]
         self._assert_command_is(
             self._surround_command(
-                MOCK_COMMAND_LINE, '; \nif [ -f "foo"*"bar" ] ; then cp "foo"*"bar" "foo_x_bar" ; fi'
+                MOCK_COMMAND_LINE, '; \nif [ -f "foo"*"bar" -a -f "foo_x_bar" ] ; then cp "foo"*"bar" "foo_x_bar" ; fi'
             )
         )