diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py
index 249b34f2b484..391dbbc11434 100644
--- a/lib/galaxy/tools/actions/__init__.py
+++ b/lib/galaxy/tools/actions/__init__.py
@@ -850,6 +850,7 @@ def _new_job_for_session(self, trans, tool, history):
                 job.session_id = model.cached_id(galaxy_session)
         if trans.user is not None:
             job.user_id = model.cached_id(trans.user)
+            job.user = trans.user
         if history:
             job.history_id = model.cached_id(history)
         job.tool_id = tool.id
diff --git a/lib/galaxy_test/api/test_workflows.py b/lib/galaxy_test/api/test_workflows.py
index af7633fb1882..393546d95ec6 100644
--- a/lib/galaxy_test/api/test_workflows.py
+++ b/lib/galaxy_test/api/test_workflows.py
@@ -5884,6 +5884,44 @@ def test_run_remove_tag_on_collection_output(self):
             ), details_collection_without_tag
             assert len(details_collection_without_tag["tags"]) == 0, details_collection_without_tag
 
+    @skip_without_tool("collection_creates_pair")
+    @skip_without_tool("cat")
+    def test_run_add_tag_on_database_operation_output(self):
+        with self.dataset_populator.test_history() as history_id:
+            self._run_jobs(
+                """
+class: GalaxyWorkflow
+inputs:
+  input1: data_collection
+steps:
+  extrat:
+    tool_id: __EXTRACT_DATASET__
+    in:
+      input: input1
+    outputs:
+      output:
+        add_tags:
+          - "name:foo"
+""",
+                test_data="""
+input1:
+  collection_type: list
+  name: the_dataset_list
+  elements:
+    - identifier: el1
+      value: 1.fastq
+      type: File
+""",
+                history_id=history_id,
+                round_trip_format_conversion=True,
+            )
+            details_dataset_with_tag = self.dataset_populator.get_history_dataset_details(
+                history_id, hid=3, wait=True, assert_ok=True
+            )
+
+            assert details_dataset_with_tag["history_content_type"] == "dataset", details_dataset_with_tag
+            assert details_dataset_with_tag["tags"][0] == "name:foo", details_dataset_with_tag
+
     @skip_without_tool("cat1")
     def test_run_with_runtime_pja(self):
         workflow = self.workflow_populator.load_workflow(name="test_for_pja_runtime")