diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 7e582782e911..7c239fd720cd 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -8002,10 +8002,20 @@ def copy_to(self, copied_step, step_mapping, user=None):
             copied_step.annotations = annotations
 
         if subworkflow := self.subworkflow:
-            copied_subworkflow = subworkflow.copy()
-            copied_step.subworkflow = copied_subworkflow
-            for subworkflow_step, copied_subworkflow_step in zip(subworkflow.steps, copied_subworkflow.steps):
-                subworkflow_step_mapping[subworkflow_step.id] = copied_subworkflow_step
+            stored_subworkflow = subworkflow.stored_workflow
+            if stored_subworkflow and stored_subworkflow.user == user:
+                # This should be fine and reduces the number of stored subworkflows
+                copied_step.subworkflow = subworkflow
+            else:
+                # Can this even happen, building a workflow with a subworkflow you don't own ?
+                copied_subworkflow = subworkflow.copy()
+                stored_workflow = StoredWorkflow(
+                    user, name=copied_subworkflow.name, workflow=copied_subworkflow, hidden=True
+                )
+                copied_subworkflow.stored_workflow = stored_workflow
+                copied_step.subworkflow = copied_subworkflow
+                for subworkflow_step, copied_subworkflow_step in zip(subworkflow.steps, copied_subworkflow.steps):
+                    subworkflow_step_mapping[subworkflow_step.id] = copied_subworkflow_step
 
         for old_conn, new_conn in zip(self.input_connections, copied_step.input_connections):
             new_conn.input_step_input = copied_step.get_or_add_input(old_conn.input_name)
diff --git a/test/unit/data/test_galaxy_mapping.py b/test/unit/data/test_galaxy_mapping.py
index 7e6c35d757f1..aa4ff3a86b3a 100644
--- a/test/unit/data/test_galaxy_mapping.py
+++ b/test/unit/data/test_galaxy_mapping.py
@@ -881,6 +881,9 @@ def test_workflows(self):
         assert loaded_invocation
         assert loaded_invocation.history.id == history_id
 
+        # recover user after expunge
+        user = loaded_invocation.history.user
+
         step_1, step_2 = loaded_invocation.workflow.steps
 
         assert not step_1.subworkflow