diff --git a/lib/galaxy/tool_util/parser/interface.py b/lib/galaxy/tool_util/parser/interface.py
index af72bf4a4825..4ddcf8cea278 100644
--- a/lib/galaxy/tool_util/parser/interface.py
+++ b/lib/galaxy/tool_util/parser/interface.py
@@ -37,6 +37,12 @@
         ResourceRequirement,
         ToolRequirements,
     )
+    from galaxy.tool_util.parser.output_objects import (
+        ToolOutput,
+        ToolOutputCollection,
+    )
+    from galaxy.tools import Tool
+
 
 NOT_IMPLEMENTED_MESSAGE = "Galaxy tool format does not yet support this tool feature."
 
@@ -331,7 +337,7 @@ def parse_provided_metadata_file(self):
         return "galaxy.json"
 
     @abstractmethod
-    def parse_outputs(self, tool):
+    def parse_outputs(self, tool: "Tool") -> Tuple[Dict[str, "ToolOutput"], Dict[str, "ToolOutputCollection"]]:
         """Return a pair of output and output collections ordered
         dictionaries for use by Tool.
         """
diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py
index bb62f87e05d5..757483abab97 100644
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -83,6 +83,10 @@
     PageSource,
     ToolSource,
 )
+from galaxy.tool_util.parser.output_objects import (
+    ToolOutput,
+    ToolOutputCollection,
+)
 from galaxy.tool_util.parser.util import (
     parse_profile_version,
     parse_tool_version_with_defaults,
@@ -847,6 +851,8 @@ def __init__(
         self.tool_errors = None
         # Parse XML element containing configuration
         self.tool_source = tool_source
+        self.outputs: Dict[str, ToolOutput] = {}
+        self.output_collections: Dict[str, ToolOutputCollection] = {}
         self._is_workflow_compatible = None
         self.__help = None
         self.__tests: Optional[str] = None
diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py
index b8be3ecff2a0..fd5f97686fba 100644
--- a/lib/galaxy/tools/actions/__init__.py
+++ b/lib/galaxy/tools/actions/__init__.py
@@ -3,13 +3,13 @@
 import os
 import re
 from abc import abstractmethod
-from collections import UserDict
 from json import dumps
 from typing import (
     Any,
     cast,
     Dict,
     List,
+    MutableMapping,
     Optional,
     Set,
     Tuple,
@@ -1157,7 +1157,7 @@ def determine_output_format(
     output: "ToolOutput",
     parameter_context,
     input_datasets,
-    input_dataset_collections: UserDict[str, model.HistoryDatasetCollectionAssociation],
+    input_dataset_collections: MutableMapping[str, model.HistoryDatasetCollectionAssociation],
     random_input_ext,
     python_template_version="3",
     execution_cache=None,
diff --git a/lib/galaxy/tools/evaluation.py b/lib/galaxy/tools/evaluation.py
index 582bd65be06e..060072b25882 100644
--- a/lib/galaxy/tools/evaluation.py
+++ b/lib/galaxy/tools/evaluation.py
@@ -33,6 +33,7 @@
     MinimalToolApp,
 )
 from galaxy.tool_util.data import TabularToolDataTable
+from galaxy.tools.actions import determine_output_format
 from galaxy.tools.parameters import (
     visit_input_values,
     wrapped_json,
@@ -130,7 +131,7 @@ class ToolEvaluator:
     job: model.Job
     materialize_datasets: bool = True
 
-    def __init__(self, app: MinimalToolApp, tool, job, local_working_directory):
+    def __init__(self, app: MinimalToolApp, tool: "Tool", job, local_working_directory):
         self.app = app
         self.job = job
         self.tool = tool
@@ -186,6 +187,9 @@ def set_compute_environment(self, compute_environment: ComputeEnvironment, get_s
             out_data,
             output_collections=out_collections,
         )
+        # late update of format_source outputs
+        self._eval_format_source(job, inp_data, out_data)
+
         self.execute_tool_hooks(inp_data=inp_data, out_data=out_data, incoming=incoming)
 
     def execute_tool_hooks(self, inp_data, out_data, incoming):
@@ -275,6 +279,23 @@ def _materialize_objects(
 
         return undeferred_objects
 
+    def _eval_format_source(
+        self,
+        job: model.Job,
+        inp_data: Dict[str, Optional[model.DatasetInstance]],
+        out_data: Dict[str, model.DatasetInstance],
+    ):
+        for output_name, output in out_data.items():
+            if (
+                (tool_output := self.tool.outputs.get(output_name))
+                and (tool_output.format_source or tool_output.change_format)
+                and output.extension == "expression.json"
+            ):
+                input_collections = {jtidca.name: jtidca.dataset_collection for jtidca in job.input_dataset_collections}
+                ext = determine_output_format(tool_output, self.param_dict, inp_data, input_collections, None)
+                if ext:
+                    output.extension = ext
+
     def _replaced_deferred_objects(
         self,
         inp_data: Dict[str, Optional[model.DatasetInstance]],