diff --git a/client/src/components/Common/DelayedInput.vue b/client/src/components/Common/DelayedInput.vue
index bb7691bbea1c..bb36d5f8fb92 100644
--- a/client/src/components/Common/DelayedInput.vue
+++ b/client/src/components/Common/DelayedInput.vue
@@ -13,7 +13,7 @@
         <b-input-group-append>
             <b-button
                 v-if="enableAdvanced"
-                v-b-tooltip.hover
+                v-b-tooltip.hover.noninteractive
                 aria-haspopup="true"
                 size="sm"
                 :pressed="showAdvanced"
@@ -25,7 +25,7 @@
                 <icon v-else fixed-width icon="angle-double-down" />
             </b-button>
             <b-button
-                v-b-tooltip.hover
+                v-b-tooltip.hover.noninteractive
                 aria-haspopup="true"
                 class="search-clear"
                 size="sm"
diff --git a/client/src/components/History/Content/model/StatesInfo.vue b/client/src/components/History/Content/model/StatesInfo.vue
index 777c4ab35cc1..c12f5644326b 100644
--- a/client/src/components/History/Content/model/StatesInfo.vue
+++ b/client/src/components/History/Content/model/StatesInfo.vue
@@ -46,7 +46,7 @@ function onFilter(value: string) {
         <p>Here are all available item states in Galaxy:</p>
         <p><i>(Note that the colors for each state correspond to content item state colors in the history)</i></p>
         <dl v-for="(state, key, index) in states" :key="index">
-            <b-alert :variant="state.status || 'success'" show>
+            <div :class="['alert', 'content-item', 'alert-' + state.status]" :data-state="key">
                 <dt>
                     <a class="text-decoration-none" href="javascript:void(0)" @click="onFilter(key)"
                         ><code>{{ key }}</code></a
@@ -54,7 +54,7 @@ function onFilter(value: string) {
                     <icon v-if="state.icon" :icon="state.icon" />
                 </dt>
                 <dd>{{ helpText[key] || state.text }}</dd>
-            </b-alert>
+            </div>
         </dl>
     </b-modal>
 </template>
diff --git a/client/src/components/Panels/Buttons/FavoritesButton.vue b/client/src/components/Panels/Buttons/FavoritesButton.vue
index b8e8b2d531fe..cf2d29d51e31 100644
--- a/client/src/components/Panels/Buttons/FavoritesButton.vue
+++ b/client/src/components/Panels/Buttons/FavoritesButton.vue
@@ -1,6 +1,6 @@
 <template>
     <b-button
-        v-b-tooltip.hover
+        v-b-tooltip.hover.noninteractive
         class="panel-header-button-toolbox"
         size="sm"
         variant="link"
diff --git a/client/src/components/Panels/Buttons/PanelViewButton.vue b/client/src/components/Panels/Buttons/PanelViewButton.vue
index 71809a564960..e68a2709b106 100644
--- a/client/src/components/Panels/Buttons/PanelViewButton.vue
+++ b/client/src/components/Panels/Buttons/PanelViewButton.vue
@@ -1,6 +1,6 @@
 <template>
     <b-dropdown
-        v-b-tooltip.hover
+        v-b-tooltip.hover.noninteractive
         right
         title="Show panel options"
         variant="link"
diff --git a/client/src/components/Panels/Common/ToolPanelLabel.vue b/client/src/components/Panels/Common/ToolPanelLabel.vue
index 791925ac8d75..13dc23970c47 100644
--- a/client/src/components/Panels/Common/ToolPanelLabel.vue
+++ b/client/src/components/Panels/Common/ToolPanelLabel.vue
@@ -1,5 +1,5 @@
 <template>
-    <div v-b-tooltip.topright.hover class="tool-panel-label" tabindex="0" :title="description">
+    <div v-b-tooltip.topright.hover.noninteractive class="tool-panel-label" tabindex="0" :title="description">
         {{ definition.text }}
         <ToolPanelLinks :links="definition.links" />
     </div>
diff --git a/client/src/components/Panels/Common/ToolSection.vue b/client/src/components/Panels/Common/ToolSection.vue
index 15cdae2de674..94de1ff9fed9 100644
--- a/client/src/components/Panels/Common/ToolSection.vue
+++ b/client/src/components/Panels/Common/ToolSection.vue
@@ -1,7 +1,7 @@
 <template>
     <div v-if="isSection && hasElements" class="tool-panel-section">
         <div
-            v-b-tooltip.topright.hover
+            v-b-tooltip.topright.hover.noninteractive
             :class="['toolSectionTitle', `tool-menu-section-${sectionName}`]"
             :title="title">
             <a class="title-link" href="javascript:void(0)" @click="toggleMenu()">
diff --git a/client/src/components/Panels/ToolBoxWorkflow.vue b/client/src/components/Panels/ToolBoxWorkflow.vue
index 0c66d78de495..c23f5c7b4e0e 100644
--- a/client/src/components/Panels/ToolBoxWorkflow.vue
+++ b/client/src/components/Panels/ToolBoxWorkflow.vue
@@ -22,9 +22,19 @@
                 placeholder="search tools"
                 :toolbox="workflowTools"
                 :query="query"
+                :query-pending="queryPending"
                 @onQuery="onQuery"
                 @onResults="onResults" />
-            <div v-if="queryTooShort" class="pb-2">
+            <div v-if="closestTerm" class="pb-2">
+                <b-badge class="alert-danger w-100">
+                    Did you mean:
+                    <i>
+                        <a href="javascript:void(0)" @click="onQuery(closestTerm)">{{ closestTerm }}</a>
+                    </i>
+                    ?
+                </b-badge>
+            </div>
+            <div v-else-if="queryTooShort" class="pb-2">
                 <b-badge class="alert-danger w-100">Search string too short!</b-badge>
             </div>
             <div v-else-if="noResults" class="pb-2">
@@ -40,21 +50,21 @@
                     :category="category"
                     tool-key="name"
                     :section-name="category.name"
-                    :query-filter="query"
+                    :query-filter="queryFilter"
                     :disable-filter="true"
                     @onClick="onInsertModule" />
                 <ToolSection
                     v-if="hasDataManagerSection"
                     :key="dataManagerSection.id"
                     :category="dataManagerSection"
-                    :query-filter="query"
+                    :query-filter="queryFilter"
                     :disable-filter="true"
                     @onClick="onInsertTool" />
                 <ToolSection
                     v-for="section in sections"
                     :key="section.id"
                     :category="section"
-                    :query-filter="query"
+                    :query-filter="queryFilter"
                     @onClick="onInsertTool" />
                 <ToolSection
                     v-if="hasWorkflowSection"
@@ -64,7 +74,7 @@
                     :sort-items="false"
                     operation-icon="fa fa-files-o"
                     operation-title="Insert individual steps."
-                    :query-filter="query"
+                    :query-filter="queryFilter"
                     :disable-filter="true"
                     @onClick="onInsertWorkflow"
                     @onOperation="onInsertWorkflowSteps" />
@@ -114,7 +124,10 @@ export default {
     },
     data() {
         return {
+            closestTerm: null,
             query: null,
+            queryPending: false,
+            queryFilter: null,
             results: null,
         };
     },
@@ -167,9 +180,13 @@ export default {
     methods: {
         onQuery(query) {
             this.query = query;
+            this.queryPending = true;
         },
-        onResults(results) {
+        onResults(results, closestTerm = null) {
             this.results = results;
+            this.closestTerm = closestTerm;
+            this.queryFilter = !this.noResults ? this.query : null;
+            this.queryPending = false;
         },
         onInsertTool(tool, evt) {
             evt.preventDefault();
diff --git a/client/src/components/Upload/UploadButton.vue b/client/src/components/Upload/UploadButton.vue
index b890877bd40a..ade8abee45b5 100644
--- a/client/src/components/Upload/UploadButton.vue
+++ b/client/src/components/Upload/UploadButton.vue
@@ -1,7 +1,7 @@
 <template>
     <b-button
         id="activity-upload"
-        v-b-tooltip.hover.bottom
+        v-b-tooltip.hover.noninteractive.bottom
         :aria-label="title | localize"
         :title="title | localize"
         class="upload-button"
diff --git a/client/src/components/Workflow/WorkflowDropdown.vue b/client/src/components/Workflow/WorkflowDropdown.vue
index 5b83a81ffbf8..d308a1aea524 100644
--- a/client/src/components/Workflow/WorkflowDropdown.vue
+++ b/client/src/components/Workflow/WorkflowDropdown.vue
@@ -10,21 +10,21 @@
             @dragend="onDragEnd">
             <Icon icon="caret-down" class="fa-lg" />
             <span class="workflow-dropdown-name">{{ workflow.name }}</span>
+            <span
+                v-if="sourceType.includes('trs')"
+                v-b-tooltip.hover
+                aria-haspopup="true"
+                :title="getWorkflowTooltip(sourceType, workflow)">
+                <Icon fixed-width icon="check" class="mr-1 workflow-trs-icon" />
+            </span>
+            <span
+                v-if="sourceType == 'url'"
+                v-b-tooltip.hover
+                aria-haspopup="true"
+                :title="getWorkflowTooltip(sourceType, workflow)">
+                <Icon fixed-width icon="link" class="mr-1 workflow-external-link" />
+            </span>
         </b-link>
-        <span
-            v-if="sourceType.includes('trs')"
-            v-b-tooltip.hover
-            aria-haspopup="true"
-            :title="getWorkflowTooltip(sourceType, workflow)">
-            <Icon fixed-width icon="check" class="mr-1 workflow-trs-icon" />
-        </span>
-        <span
-            v-if="sourceType == 'url'"
-            v-b-tooltip.hover
-            aria-haspopup="true"
-            :title="getWorkflowTooltip(sourceType, workflow)">
-            <Icon fixed-width icon="link" class="mr-1 workflow-external-link" />
-        </span>
         <p v-if="workflow.description" class="workflow-dropdown-description">
             <TextSummary :description="workflow.description" :show-details.sync="showDetails" />
         </p>
diff --git a/client/src/onload/index.js b/client/src/onload/index.js
index 110bdb6b3966..a658454d3e20 100644
--- a/client/src/onload/index.js
+++ b/client/src/onload/index.js
@@ -8,6 +8,7 @@ import "scss/base.scss";
 import "./publicPath";
 // Default Font
 import "@fontsource/atkinson-hyperlegible";
+import "@fontsource/atkinson-hyperlegible/700.css";
 
 // Client-side configuration variables (based on environment)
 import { library } from "@fortawesome/fontawesome-svg-core";
diff --git a/client/src/schema/schema.ts b/client/src/schema/schema.ts
index ef9413eeb48f..fba346ae618c 100644
--- a/client/src/schema/schema.ts
+++ b/client/src/schema/schema.ts
@@ -4282,8 +4282,7 @@ export interface components {
              * @description The current state of this dataset.
              */
             state: components["schemas"]["DatasetState"];
-            /** Tags */
-            tags: string;
+            tags: components["schemas"]["TagCollection"];
             /**
              * Type
              * @enum {string}
@@ -8494,14 +8493,11 @@ export interface components {
         TaggableItemClass:
             | "History"
             | "HistoryDatasetAssociation"
+            | "HistoryDatasetCollectionAssociation"
             | "LibraryDatasetDatasetAssociation"
             | "Page"
-            | "WorkflowStep"
             | "StoredWorkflow"
-            | "Visualization"
-            | "HistoryDatasetCollection"
-            | "LibraryDatasetCollection"
-            | "Tool";
+            | "Visualization";
         /** ToolDataDetails */
         ToolDataDetails: {
             /**
diff --git a/config/plugins/visualizations/h5web/package.json b/config/plugins/visualizations/h5web/package.json
index 145aa121d70a..bf378c84849c 100644
--- a/config/plugins/visualizations/h5web/package.json
+++ b/config/plugins/visualizations/h5web/package.json
@@ -1,27 +1,23 @@
 {
     "name": "visualization",
-    "version": "0.1.0",
+    "version": "0.2.0",
     "keywords": [
         "galaxy",
-        "visualization"
+        "visualization",
+        "h5web"
     ],
     "license": "AFL-3.0",
     "dependencies": {
-        "@galaxyproject/charts": "^0.0.6",
-        "@h5web/app": "^7.1.0",
-        "backbone": "^1.3.3",
+        "@h5web/app": "^8.0.0",
+        "buffer": "^6.0.3",
         "normalize.css": "^8.0.1",
-        "react": "17.0.2",
-        "react-dom": "17.0.2"
+        "react": "18.2.0",
+        "react-dom": "18.2.0"
     },
     "scripts": {
         "build": "parcel build src/script.js --dist-dir static"
     },
     "devDependencies": {
-        "@types/react": "^17.0.52",
-        "@types/react-dom": "^17.0.18",
-        "buffer": "^5.7.1",
-        "parcel": "^2.0.0",
-        "typescript": "4.9.3"
+        "parcel": "^2.9.3"
     }
 }
diff --git a/config/plugins/visualizations/h5web/src/script.js b/config/plugins/visualizations/h5web/src/script.js
index a596cf214d67..a1a944ec045a 100644
--- a/config/plugins/visualizations/h5web/src/script.js
+++ b/config/plugins/visualizations/h5web/src/script.js
@@ -8,46 +8,51 @@
  * point.
  */
 
-import './styles.css';
-import React, { StrictMode } from 'react'
-import {render as reactRender} from 'react-dom'
-import {App, H5GroveProvider} from '@h5web/app'
+import "./styles.css";
+import React from "react";
+import { createRoot } from "react-dom/client";
+import { App, H5GroveProvider } from "@h5web/app";
 
 /* This will be part of the charts/viz standard lib in 23.1 */
 const slashCleanup = /(\/)+/g;
 function prefixedDownloadUrl(root, path) {
-    return `${root}/${path}`.replace(slashCleanup, "/");
+  return `${root}/${path}`.replace(slashCleanup, "/");
 }
 
 function MyApp(props) {
-    return (
-          <H5GroveProvider
-            url={props.url}
-            filepath={props.name}
-            axiosConfig={{ params: { file: props.name } }}
-          >
-            <App explorerOpen={props.explorer} />
-          </H5GroveProvider>
-      );
+  return (
+    <H5GroveProvider
+      url={props.url}
+      filepath={props.name}
+      axiosConfig={{ params: { file: props.name } }}
+    >
+      <App explorerOpen={props.explorer} />
+    </H5GroveProvider>
+  );
 }
 
 window.bundleEntries = window.bundleEntries || {};
 window.bundleEntries.load = function (options) {
-    var dataset = options.dataset;
-    var settings = options.chart.settings;
-    var explorer = settings.get('explorer');
-    var url = window.location.origin + prefixedDownloadUrl(options.root, "/api/datasets/" + dataset.id + "/content");
-    reactRender(
-      <MyApp
-        url={url}
-        name={dataset.name}
-        filepath={dataset.file_name}
-        explorer={explorer}
-      />,
-      document.getElementById(options.target)
-    )
-    options.chart.state('ok', 'Chart drawn.');
-    options.process.resolve();
+  const dataset = options.dataset;
+  const settings = options.chart.settings;
+  const explorer = settings.get("explorer");
+  const url =
+    window.location.origin +
+    prefixedDownloadUrl(
+      options.root,
+      "/api/datasets/" + dataset.id + "/content"
+    );
+  const root = createRoot(document.getElementById(options.target));
+  root.render(
+    <MyApp
+      url={url}
+      name={dataset.name}
+      filepath={dataset.file_name}
+      explorer={explorer}
+    />
+  );
+  options.chart.state("ok", "Chart drawn.");
+  options.process.resolve();
 };
 
 export default MyApp;
diff --git a/lib/galaxy/app_unittest_utils/galaxy_mock.py b/lib/galaxy/app_unittest_utils/galaxy_mock.py
index 049cad5908ca..462bf7094512 100644
--- a/lib/galaxy/app_unittest_utils/galaxy_mock.py
+++ b/lib/galaxy/app_unittest_utils/galaxy_mock.py
@@ -293,9 +293,13 @@ def __init__(self, app=None, user=None, history=None, **kwargs):
         self.security = self.app.security
         self.history = history
 
-        self.request: Any = Bunch(headers={}, body=None)
+        self.request: Any = Bunch(headers={}, body=None, host="request.host")
         self.response: Any = Bunch(headers={}, set_content_type=lambda i: None)
 
+    @property
+    def tag_handler(self):
+        return self.app.tag_handler
+
     def check_csrf_token(self, payload):
         pass
 
diff --git a/lib/galaxy/datatypes/sequence.py b/lib/galaxy/datatypes/sequence.py
index 5fe01c38f9fc..89746eec5c3e 100644
--- a/lib/galaxy/datatypes/sequence.py
+++ b/lib/galaxy/datatypes/sequence.py
@@ -175,7 +175,7 @@ def do_slow_split(cls, input_datasets, subdir_generator_function, split_params):
     def do_fast_split(cls, input_datasets, toc_file_datasets, subdir_generator_function, split_params):
         data = json.load(open(toc_file_datasets[0].file_name))
         sections = data["sections"]
-        total_sequences = int(0)
+        total_sequences = 0
         for section in sections:
             total_sequences += int(section["sequences"])
         sequences_per_file = cls.get_sequences_per_file(total_sequences, split_params)
@@ -247,7 +247,7 @@ def get_split_commands_with_toc(
         sections = toc_file["sections"]
         result = []
 
-        current_sequence = int(0)
+        current_sequence = 0
         i = 0
         # skip to the section that contains my starting sequence
         while i < len(sections) and start_sequence >= current_sequence + int(sections[i]["sequences"]):
diff --git a/lib/galaxy/dependencies/pinned-lint-requirements.txt b/lib/galaxy/dependencies/pinned-lint-requirements.txt
index 0685b383b397..4349574508b8 100644
--- a/lib/galaxy/dependencies/pinned-lint-requirements.txt
+++ b/lib/galaxy/dependencies/pinned-lint-requirements.txt
@@ -5,6 +5,6 @@ importlib-metadata==4.2.0
 mccabe==0.7.0
 pycodestyle==2.9.1
 pyflakes==2.5.0
-ruff==0.0.280
+ruff==0.0.283
 typing_extensions==4.7.1
 zipp==3.15.0
diff --git a/lib/galaxy/job_execution/actions/post.py b/lib/galaxy/job_execution/actions/post.py
index f8edd4704774..b151d7ba6e3e 100644
--- a/lib/galaxy/job_execution/actions/post.py
+++ b/lib/galaxy/job_execution/actions/post.py
@@ -448,13 +448,13 @@ class TagDatasetAction(DefaultJobAction):
     def execute_on_mapped_over(
         cls, trans, sa_session, action, step_inputs, step_outputs, replacement_dict, final_job_state=None
     ):
-        tag_handler = trans.app.tag_handler.create_tag_handler_session()
         if action.action_arguments:
             tags = [
                 t.replace("#", "name:") if t.startswith("#") else t
                 for t in [t.strip() for t in action.action_arguments.get("tags", "").split(",") if t.strip()]
             ]
-            if tags:
+            if tags and step_outputs:
+                tag_handler = trans.tag_handler
                 for name, step_output in step_outputs.items():
                     if action.output_name == "" or name == action.output_name:
                         cls._execute(tag_handler, trans.user, step_output, tags)
@@ -462,12 +462,12 @@ def execute_on_mapped_over(
     @classmethod
     def execute(cls, app, sa_session, action, job, replacement_dict, final_job_state=None):
         if action.action_arguments:
-            tag_handler = app.tag_handler.create_tag_handler_session()
             tags = [
                 t.replace("#", "name:") if t.startswith("#") else t
                 for t in [t.strip() for t in action.action_arguments.get("tags", "").split(",") if t.strip()]
             ]
             if tags:
+                tag_handler = app.tag_handler.create_tag_handler_session(job.galaxy_session)
                 for dataset_assoc in job.output_datasets:
                     if action.output_name == "" or dataset_assoc.name == action.output_name:
                         cls._execute(tag_handler, job.user, dataset_assoc.dataset, tags)
diff --git a/lib/galaxy/job_execution/output_collect.py b/lib/galaxy/job_execution/output_collect.py
index 826218c42b50..98f5dad4c1f3 100644
--- a/lib/galaxy/job_execution/output_collect.py
+++ b/lib/galaxy/job_execution/output_collect.py
@@ -244,14 +244,14 @@ def __init__(
     @property
     def tag_handler(self):
         if self._tag_handler is None:
-            self._tag_handler = self.app.tag_handler.create_tag_handler_session()
+            self._tag_handler = self.app.tag_handler.create_tag_handler_session(self.job.galaxy_session)
         return self._tag_handler
 
     @property
     def work_context(self):
         from galaxy.work.context import WorkRequestContext
 
-        return WorkRequestContext(self.app, user=self.user)
+        return WorkRequestContext(self.app, user=self.user, galaxy_session=self.job.galaxy_session)
 
     @property
     def sa_session(self) -> ScopedSession:
diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index 786947e08dce..318a410d4221 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -1058,7 +1058,7 @@ def tool_directory(self):
     def job_io(self):
         if self._job_io is None:
             job = self.get_job()
-            work_request = WorkRequestContext(self.app, user=job.user)
+            work_request = WorkRequestContext(self.app, user=job.user, galaxy_session=job.galaxy_session)
             user_context = ProvidesUserFileSourcesUserContext(work_request)
             tool_source = self.tool and self.tool.tool_source.to_string()
             self._job_io = JobIO(
@@ -1913,7 +1913,7 @@ def fail(message=job.info, exception=None):
                     app=self.app,
                     import_options=import_options,
                     user=job.user,
-                    tag_handler=self.app.tag_handler.create_tag_handler_session(),
+                    tag_handler=self.app.tag_handler.create_tag_handler_session(job.galaxy_session),
                 )
                 import_model_store.perform_import(history=job.history, job=job)
                 if job.state == job.states.ERROR:
diff --git a/lib/galaxy/managers/citations.py b/lib/galaxy/managers/citations.py
index 64e8ca91754d..09a4e137cbee 100644
--- a/lib/galaxy/managers/citations.py
+++ b/lib/galaxy/managers/citations.py
@@ -152,12 +152,10 @@ def to_bibtex(self):
                 log.exception("Failed to fetch bibtex for DOI %s", self.__doi)
 
         if self.raw_bibtex is DoiCitation.BIBTEX_UNSET:
-            return """@MISC{{{doi},
-                DOI = {{{doi}}},
+            return f"""@MISC{{{self.__doi},
+                DOI = {{{self.__doi}}},
                 note = {{Failed to fetch BibTeX for DOI.}}
-            }}""".format(
-                doi=self.__doi
-            )
+            }}"""
         else:
             return self.raw_bibtex
 
diff --git a/lib/galaxy/managers/cloud.py b/lib/galaxy/managers/cloud.py
index 671ec216989e..2a26640c9cee 100644
--- a/lib/galaxy/managers/cloud.py
+++ b/lib/galaxy/managers/cloud.py
@@ -86,7 +86,7 @@ def configure_provider(provider, credentials):
             if len(missing_credentials) > 0:
                 raise RequestParameterMissingException(
                     "The following required key(s) are missing from the provided "
-                    "credentials object: {}".format(missing_credentials)
+                    f"credentials object: {missing_credentials}"
                 )
             session_token = credentials.get("SessionToken")
             config = {"aws_access_key": access, "aws_secret_key": secret, "aws_session_token": session_token}
@@ -107,7 +107,7 @@ def configure_provider(provider, credentials):
             if len(missing_credentials) > 0:
                 raise RequestParameterMissingException(
                     "The following required key(s) are missing from the provided "
-                    "credentials object: {}".format(missing_credentials)
+                    f"credentials object: {missing_credentials}"
                 )
 
             config = {
@@ -145,7 +145,7 @@ def configure_provider(provider, credentials):
             if len(missing_credentials) > 0:
                 raise RequestParameterMissingException(
                     "The following required key(s) are missing from the provided "
-                    "credentials object: {}".format(missing_credentials)
+                    f"credentials object: {missing_credentials}"
                 )
             config = {
                 "os_username": username,
@@ -161,8 +161,8 @@ def configure_provider(provider, credentials):
             connection = CloudProviderFactory().create_provider(ProviderList.GCP, config)
         else:
             raise RequestParameterInvalidException(
-                "Unrecognized provider '{}'; the following are the supported "
-                "providers: {}.".format(provider, SUPPORTED_PROVIDERS.keys())
+                f"Unrecognized provider '{provider}'; the following are the supported "
+                f"providers: {SUPPORTED_PROVIDERS.keys()}."
             )
 
         # The authorization-assertion mechanism of Cloudbridge assumes a user has an elevated privileges,
diff --git a/lib/galaxy/managers/collections.py b/lib/galaxy/managers/collections.py
index e9684699b6d4..30776a1d8161 100644
--- a/lib/galaxy/managers/collections.py
+++ b/lib/galaxy/managers/collections.py
@@ -22,6 +22,11 @@
     RequestParameterInvalidException,
 )
 from galaxy.managers.collections_util import validate_input_element_identifiers
+from galaxy.managers.context import (
+    ProvidesAppContext,
+    ProvidesHistoryContext,
+    ProvidesUserContext,
+)
 from galaxy.managers.hdas import (
     HDAManager,
     HistoryDatasetAssociationNoHistoryException,
@@ -35,7 +40,6 @@
 from galaxy.model.dataset_collections.registry import DATASET_COLLECTION_TYPES_REGISTRY
 from galaxy.model.dataset_collections.type_description import COLLECTION_TYPE_DESCRIPTION_FACTORY
 from galaxy.model.mapping import GalaxyModelMapping
-from galaxy.model.tags import GalaxyTagHandler
 from galaxy.schema.schema import DatasetCollectionInstanceType
 from galaxy.schema.tasks import PrepareDatasetCollectionDownload
 from galaxy.security.idencoding import IdEncodingHelper
@@ -65,7 +69,6 @@ def __init__(
         security: IdEncodingHelper,
         hda_manager: HDAManager,
         history_manager: HistoryManager,
-        tag_handler: GalaxyTagHandler,
         ldda_manager: LDDAManager,
         short_term_storage_monitor: ShortTermStorageMonitor,
     ):
@@ -74,15 +77,13 @@ def __init__(
         self.model = model
         self.security = security
         self.short_term_storage_monitor = short_term_storage_monitor
-
         self.hda_manager = hda_manager
         self.history_manager = history_manager
-        self.tag_handler = tag_handler.create_tag_handler_session()
         self.ldda_manager = ldda_manager
 
     def precreate_dataset_collection_instance(
         self,
-        trans,
+        trans: ProvidesHistoryContext,
         parent,
         name,
         structure,
@@ -158,7 +159,7 @@ def precreate_dataset_collection(
 
     def create(
         self,
-        trans,
+        trans: ProvidesHistoryContext,
         parent,
         name,
         collection_type,
@@ -222,7 +223,7 @@ def create(
 
     def _create_instance_for_collection(
         self,
-        trans,
+        trans: ProvidesHistoryContext,
         parent,
         name,
         dataset_collection,
@@ -270,14 +271,14 @@ def _create_instance_for_collection(
         # values.
         if isinstance(tags, list):
             assert implicit_inputs is None, implicit_inputs
-            tags = self.tag_handler.add_tags_from_list(trans.user, dataset_collection_instance, tags, flush=False)
+            tags = trans.tag_handler.add_tags_from_list(trans.user, dataset_collection_instance, tags, flush=False)
         else:
             tags = self._append_tags(dataset_collection_instance, implicit_inputs, tags)
         return self.__persist(dataset_collection_instance, flush=flush)
 
     def create_dataset_collection(
         self,
-        trans,
+        trans: ProvidesHistoryContext,
         collection_type,
         element_identifiers=None,
         elements=None,
@@ -324,7 +325,9 @@ def create_dataset_collection(
         dataset_collection.collection_type = collection_type
         return dataset_collection
 
-    def get_converters_for_collection(self, trans, id, datatypes_registry: Registry, instance_type="history"):
+    def get_converters_for_collection(
+        self, trans: ProvidesHistoryContext, id, datatypes_registry: Registry, instance_type="history"
+    ):
         dataset_collection_instance = self.get_dataset_collection_instance(
             trans, id=id, instance_type=instance_type, check_ownership=True
         )
@@ -360,7 +363,7 @@ def get_converters_for_collection(self, trans, id, datatypes_registry: Registry,
 
     def _element_identifiers_to_elements(
         self,
-        trans,
+        trans: ProvidesHistoryContext,
         collection_type_description,
         element_identifiers,
         hide_source_items=False,
@@ -405,7 +408,7 @@ def _append_tags(self, dataset_collection_instance, implicit_inputs=None, tags=N
     def collection_builder_for(self, dataset_collection):
         return builder.BoundCollectionBuilder(dataset_collection)
 
-    def delete(self, trans, instance_type, id, recursive=False, purge=False):
+    def delete(self, trans: ProvidesHistoryContext, instance_type, id, recursive=False, purge=False):
         dataset_collection_instance = self.get_dataset_collection_instance(
             trans, instance_type, id, check_ownership=True
         )
@@ -431,7 +434,7 @@ def delete(self, trans, instance_type, id, recursive=False, purge=False):
             trans.sa_session.commit()
         return async_result
 
-    def update(self, trans, instance_type, id, payload):
+    def update(self, trans: ProvidesHistoryContext, instance_type, id, payload):
         dataset_collection_instance = self.get_dataset_collection_instance(
             trans, instance_type, id, check_ownership=True
         )
@@ -447,7 +450,15 @@ def update(self, trans, instance_type, id, payload):
         changed = self._set_from_dict(trans, dataset_collection_instance, payload)
         return changed
 
-    def copy(self, trans, parent, source, encoded_source_id, copy_elements=False, dataset_instance_attributes=None):
+    def copy(
+        self,
+        trans: ProvidesHistoryContext,
+        parent,
+        source,
+        encoded_source_id,
+        copy_elements=False,
+        dataset_instance_attributes=None,
+    ):
         """
         PRECONDITION: security checks on ability to add to parent occurred
         during load.
@@ -467,7 +478,7 @@ def copy(self, trans, parent, source, encoded_source_id, copy_elements=False, da
             trans.sa_session.commit()
         return new_hdca
 
-    def _set_from_dict(self, trans, dataset_collection_instance, new_data):
+    def _set_from_dict(self, trans: ProvidesUserContext, dataset_collection_instance, new_data):
         # send what we can down into the model
         changed = dataset_collection_instance.set_from_dict(new_data)
 
@@ -481,10 +492,12 @@ def _set_from_dict(self, trans, dataset_collection_instance, new_data):
         # the api promises a list of changed fields, but tags are not marked as changed to avoid the
         # flush, so we must handle changed tag responses manually
         new_tags = None
-        if "tags" in new_data.keys() and trans.get_user():
+        if "tags" in new_data.keys():
             # set_tags_from_list will flush on its own, no need to add to 'changed' here and incur a second flush.
-            new_tags = self.tag_handler.set_tags_from_list(
-                trans.get_user(), dataset_collection_instance, new_data["tags"]
+            new_tags = trans.tag_handler.set_tags_from_list(
+                trans.user,
+                dataset_collection_instance,
+                new_data["tags"],
             )
 
         if changed.keys():
@@ -634,14 +647,14 @@ def __load_element(self, trans, element_identifier, hide_source_items, copy_elem
                 hda.id, user=trans.user, current_history=history or trans.history
             ):
                 hda.visible = False
-            self.tag_handler.apply_item_tags(user=trans.user, item=element, tags_str=tag_str, flush=False)
+            trans.tag_handler.apply_item_tags(user=trans.user, item=element, tags_str=tag_str, flush=False)
             return element
         elif src_type == "ldda":
             element2 = self.ldda_manager.get(trans, element_id, check_accessible=True)
             element3 = element2.to_history_dataset_association(
                 history or trans.history, add_to_history=True, visible=not hide_source_items
             )
-            self.tag_handler.apply_item_tags(user=trans.user, item=element3, tags_str=tag_str, flush=False)
+            trans.tag_handler.apply_item_tags(user=trans.user, item=element3, tags_str=tag_str, flush=False)
             return element3
         elif src_type == "hdca":
             # TODO: Option to copy? Force copy? Copy or allow if not owned?
@@ -658,18 +671,18 @@ def match_collections(self, collections_to_match):
 
     @overload
     def get_dataset_collection_instance(
-        self, trans, instance_type: Literal["history"], id, **kwds: Any
+        self, trans: ProvidesHistoryContext, instance_type: Literal["history"], id, **kwds: Any
     ) -> model.HistoryDatasetCollectionAssociation:
         ...
 
     @overload
     def get_dataset_collection_instance(
-        self, trans, instance_type: Literal["library"], id, **kwds: Any
+        self, trans: ProvidesHistoryContext, instance_type: Literal["library"], id, **kwds: Any
     ) -> model.LibraryDatasetCollectionAssociation:
         ...
 
     def get_dataset_collection_instance(
-        self, trans, instance_type: DatasetCollectionInstanceType, id, **kwds: Any
+        self, trans: ProvidesHistoryContext, instance_type: DatasetCollectionInstanceType, id, **kwds: Any
     ) -> Union[model.HistoryDatasetCollectionAssociation, model.LibraryDatasetCollectionAssociation]:
         """ """
         if instance_type == "history":
@@ -784,7 +797,7 @@ def __init_rule_data(self, elements, collection_type_description, parent_identif
         return data, sources
 
     def __get_history_collection_instance(
-        self, trans, id, check_ownership=False, check_accessible=True
+        self, trans: ProvidesHistoryContext, id, check_ownership=False, check_accessible=True
     ) -> model.HistoryDatasetCollectionAssociation:
         instance_id = trans.app.security.decode_id(id) if isinstance(id, str) else id
         collection_instance = trans.sa_session.query(trans.app.model.HistoryDatasetCollectionAssociation).get(
@@ -792,6 +805,7 @@ def __get_history_collection_instance(
         )
         if not collection_instance:
             raise RequestParameterInvalidException("History dataset collection association not found")
+        # TODO: that sure looks like a bug, we can't check ownership using the history of the object we're checking ownership for ...
         history = getattr(trans, "history", collection_instance.history)
         if check_ownership:
             self.history_manager.error_unless_owner(collection_instance.history, trans.user, current_history=history)
@@ -802,7 +816,7 @@ def __get_history_collection_instance(
         return collection_instance
 
     def __get_library_collection_instance(
-        self, trans, id, check_ownership=False, check_accessible=True
+        self, trans: ProvidesHistoryContext, id, check_ownership=False, check_accessible=True
     ) -> model.LibraryDatasetCollectionAssociation:
         if check_ownership:
             raise NotImplementedError(
@@ -823,7 +837,7 @@ def __get_library_collection_instance(
                 )
         return collection_instance
 
-    def get_collection_contents(self, trans, parent_id, limit=None, offset=None):
+    def get_collection_contents(self, trans: ProvidesAppContext, parent_id, limit=None, offset=None):
         """Find first level of collection contents by containing collection parent_id"""
         contents_qry = self._get_collection_contents_qry(parent_id, limit=limit, offset=offset)
         contents = contents_qry.with_session(trans.sa_session()).all()
diff --git a/lib/galaxy/managers/context.py b/lib/galaxy/managers/context.py
index 4ee1a1337cf4..2a19c85ef533 100644
--- a/lib/galaxy/managers/context.py
+++ b/lib/galaxy/managers/context.py
@@ -39,6 +39,7 @@
 from json import dumps
 from typing import (
     Callable,
+    cast,
     List,
     Optional,
 )
@@ -49,15 +50,18 @@
 )
 from galaxy.model import (
     Dataset,
+    GalaxySession,
     History,
     HistoryDatasetAssociation,
     Role,
+    User,
 )
 from galaxy.model.base import (
     ModelMapping,
     transaction,
 )
 from galaxy.model.scoped_session import galaxy_scoped_session
+from galaxy.model.tags import GalaxyTagHandlerSession
 from galaxy.schema.tasks import RequestUser
 from galaxy.security.idencoding import IdEncodingHelper
 from galaxy.security.vault import UserVaultWrapper
@@ -206,6 +210,15 @@ class ProvidesUserContext(ProvidesAppContext):
     properties.
     """
 
+    galaxy_session: Optional[GalaxySession] = None
+    _tag_handler: Optional[GalaxyTagHandlerSession] = None
+
+    @property
+    def tag_handler(self):
+        if self._tag_handler is None:
+            self._tag_handler = self.app.tag_handler.create_tag_handler_session(self.galaxy_session)
+        return self._tag_handler
+
     @property
     def async_request_user(self) -> RequestUser:
         if self.user is None:
@@ -221,6 +234,10 @@ def user_vault(self):
         """Provide access to a user's personal vault."""
         return UserVaultWrapper(self.app.vault, self.user)
 
+    def get_user(self) -> Optional[User]:
+        user = cast(Optional[User], self.user or self.galaxy_session and self.galaxy_session.user)
+        return user
+
     @property
     def anonymous(self) -> bool:
         return self.user is None
diff --git a/lib/galaxy/managers/hdas.py b/lib/galaxy/managers/hdas.py
index 5b8aee4aec7e..589c36b4092c 100644
--- a/lib/galaxy/managers/hdas.py
+++ b/lib/galaxy/managers/hdas.py
@@ -42,7 +42,6 @@
 )
 from galaxy.model.base import transaction
 from galaxy.model.deferred import materializer_factory
-from galaxy.model.tags import GalaxyTagHandler
 from galaxy.schema.schema import DatasetSourceType
 from galaxy.schema.storage_cleaner import (
     CleanableItemsSummary,
@@ -71,7 +70,6 @@ class HistoryDatasetAssociationNoHistoryException(Exception):
 class HDAManager(
     datasets.DatasetAssociationManager,
     secured.OwnableManagerMixin,
-    taggable.TaggableManagerMixin,
     annotatable.AnnotatableManagerMixin,
 ):
     """
@@ -93,14 +91,12 @@ def __init__(
         app: MinimalManagerApp,
         user_manager: users.UserManager,
         ldda_manager: lddas.LDDAManager,
-        tag_handler: GalaxyTagHandler,
     ):
         """
         Set up and initialize other managers needed by hdas.
         """
         super().__init__(app)
         self.user_manager = user_manager
-        self.tag_handler = tag_handler
         self.ldda_manager = ldda_manager
 
     def get_owned_ids(self, object_ids, history=None):
@@ -708,7 +704,6 @@ class HDADeserializer(
     def __init__(self, app: MinimalManagerApp):
         super().__init__(app)
         self.hda_manager = self.manager
-        self.tag_handler = app.tag_handler
 
     def add_deserializers(self):
         super().add_deserializers()
diff --git a/lib/galaxy/managers/hdcas.py b/lib/galaxy/managers/hdcas.py
index d30663009d20..c3136f03d941 100644
--- a/lib/galaxy/managers/hdcas.py
+++ b/lib/galaxy/managers/hdcas.py
@@ -59,7 +59,6 @@ class HDCAManager(
     secured.AccessibleManagerMixin,
     secured.OwnableManagerMixin,
     deletable.PurgableManagerMixin,
-    taggable.TaggableManagerMixin,
     annotatable.AnnotatableManagerMixin,
 ):
     """
diff --git a/lib/galaxy/managers/library_datasets.py b/lib/galaxy/managers/library_datasets.py
index 1f06a6cfb4f3..091946714041 100644
--- a/lib/galaxy/managers/library_datasets.py
+++ b/lib/galaxy/managers/library_datasets.py
@@ -12,7 +12,7 @@
     RequestParameterInvalidException,
 )
 from galaxy.managers import datasets
-from galaxy.model import tags
+from galaxy.managers.context import ProvidesUserContext
 from galaxy.model.base import transaction
 from galaxy.structured_app import MinimalManagerApp
 from galaxy.util import validation
@@ -27,7 +27,6 @@ class LibraryDatasetsManager(datasets.DatasetAssociationManager):
 
     def __init__(self, app: MinimalManagerApp):
         self.app = app
-        self.tag_handler = tags.GalaxyTagHandler(app.model.context)
 
     def get(self, trans, decoded_library_dataset_id, check_accessible=True):
         """
@@ -82,7 +81,7 @@ def update(self, trans, ld, payload):
         self._set_from_dict(trans, ldda, payload)
         return ld
 
-    def _set_from_dict(self, trans, ldda, new_data):
+    def _set_from_dict(self, trans: ProvidesUserContext, ldda, new_data):
         changed = False
         new_name = new_data.get("name", None)
         if new_name is not None and new_name != ldda.name:
@@ -109,10 +108,10 @@ def _set_from_dict(self, trans, ldda, new_data):
             changed = True
         new_tags = new_data.get("tags", None)
         if new_tags is not None and new_tags != ldda.tags:
-            self.tag_handler.delete_item_tags(item=ldda, user=trans.user)
-            tag_list = self.tag_handler.parse_tags_list(new_tags)
+            trans.tag_handler.delete_item_tags(item=ldda, user=trans.user)
+            tag_list = trans.tag_handler.parse_tags_list(new_tags)
             for tag in tag_list:
-                self.tag_handler.apply_item_tag(item=ldda, user=trans.user, name=tag[0], value=tag[1])
+                trans.tag_handler.apply_item_tag(item=ldda, user=trans.user, name=tag[0], value=tag[1])
             changed = True
         if changed:
             ldda.update_parent_folder_update_times()
@@ -251,7 +250,7 @@ def serialize(self, trans, ld):
             current_user_roles, ld
         )
         rval["is_unrestricted"] = trans.app.security_agent.dataset_is_public(ldda.dataset)
-        rval["tags"] = self.tag_handler.get_tags_str(ldda.tags)
+        rval["tags"] = trans.tag_handler.get_tags_list(ldda.tags)
 
         #  Manage dataset permission is always attached to the dataset itself, not the the ld or ldda to maintain consistency
         rval["can_user_manage"] = trans.user_is_admin or trans.app.security_agent.can_manage_dataset(
diff --git a/lib/galaxy/managers/notification.py b/lib/galaxy/managers/notification.py
index 3e12fada5dd5..7a8180bb7aaf 100644
--- a/lib/galaxy/managers/notification.py
+++ b/lib/galaxy/managers/notification.py
@@ -323,7 +323,7 @@ def _create_notification_model(self, payload: NotificationCreateData):
             payload.source,
             payload.category,
             payload.variant,
-            payload.content.json(),
+            payload.content.dict(),
         )
         notification.publication_time = payload.publication_time
         notification.expiration_time = payload.expiration_time
diff --git a/lib/galaxy/managers/sharable.py b/lib/galaxy/managers/sharable.py
index 3c1f908da134..92499743b3d0 100644
--- a/lib/galaxy/managers/sharable.py
+++ b/lib/galaxy/managers/sharable.py
@@ -58,7 +58,6 @@ class SharableModelManager(
     base.ModelManager,
     secured.OwnableManagerMixin,
     secured.AccessibleManagerMixin,
-    taggable.TaggableManagerMixin,
     annotatable.AnnotatableManagerMixin,
     ratable.RatableManagerMixin,
 ):
diff --git a/lib/galaxy/managers/taggable.py b/lib/galaxy/managers/taggable.py
index bcd9e7388190..fae4b6440185 100644
--- a/lib/galaxy/managers/taggable.py
+++ b/lib/galaxy/managers/taggable.py
@@ -6,7 +6,7 @@
 
 import logging
 import re
-from typing import Type
+from typing import Optional
 
 from sqlalchemy import (
     func,
@@ -14,8 +14,8 @@
 )
 
 from galaxy import model
+from galaxy.managers.context import ProvidesUserContext
 from galaxy.model.tags import GalaxyTagHandler
-from galaxy.util import unicodify
 from .base import (
     ModelValidator,
     raise_filter_err,
@@ -42,42 +42,6 @@ def _tags_to_strings(item):
     return sorted(tag_list, key=lambda str: re.sub("^name:", "#", str))
 
 
-def _tags_from_strings(item, tag_handler, new_tags_list, user=None):
-    # TODO: have to assume trans.user here...
-    if not user:
-        # raise galaxy_exceptions.RequestParameterMissingException( 'User required for tags on ' + str( item ) )
-        # TODO: this becomes a 'silent failure' - no tags are set. This is a questionable approach but
-        # I haven't found a better one for anon users copying items with tags
-        return
-    # TODO: duped from tags manager - de-dupe when moved to taggable mixin
-    tag_handler.delete_item_tags(user, item)
-    new_tags_str = ",".join(new_tags_list)
-    tag_handler.apply_item_tags(user, item, unicodify(new_tags_str, "utf-8"))
-    # TODO:!! does the creation of new_tags_list mean there are now more and more unused tag rows in the db?
-
-
-class TaggableManagerMixin:
-    tag_assoc: Type[model.ItemTagAssociation]
-    tag_handler: GalaxyTagHandler
-
-    # TODO: most of this can be done by delegating to the GalaxyTagHandler?
-    def get_tags(self, item):
-        """
-        Return a list of tag strings.
-        """
-        return _tags_to_strings(item)
-
-    def set_tags(self, item, new_tags, user=None):
-        """
-        Set an `item`'s tags from a list of strings.
-        """
-        return _tags_from_strings(item, self.tag_handler, new_tags, user=user)
-
-    # def tags_by_user( self, user, **kwargs ):
-    # TODO: here or GalaxyTagHandler
-    #    pass
-
-
 class TaggableSerializerMixin:
     def add_serializers(self):
         self.serializers["tags"] = self.serialize_tags
@@ -96,14 +60,16 @@ class TaggableDeserializerMixin:
     def add_deserializers(self):
         self.deserializers["tags"] = self.deserialize_tags
 
-    def deserialize_tags(self, item, key, val, user=None, **context):
+    def deserialize_tags(
+        self, item, key, val, *, user: Optional[model.User] = None, trans: ProvidesUserContext, **context
+    ):
         """
         Make sure `val` is a valid list of tag strings and assign them.
 
         Note: this will erase any previous tags.
         """
         new_tags_list = self.validate.basestring_list(key, val)
-        _tags_from_strings(item, self.tag_handler, new_tags_list, user=user)
+        trans.tag_handler.set_tags_from_list(user=user, item=item, new_tags_list=new_tags_list)
         return item.tags
 
 
diff --git a/lib/galaxy/managers/tags.py b/lib/galaxy/managers/tags.py
index 5eec0e6b53e7..1bf1ec953392 100644
--- a/lib/galaxy/managers/tags.py
+++ b/lib/galaxy/managers/tags.py
@@ -4,7 +4,6 @@
 from pydantic import Field
 
 from galaxy.managers.context import ProvidesUserContext
-from galaxy.model import ItemTagAssociation
 from galaxy.model.base import transaction
 from galaxy.model.tags import GalaxyTagHandlerSession
 from galaxy.schema.fields import DecodedDatabaseIdField
@@ -13,10 +12,15 @@
     TagCollection,
 )
 
-taggable_item_names = {item: item for item in ItemTagAssociation.associated_item_names}
-# This Enum is generated dynamically and mypy can not statically infer its real type,
-# so it should be ignored. See: https://github.com/python/mypy/issues/4865#issuecomment-592560696
-TaggableItemClass = Enum("TaggableItemClass", taggable_item_names)  # type: ignore[misc]
+
+class TaggableItemClass(Enum):
+    History = "History"
+    HistoryDatasetAssociation = "HistoryDatasetAssociation"
+    HistoryDatasetCollectionAssociation = "HistoryDatasetCollectionAssociation"
+    LibraryDatasetDatasetAssociation = "LibraryDatasetDatasetAssociation"
+    Page = "Page"
+    StoredWorkflow = "StoredWorkflow"
+    Visualization = "Visualization"
 
 
 class ItemTagsPayload(Model):
@@ -42,17 +46,15 @@ class TagsManager:
 
     def update(self, trans: ProvidesUserContext, payload: ItemTagsPayload) -> None:
         """Apply a new set of tags to an item; previous tags are deleted."""
-        sa_session = trans.sa_session
         user = trans.user
-        tag_handler = GalaxyTagHandlerSession(sa_session)
         new_tags: Optional[str] = None
         if payload.item_tags and len(payload.item_tags.__root__) > 0:
             new_tags = ",".join(payload.item_tags.__root__)
-        item = self._get_item(tag_handler, payload)
-        tag_handler.delete_item_tags(user, item)
-        tag_handler.apply_item_tags(user, item, new_tags)
-        with transaction(sa_session):
-            sa_session.commit()
+        item = self._get_item(trans.tag_handler, payload)
+        trans.tag_handler.delete_item_tags(user, item)
+        trans.tag_handler.apply_item_tags(user, item, new_tags)
+        with transaction(trans.sa_session):
+            trans.sa_session.commit()
 
     def _get_item(self, tag_handler: GalaxyTagHandlerSession, payload: ItemTagsPayload):
         """
diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py
index 077165451be7..a268c4ce73c4 100644
--- a/lib/galaxy/managers/workflows.py
+++ b/lib/galaxy/managers/workflows.py
@@ -639,7 +639,7 @@ def build_workflow_from_raw_description(
             annotation = sanitize_html(data["annotation"])
             self.add_item_annotation(trans.sa_session, stored.user, stored, annotation)
         workflow_tags = data.get("tags", [])
-        trans.app.tag_handler.set_tags_from_list(user=trans.user, item=stored, new_tags_list=workflow_tags)
+        trans.tag_handler.set_tags_from_list(user=trans.user, item=stored, new_tags_list=workflow_tags)
 
         # Persist
         trans.sa_session.add(stored)
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 2ad0ac59b707..89af15735e0e 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -119,6 +119,7 @@
 import galaxy.util
 from galaxy.model.base import transaction
 from galaxy.model.custom_types import (
+    DoubleEncodedJsonType,
     JSONType,
     MetadataType,
     MutableJSONType,
@@ -2782,7 +2783,9 @@ class Notification(Base, Dictifiable, RepresentById):
     variant = Column(
         String(16), index=True
     )  # Defines the 'importance' of the notification ('info', 'warning', 'urgent', etc.). Used for filtering, highlight rendering, etc
-    content = Column(JSONType)  # Structured content in JSON. Depending on the category of the notification
+    # A bug in early 23.1 led to values being stored as json string, so we use this special type to process the result value twice.
+    # content should always be a dict
+    content = Column(DoubleEncodedJsonType)
 
     user_notification_associations = relationship("UserNotificationAssociation", back_populates="notification")
 
@@ -9821,13 +9824,11 @@ def __str__(self):
 class ItemTagAssociation(Dictifiable):
     dict_collection_visible_keys = ["id", "user_tname", "user_value"]
     dict_element_visible_keys = dict_collection_visible_keys
-    associated_item_names: List[str] = []
     user_tname: Column
     user_value = Column(TrimmedString(255), index=True)
 
     def __init_subclass__(cls, **kwargs):
         super().__init_subclass__(**kwargs)
-        cls.associated_item_names.append(cls.__name__.replace("TagAssociation", ""))
 
     def copy(self, cls=None):
         if cls:
diff --git a/lib/galaxy/model/custom_types.py b/lib/galaxy/model/custom_types.py
index 8f292a792a43..8152802adff8 100644
--- a/lib/galaxy/model/custom_types.py
+++ b/lib/galaxy/model/custom_types.py
@@ -116,6 +116,17 @@ def compare_values(self, x, y):
         return x == y
 
 
+class DoubleEncodedJsonType(JSONType):
+    def process_result_value(self, value, dialect):
+        value = super().process_result_value(value, dialect)
+        if isinstance(value, str):
+            try:
+                return json.loads(value)
+            except ValueError:
+                return value
+        return value
+
+
 class MutableJSONType(JSONType):
     """Associated with MutationObj"""
 
diff --git a/lib/galaxy/model/store/__init__.py b/lib/galaxy/model/store/__init__.py
index be285bbce3d5..8be4bbe25c6a 100644
--- a/lib/galaxy/model/store/__init__.py
+++ b/lib/galaxy/model/store/__init__.py
@@ -2936,7 +2936,7 @@ def source_to_import_store(
     else:
         source_uri: str = str(source)
         delete = False
-        tag_handler = app.tag_handler.create_tag_handler_session()
+        tag_handler = app.tag_handler.create_tag_handler_session(galaxy_session=None)
         if source_uri.startswith("file://"):
             source_uri = source_uri[len("file://") :]
         if "://" in source_uri:
diff --git a/lib/galaxy/model/store/discover.py b/lib/galaxy/model/store/discover.py
index 19a4c1e84194..832357acb45b 100644
--- a/lib/galaxy/model/store/discover.py
+++ b/lib/galaxy/model/store/discover.py
@@ -60,6 +60,9 @@ class ModelPersistenceContext(metaclass=abc.ABCMeta):
     max_discovered_files = float("inf")
     discovered_file_count: int
 
+    def get_job(self) -> Optional[galaxy.model.Job]:
+        return getattr(self, "job", None)
+
     def create_dataset(
         self,
         ext,
@@ -161,7 +164,7 @@ def create_dataset(
             primary_data.created_from_basename = created_from_basename
 
         if tag_list:
-            job = getattr(self, "job", None)
+            job = self.get_job()
             self.tag_handler.add_tags_from_list(job and job.user, primary_data, tag_list, flush=False)
 
         # If match specified a name use otherwise generate one from
@@ -573,7 +576,7 @@ def __init__(self, object_store, export_store: "ModelExportStore", working_direc
 
     @property
     def tag_handler(self):
-        return GalaxySessionlessTagHandler(self.sa_session)
+        return GalaxySessionlessTagHandler(self.sa_session, galaxy_session=None)
 
     @property
     def sa_session(self):
diff --git a/lib/galaxy/model/tags.py b/lib/galaxy/model/tags.py
index ae6a6c7dd2b1..1b378a534643 100644
--- a/lib/galaxy/model/tags.py
+++ b/lib/galaxy/model/tags.py
@@ -5,6 +5,7 @@
     List,
     Optional,
     Tuple,
+    TYPE_CHECKING,
 )
 
 from sqlalchemy.exc import IntegrityError
@@ -13,6 +14,7 @@
 from sqlalchemy.sql.expression import func
 
 import galaxy.model
+from galaxy.exceptions import ItemOwnershipException
 from galaxy.model.base import transaction
 from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.util import (
@@ -20,6 +22,13 @@
     unicodify,
 )
 
+if TYPE_CHECKING:
+    from galaxy.model import (
+        GalaxySession,
+        Tag,
+        User,
+    )
+
 log = logging.getLogger(__name__)
 
 
@@ -36,7 +45,7 @@ class TagHandler:
     Manages CRUD operations related to tagging objects.
     """
 
-    def __init__(self, sa_session: galaxy_scoped_session) -> None:
+    def __init__(self, sa_session: galaxy_scoped_session, galaxy_session=None) -> None:
         self.sa_session = sa_session
         # Minimum tag length.
         self.min_tag_len = 1
@@ -50,10 +59,13 @@ def __init__(self, sa_session: galaxy_scoped_session) -> None:
         self.key_value_separators = "=:"
         # Initialize with known classes - add to this in subclasses.
         self.item_tag_assoc_info: Dict[str, ItemTagAssocInfo] = {}
+        # Can't include type annotation in signature, because lagom will attempt to look up
+        # GalaxySession, but can't find it due to the circular import
+        self.galaxy_session: Optional["GalaxySession"] = galaxy_session
 
-    def create_tag_handler_session(self):
+    def create_tag_handler_session(self, galaxy_session: Optional["GalaxySession"]):
         # Creates a transient tag handler that avoids repeated flushes
-        return GalaxyTagHandlerSession(self.sa_session)
+        return GalaxyTagHandlerSession(self.sa_session, galaxy_session=galaxy_session)
 
     def add_tags_from_list(self, user, item, new_tags_list, flush=True):
         new_tags_set = set(new_tags_list)
@@ -68,7 +80,13 @@ def remove_tags_from_list(self, user, item, tag_to_remove_list, flush=True):
             tags_set -= tag_to_remove_set
         return self.set_tags_from_list(user, item, tags_set, flush=flush)
 
-    def set_tags_from_list(self, user, item, new_tags_list, flush=True):
+    def set_tags_from_list(
+        self,
+        user,
+        item,
+        new_tags_list,
+        flush=True,
+    ):
         # precondition: item is already security checked against user
         # precondition: incoming tags is a list of sanitized/formatted strings
         self.delete_item_tags(user, item)
@@ -125,8 +143,9 @@ def get_tool_tags(self):
             tags.append(self.get_tag_by_id(tag_id))
         return tags
 
-    def remove_item_tag(self, user, item, tag_name):
+    def remove_item_tag(self, user: "User", item, tag_name: str):
         """Remove a tag from an item."""
+        self._ensure_user_owns_item(user, item)
         # Get item tag association.
         item_tag_assoc = self._get_item_tag_assoc(user, item, tag_name)
         # Remove association.
@@ -137,8 +156,9 @@ def remove_item_tag(self, user, item, tag_name):
             return True
         return False
 
-    def delete_item_tags(self, user, item):
+    def delete_item_tags(self, user: Optional["User"], item):
         """Delete tags from an item."""
+        self._ensure_user_owns_item(user, item)
         # Delete item-tag associations.
         for tag in item.tags:
             if tag.id:
@@ -147,6 +167,31 @@ def delete_item_tags(self, user, item):
         # Delete tags from item.
         del item.tags[:]
 
+    def _ensure_user_owns_item(self, user: Optional["User"], item):
+        """Raises exception if user does not own item.
+        Notice that even admin users cannot directly modify tags on items they do not own.
+        To modify tags on items they don't own, admin users must impersonate the item's owner.
+        """
+        if getattr(item, "id", None) is None:
+            # Item is not persisted, likely it is being copied from an existing, so no need
+            # to check ownership at this point.
+            return
+        # Prefer checking ownership via history (or associated history).
+        # When checking multiple items in batch this should save a few lazy-loads
+        is_owner = False
+        history = item if isinstance(item, galaxy.model.History) else getattr(item, "history", None)
+        if not user:
+            if self.galaxy_session and history:
+                # anon users can only tag histories and history items,
+                # and should only have a single history
+                if history == self.galaxy_session.current_history:
+                    return
+            raise ItemOwnershipException("User does not own item.")
+        user_id = history.user_id if history else getattr(item, "user_id", None)
+        is_owner = user_id == user.id
+        if not is_owner:
+            raise ItemOwnershipException("User does not own item.")
+
     def item_has_tag(self, user, item, tag):
         """Returns true if item is has a given tag."""
         # Get tag name.
@@ -162,7 +207,15 @@ def item_has_tag(self, user, item, tag):
             return True
         return False
 
-    def apply_item_tag(self, user, item, name, value=None, flush=True):
+    def apply_item_tag(
+        self,
+        user: Optional["User"],
+        item,
+        name,
+        value=None,
+        flush=True,
+    ):
+        self._ensure_user_owns_item(user, item)
         # Use lowercase name for searching/creating tag.
         if name is None:
             return
@@ -197,27 +250,40 @@ def apply_item_tag(self, user, item, name, value=None, flush=True):
                 self.sa_session.commit()
         return item_tag_assoc
 
-    def apply_item_tags(self, user, item, tags_str, flush=True):
+    def apply_item_tags(
+        self,
+        user: Optional["User"],
+        item,
+        tags_str: Optional[str],
+        flush=True,
+    ):
         """Apply tags to an item."""
+        self._ensure_user_owns_item(user, item)
         # Parse tags.
         parsed_tags = self.parse_tags(tags_str)
         # Apply each tag.
         for name, value in parsed_tags:
             self.apply_item_tag(user, item, name, value, flush=flush)
 
-    def get_tags_str(self, tags):
-        """Build a string from an item's tags."""
-        # Return empty string if there are no tags.
+    def get_tags_list(self, tags) -> List[str]:
+        """Build a list of tags from an item's tags."""
+        # Return empty list if there are no tags.
         if not tags:
-            return ""
-        # Create string of tags.
-        tags_str_list = list()
+            return []
+        # Create list of tags.
+        tags_list: List[str] = []
         for tag in tags:
             tag_str = tag.user_tname
             if tag.value is not None:
                 tag_str += f":{tag.user_value}"
-            tags_str_list.append(tag_str)
-        return ", ".join(tags_str_list)
+            tags_list.append(tag_str)
+        return tags_list
+
+    def get_tags_str(self, tags):
+        """Build a string from an item's tags."""
+        tags_list = self.get_tags_list(tags)
+        # Return empty string if there are no tags.
+        return ", ".join(tags_list)
 
     def get_tag_by_id(self, tag_id):
         """Get a Tag object from a tag id."""
@@ -377,47 +443,58 @@ def _get_name_value_pair(self, tag_str) -> List[Optional[str]]:
 
 
 class GalaxyTagHandler(TagHandler):
-    def __init__(self, sa_session: galaxy_scoped_session):
+    _item_tag_assoc_info: Dict[str, ItemTagAssocInfo] = {}
+
+    def __init__(self, sa_session: galaxy_scoped_session, galaxy_session=None):
+        TagHandler.__init__(self, sa_session, galaxy_session=galaxy_session)
+        if not GalaxyTagHandler._item_tag_assoc_info:
+            GalaxyTagHandler.init_tag_associations()
+        self.item_tag_assoc_info = GalaxyTagHandler._item_tag_assoc_info
+
+    @classmethod
+    def init_tag_associations(cls):
         from galaxy import model
 
-        TagHandler.__init__(self, sa_session)
-        self.item_tag_assoc_info["History"] = ItemTagAssocInfo(
-            model.History, model.HistoryTagAssociation, model.HistoryTagAssociation.history_id
-        )
-        self.item_tag_assoc_info["HistoryDatasetAssociation"] = ItemTagAssocInfo(
-            model.HistoryDatasetAssociation,
-            model.HistoryDatasetAssociationTagAssociation,
-            model.HistoryDatasetAssociationTagAssociation.history_dataset_association_id,
-        )
-        self.item_tag_assoc_info["HistoryDatasetCollectionAssociation"] = ItemTagAssocInfo(
-            model.HistoryDatasetCollectionAssociation,
-            model.HistoryDatasetCollectionTagAssociation,
-            model.HistoryDatasetCollectionTagAssociation.history_dataset_collection_id,
-        )
-        self.item_tag_assoc_info["LibraryDatasetDatasetAssociation"] = ItemTagAssocInfo(
-            model.LibraryDatasetDatasetAssociation,
-            model.LibraryDatasetDatasetAssociationTagAssociation,
-            model.LibraryDatasetDatasetAssociationTagAssociation.library_dataset_dataset_association_id,
-        )
-        self.item_tag_assoc_info["Page"] = ItemTagAssocInfo(
-            model.Page, model.PageTagAssociation, model.PageTagAssociation.page_id
-        )
-        self.item_tag_assoc_info["StoredWorkflow"] = ItemTagAssocInfo(
-            model.StoredWorkflow,
-            model.StoredWorkflowTagAssociation,
-            model.StoredWorkflowTagAssociation.stored_workflow_id,
-        )
-        self.item_tag_assoc_info["Visualization"] = ItemTagAssocInfo(
-            model.Visualization, model.VisualizationTagAssociation, model.VisualizationTagAssociation.visualization_id
-        )
+        cls._item_tag_assoc_info = {
+            "History": ItemTagAssocInfo(
+                model.History, model.HistoryTagAssociation, model.HistoryTagAssociation.history_id
+            ),
+            "HistoryDatasetAssociation": ItemTagAssocInfo(
+                model.HistoryDatasetAssociation,
+                model.HistoryDatasetAssociationTagAssociation,
+                model.HistoryDatasetAssociationTagAssociation.history_dataset_association_id,
+            ),
+            "HistoryDatasetCollectionAssociation": ItemTagAssocInfo(
+                model.HistoryDatasetCollectionAssociation,
+                model.HistoryDatasetCollectionTagAssociation,
+                model.HistoryDatasetCollectionTagAssociation.history_dataset_collection_id,
+            ),
+            "LibraryDatasetDatasetAssociation": ItemTagAssocInfo(
+                model.LibraryDatasetDatasetAssociation,
+                model.LibraryDatasetDatasetAssociationTagAssociation,
+                model.LibraryDatasetDatasetAssociationTagAssociation.library_dataset_dataset_association_id,
+            ),
+            "Page": ItemTagAssocInfo(model.Page, model.PageTagAssociation, model.PageTagAssociation.page_id),
+            "StoredWorkflow": ItemTagAssocInfo(
+                model.StoredWorkflow,
+                model.StoredWorkflowTagAssociation,
+                model.StoredWorkflowTagAssociation.stored_workflow_id,
+            ),
+            "Visualization": ItemTagAssocInfo(
+                model.Visualization,
+                model.VisualizationTagAssociation,
+                model.VisualizationTagAssociation.visualization_id,
+            ),
+        }
+        return cls._item_tag_assoc_info
 
 
 class GalaxyTagHandlerSession(GalaxyTagHandler):
     """Like GalaxyTagHandler, but avoids one flush per created tag."""
 
-    def __init__(self, sa_session):
-        super().__init__(sa_session)
-        self.created_tags = {}
+    def __init__(self, sa_session, galaxy_session: Optional["GalaxySession"]):
+        super().__init__(sa_session, galaxy_session)
+        self.created_tags: Dict[str, "Tag"] = {}
 
     def _get_tag(self, tag_name):
         """Get tag from cache or database."""
@@ -432,6 +509,10 @@ def _create_tag_instance(self, tag_name):
 
 
 class GalaxySessionlessTagHandler(GalaxyTagHandlerSession):
+    def _ensure_user_owns_item(self, user: Optional["User"], item):
+        # In sessionless mode we don't need to check ownership, we're only exporting
+        pass
+
     def _get_tag(self, tag_name):
         """Get tag from cache or database."""
         # Short-circuit session access
diff --git a/lib/galaxy/objectstore/pithos.py b/lib/galaxy/objectstore/pithos.py
index 63b81bbc2815..6dfa71b239fb 100644
--- a/lib/galaxy/objectstore/pithos.py
+++ b/lib/galaxy/objectstore/pithos.py
@@ -320,10 +320,7 @@ def _size(self, obj, **kwargs) -> int:
             try:
                 return os.path.getsize(self._get_cache_path(path))
             except OSError as ex:
-                log.warning(
-                    "Could not get size of file {path} in local cache,"
-                    "will try Pithos. Error: {err}".format(path=path, err=ex)
-                )
+                log.warning(f"Could not get size of file {path} in local cache," f"will try Pithos. Error: {ex}")
         try:
             file = self.pithos.get_object_info(path)
         except ClientError as ce:
@@ -406,10 +403,7 @@ def _update_from_file(self, obj, **kwargs):
         if kwargs.get("create"):
             self._create(obj, **kwargs)
         if not self._exists(obj, **kwargs):
-            raise ObjectNotFound(
-                "objectstore.update_from_file, object does not exist: {obj}, "
-                "kwargs: {kwargs}".format(obj=obj, kwargs=kwargs)
-            )
+            raise ObjectNotFound(f"objectstore.update_from_file, object does not exist: {obj}, " f"kwargs: {kwargs}")
 
         path = self._construct_path(obj, **kwargs)
         cache_path = self._get_cache_path(path)
diff --git a/lib/galaxy/queues.py b/lib/galaxy/queues.py
index 75fd4fa5fb53..d68012beb5c7 100644
--- a/lib/galaxy/queues.py
+++ b/lib/galaxy/queues.py
@@ -23,8 +23,7 @@ def all_control_queues_for_declare(application_stack):
     """
     # Get all active processes and construct queues for each process
     process_names = (
-        "{p.server_name}@{p.hostname}".format(p=p)
-        for p in application_stack.app.database_heartbeat.get_active_processes()
+        f"{p.server_name}@{p.hostname}" for p in application_stack.app.database_heartbeat.get_active_processes()
     )
     return [Queue(f"control.{server_name}", galaxy_exchange, routing_key="control.*") for server_name in process_names]
 
diff --git a/lib/galaxy/schema/notifications.py b/lib/galaxy/schema/notifications.py
index 3b5c422592cb..30bb8c1aa1a9 100644
--- a/lib/galaxy/schema/notifications.py
+++ b/lib/galaxy/schema/notifications.py
@@ -1,8 +1,6 @@
-import json
 from datetime import datetime
 from enum import Enum
 from typing import (
-    Any,
     Dict,
     List,
     Optional,
@@ -14,7 +12,6 @@
     Field,
     Required,
 )
-from pydantic.utils import GetterDict
 from typing_extensions import (
     Annotated,
     Literal,
@@ -125,21 +122,6 @@ class NewSharedItemNotificationContent(Model):
     ),
 ]
 
-
-class NotificationGetter(GetterDict):
-    """Helper to convert a Notification ORM model into a NotificationResponse.
-    For more information: https://docs.pydantic.dev/usage/models/#data-binding
-    """
-
-    def get(self, key: Any, default: Any = None) -> Any:
-        # The `content` column of the ORM model is a JSON string that needs to be passed
-        # as a dictionary to the `from_orm` constructor to build the correct `AnyNotificationContent`.
-        if key in {"content"} and isinstance(self._obj.content, str):
-            return json.loads(self._obj.content)
-
-        return super().get(key, default)
-
-
 NotificationIdField = Field(
     Required,
     title="ID",
@@ -204,7 +186,6 @@ class NotificationResponse(Model):
 
     class Config:
         orm_mode = True
-        getter_dict = NotificationGetter
 
 
 class UserNotificationResponse(NotificationResponse):
diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py
index 55efaab25fce..f9c45f9a4d13 100644
--- a/lib/galaxy/schema/schema.py
+++ b/lib/galaxy/schema/schema.py
@@ -3050,7 +3050,7 @@ class FileLibraryFolderItem(LibraryFolderItemBase):
     file_size: str
     raw_size: int
     ldda_id: EncodedDatabaseIdField
-    tags: str
+    tags: TagCollection
     message: Optional[str]
 
 
diff --git a/lib/galaxy/tool_util/deps/brew_exts.py b/lib/galaxy/tool_util/deps/brew_exts.py
index 7e8b67991678..852d7425a908 100755
--- a/lib/galaxy/tool_util/deps/brew_exts.py
+++ b/lib/galaxy/tool_util/deps/brew_exts.py
@@ -160,14 +160,14 @@ def __init__(self, command, stdout, stderr):
         self.stdout = stdout
         self.stderr = stderr
         self.message = (
-            "Failed to execute command-line {}, stderr was:\n"
+            f"Failed to execute command-line {command}, stderr was:\n"
             "-------->>begin stderr<<--------\n"
-            "{}\n"
+            f"{stderr}\n"
             "-------->>end stderr<<--------\n"
             "-------->>begin stdout<<--------\n"
-            "{}\n"
+            f"{stdout}\n"
             "-------->>end stdout<<--------\n"
-        ).format(command, stderr, stdout)
+        )
 
     def __str__(self):
         return self.message
diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py
index 943c62c1f3e5..53e8f1e0c4cb 100644
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -3724,7 +3724,12 @@ def produce_outputs(self, trans, out_data, output_collections, incoming, history
         def copy_dataset(dataset, tags):
             copied_dataset = dataset.copy(copy_tags=dataset.tags, flush=False)
             if tags is not None:
-                tag_handler.set_tags_from_list(trans.get_user(), copied_dataset, tags, flush=False)
+                tag_handler.set_tags_from_list(
+                    trans.get_user(),
+                    copied_dataset,
+                    tags,
+                    flush=False,
+                )
             copied_dataset.history_id = history.id
             copied_datasets.append(copied_dataset)
             return copied_dataset
@@ -3772,7 +3777,10 @@ def add_copied_value_to_new_elements(new_tags_dict, dce):
                             old_tags = old_tags - set(new_tags)
                         new_tags = old_tags
                     tag_handler.add_tags_from_list(
-                        user=history.user, item=copied_value, new_tags_list=new_tags, flush=False
+                        user=history.user,
+                        item=copied_value,
+                        new_tags_list=new_tags,
+                        flush=False,
                     )
             else:
                 # We have a collection, and we copy the elements so that we don't manipulate the original tags
diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py
index 04079adf958d..934604a8ef8e 100644
--- a/lib/galaxy/tools/actions/__init__.py
+++ b/lib/galaxy/tools/actions/__init__.py
@@ -977,7 +977,7 @@ def __init__(
         hdca_tags,
     ):
         self.trans = trans
-        self.tag_handler = trans.app.tag_handler.create_tag_handler_session()
+        self.tag_handler = trans.tag_handler
         self.history = history
         self.tool = tool
         self.tool_action = tool_action
diff --git a/lib/galaxy/tools/actions/model_operations.py b/lib/galaxy/tools/actions/model_operations.py
index f963321283cc..6b6dac0af782 100644
--- a/lib/galaxy/tools/actions/model_operations.py
+++ b/lib/galaxy/tools/actions/model_operations.py
@@ -1,4 +1,5 @@
 import logging
+from typing import TYPE_CHECKING
 
 from galaxy.tools.actions import (
     DefaultToolAction,
@@ -6,6 +7,9 @@
     ToolExecutionCache,
 )
 
+if TYPE_CHECKING:
+    from galaxy.managers.context import ProvidesUserContext
+
 log = logging.getLogger(__name__)
 
 
@@ -103,8 +107,10 @@ def execute(
         log.info(f"Calling produce_outputs, tool is {tool}")
         return job, out_data, history
 
-    def _produce_outputs(self, trans, tool, out_data, output_collections, incoming, history, tags, hdca_tags):
-        tag_handler = trans.app.tag_handler.create_tag_handler_session()
+    def _produce_outputs(
+        self, trans: "ProvidesUserContext", tool, out_data, output_collections, incoming, history, tags, hdca_tags
+    ):
+        tag_handler = trans.tag_handler
         tool.produce_outputs(
             trans,
             out_data,
diff --git a/lib/galaxy/tools/actions/upload_common.py b/lib/galaxy/tools/actions/upload_common.py
index d0859f7ba72b..14e4861a9ad8 100644
--- a/lib/galaxy/tools/actions/upload_common.py
+++ b/lib/galaxy/tools/actions/upload_common.py
@@ -22,12 +22,12 @@
     stream_to_file,
     validate_non_local,
 )
+from galaxy.managers.context import ProvidesUserContext
 from galaxy.model import (
     FormDefinition,
     LibraryDataset,
     LibraryFolder,
     Role,
-    tags,
 )
 from galaxy.model.base import transaction
 from galaxy.util import is_url
@@ -261,8 +261,10 @@ def __new_library_upload(trans, cntrller, uploaded_dataset, library_bunch, tag_h
     return ldda
 
 
-def new_upload(trans, cntrller, uploaded_dataset, library_bunch=None, history=None, state=None, tag_list=None):
-    tag_handler = tags.GalaxyTagHandlerSession(trans.sa_session)
+def new_upload(
+    trans: ProvidesUserContext, cntrller, uploaded_dataset, library_bunch=None, history=None, state=None, tag_list=None
+):
+    tag_handler = trans.tag_handler
     if library_bunch:
         upload_target_dataset_instance = __new_library_upload(
             trans, cntrller, uploaded_dataset, library_bunch, tag_handler, state
diff --git a/lib/galaxy/tools/evaluation.py b/lib/galaxy/tools/evaluation.py
index c0be8794f9e9..b53bc48f51c8 100644
--- a/lib/galaxy/tools/evaluation.py
+++ b/lib/galaxy/tools/evaluation.py
@@ -268,7 +268,9 @@ def replace_deferred(input, value, context, prefixed_name=None, **kwargs):
         visit_input_values(self.tool.inputs, incoming, replace_deferred)
 
     def _validate_incoming(self, incoming: dict):
-        request_context = WorkRequestContext(app=self.app, user=self._user, history=self._history)
+        request_context = WorkRequestContext(
+            app=self.app, user=self._user, history=self._history, galaxy_session=self.job.galaxy_session
+        )
 
         def validate_inputs(input, value, context, **kwargs):
             value = input.from_json(value, request_context, context)
diff --git a/lib/galaxy/tools/imp_exp/__init__.py b/lib/galaxy/tools/imp_exp/__init__.py
index 3a70bbb530c2..4c8c53a91141 100644
--- a/lib/galaxy/tools/imp_exp/__init__.py
+++ b/lib/galaxy/tools/imp_exp/__init__.py
@@ -65,7 +65,10 @@ def cleanup_after_job(self):
                     "history import archive directory",
                 )
             model_store = store.get_import_model_store_for_directory(
-                archive_dir, app=self.app, user=user, tag_handler=self.app.tag_handler.create_tag_handler_session()
+                archive_dir,
+                app=self.app,
+                user=user,
+                tag_handler=self.app.tag_handler.create_tag_handler_session(jiha.job.galaxy_session),
             )
             job = jiha.job
             with model_store.target_history(default_history=job.history) as new_history:
diff --git a/lib/galaxy/util/__init__.py b/lib/galaxy/util/__init__.py
index 7d3df0d4ded1..680df532d56c 100644
--- a/lib/galaxy/util/__init__.py
+++ b/lib/galaxy/util/__init__.py
@@ -544,7 +544,7 @@ def pretty_print_time_interval(time=False, precise=False, utc=False):
         now = datetime.utcnow()
     else:
         now = datetime.now()
-    if type(time) is int:
+    if isinstance(time, (int, float)):
         diff = now - datetime.fromtimestamp(time)
     elif isinstance(time, datetime):
         diff = now - time
diff --git a/lib/galaxy/util/commands.py b/lib/galaxy/util/commands.py
index 679aba33f4f4..321dd71f2fad 100644
--- a/lib/galaxy/util/commands.py
+++ b/lib/galaxy/util/commands.py
@@ -155,14 +155,14 @@ def __init__(self, command, stdout, stderr, returncode):
         self.stderr = stderr
         self.returncode = returncode
         self.message = (
-            "Failed to execute command-line {}, stderr was:\n"
+            f"Failed to execute command-line {command}, stderr was:\n"
             "-------->>begin stderr<<--------\n"
-            "{}\n"
+            f"{stderr}\n"
             "-------->>end stderr<<--------\n"
             "-------->>begin stdout<<--------\n"
-            "{}\n"
+            f"{stdout}\n"
             "-------->>end stdout<<--------\n"
-        ).format(command, stderr, stdout)
+        )
 
     def __str__(self):
         """Return a verbose error message indicating the command problem."""
diff --git a/lib/galaxy/webapps/base/controller.py b/lib/galaxy/webapps/base/controller.py
index 1bec1ae6f993..ad579fc01595 100644
--- a/lib/galaxy/webapps/base/controller.py
+++ b/lib/galaxy/webapps/base/controller.py
@@ -35,7 +35,6 @@
     ExtendedMetadataIndex,
     HistoryDatasetAssociation,
     LibraryDatasetDatasetAssociation,
-    tags,
 )
 from galaxy.model.base import transaction
 from galaxy.model.item_attrs import UsesAnnotations
@@ -1389,9 +1388,6 @@ def get_item(self, trans, id):
 
 
 class UsesTagsMixin(SharableItemSecurityMixin):
-    def get_tag_handler(self, trans) -> tags.GalaxyTagHandler:
-        return trans.app.tag_handler
-
     def _get_user_tags(self, trans, item_class_name, id):
         user = trans.user
         tagged_item = self._get_tagged_item(trans, item_class_name, id)
@@ -1407,7 +1403,7 @@ def _remove_items_tag(self, trans, item_class_name, id, tag_name):
         """Remove a tag from an item."""
         user = trans.user
         tagged_item = self._get_tagged_item(trans, item_class_name, id)
-        deleted = tagged_item and self.get_tag_handler(trans).remove_item_tag(trans, user, tagged_item, tag_name)
+        deleted = tagged_item and trans.tag_handler.remove_item_tag(user, tagged_item, tag_name)
         with transaction(trans.sa_session):
             trans.sa_session.commit()
         return deleted
@@ -1415,7 +1411,7 @@ def _remove_items_tag(self, trans, item_class_name, id, tag_name):
     def _apply_item_tag(self, trans, item_class_name, id, tag_name, tag_value=None):
         user = trans.user
         tagged_item = self._get_tagged_item(trans, item_class_name, id)
-        tag_assoc = self.get_tag_handler(trans).apply_item_tag(user, tagged_item, tag_name, tag_value)
+        tag_assoc = trans.tag_handler.apply_item_tag(user, tagged_item, tag_name, tag_value)
         with transaction(trans.sa_session):
             trans.sa_session.commit()
         return tag_assoc
@@ -1424,11 +1420,10 @@ def _get_item_tag_assoc(self, trans, item_class_name, id, tag_name):
         user = trans.user
         tagged_item = self._get_tagged_item(trans, item_class_name, id)
         log.debug(f"In get_item_tag_assoc with tagged_item {tagged_item}")
-        return self.get_tag_handler(trans)._get_item_tag_assoc(user, tagged_item, tag_name)
+        return trans.tag_handler._get_item_tag_assoc(user, tagged_item, tag_name)
 
     def set_tags_from_list(self, trans, item, new_tags_list, user=None):
-        tag_handler = tags.GalaxyTagHandler(trans.app.model.context)
-        return tag_handler.set_tags_from_list(user, item, new_tags_list)
+        return trans.tag_handler.set_tags_from_list(user, item, new_tags_list)
 
     def get_user_tags_used(self, trans, user=None):
         """
@@ -1444,7 +1439,7 @@ def get_user_tags_used(self, trans, user=None):
             return []
 
         # get all the taggable model TagAssociations
-        tag_models = [v.tag_assoc_class for v in trans.app.tag_handler.item_tag_assoc_info.values()]
+        tag_models = [v.tag_assoc_class for v in trans.tag_handler.item_tag_assoc_info.values()]
         # create a union of subqueries for each for this user - getting only the tname and user_value
         all_tags_query = None
         for tag_model in tag_models:
diff --git a/lib/galaxy/webapps/base/webapp.py b/lib/galaxy/webapps/base/webapp.py
index 4e2cac6e5086..93ca937e0484 100644
--- a/lib/galaxy/webapps/base/webapp.py
+++ b/lib/galaxy/webapps/base/webapp.py
@@ -147,7 +147,7 @@ def build_apispec(self):
             methods = []
             if rule.conditions:
                 m = rule.conditions.get("method", [])
-                methods = type(m) is str and [m] or m
+                methods = [m] if isinstance(m, str) else m
             # Find the controller class
             if controller not in self.api_controllers:
                 # Only happens when removing a controller after porting to FastAPI.
diff --git a/lib/galaxy/webapps/galaxy/api/cloudauthz.py b/lib/galaxy/webapps/galaxy/api/cloudauthz.py
index 36616b095afc..fb0227f1f906 100644
--- a/lib/galaxy/webapps/galaxy/api/cloudauthz.py
+++ b/lib/galaxy/webapps/galaxy/api/cloudauthz.py
@@ -124,8 +124,7 @@ def create(self, trans, payload, **kwargs):
         if not isinstance(config, dict):
             log.debug(msg_template.format(f"invalid config type `{type(config)}`, expect `dict`"))
             raise RequestParameterInvalidException(
-                "Invalid type for the required `config` variable; expect `dict` "
-                "but received `{}`.".format(type(config))
+                "Invalid type for the required `config` variable; expect `dict` " f"but received `{type(config)}`."
             )
         if authn_id:
             try:
diff --git a/lib/galaxy/webapps/galaxy/api/library_contents.py b/lib/galaxy/webapps/galaxy/api/library_contents.py
index c152759b66c4..a0e21e1ad6d6 100644
--- a/lib/galaxy/webapps/galaxy/api/library_contents.py
+++ b/lib/galaxy/webapps/galaxy/api/library_contents.py
@@ -183,7 +183,7 @@ def show(self, trans, id, library_id, **kwd):
             rval["parent_library_id"] = trans.security.encode_id(rval["parent_library_id"])
 
             tag_manager = tags.GalaxyTagHandler(trans.sa_session)
-            rval["tags"] = tag_manager.get_tags_str(content.library_dataset_dataset_association.tags)
+            rval["tags"] = tag_manager.get_tags_list(content.library_dataset_dataset_association.tags)
         return rval
 
     @expose_api
@@ -371,9 +371,9 @@ def _upload_library_dataset(self, trans, folder_id: int, **kwd):
                 trans, folder_id=folder.id, replace_dataset=replace_dataset, **kwd
             )
             if created_outputs_dict:
-                if type(created_outputs_dict) == str:
+                if isinstance(created_outputs_dict, str):
                     return 400, created_outputs_dict
-                elif type(created_outputs_dict) == tuple:
+                elif isinstance(created_outputs_dict, tuple):
                     return created_outputs_dict[0], created_outputs_dict[1]
                 return 200, created_outputs_dict
             else:
diff --git a/lib/galaxy/webapps/galaxy/api/users.py b/lib/galaxy/webapps/galaxy/api/users.py
index cd2768f5e13c..3cbbb57751c2 100644
--- a/lib/galaxy/webapps/galaxy/api/users.py
+++ b/lib/galaxy/webapps/galaxy/api/users.py
@@ -1087,7 +1087,7 @@ def set_toolbox_filters(self, trans, id, payload=None, **kwd):
             for prefixed_name in payload:
                 if prefixed_name.startswith(filter_type):
                     filter_selection = payload.get(prefixed_name)
-                    if type(filter_selection) != bool:
+                    if not isinstance(filter_selection, bool):
                         raise exceptions.RequestParameterInvalidException(
                             "Please specify the filter selection as boolean value."
                         )
diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py
index 8c99c6cb983e..dc3713100660 100644
--- a/lib/galaxy/webapps/galaxy/api/workflows.py
+++ b/lib/galaxy/webapps/galaxy/api/workflows.py
@@ -524,8 +524,10 @@ def update(self, trans: GalaxyWebTransaction, id, payload, **kwds):
                         require_flush = True
             # set tags
             if "tags" in workflow_dict:
-                trans.app.tag_handler.set_tags_from_list(
-                    user=trans.user, item=stored_workflow, new_tags_list=workflow_dict["tags"]
+                trans.tag_handler.set_tags_from_list(
+                    user=trans.user,
+                    item=stored_workflow,
+                    new_tags_list=workflow_dict["tags"],
                 )
 
             if require_flush:
diff --git a/lib/galaxy/webapps/galaxy/controllers/authnz.py b/lib/galaxy/webapps/galaxy/controllers/authnz.py
index 71bd0fbcf7e4..88af61d8fa11 100644
--- a/lib/galaxy/webapps/galaxy/controllers/authnz.py
+++ b/lib/galaxy/webapps/galaxy/controllers/authnz.py
@@ -101,9 +101,9 @@ def callback(self, trans, provider, idphint=None, **kwargs):
                 " Error message: {}".format(provider, user, kwargs.get("error", "None"))
             )
             return trans.show_error_message(
-                "Failed to handle authentication callback from {}. "
+                f"Failed to handle authentication callback from {provider}. "
                 "Please try again, and if the problem persists, contact "
-                "the Galaxy instance admin".format(provider)
+                "the Galaxy instance admin"
             )
         try:
             success, message, (redirect_url, user) = trans.app.authnz_manager.callback(
diff --git a/lib/galaxy/webapps/galaxy/controllers/tag.py b/lib/galaxy/webapps/galaxy/controllers/tag.py
index 2271945eae79..dcd3513daf14 100644
--- a/lib/galaxy/webapps/galaxy/controllers/tag.py
+++ b/lib/galaxy/webapps/galaxy/controllers/tag.py
@@ -30,7 +30,7 @@ def add_tag_async(self, trans, item_id=None, item_class=None, new_tag=None, cont
         # Apply tag.
         item = self._get_item(trans, item_class, trans.security.decode_id(item_id))
         user = trans.user
-        self.get_tag_handler(trans).apply_item_tags(user, item, new_tag)
+        trans.tag_handler.apply_item_tags(user, item, new_tag)
         with transaction(trans.sa_session):
             trans.sa_session.commit()
         # Log.
@@ -46,7 +46,7 @@ def remove_tag_async(self, trans, item_id=None, item_class=None, tag_name=None,
         # Remove tag.
         item = self._get_item(trans, item_class, trans.security.decode_id(item_id))
         user = trans.user
-        self.get_tag_handler(trans).remove_item_tag(user, item, tag_name)
+        trans.tag_handler.remove_item_tag(user, item, tag_name)
         with transaction(trans.sa_session):
             trans.sa_session.commit()
         # Log.
@@ -82,7 +82,7 @@ def _get_tag_autocomplete_names(self, trans, q, limit, timestamp, user=None, ite
             raise RuntimeError("Both item and item_class cannot be None")
         elif item is not None:
             item_class = item.__class__
-        item_tag_assoc_class = self.get_tag_handler(trans).get_tag_assoc_class(item_class)
+        item_tag_assoc_class = trans.tag_handler.get_tag_assoc_class(item_class)
         # Build select statement.
         from_obj = item_tag_assoc_class.table.join(item_class.table).join(trans.app.model.Tag.table)
         where_clause = and_(
@@ -101,9 +101,9 @@ def _get_tag_autocomplete_names(self, trans, q, limit, timestamp, user=None, ite
         # Create and return autocomplete data.
         ac_data = "#Header|Your Tags\n"
         for row in result_set:
-            tag = self.get_tag_handler(trans).get_tag_by_id(row[0])
+            tag = trans.tag_handler.get_tag_by_id(row[0])
             # Exclude tags that are already applied to the item.
-            if (item is not None) and (self.get_tag_handler(trans).item_has_tag(trans.user, item, tag)):
+            if (item is not None) and (trans.tag_handler.item_has_tag(trans.user, item, tag)):
                 continue
             # Add tag to autocomplete data. Use the most frequent name that user
             # has employed for the tag.
@@ -119,7 +119,7 @@ def _get_tag_autocomplete_values(self, trans, q, limit, timestamp, user=None, it
         tag_name_and_value = q.split(":")
         tag_name = tag_name_and_value[0]
         tag_value = tag_name_and_value[1]
-        tag = self.get_tag_handler(trans).get_tag_by_name(tag_name)
+        tag = trans.tag_handler.get_tag_by_name(tag_name)
         # Don't autocomplete if tag doesn't exist.
         if tag is None:
             return ""
@@ -128,7 +128,7 @@ def _get_tag_autocomplete_values(self, trans, q, limit, timestamp, user=None, it
             raise RuntimeError("Both item and item_class cannot be None")
         elif item is not None:
             item_class = item.__class__
-        item_tag_assoc_class = self.get_tag_handler(trans).get_tag_assoc_class(item_class)
+        item_tag_assoc_class = trans.tag_handler.get_tag_assoc_class(item_class)
         # Build select statement.
         from_obj = item_tag_assoc_class.table.join(item_class.table).join(trans.app.model.Tag.table)
         where_clause = and_(
@@ -179,6 +179,6 @@ def _get_item(self, trans, item_class_name, id):
         """
         Get an item based on type and id.
         """
-        item_class = self.get_tag_handler(trans).item_tag_assoc_info[item_class_name].item_class
+        item_class = trans.tag_handler.item_tag_assoc_info[item_class_name].item_class
         item = trans.sa_session.query(item_class).filter(item_class.id == id)[0]
         return item
diff --git a/lib/galaxy/webapps/galaxy/services/history_contents.py b/lib/galaxy/webapps/galaxy/services/history_contents.py
index e063550771ed..436f6f7a9519 100644
--- a/lib/galaxy/webapps/galaxy/services/history_contents.py
+++ b/lib/galaxy/webapps/galaxy/services/history_contents.py
@@ -43,7 +43,10 @@
     api_payload_to_create_params,
     dictify_dataset_collection_instance,
 )
-from galaxy.managers.context import ProvidesHistoryContext
+from galaxy.managers.context import (
+    ProvidesHistoryContext,
+    ProvidesUserContext,
+)
 from galaxy.managers.genomes import GenomesManager
 from galaxy.managers.history_contents import (
     HistoryContentsFilters,
@@ -1391,10 +1394,8 @@ def __init__(
                 item, params, trans
             ),
             HistoryContentItemOperation.change_dbkey: lambda item, params, trans: self._change_dbkey(item, params),
-            HistoryContentItemOperation.add_tags: lambda item, params, trans: self._add_tags(item, trans.user, params),
-            HistoryContentItemOperation.remove_tags: lambda item, params, trans: self._remove_tags(
-                item, trans.user, params
-            ),
+            HistoryContentItemOperation.add_tags: lambda item, params, trans: self._add_tags(trans, item, params),
+            HistoryContentItemOperation.remove_tags: lambda item, params, trans: self._remove_tags(trans, item, params),
         }
 
     def apply(
@@ -1489,10 +1490,8 @@ def _change_dbkey(self, item: HistoryItemModel, params: ChangeDbkeyOperationPara
             for dataset_instance in item.dataset_instances:
                 dataset_instance.set_dbkey(params.dbkey)
 
-    def _add_tags(self, item: HistoryItemModel, user: User, params: TagOperationParams):
-        manager = self._get_item_manager(item)
-        manager.tag_handler.add_tags_from_list(user, item, params.tags, flush=self.flush)
+    def _add_tags(self, trans: ProvidesUserContext, item: HistoryItemModel, params: TagOperationParams):
+        trans.tag_handler.add_tags_from_list(trans.user, item, params.tags, flush=self.flush)
 
-    def _remove_tags(self, item: HistoryItemModel, user: User, params: TagOperationParams):
-        manager = self._get_item_manager(item)
-        manager.tag_handler.remove_tags_from_list(user, item, params.tags, flush=self.flush)
+    def _remove_tags(self, trans: ProvidesUserContext, item: HistoryItemModel, params: TagOperationParams):
+        trans.tag_handler.remove_tags_from_list(trans.user, item, params.tags, flush=self.flush)
diff --git a/lib/galaxy/webapps/galaxy/services/library_folder_contents.py b/lib/galaxy/webapps/galaxy/services/library_folder_contents.py
index 1b601991efd3..6629559140fa 100644
--- a/lib/galaxy/webapps/galaxy/services/library_folder_contents.py
+++ b/lib/galaxy/webapps/galaxy/services/library_folder_contents.py
@@ -183,7 +183,7 @@ def _serialize_library_dataset(
             file_size=util.nice_size(raw_size),
             raw_size=raw_size,
             ldda_id=ldda.id,
-            tags=tag_manager.get_tags_str(ldda.tags),
+            tags=tag_manager.get_tags_list(ldda.tags),
             message=ldda.message or ldda.info,
         )
         return dataset_item
diff --git a/lib/galaxy/work/context.py b/lib/galaxy/work/context.py
index f78db6bf28cd..310779e504e3 100644
--- a/lib/galaxy/work/context.py
+++ b/lib/galaxy/work/context.py
@@ -1,8 +1,15 @@
 import abc
-from typing import Optional
+from typing import (
+    List,
+    Optional,
+)
 
 from galaxy.managers.context import ProvidesHistoryContext
-from galaxy.model import History
+from galaxy.model import (
+    GalaxySession,
+    History,
+    Role,
+)
 from galaxy.model.base import transaction
 
 
@@ -19,13 +26,22 @@ class WorkRequestContext(ProvidesHistoryContext):
     objects.
     """
 
-    def __init__(self, app, user=None, history=None, workflow_building_mode=False, url_builder=None):
+    def __init__(
+        self,
+        app,
+        user=None,
+        history=None,
+        workflow_building_mode=False,
+        url_builder=None,
+        galaxy_session: Optional[GalaxySession] = None,
+    ):
         self._app = app
         self.__user = user
-        self.__user_current_roles = None
+        self.__user_current_roles: Optional[List[Role]] = None
         self.__history = history
         self._url_builder = url_builder
         self.workflow_building_mode = workflow_building_mode
+        self.galaxy_session = galaxy_session
 
     @property
     def app(self):
@@ -88,10 +104,9 @@ def get_content_type(self):
 
 
 class SessionRequestContext(WorkRequestContext):
-    """Like WorkRequestContext, but provides access to galaxy session and request."""
+    """Like WorkRequestContext, but provides access to request."""
 
     def __init__(self, **kwargs):
-        self.galaxy_session = kwargs.pop("galaxy_session", None)
         self.__request: GalaxyAbstractRequest = kwargs.pop("request")
         self.__response: GalaxyAbstractResponse = kwargs.pop("response")
         super().__init__(**kwargs)
@@ -112,7 +127,7 @@ def get_galaxy_session(self):
         return self.galaxy_session
 
     def set_history(self, history):
-        if history and not history.deleted:
+        if history and not history.deleted and self.galaxy_session:
             self.galaxy_session.current_history = history
         self.sa_session.add(self.galaxy_session)
         with transaction(self.sa_session):
@@ -134,4 +149,5 @@ def proxy_work_context_for_history(
         history=history or trans.history,
         url_builder=trans.url_builder,
         workflow_building_mode=workflow_building_mode,
+        galaxy_session=trans.galaxy_session,
     )
diff --git a/lib/galaxy_test/api/test_datasets.py b/lib/galaxy_test/api/test_datasets.py
index 9fedb5116ccf..fdd139d78640 100644
--- a/lib/galaxy_test/api/test_datasets.py
+++ b/lib/galaxy_test/api/test_datasets.py
@@ -1,4 +1,5 @@
 import textwrap
+import urllib
 import zipfile
 from io import BytesIO
 from typing import (
@@ -530,6 +531,39 @@ def test_tag_change(self, history_id):
         assert "name:tag_d" in updated_hda["tags"]
         assert "name:tag_e" in updated_hda["tags"]
 
+    def test_anon_tag_permissions(self):
+        with self._different_user(anon=True):
+            history_id = self._get(urllib.parse.urljoin(self.url, "history/current_history_json")).json()["id"]
+            hda_id = self.dataset_populator.new_dataset(history_id, content="abc", wait=True)["id"]
+            payload = {
+                "item_id": hda_id,
+                "item_class": "HistoryDatasetAssociation",
+                "item_tags": ["cool:tag_a", "cool:tag_b", "tag_c", "name:tag_d", "#tag_e"],
+            }
+            put_response = self._put("tags", data=payload, json=True)
+            updated_hda = self._get(f"histories/{history_id}/contents/{hda_id}").json()
+            assert len(updated_hda["tags"]) == 5
+            # ensure we can remove these tags again
+            payload = {
+                "item_id": hda_id,
+                "item_class": "HistoryDatasetAssociation",
+                "item_tags": [],
+            }
+            put_response = self._put("tags", data=payload, json=True)
+            put_response.raise_for_status()
+            updated_hda = self._get(f"histories/{history_id}/contents/{hda_id}").json()
+            assert len(updated_hda["tags"]) == 0
+        with self._different_user(anon=True):
+            # another anon user can't modify tags
+            payload = {
+                "item_id": hda_id,
+                "item_class": "HistoryDatasetAssociation",
+                "item_tags": ["cool:tag_a", "cool:tag_b", "tag_c", "name:tag_d", "#tag_e"],
+            }
+            put_response = self._put("tags", data=payload, json=True)
+            updated_hda = self._get(f"histories/{history_id}/contents/{hda_id}").json()
+            assert len(updated_hda["tags"]) == 0
+
     @skip_without_tool("cat_data_and_sleep")
     def test_update_datatype(self, history_id):
         hda_id = self.dataset_populator.new_dataset(history_id)["id"]
diff --git a/lib/galaxy_test/api/test_drs.py b/lib/galaxy_test/api/test_drs.py
index c221e486a03c..00921aca35fe 100644
--- a/lib/galaxy_test/api/test_drs.py
+++ b/lib/galaxy_test/api/test_drs.py
@@ -130,7 +130,7 @@ def test_exception_handling_schema(self):
         for method in HTTP_METHODS:
             api_url = self._url_join(f"ga4gh/drs/v1/objects/{drs_id}/access/fakeid")
             error_response = method(api_url)
-            assert type(error_response.status_code) == int
+            assert isinstance(error_response.status_code, int)
             assert error_response.status_code == 404
             error_as_dict = error_response.json()
             assert "status_code" in error_as_dict
diff --git a/lib/galaxy_test/api/test_libraries.py b/lib/galaxy_test/api/test_libraries.py
index 816efff4cadd..cd1c67926d29 100644
--- a/lib/galaxy_test/api/test_libraries.py
+++ b/lib/galaxy_test/api/test_libraries.py
@@ -458,7 +458,7 @@ def test_update_dataset_tags(self):
         data = {"tags": ["#Lancelot", "name:Holy Grail", "blue"]}
         ld_updated = self._patch_library_dataset(ld["id"], data)
         self._assert_has_keys(ld_updated, "tags")
-        assert ld_updated["tags"] == "name:Lancelot, name:HolyGrail, blue"
+        assert ld_updated["tags"] == ["name:Lancelot", "name:HolyGrail", "blue"]
 
     @requires_new_library
     def test_invalid_update_dataset_in_folder(self):
diff --git a/lib/galaxy_test/api/test_tags.py b/lib/galaxy_test/api/test_tags.py
new file mode 100644
index 000000000000..f8908968f3b8
--- /dev/null
+++ b/lib/galaxy_test/api/test_tags.py
@@ -0,0 +1,170 @@
+import json
+from typing import List
+from unittest import SkipTest
+from uuid import uuid4
+
+from galaxy_test.base.populators import (
+    DatasetCollectionPopulator,
+    DatasetPopulator,
+    LibraryPopulator,
+    WorkflowPopulator,
+)
+from ._framework import ApiTestCase
+
+
+class TagsApiTests(ApiTestCase):
+    api_name: str
+    item_class: str
+    publishable: bool
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    def create_item(self) -> str:
+        """Creates a taggable item and returns it's ID."""
+        raise SkipTest("Abstract")
+
+    def test_tags_on_item(self):
+        item_id = self.create_item()
+        self._assert_tags_in_item(item_id, [])
+
+        new_tags = ["APITag"]
+        update_history_tags_response = self._update_tags_using_tags_api(item_id, new_tags)
+        self._assert_status_code_is(update_history_tags_response, 204)
+        self._assert_tags_in_item(item_id, new_tags)
+
+        # other users can't create or update tags
+        with self._different_user():
+            new_tags = ["otherAPITag"]
+            update_history_tags_response = self._update_tags_using_tags_api(item_id, new_tags)
+            self._assert_status_code_is(update_history_tags_response, 403)
+
+    def _update_tags_using_tags_api(self, item_id, new_tags):
+        payload = {"item_class": self.item_class, "item_id": item_id, "item_tags": new_tags}
+        update_history_tags_response = self._put("tags", data=payload, json=True)
+        return update_history_tags_response
+
+    def _get_item(self, item_id: str):
+        item_response = self._get(f"{self.api_name}/{item_id}")
+        self._assert_status_code_is(item_response, 200)
+        item = item_response.json()
+        return item
+
+    def _assert_tags_in_item(self, item_id, expected_tags: List[str]):
+        item = self._get_item(item_id)
+        assert "tags" in item
+        assert item["tags"] == expected_tags
+
+
+class TestHistoryTags(TagsApiTests):
+    api_name = "histories"
+    item_class = "History"
+
+    def create_item(self) -> str:
+        item_id = self.dataset_populator.new_history(name=f"history-{uuid4()}")
+        return item_id
+
+
+class TestDatasetTags(TagsApiTests):
+    api_name = "datasets"
+    item_class = "HistoryDatasetAssociation"
+
+    def create_item(self) -> str:
+        history_id = self.dataset_populator.new_history(name=f"history-{uuid4()}")
+        item_id = self.dataset_populator.new_dataset(history_id=history_id)
+        return item_id["id"]
+
+
+class TestDatasetCollectionTags(TagsApiTests):
+    api_name = "dataset_collections"
+    item_class = "HistoryDatasetCollectionAssociation"
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
+
+    def create_item(self) -> str:
+        history_id = self.dataset_populator.new_history(name=f"history-{uuid4()}")
+        response = self.dataset_collection_populator.create_list_in_history(history_id)
+        self._assert_status_code_is(response, 200)
+        hdca = response.json()["outputs"][0]
+        return hdca["id"]
+
+
+class TestLibraryDatasetTags(TagsApiTests):
+    api_name = "libraries"
+    item_class = "LibraryDatasetDatasetAssociation"
+
+    def setUp(self):
+        super().setUp()
+        self.library_populator = LibraryPopulator(self.galaxy_interactor)
+
+    def create_item(self) -> str:
+        ld = self.library_populator.new_library_dataset(name=f"test-library-dataset-{uuid4()}")
+        self.library_id = ld["parent_library_id"]
+        return ld["id"]
+
+    def _get_item(self, item_id: str):
+        assert self.library_id is not None, "Library ID not set"
+        item_response = self._get(f"{self.api_name}/{self.library_id}/contents/{item_id}")
+        self._assert_status_code_is(item_response, 200)
+        item = item_response.json()
+        return item
+
+
+class TestPageTags(TagsApiTests):
+    api_name = "pages"
+    item_class = "Page"
+
+    def create_item(self) -> str:
+        page = self.dataset_populator.new_page(slug=f"page-{uuid4()}")
+        return page["id"]
+
+
+class TestWorkflowTags(TagsApiTests):
+    api_name = "workflows"
+    item_class = "StoredWorkflow"
+
+    def setUp(self):
+        super().setUp()
+        self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
+
+    def create_item(self) -> str:
+        workflow = self.workflow_populator.load_workflow(name=f"workflow-{uuid4()}")
+        data = dict(
+            workflow=json.dumps(workflow),
+        )
+        upload_response = self._post("workflows", data=data)
+
+        self._assert_status_code_is(upload_response, 200)
+        return upload_response.json()["id"]
+
+
+class TestVisualizationTags(TagsApiTests):
+    api_name = "visualizations"
+    item_class = "Visualization"
+
+    def create_item(self) -> str:
+        uuid_str = str(uuid4())
+
+        title = f"Test Visualization {uuid_str}"
+        slug = f"test-visualization-{uuid_str}"
+        config = json.dumps(
+            {
+                "x": 10,
+                "y": 12,
+            }
+        )
+        create_payload = {
+            "title": title,
+            "slug": slug,
+            "type": "example",
+            "dbkey": "hg17",
+            "annotation": "this is a test visualization for tags",
+            "config": config,
+        }
+        response = self._post("visualizations", data=create_payload)
+        self._assert_status_code_is(response, 200)
+        viz = response.json()
+        return viz["id"]
diff --git a/packages/selenium/setup.cfg b/packages/selenium/setup.cfg
index 8432f9e7fad1..a95b3a449dfa 100644
--- a/packages/selenium/setup.cfg
+++ b/packages/selenium/setup.cfg
@@ -37,7 +37,7 @@ install_requires =
     axe-selenium-python
     PyYAML
     requests
-    selenium
+    selenium!=4.11.0,!=4.11.1,!=4.11.2
 packages = find:
 python_requires = >=3.7
 
diff --git a/packages/test_selenium/setup.cfg b/packages/test_selenium/setup.cfg
index 086d0243571a..dca41568d040 100644
--- a/packages/test_selenium/setup.cfg
+++ b/packages/test_selenium/setup.cfg
@@ -41,7 +41,7 @@ install_requires =
     pytest
     PyYAML
     requests
-    selenium
+    selenium!=4.11.0,!=4.11.1,!=4.11.2
 packages = find:
 python_requires = >=3.7
 
diff --git a/scripts/api/common.py b/scripts/api/common.py
index 1a99d4af6c32..1b5a09e44070 100644
--- a/scripts/api/common.py
+++ b/scripts/api/common.py
@@ -109,7 +109,7 @@ def display(api_key, url, return_formatted=True):
         print("------------------")
         for k, v in r.items():
             print(f"{k}: {v}")
-    elif type(r) == str:
+    elif isinstance(r, str):
         print(r)
     else:
         print("response is unknown type: %s" % type(r))
diff --git a/test/integration/test_cli_runners.py b/test/integration/test_cli_runners.py
index 3dd4ed613792..b240ebaf7cc3 100644
--- a/test/integration/test_cli_runners.py
+++ b/test/integration/test_cli_runners.py
@@ -43,7 +43,7 @@ def start_ssh_docker(container_name, jobs_directory, port=10022, image="agaveapi
         "--rm",
         "--privileged",  # for torque
         "-v",
-        "{jobs_directory}:{jobs_directory}".format(jobs_directory=jobs_directory),
+        f"{jobs_directory}:{jobs_directory}",
         "-v",
         f"{ssh_keys.public_key_file}:/home/testuser/.ssh/authorized_keys",
         "--ulimit",
diff --git a/test/integration/test_workflow_refactoring.py b/test/integration/test_workflow_refactoring.py
index 4048fcc9743a..730106e58cdc 100644
--- a/test/integration/test_workflow_refactoring.py
+++ b/test/integration/test_workflow_refactoring.py
@@ -899,6 +899,10 @@ def __init__(self, app, user):
         self.history = None
         self.workflow_building_mode = workflow_building_modes.ENABLED
 
+    @property
+    def galaxy_session(self):
+        return None
+
     @property
     def app(self):
         return self._app
diff --git a/test/unit/app/managers/base.py b/test/unit/app/managers/base.py
index 5c0aa7b414d9..5251fc06730a 100644
--- a/test/unit/app/managers/base.py
+++ b/test/unit/app/managers/base.py
@@ -2,12 +2,14 @@
 """
 
 import json
+from typing import cast
 
 import sqlalchemy
 
 from galaxy.app_unittest_utils import galaxy_mock
 from galaxy.managers.users import UserManager
 from galaxy.util.unittest import TestCase
+from galaxy.work.context import SessionRequestContext
 
 # =============================================================================
 admin_email = "admin@admin.admin"
@@ -33,7 +35,9 @@ def setUp(self):
 
     def set_up_mocks(self):
         admin_users_list = [u for u in admin_users.split(",") if u]
-        self.trans = galaxy_mock.MockTrans(admin_users=admin_users, admin_users_list=admin_users_list)
+        self.trans = cast(
+            SessionRequestContext, galaxy_mock.MockTrans(admin_users=admin_users, admin_users_list=admin_users_list)
+        )
         self.app = self.trans.app
 
     def set_up_managers(self):
@@ -99,7 +103,7 @@ def assertIsJsonifyable(self, item):
 
 
 class CreatesCollectionsMixin:
-    trans: galaxy_mock.MockTrans
+    trans: SessionRequestContext
 
     def build_element_identifiers(self, elements):
         identifier_list = []
diff --git a/test/unit/app/managers/test_CollectionManager.py b/test/unit/app/managers/test_CollectionManager.py
index feb7da6e9cfb..3e7e17a8bbcb 100644
--- a/test/unit/app/managers/test_CollectionManager.py
+++ b/test/unit/app/managers/test_CollectionManager.py
@@ -28,6 +28,7 @@ def set_up_managers(self):
 
     def test_create_simple_list(self):
         owner = self.user_manager.create(**user2_data)
+        self.trans.set_user(owner)
 
         history = self.history_manager.create(name="history1", user=owner)
 
@@ -80,6 +81,7 @@ def test_create_simple_list(self):
 
     def test_update_from_dict(self):
         owner = self.user_manager.create(**user2_data)
+        self.trans.set_user(owner)
 
         history = self.history_manager.create(name="history1", user=owner)
 
diff --git a/test/unit/app/managers/test_HDAManager.py b/test/unit/app/managers/test_HDAManager.py
index 38a241fd04a8..5692d83eec9e 100644
--- a/test/unit/app/managers/test_HDAManager.py
+++ b/test/unit/app/managers/test_HDAManager.py
@@ -90,18 +90,6 @@ def test_create(self):
         assert hda3.history is None, "history will be None"
         assert hda3.hid is None, "should allow setting hid to None (or any other value)"
 
-    def test_hda_tags(self):
-        owner = self.user_manager.create(**user2_data)
-        history1 = self.history_manager.create(name="history1", user=owner)
-        dataset1 = self.dataset_manager.create()
-        hda1 = self.hda_manager.create(history=history1, dataset=dataset1)
-
-        self.log("should be able to set tags on an hda")
-        tags_to_set = ["tag-one", "tag-two"]
-        self.hda_manager.set_tags(hda1, tags_to_set, user=owner)
-        tag_str_array = self.hda_manager.get_tags(hda1)
-        assert sorted(tags_to_set) == sorted(tag_str_array)
-
     def test_hda_annotation(self):
         owner = self.user_manager.create(**user2_data)
         history1 = self.history_manager.create(name="history1", user=owner)
@@ -131,11 +119,9 @@ def test_copy_from_hda(self):
         self.log("tags should be copied between HDAs")
         tagged = self.hda_manager.create(history=history1, dataset=self.dataset_manager.create())
         tags_to_set = ["tag-one", "tag-two"]
-        self.hda_manager.set_tags(tagged, tags_to_set, user=owner)
-
+        self.app.tag_handler.add_tags_from_list(owner, tagged, tags_to_set)
         hda2 = self.hda_manager.copy(tagged, history=history1)
-        tag_str_array = self.hda_manager.get_tags(hda2)
-        assert sorted(tags_to_set) == sorted(tag_str_array)
+        assert tagged.make_tag_string_list() == hda2.make_tag_string_list()
 
         self.log("annotations should be copied between HDAs")
         annotated = self.hda_manager.create(history=history1, dataset=self.dataset_manager.create())
diff --git a/test/unit/app/managers/test_HDCAManager.py b/test/unit/app/managers/test_HDCAManager.py
index 1388953372d0..23ea050b2def 100644
--- a/test/unit/app/managers/test_HDCAManager.py
+++ b/test/unit/app/managers/test_HDCAManager.py
@@ -31,6 +31,7 @@ def set_up_managers(self):
     def _create_history(self, user_data=None, **kwargs):
         user_data = user_data or user2_data
         owner = self.user_manager.create(**user_data)
+        self.trans.set_user(owner)
         return self.history_manager.create(user=owner, **kwargs)
 
     def _create_hda(self, history, dataset=None, **kwargs):
diff --git a/test/unit/app/managers/test_HistoryContentsManager.py b/test/unit/app/managers/test_HistoryContentsManager.py
index 9c20987d115b..367d80241388 100644
--- a/test/unit/app/managers/test_HistoryContentsManager.py
+++ b/test/unit/app/managers/test_HistoryContentsManager.py
@@ -56,6 +56,7 @@ def add_list_collection_to_history(self, history, hdas, name="test collection",
 class TestHistoryAsContainer(HistoryAsContainerBaseTestCase):
     def test_contents(self):
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
 
         self.log("calling contents on an empty history should return an empty list")
@@ -74,6 +75,7 @@ def test_contents(self):
 
     def test_contained(self):
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
 
         self.log("calling contained on an empty history should return an empty list")
@@ -86,6 +88,7 @@ def test_contained(self):
 
     def test_copy_elements_on_collection_creation(self):
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
         hdas = [self.add_hda_to_history(history, name=("hda-" + str(x))) for x in range(3)]
         hdca = self.add_list_collection_to_history(history, hdas)
@@ -96,6 +99,7 @@ def test_copy_elements_on_collection_creation(self):
 
     def test_subcontainers(self):
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
 
         self.log("calling subcontainers on an empty history should return an empty list")
@@ -109,6 +113,7 @@ def test_subcontainers(self):
 
     def test_limit_and_offset(self):
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
         contents = []
         contents.extend([self.add_hda_to_history(history, name=("hda-" + str(x))) for x in range(3)])
@@ -130,6 +135,7 @@ def test_limit_and_offset(self):
     def test_orm_filtering(self):
         parse_filter = self.history_contents_filters.parse_filter
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
         contents = []
         contents.extend([self.add_hda_to_history(history, name=("hda-" + str(x))) for x in range(3)])
@@ -228,6 +234,7 @@ def test_orm_filtering(self):
 
     def test_order_by(self):
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
         contents = []
         contents.extend([self.add_hda_to_history(history, name=("hda-" + str(x))) for x in range(3)])
@@ -263,6 +270,7 @@ def get_create_time(item):
 
     def test_update_time_filter(self):
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
         contents = []
         contents.extend([self.add_hda_to_history(history, name=("hda-" + str(x))) for x in range(3)])
@@ -292,6 +300,7 @@ def get_update_time(item):
     def test_filtered_counting(self):
         parse_filter = self.history_contents_filters.parse_filter
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
         contents = []
         contents.extend([self.add_hda_to_history(history, name=("hda-" + str(x))) for x in range(3)])
@@ -324,6 +333,7 @@ def test_filtered_counting(self):
 
     def test_type_id(self):
         user2 = self.user_manager.create(**user2_data)
+        self.trans.set_user(user2)
         history = self.history_manager.create(name="history", user=user2)
         contents = []
         contents.extend([self.add_hda_to_history(history, name=("hda-" + str(x))) for x in range(3)])
diff --git a/test/unit/app/managers/test_HistoryManager.py b/test/unit/app/managers/test_HistoryManager.py
index 2cba2e80558e..f24ff8ffc3e6 100644
--- a/test/unit/app/managers/test_HistoryManager.py
+++ b/test/unit/app/managers/test_HistoryManager.py
@@ -95,13 +95,13 @@ def test_copy(self):
         history1 = self.history_manager.create(name="history1", user=user2)
         tags = ["tag-one"]
         annotation = "history annotation"
-        self.history_manager.set_tags(history1, tags, user=user2)
+        self.history_manager.tag_handler.set_tags_from_list(user=user2, item=history1, new_tags_list=tags)
         self.history_manager.annotate(history1, annotation, user=user2)
 
         hda = self.add_hda_to_history(history1, name="wat")
         hda_tags = ["tag-one", "tag-two"]
         hda_annotation = "annotation"
-        self.hda_manager.set_tags(hda, hda_tags, user=user2)
+        self.app.tag_handler.set_tags_from_list(user=user2, item=hda, new_tags_list=hda_tags)
         self.hda_manager.annotate(hda, hda_annotation, user=user2)
 
         history2 = self.history_manager.copy(history1, user=user3)
@@ -112,8 +112,7 @@ def test_copy(self):
         assert history2 != history1
 
         copied_hda = history2.datasets[0]
-        copied_hda_tags = self.hda_manager.get_tags(copied_hda)
-        assert sorted(hda_tags) == sorted(copied_hda_tags)
+        assert hda.make_tag_string_list() == copied_hda.make_tag_string_list()
         copied_hda_annotation = self.hda_manager.annotation(copied_hda)
         assert hda_annotation == copied_hda_annotation
 
diff --git a/test/unit/app/managers/test_NotificationManager.py b/test/unit/app/managers/test_NotificationManager.py
index 19403545c431..99d493b3ef3b 100644
--- a/test/unit/app/managers/test_NotificationManager.py
+++ b/test/unit/app/managers/test_NotificationManager.py
@@ -1,4 +1,3 @@
-import json
 from datetime import (
     datetime,
     timedelta,
@@ -109,7 +108,7 @@ def _assert_notification_expected(self, actual_notification: Any, expected_notif
             assert actual_notification.expiration_time is not None
 
         assert actual_notification.content
-        user_notification_content = json.loads(actual_notification.content)
+        user_notification_content = actual_notification.content
         assert user_notification_content["category"] == expected_notification["content"]["category"]
         assert user_notification_content["subject"] == expected_notification["content"]["subject"]
         assert user_notification_content["message"] == expected_notification["content"]["message"]
@@ -122,7 +121,7 @@ def test_create_notification_broadcast(self):
 
         assert actual_notification.id
         assert actual_notification.category == "broadcast"
-        notification_content = json.loads(actual_notification.content)
+        notification_content = actual_notification.content
         assert notification_content["category"] == "broadcast"
         assert notification_content["subject"] == "Testing Broadcast Subject"
         assert notification_content["message"] == "Testing Broadcast Message"
@@ -161,16 +160,16 @@ def test_get_all_broadcasted_notifications(self):
 
         notifications = self.notification_manager.get_all_broadcasted_notifications()
         assert len(notifications) == 1
-        assert json.loads(notifications[0].content)["subject"] == "Recent Notification"
+        assert notifications[0].content["subject"] == "Recent Notification"
 
         notifications = self.notification_manager.get_all_broadcasted_notifications(since=next_week, active_only=False)
         assert len(notifications) == 2
-        assert json.loads(notifications[0].content)["subject"] == "Scheduled Next Week Notification"
-        assert json.loads(notifications[1].content)["subject"] == "Scheduled Next Month Notification"
+        assert notifications[0].content["subject"] == "Scheduled Next Week Notification"
+        assert notifications[1].content["subject"] == "Scheduled Next Month Notification"
 
         notifications = self.notification_manager.get_all_broadcasted_notifications(since=next_month, active_only=False)
         assert len(notifications) == 1
-        assert json.loads(notifications[0].content)["subject"] == "Scheduled Next Month Notification"
+        assert notifications[0].content["subject"] == "Scheduled Next Month Notification"
 
     def test_update_broadcasted_notification(self):
         next_month = datetime.utcnow() + timedelta(days=30)
@@ -195,7 +194,7 @@ def test_update_broadcasted_notification(self):
         assert updated_notification.source == update_request.source
         assert updated_notification.variant == update_request.variant
         assert updated_notification.publication_time == update_request.publication_time
-        content = json.loads(updated_notification.content)
+        content = updated_notification.content
         assert content["subject"] == expected_content.subject
         assert content["message"] == expected_content.message
 
diff --git a/test/unit/app/managers/test_UserManager.py b/test/unit/app/managers/test_UserManager.py
index 0fa3231148d1..b6c24c7ed276 100644
--- a/test/unit/app/managers/test_UserManager.py
+++ b/test/unit/app/managers/test_UserManager.py
@@ -198,7 +198,6 @@ def testable_url_for(*a, **k):
     def test_activation_email(self):
         self.log("should produce the activation email")
         self.user_manager.create(email="user@nopassword.com", username="nopassword")
-        self.trans.request.host = "request.host"
 
         def validate_send_email(frm, to, subject, body, config, html=None):
             assert frm == "email_from"
@@ -222,7 +221,6 @@ def validate_send_email(frm, to, subject, body, config, html=None):
     def test_reset_email(self):
         self.log("should produce the password reset email")
         self.user_manager.create(email="user@nopassword.com", username="nopassword")
-        self.trans.request.host = "request.host"
 
         def validate_send_email(frm, to, subject, body, config, html=None):
             assert frm == "email_from"
diff --git a/test/unit/app/tools/test_actions.py b/test/unit/app/tools/test_actions.py
index e5f6b1990286..85cc8ca63702 100644
--- a/test/unit/app/tools/test_actions.py
+++ b/test/unit/app/tools/test_actions.py
@@ -259,6 +259,10 @@ def __init__(self, app, history, user=None):
         self._user_is_active = True
         self.user_is_admin = False
 
+    @property
+    def tag_handler(self):
+        return self.app.tag_handler
+
     def get_user_is_active(self):
         # NOTE: the real user_is_active also checks whether activation is enabled in the config
         return self._user_is_active
diff --git a/test/unit/app/tools/test_execution.py b/test/unit/app/tools/test_execution.py
index 516a75d24bfc..0079a75dcb71 100644
--- a/test/unit/app/tools/test_execution.py
+++ b/test/unit/app/tools/test_execution.py
@@ -206,6 +206,7 @@ def __init__(self, app, history):
         self.webapp = Bunch(name="galaxy")
         self.sa_session = self.app.model.context
         self.url_builder = None
+        self.galaxy_session = None
 
     def get_history(self, **kwargs):
         return self.history
diff --git a/test/unit/config/test_load_config.py b/test/unit/config/test_load_config.py
index 13e4cd8754d3..7ab70339437f 100644
--- a/test/unit/config/test_load_config.py
+++ b/test/unit/config/test_load_config.py
@@ -47,10 +47,10 @@ def test_load_config_from_schema(mock_init):
     assert config.property5 is None
     assert config.property6 is None
 
-    assert type(config.property1) is str
-    assert type(config.property2) is int
-    assert type(config.property3) is float
-    assert type(config.property4) is bool
+    assert isinstance(config.property1, str)
+    assert isinstance(config.property2, int)
+    assert isinstance(config.property3, float)
+    assert isinstance(config.property4, bool)
 
 
 def test_update_raw_config_from_kwargs(mock_init):
@@ -64,10 +64,10 @@ def test_update_raw_config_from_kwargs(mock_init):
     assert config._raw_config["property5"] is None  # no change
     assert config._raw_config["property6"] is None  # no change
 
-    assert type(config._raw_config["property1"]) is str
-    assert type(config._raw_config["property2"]) is int
-    assert type(config._raw_config["property3"]) is float
-    assert type(config._raw_config["property4"]) is bool
+    assert isinstance(config._raw_config["property1"], str)
+    assert isinstance(config._raw_config["property2"], int)
+    assert isinstance(config._raw_config["property3"], float)
+    assert isinstance(config._raw_config["property4"], bool)
 
 
 def test_update_raw_config_from_string_kwargs(mock_init):
@@ -80,10 +80,10 @@ def test_update_raw_config_from_string_kwargs(mock_init):
     assert config._raw_config["property3"] == 2.0  # updated
     assert config._raw_config["property4"] is False  # updated
 
-    assert type(config._raw_config["property1"]) is str
-    assert type(config._raw_config["property2"]) is int
-    assert type(config._raw_config["property3"]) is float
-    assert type(config._raw_config["property4"]) is bool
+    assert isinstance(config._raw_config["property1"], str)
+    assert isinstance(config._raw_config["property2"], int)
+    assert isinstance(config._raw_config["property3"], float)
+    assert isinstance(config._raw_config["property4"], bool)
 
 
 def test_update_raw_config_from_kwargs_with_none(mock_init):
@@ -111,7 +111,7 @@ def test_update_raw_config_from_kwargs_falsy_not_none(mock_init):
     config = BaseAppConfiguration(property1=0)
 
     assert config._raw_config["property1"] == "0"  # updated
-    assert type(config._raw_config["property1"]) is str  # and converted to str
+    assert isinstance(config._raw_config["property1"], str)  # and converted to str
 
 
 def test_unset_renamed_option_set_by_old_option(mock_init, monkeypatch):
diff --git a/test/unit/config/test_schema.py b/test/unit/config/test_schema.py
index 1b6b11b4c86a..820f74300e25 100644
--- a/test/unit/config/test_schema.py
+++ b/test/unit/config/test_schema.py
@@ -60,10 +60,10 @@ def mock_init(self, stream):
     assert schema.defaults["property5"] is None
     assert schema.defaults["property6"] is None
 
-    assert type(schema.defaults["property1"]) is str
-    assert type(schema.defaults["property2"]) is int
-    assert type(schema.defaults["property3"]) is float
-    assert type(schema.defaults["property4"]) is bool
+    assert isinstance(schema.defaults["property1"], str)
+    assert isinstance(schema.defaults["property2"], int)
+    assert isinstance(schema.defaults["property3"], float)
+    assert isinstance(schema.defaults["property4"], bool)
 
     assert len(schema.reloadable_options) == 2
     assert "property2" in schema.reloadable_options
diff --git a/test/unit/data/model/test_model_discovery.py b/test/unit/data/model/test_model_discovery.py
index ba32fdaf3958..955f7d1c1fd5 100644
--- a/test/unit/data/model/test_model_discovery.py
+++ b/test/unit/data/model/test_model_discovery.py
@@ -252,7 +252,11 @@ def _import_directory_to_history(app, target, work_directory):
 
     import_options = store.ImportOptions(allow_dataset_object_edit=True)
     import_model_store = store.get_import_model_store_for_directory(
-        target, app=app, user=u, import_options=import_options, tag_handler=app.tag_handler.create_tag_handler_session()
+        target,
+        app=app,
+        user=u,
+        import_options=import_options,
+        tag_handler=app.tag_handler.create_tag_handler_session(None),
     )
     with import_model_store.target_history(default_history=import_history):
         import_model_store.perform_import(import_history)