diff --git a/lib/galaxy/authnz/custos_authnz.py b/lib/galaxy/authnz/custos_authnz.py
index f3aa897a7c1..6a82db7243d 100644
--- a/lib/galaxy/authnz/custos_authnz.py
+++ b/lib/galaxy/authnz/custos_authnz.py
@@ -22,6 +22,7 @@ from galaxy.model import (
     CustosAuthnzToken,
     User,
 )
+from galaxy.model.orm.util import add_object_to_object_session
 from ..authnz import IdentityProvider
 
 log = logging.getLogger(__name__)
@@ -186,7 +187,6 @@ class CustosAuthnz(IdentityProvider):
             trans.app.user_manager.send_activation_email(trans, email, username)
 
         custos_authnz_token = CustosAuthnzToken(
-            user=user,
             external_user_id=user_id,
             provider=self.config["provider"],
             access_token=access_token,
@@ -195,6 +195,8 @@ class CustosAuthnz(IdentityProvider):
             expiration_time=expiration_time,
             refresh_expiration_time=refresh_expiration_time,
         )
+        add_object_to_object_session(custos_authnz_token, user)
+        custos_authnz_token.user = user
 
         trans.sa_session.add(user)
         trans.sa_session.add(custos_authnz_token)
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index e133999e0e1..4658cd1e92c 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -1721,6 +1721,7 @@ class Task(Base, JobLike, RepresentById):
         self.parameters = []
         self.state = Task.states.NEW
         self.working_directory = working_directory
+        add_object_to_object_session(self, job)
         self.job = job
         self.prepare_input_files_cmd = prepare_files_cmd
         self._init_metrics()
@@ -1980,6 +1981,7 @@ class JobToInputLibraryDatasetAssociation(Base, RepresentById):
 
     def __init__(self, name, dataset):
         self.name = name
+        add_object_to_object_session(self, dataset)
         self.dataset = dataset
 
 
@@ -1997,6 +1999,7 @@ class JobToOutputLibraryDatasetAssociation(Base, RepresentById):
 
     def __init__(self, name, dataset):
         self.name = name
+        add_object_to_object_session(self, dataset)
         self.dataset = dataset
 
 
@@ -2140,6 +2143,7 @@ class JobExternalOutputMetadata(Base, RepresentById):
     job = relationship("Job", back_populates="external_output_metadata")
 
     def __init__(self, job=None, dataset=None):
+        add_object_to_object_session(self, job)
         self.job = job
         if isinstance(dataset, galaxy.model.HistoryDatasetAssociation):
             self.history_dataset_association = dataset
@@ -2280,6 +2284,8 @@ class JobContainerAssociation(Base, RepresentById):
     job = relationship("Job", back_populates="container")
 
     def __init__(self, **kwd):
+        if "job" in kwd:
+            add_object_to_object_session(self, kwd["job"])
         super().__init__(**kwd)
         self.container_info = self.container_info or {}
 
@@ -2371,6 +2377,7 @@ class UserGroupAssociation(Base, RepresentById):
     group = relationship("Group", back_populates="users")
 
     def __init__(self, user, group):
+        add_object_to_object_session(self, user)
         self.user = user
         self.group = group
 
@@ -3052,6 +3059,7 @@ class UserQuotaAssociation(Base, Dictifiable, RepresentById):
     dict_element_visible_keys = ["user"]
 
     def __init__(self, user, quota):
+        add_object_to_object_session(self, user)
         self.user = user
         self.quota = quota
 
@@ -3070,6 +3078,7 @@ class GroupQuotaAssociation(Base, Dictifiable, RepresentById):
     dict_element_visible_keys = ["group"]
 
     def __init__(self, group, quota):
+        add_object_to_object_session(self, group)
         self.group = group
         self.quota = quota
 
@@ -3257,6 +3266,7 @@ class LibraryDatasetDatasetAssociationPermissions(Base, RepresentById):
     def __init__(self, action, library_item, role):
         self.action = action
         if isinstance(library_item, LibraryDatasetDatasetAssociation):
