diff --git a/lib/galaxy/managers/jobs.py b/lib/galaxy/managers/jobs.py
index bf26cad31a0a..cd91861a7408 100644
--- a/lib/galaxy/managers/jobs.py
+++ b/lib/galaxy/managers/jobs.py
@@ -1,13 +1,14 @@
 import json
 import logging
-import typing
 from datetime import (
     date,
     datetime,
 )
 from typing import (
     cast,
+    Dict,
     List,
+    Optional,
 )
 
 import sqlalchemy
@@ -38,6 +39,7 @@
     Safety,
 )
 from galaxy.managers.collections import DatasetCollectionManager
+from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.datasets import DatasetManager
 from galaxy.managers.hdas import HDAManager
 from galaxy.managers.lddas import LDDAManager
@@ -106,7 +108,9 @@ def __init__(self, app: StructuredApp):
         self.app = app
         self.dataset_manager = DatasetManager(app)
 
-    def index_query(self, trans, payload: JobIndexQueryPayload) -> sqlalchemy.engine.Result:
+    def index_query(
+        self, trans: ProvidesUserContext, payload: JobIndexQueryPayload
+    ) -> Optional[sqlalchemy.engine.ScalarResult]:
         """The caller is responsible for security checks on the resulting job if
         history_id, invocation_id, or implicit_collection_jobs_id is set.
         Otherwise this will only return the user's jobs or all jobs if the requesting
@@ -122,6 +126,13 @@ def index_query(self, trans, payload: JobIndexQueryPayload) -> sqlalchemy.engine
         search = payload.search
         order_by = payload.order_by
 
+        if trans.user is None:
+            # If the user is anonymous we can only return jobs for the current session history
+            if trans.galaxy_session and trans.galaxy_session.current_history_id:
+                history_id = trans.galaxy_session.current_history_id
+            else:
+                return None
+
         def build_and_apply_filters(stmt, objects, filter_func):
             if objects is not None:
                 if isinstance(objects, (str, date, datetime)):
@@ -208,7 +219,7 @@ def add_search_criteria(stmt):
             if user_details:
                 stmt = stmt.outerjoin(Job.user)
         else:
-            if history_id is None and invocation_id is None and implicit_collection_jobs_id is None:
+            if history_id is None and invocation_id is None and implicit_collection_jobs_id is None and trans.user:
                 stmt = stmt.where(Job.user_id == trans.user.id)
             # caller better check security
 
@@ -647,7 +658,7 @@ def _build_stmt_for_dce(self, stmt, data_conditions, used_ids, k, v):
         return stmt
 
 
-def view_show_job(trans, job: Job, full: bool) -> typing.Dict:
+def view_show_job(trans, job: Job, full: bool) -> Dict:
     is_admin = trans.user_is_admin
     job_dict = job.to_dict("element", system_details=is_admin)
     if trans.app.config.expose_dataset_path and "command_line" not in job_dict:
diff --git a/lib/galaxy/webapps/galaxy/services/jobs.py b/lib/galaxy/webapps/galaxy/services/jobs.py
index 240d841a67ea..727497d12af8 100644
--- a/lib/galaxy/webapps/galaxy/services/jobs.py
+++ b/lib/galaxy/webapps/galaxy/services/jobs.py
@@ -81,7 +81,9 @@ def index(
             or payload.history_id is not None
         )
         jobs = self.job_manager.index_query(trans, payload)
-        out = []
+        out: List[Dict[str, Any]] = []
+        if jobs is None:
+            return out
         for job in jobs.yield_per(model.YIELD_PER_ROWS):
             # TODO: optimize if this crucial
             if check_security_of_jobs and not security_check(trans, job.history, check_accessible=True):