Merge pull request #22360 from mvdbeek/pja-job-cache-fix

[26.0] Fix HideDatasetAction not applied for cached workflow jobs

lib/galaxy/jobs/handler.py

@@ -31,6 +31,7 @@ from sqlalchemy.sql.expression import (
 
 from galaxy import model
 from galaxy.exceptions import ObjectNotFound
+from galaxy.job_execution.actions.post import ActionBox
 from galaxy.jobs import (
     JobQueueI,
     JobWrapper,
@@ -554,6 +555,12 @@ class JobHandlerQueue(BaseJobHandlerQueue):
                     copied_from_job = self.sa_session.get(model.Job, job.copied_from_job_id)
                     assert copied_from_job is not None
                     job.copy_from_job(copied_from_job)
+                    # Execute non-immediate post-job actions (e.g. HideDatasetAction)
+                    # that were registered during workflow step execution but are
+                    # normally only run at job completion via the runner finish path.
+                    for pjaa in job.post_job_actions:
+                        if pjaa.post_job_action.action_type not in ActionBox.immediate_actions:
+                            ActionBox.execute(self.app, self.sa_session, pjaa.post_job_action, job)
                     continue
                 job_state = self.__check_job_state(job)
                 if job_state == JOB_WAIT:

lib/galaxy_test/api/test_workflows.py

@@ -6757,6 +6757,48 @@ steps:
                 first_wf_output["file_name"] == second_wf_output["file_name"]
             ), f"first output:\n{first_wf_output}\nsecond output:\n{second_wf_output}"
 
+    @skip_without_tool("cat1")
+    def test_workflow_rerun_with_use_cached_job_hides_output(self, history_id: str):
+        run_summary = self._run_workflow(
+            """
+class: GalaxyWorkflow
+inputs:
+  input1: data
+steps:
+  first_cat:
+    tool_id: cat1
+    in:
+      input1: input1
+    outputs:
+      out_file1:
+        hide: true
+""",
+            test_data="""
+input1:
+  value: 1.fasta
+  type: File
+  name: fasta1
+""",
+            history_id=history_id,
+        )
+        first_output = self.dataset_populator.get_history_dataset_details(history_id=history_id, hid=2)
+        assert not first_output["visible"], f"Expected output to be hidden on first run: {first_output}"
+
+        # Unhide the output so we can verify the cached rerun hides it again
+        # (rather than just preserving already-hidden state)
+        self.dataset_populator.update_dataset(first_output["id"], {"visible": True})
+        first_output = self.dataset_populator.get_history_dataset_details(history_id=history_id, hid=2)
+        assert first_output["visible"]
+
+        rerun_summary = self.workflow_populator.rerun(run_summary, use_cached_job=True)
+        # Verify job was actually cached
+        for job in rerun_summary.jobs:
+            job_details = self.dataset_populator.get_job_details(job["id"], full=True).json()
+            assert job_details["copied_from_job_id"], f"Expected job to be cached: {job_details}"
+
+        cached_output = self.dataset_populator.get_history_dataset_details(history_id=history_id, hid=3)
+        assert not cached_output["visible"], f"Expected output to be hidden with cached job: {cached_output}"
+
     @skip_without_tool("cat1")
     @skip_without_tool("identifier_multiple")
     def test_workflow_rerun_with_cached_job_consumes_implicit_hdca(self, history_id: str):

lib/galaxy_test/base/populators.py

@@ -2761,7 +2761,13 @@ class BaseWorkflowPopulator(BasePopulator):
             expected_response=expected_response,
         )
 
-    def rerun(self, run_jobs_summary: "RunJobsSummary", wait: bool = True, assert_ok: bool = True) -> "RunJobsSummary":
+    def rerun(
+        self,
+        run_jobs_summary: "RunJobsSummary",
+        wait: bool = True,
+        assert_ok: bool = True,
+        use_cached_job: bool = False,
+    ) -> "RunJobsSummary":
         history_id = run_jobs_summary.history_id
         invocation_id = run_jobs_summary.invocation_id
         inputs = run_jobs_summary.inputs
@@ -2769,6 +2775,8 @@ class BaseWorkflowPopulator(BasePopulator):
         workflow_id = workflow_request["workflow_id"]
         assert workflow_request["history_id"] == history_id
         assert workflow_request["instance"] is True
+        if use_cached_job:
+            workflow_request["use_cached_job"] = True
         return self._request_to_summary(
             history_id,
             workflow_id,