+            add_object_to_object_session(self, library_item)
             self.library_dataset_dataset_association = library_item
         else:
             raise Exception(f"Invalid LibraryDatasetDatasetAssociation specified: {library_item.__class__.__name__}")
@@ -3291,6 +3301,7 @@ class DefaultHistoryPermissions(Base, RepresentById):
     role = relationship("Role")
 
     def __init__(self, history, action, role):
+        add_object_to_object_session(self, history)
         self.history = history
         self.action = action
         self.role = role
@@ -5392,6 +5403,7 @@ class ImplicitlyConvertedDatasetAssociation(Base, RepresentById):
         self, id=None, parent=None, dataset=None, file_type=None, deleted=False, purged=False, metadata_safe=True
     ):
         self.id = id
+        add_object_to_object_session(self, dataset)
         if isinstance(dataset, HistoryDatasetAssociation):
             self.dataset = dataset
         elif isinstance(dataset, LibraryDatasetDatasetAssociation):
@@ -6471,6 +6483,7 @@ class GalaxySessionToHistoryAssociation(Base, RepresentById):
 
     def __init__(self, galaxy_session, history):
         self.galaxy_session = galaxy_session
+        add_object_to_object_session(self, history)
         self.history = history
 
 
@@ -6573,6 +6586,7 @@ class StoredWorkflow(Base, HasTags, Dictifiable, RepresentById):
         workflow=None,
         hidden=False,
     ):
+        add_object_to_object_session(self, user)
         self.user = user
         self.name = name
         self.slug = slug
@@ -6891,6 +6905,7 @@ class WorkflowStep(Base, RepresentById):
         conn = WorkflowStepConnection()
         conn.input_step_input = step_input
         conn.output_name = output_name
+        add_object_to_object_session(conn, output_step)
         conn.output_step = output_step
         if input_subworkflow_step_index is not None:
             input_subworkflow_step = self.subworkflow.step_by_index(input_subworkflow_step_index)
@@ -7059,6 +7074,7 @@ class WorkflowStepInput(Base, RepresentById):
     )
 
     def __init__(self, workflow_step):
+        add_object_to_object_session(self, workflow_step)
         self.workflow_step = workflow_step
         self.default_value_set = False
 
@@ -7247,6 +7263,7 @@ class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, RepresentBy
         assoc = WorkflowInvocationToSubworkflowInvocationAssociation()
         assoc.workflow_invocation = self
         assoc.workflow_step = step
+        add_object_to_object_session(subworkflow_invocation, self.history)
         subworkflow_invocation.history = self.history
         subworkflow_invocation.workflow = step.subworkflow
         assoc.subworkflow_invocation = subworkflow_invocation
