diff --git a/lib/galaxy/job_execution/output_collect.py b/lib/galaxy/job_execution/output_collect.py
index ac35b639917..80761493c0f 100644
--- a/lib/galaxy/job_execution/output_collect.py
+++ b/lib/galaxy/job_execution/output_collect.py
@@ -547,6 +547,7 @@ def collect_primary_datasets(job_context: Union[JobContext, SessionlessJobContex
             outdata.init_meta()
             outdata.set_meta()
             outdata.set_peek()
+            outdata.discovered = True
             sa_session = job_context.sa_session
             if sa_session:
                 sa_session.add(outdata)
diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index 72a614f8663..803da2dccc1 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -1852,6 +1852,9 @@ class JobWrapper(HasResourceParameters):
                 job.job_messages = [str(e)]
 
             for dataset_assoc in output_dataset_associations:
+                if getattr(dataset_assoc.dataset, "discovered", False):
+                    # skip outputs that have been discovered
+                    continue
                 context = self.get_dataset_finish_context(job_context, dataset_assoc)
                 # should this also be checking library associations? - can a library item be added from a history before the job has ended? -
                 # lets not allow this to occur
diff --git a/lib/galaxy/model/store/discover.py b/lib/galaxy/model/store/discover.py
index 9dd5a6000bc..67b6bb85c75 100644
--- a/lib/galaxy/model/store/discover.py
+++ b/lib/galaxy/model/store/discover.py
@@ -815,6 +815,7 @@ def persist_hdas(elements, model_persistence_context, final_job_state="ok"):
                     final_job_state=state,
                     storage_callbacks=storage_callbacks,
                 )
+                dataset.discovered = True
                 if not hda_id:
                     datasets.append(dataset)