@@ -9026,6 +9043,7 @@ class HistoryDatasetAssociationRatingAssociation(ItemRatingAssociation, Represen
     user = relationship("User")
 
     def _set_item(self, history_dataset_association):
+        add_object_to_object_session(self, history_dataset_association)
         self.history_dataset_association = history_dataset_association
 
 
@@ -9040,6 +9058,7 @@ class StoredWorkflowRatingAssociation(ItemRatingAssociation, RepresentById):
     user = relationship("User")
 
     def _set_item(self, stored_workflow):
+        add_object_to_object_session(self, stored_workflow)
         self.stored_workflow = stored_workflow
 
 
@@ -9054,6 +9073,7 @@ class PageRatingAssociation(ItemRatingAssociation, RepresentById):
     user = relationship("User")
 
     def _set_item(self, page):
+        add_object_to_object_session(self, page)
         self.page = page
 
 
@@ -9068,6 +9088,7 @@ class VisualizationRatingAssociation(ItemRatingAssociation, RepresentById):
     user = relationship("User")
 
     def _set_item(self, visualization):
+        add_object_to_object_session(self, visualization)
         self.visualization = visualization
 
 
@@ -9082,6 +9103,7 @@ class HistoryDatasetCollectionRatingAssociation(ItemRatingAssociation, Represent
     user = relationship("User")
 
     def _set_item(self, dataset_collection):
+        add_object_to_object_session(self, dataset_collection)
         self.dataset_collection = dataset_collection
 
 
@@ -9096,6 +9118,7 @@ class LibraryDatasetCollectionRatingAssociation(ItemRatingAssociation, Represent
     user = relationship("User")
 
     def _set_item(self, dataset_collection):
+        add_object_to_object_session(self, dataset_collection)
         self.dataset_collection = dataset_collection
 
 
diff --git a/lib/tool_shed/webapp/model/__init__.py b/lib/tool_shed/webapp/model/__init__.py
index 07dd38feac5..52def04404b 100644
--- a/lib/tool_shed/webapp/model/__init__.py
+++ b/lib/tool_shed/webapp/model/__init__.py
@@ -39,6 +39,7 @@ from galaxy.model.custom_types import (
     TrimmedString,
 )
 from galaxy.model.orm.now import now
+from galaxy.model.orm.util import add_object_to_object_session
 from galaxy.security.validate_user_input import validate_password_str
 from galaxy.util import unique_id
 from galaxy.util.bunch import Bunch
@@ -182,6 +183,7 @@ class PasswordResetToken(Base, _HasTable):
             self.token = token
         else:
             self.token = unique_id()
+        add_object_to_object_session(self, user)
         self.user = user
         self.expiration_time = now() + timedelta(hours=24)
 
@@ -252,6 +254,7 @@ class UserGroupAssociation(Base, _HasTable):
     group = relationship("Group", back_populates="users")
 
     def __init__(self, user, group):
+        add_object_to_object_session(self, user)
         self.user = user
         self.group = group
 
@@ -268,7 +271,9 @@ class UserRoleAssociation(Base, _HasTable):
     role = relationship("Role", back_populates="users")
 
     def __init__(self, user, role):
+        add_object_to_object_session(self, user)
         self.user = user
+        add_object_to_object_session(self, role)
         self.role = role
 
 
@@ -300,6 +305,7 @@ class RepositoryRoleAssociation(Base, _HasTable):
     role = relationship("Role", back_populates="repositories")
 
     def __init__(self, repository, role):
+        add_object_to_object_session(self, repository)
         self.repository = repository
         self.role = role
 
diff --git a/test/unit/app/tools/test_execution.py b/test/unit/app/tools/test_execution.py
index 0f22d7471ee..bc12e9c4d37 100644
--- a/test/unit/app/tools/test_execution.py
+++ b/test/unit/app/tools/test_execution.py
@@ -7,6 +7,7 @@ import webob.exc
 
 import galaxy.model
 from galaxy.app_unittest_utils import tools_support
+from galaxy.model.orm.util import add_object_to_object_session
 from galaxy.util.bunch import Bunch
 
 BASE_REPEAT_TOOL_CONTENTS = """<tool id="test_tool" name="Test Tool">
@@ -126,6 +127,7 @@ class ToolExecutionTestCase(TestCase, tools_support.UsesTools):
         hda.dataset.state = "ok"
 
         self.trans.sa_session.add(hda)
+        add_object_to_object_session(self.history, hda)
         self.history.datasets.append(hda)
         self.trans.sa_session.flush()
         return hda
diff --git a/test/unit/data/model/mapping/test_model_mapping.py b/test/unit/data/model/mapping/test_model_mapping.py
index 93e16eb9b44..59a933ac58e 100644
--- a/test/unit/data/model/mapping/test_model_mapping.py
+++ b/test/unit/data/model/mapping/test_model_mapping.py
@@ -74,6 +74,7 @@ from sqlalchemy import (
 
 from galaxy import model
 from galaxy.model.orm.now import now
+from galaxy.model.orm.util import add_object_to_object_session
 from .common import (
     AbstractBaseTest,
     collection_consists_of_objects,
@@ -1375,8 +1376,7 @@ class TestHistory(BaseTest):
         workflow_invocation,
         job,
     ):
-        obj = cls_()
-        obj.user = user
+        obj = cls_(user=user)
         obj.datasets.append(history_dataset_association)
         obj.dataset_collections.append(history_dataset_collection_association)
         obj.exports.append(job_export_history_archive)
@@ -1580,9 +1580,8 @@ class TestHistoryDatasetAssociation(BaseTest):
             icpda,
         ]
 
-        obj = cls_()
+        obj = cls_(dataset=dataset)
         obj.history = history
-        obj.dataset = dataset
         obj.copied_from_history_dataset_association = copied_from_hda
         obj.copied_from_library_dataset_dataset_association = copied_from_ldda
         obj.extended_metadata = extended_metadata
@@ -1902,6 +1901,7 @@ class TestHistoryDatasetCollectionAssociation(BaseTest):
         obj = cls_()
         obj.collection = dataset_collection
         obj.history = history
+        add_object_to_object_session(obj, history_dataset_collection_association)
         obj.copied_from_history_dataset_collection_association = history_dataset_collection_association
         obj.copied_to_history_dataset_collection_association.append(copied_to_hdca)
         obj.job = job
@@ -2146,7 +2146,6 @@ class TestImplicitCollectionJobsJobAssociation(BaseTest):
         order_index = 1
         obj = cls_()
         obj.implicit_collection_jobs = implicit_collection_jobs
-        session.add(job)  # must be bound to a session for lazy load of attributes
         obj.job = job
         obj.order_index = order_index
 
@@ -2160,7 +2159,6 @@ class TestImplicitCollectionJobsJobAssociation(BaseTest):
     def test_relationships(self, session, cls_, implicit_collection_jobs, job):
         obj = cls_()
         obj.implicit_collection_jobs = implicit_collection_jobs
-        session.add(job)  # must be bound to a session for lazy load of attributes
         obj.job = job
         obj.order_index = 1
 
@@ -2502,6 +2500,7 @@ class TestJob(BaseTest):
         obj.implicit_collection_jobs_association = implicit_collection_jobs_job_association
         obj.container = job_container_association
         obj.data_manager_association = data_manager_job_association
+        add_object_to_object_session(obj, history_dataset_collection_association)
         obj.history_dataset_collection_associations.append(history_dataset_collection_association)
         obj.workflow_invocation_step = workflow_invocation_step
 
@@ -2566,9 +2565,7 @@ class TestJobContainerAssociation(BaseTest):
         created_time = now()
         modified_time = created_time + timedelta(hours=1)
 
-        session.add(job)  # must be bound to a session for lazy load of attributes
-        obj = cls_()
-        obj.job = job
+        obj = cls_(job=job)
         obj.container_type = container_type
         obj.container_name = container_name
         obj.container_info = container_info
@@ -2586,7 +2583,6 @@ class TestJobContainerAssociation(BaseTest):
             assert stored_obj.modified_time == modified_time
 
     def test_relationships(self, session, cls_, job):
-        session.add(job)  # must be bound to a session for lazy load of attributes
         obj = cls_(job=job)
 
         with dbcleanup(session, obj) as obj_id:
@@ -3293,6 +3289,7 @@ class TestLibraryDatasetDatasetAssociation(BaseTest):
 
         obj = cls_()
         obj.library_dataset = library_dataset
+        add_object_to_object_session(obj, dataset)
         obj.dataset = dataset
         obj.create_time = create_time
         obj.copied_from_history_dataset_association = history_dataset_association
@@ -3381,6 +3378,7 @@ class TestLibraryDatasetDatasetAssociation(BaseTest):
 
         obj = cls_()
         obj.library_dataset = library_dataset
+        add_object_to_object_session(obj, dataset)
         obj.dataset = dataset
         obj.copied_from_history_dataset_association = history_dataset_association
         obj.copied_from_library_dataset_dataset_association = copied_from_ldda
@@ -3607,6 +3605,7 @@ class TestLibraryFolder(BaseTest):
         library_folder_factory,
     ):
         obj = cls_()
+        add_object_to_object_session(obj, library_folder)
         obj.parent = library_folder
         folder1 = library_folder_factory()
         obj.folders.append(folder1)
@@ -4401,6 +4400,7 @@ class TestStoredWorkflowMenuEntry(BaseTest):
         order_index = 1
         obj = cls_()
         obj.stored_workflow = stored_workflow
+        add_object_to_object_session(obj, user)
         obj.user = user
         obj.order_index = order_index
 
@@ -4414,6 +4414,7 @@ class TestStoredWorkflowMenuEntry(BaseTest):
     def test_relationships(self, session, cls_, stored_workflow, user):
         obj = cls_()
         obj.stored_workflow = stored_workflow
+        add_object_to_object_session(obj, user)
         obj.user = user
 
         with dbcleanup(session, obj) as obj_id:
@@ -5646,6 +5647,7 @@ class TestWorkflowInvocationOutputValue(BaseTest):
         persist(session, workflow_invocation_step)
 
         obj = cls_()
+        add_object_to_object_session(obj, workflow_invocation)
         obj.workflow_invocation = workflow_invocation
         obj.workflow_step = workflow_step
         obj.workflow_output = workflow_output
@@ -5677,9 +5679,6 @@ class TestWorkflowInvocationStep(BaseTest):
         update_time = create_time + timedelta(hours=1)
         state, action = "a", "b"
 
-        session.add(job)  # must be bound to a session for lazy load of attributes
-        session.add(implicit_collection_jobs)  # must be bound to a session for lazy load of attributes
-
         obj = cls_()
         obj.create_time = create_time
         obj.update_time = update_time
@@ -5714,9 +5713,6 @@ class TestWorkflowInvocationStep(BaseTest):
         workflow_invocation_step_output_dataset_association,
         workflow_invocation_output_value,
     ):
-        session.add(job)  # must be bound to a session for lazy load of attributes
-        session.add(implicit_collection_jobs)  # must be bound to a session for lazy load of attributes
-
         # setup workflow_invocation_output_value to test the output_value attribute
         output_value = workflow_invocation_output_value
         output_value.workflow_invocation = workflow_invocation
@@ -5724,6 +5720,7 @@ class TestWorkflowInvocationStep(BaseTest):
         persist(session, output_value)
 
         obj = cls_()
+        add_object_to_object_session(obj, workflow_invocation)
         obj.workflow_invocation = workflow_invocation
         obj.workflow_step = workflow_step
         obj.job = job
@@ -6952,6 +6949,7 @@ def role(session):
 @pytest.fixture
 def stored_workflow(session, user):
     instance = model.StoredWorkflow()
+    add_object_to_object_session(instance, user)
     instance.user = user
     yield from dbcleanup_wrapper(session, instance)
 
@@ -7185,6 +7183,7 @@ def workflow_request_to_input_dataset_collection_association(session):
 @pytest.fixture
 def workflow_step(session, workflow):
     instance = model.WorkflowStep()
+    add_object_to_object_session(instance, workflow)
     instance.workflow = workflow
     yield from dbcleanup_wrapper(session, instance)
 
@@ -7392,7 +7391,9 @@ def workflow_step_connection_factory():
 def workflow_step_factory(workflow):
     def make_instance(*args, **kwds):
         instance = model.WorkflowStep()
-        instance.workflow = kwds.get("workflow", workflow)
+        workflow2 = kwds.get("workflow", workflow)  # rename workflow not to confuse pytest
+        add_object_to_object_session(instance, workflow2)
+        instance.workflow = workflow2
         instance.subworkflow = kwds.get("subworkflow")
         return instance
 
diff --git a/test/unit/data/test_galaxy_mapping.py b/test/unit/data/test_galaxy_mapping.py
index 4a16c2dcafb..6b1fb7e3491 100644
--- a/test/unit/data/test_galaxy_mapping.py
+++ b/test/unit/data/test_galaxy_mapping.py
@@ -18,6 +18,10 @@ import galaxy.model.mapping as mapping
 from galaxy import model
 from galaxy.model.database_utils import create_database
 from galaxy.model.metadata import MetadataTempFile
+from galaxy.model.orm.util import (
+    add_object_to_object_session,
+    get_object_session,
+)
 from galaxy.model.security import GalaxyRBACAgent
 
 datatypes_registry = galaxy.datatypes.registry.Registry()
@@ -82,6 +86,7 @@ class MappingTests(BaseModelTestCase):
             annotated_association.annotation = "Test Annotation"
             annotated_association.user = u
             for key, value in kwds.items():
+                add_object_to_object_session(annotated_association, value)
                 setattr(annotated_association, key, value)
             self.persist(annotated_association)
             self.expunge()
@@ -90,6 +95,7 @@ class MappingTests(BaseModelTestCase):
             assert stored_annotation.user.email == "annotator@example.com"
 
         sw = model.StoredWorkflow()
+        add_object_to_object_session(sw, u)
         sw.user = u
         self.persist(sw)
         persist_and_check_annotation(model.StoredWorkflowAnnotationAssociation, stored_workflow=sw)
@@ -99,6 +105,7 @@ class MappingTests(BaseModelTestCase):
         self.persist(workflow)
 
         ws = model.WorkflowStep()
+        add_object_to_object_session(ws, workflow)
         ws.workflow = workflow
         self.persist(ws)
         persist_and_check_annotation(model.WorkflowStepAnnotationAssociation, workflow_step=ws)
@@ -152,6 +159,7 @@ class MappingTests(BaseModelTestCase):
             assert stored_rating.user.email == user_email
 
         sw = model.StoredWorkflow()
+        add_object_to_object_session(sw, u)
         sw.user = u
         self.persist(sw)
         persist_and_check_rating(model.StoredWorkflowRatingAssociation, sw)
@@ -801,8 +809,16 @@ class MappingTests(BaseModelTestCase):
 
         def workflow_from_steps(steps):
             stored_workflow = model.StoredWorkflow()
+            add_object_to_object_session(stored_workflow, user)
             stored_workflow.user = user
             workflow = model.Workflow()
+
+            if steps:
+                for step in steps:
+                    if get_object_session(step):
+                        add_object_to_object_session(workflow, step)
+                        break
+
             workflow.steps = steps
             workflow.stored_workflow = stored_workflow
             return workflow
@@ -816,6 +832,7 @@ class MappingTests(BaseModelTestCase):
         workflow_step_2 = model.WorkflowStep()
         workflow_step_2.order_index = 1
         workflow_step_2.type = "subworkflow"
+        add_object_to_object_session(workflow_step_2, child_workflow)
         workflow_step_2.subworkflow = child_workflow
 
         workflow_step_1.get_or_add_input("moo1")
@@ -830,6 +847,7 @@ class MappingTests(BaseModelTestCase):
         annotation = model.WorkflowStepAnnotationAssociation()
         annotation.annotation = "Test Step Annotation"
         annotation.user = user
+        add_object_to_object_session(annotation, workflow_step_1)
         annotation.workflow_step = workflow_step_1
         self.persist(annotation)
 
@@ -840,9 +858,11 @@ class MappingTests(BaseModelTestCase):
 
         workflow_invocation = model.WorkflowInvocation()
         workflow_invocation.uuid = invocation_uuid
+        add_object_to_object_session(workflow_invocation, h1)
         workflow_invocation.history = h1
 
         workflow_invocation_step1 = model.WorkflowInvocationStep()
+        add_object_to_object_session(workflow_invocation_step1, workflow_invocation)
         workflow_invocation_step1.workflow_invocation = workflow_invocation
         workflow_invocation_step1.workflow_step = workflow_step_1
 
@@ -850,6 +870,7 @@ class MappingTests(BaseModelTestCase):
         workflow_invocation.attach_subworkflow_invocation_for_step(workflow_step_2, subworkflow_invocation)
 
         workflow_invocation_step2 = model.WorkflowInvocationStep()
+        add_object_to_object_session(workflow_invocation_step2, workflow_invocation)
         workflow_invocation_step2.workflow_invocation = workflow_invocation
         workflow_invocation_step2.workflow_step = workflow_step_2
 
@@ -857,6 +878,7 @@ class MappingTests(BaseModelTestCase):
 
         d1 = self.new_hda(h1, name="1")
         workflow_request_dataset = model.WorkflowRequestToInputDatasetAssociation()
+        add_object_to_object_session(workflow_request_dataset, workflow_invocation)
         workflow_request_dataset.workflow_invocation = workflow_invocation
         workflow_request_dataset.workflow_step = workflow_step_1
         workflow_request_dataset.dataset = d1