Route history_update SSE events for anonymous sessions

HistoryAuditMonitor only dispatched history_update events via
push_to_user, so anonymous-owned histories (user_id IS NULL) never
produced events. With enable_sse_history_updates on, the client
disables polling and waits for SSE events — leaving anonymous history
panels frozen, which broke seven UI tests across the Playwright suites.

Extend the pipeline with a parallel galaxy_session.id-keyed route:

- SSEConnectionManager tracks a _session_connections map alongside
  _connections and exposes push_to_session; connect/disconnect/stream
  accept an optional galaxy_session_id.
- EventsService.open_stream forwards trans.galaxy_session.id so
  anonymous sessions register under their session key.
- SSEEventDispatcher.history_update and HistoryUpdatePayload gain an
  optional session_updates dict; the queue_worker handler fans out
  session-keyed events via push_to_session.
- HistoryAuditMonitor caches (user_id, session_ids) per history and
  performs one extra indexed lookup against GalaxySessionToHistoryAssociation
  only for anon histories.

galaxy_session.id never leaves the server — it's used only as an
in-memory/AMQP dispatch key; the browser-visible event payload still
contains just encoded history_ids.

Author: mvdbeek

lib/galaxy/managers/history_audit_monitor.py

@@ -31,6 +31,7 @@
 from galaxy.config import GalaxyAppConfiguration
 from galaxy.managers.sse_dispatch import SSEEventDispatcher
 from galaxy.model import (
+    GalaxySessionToHistoryAssociation,
     History,
     HistoryAudit,
 )
@@ -119,8 +120,11 @@ def __init__(
         self._exit = threading.Event()
         self._thread: Optional[threading.Thread] = None
         self._active = False
-        # Bounded LRU cache: history_id -> user_id, refreshed on miss.
-        self._history_owner_cache: OrderedDict[int, int] = OrderedDict()
+        # Bounded LRU cache: history_id -> (user_id, session_ids), refreshed on miss.
+        # For registered-owned histories: (user_id, ()); for anonymous histories:
+        # (None, (session_id, ...)) — a history can be associated with multiple
+        # sessions via GalaxySessionToHistoryAssociation.
+        self._history_owner_cache: OrderedDict[int, tuple[Optional[int], tuple[int, ...]]] = OrderedDict()
 
     def start(self) -> None:
         if self._active:
@@ -227,7 +231,7 @@ def _poll_audit_table(self) -> None:
     # --- Common dispatch logic ---
 
     def _dispatch_history_updates(self, history_ids: set[int]) -> None:
-        """Map history_ids to user_ids and send Kombu control task.
+        """Map history_ids to user_ids / session_ids and send Kombu control task.
 
         Raw integer history IDs are sent across the control queue; encoding is
         deferred to the ``history_update`` task handler on the receiving side,
@@ -239,24 +243,58 @@ def _dispatch_history_updates(self, history_ids: set[int]) -> None:
             self._refresh_owner_cache(unknown)
 
         user_updates: dict[str, list[int]] = defaultdict(list)
+        session_updates: dict[str, list[int]] = defaultdict(list)
         for history_id in history_ids:
-            user_id = self._history_owner_cache.get(history_id)
+            entry = self._history_owner_cache.get(history_id)
+            if entry is None:
+                continue
+            user_id, session_ids = entry
             if user_id is not None:
                 user_updates[str(user_id)].append(history_id)
+            else:
+                for session_id in session_ids:
+                    session_updates[str(session_id)].append(history_id)
 
-        if not user_updates:
+        if not user_updates and not session_updates:
             return
 
-        self._dispatcher.history_update(user_updates=dict(user_updates))
+        self._dispatcher.history_update(
+            user_updates=dict(user_updates),
+            session_updates=dict(session_updates) if session_updates else None,
+        )
 
     def _refresh_owner_cache(self, history_ids: set[int]) -> None:
-        """Look up user_id for given history_ids and update the bounded cache."""
+        """Look up ownership for given history_ids and update the bounded cache.
+
+        Registered-owned histories resolve with just ``History.user_id``. For
+        histories where ``user_id IS NULL`` we additionally fetch associated
+        ``galaxy_session.id`` values from ``GalaxySessionToHistoryAssociation``
+        so the anonymous SSE dispatch path can target the right browser.
+        """
         try:
-            stmt = sa_select(History.id, History.user_id).where(History.id.in_(history_ids))
             with self._model.new_session() as session:
+                stmt = sa_select(History.id, History.user_id).where(History.id.in_(history_ids))
+                anon_history_ids: set[int] = set()
                 for row in session.execute(stmt):
-                    self._history_owner_cache[row[0]] = row[1]
-                    self._history_owner_cache.move_to_end(row[0])
+                    hid, uid = row[0], row[1]
+                    self._history_owner_cache[hid] = (uid, ())
+                    self._history_owner_cache.move_to_end(hid)
+                    if uid is None:
+                        anon_history_ids.add(hid)
+
+                if anon_history_ids:
+                    assoc_stmt = sa_select(
+                        GalaxySessionToHistoryAssociation.history_id,
+                        GalaxySessionToHistoryAssociation.session_id,
+                    ).where(GalaxySessionToHistoryAssociation.history_id.in_(anon_history_ids))
+                    sessions_by_history: dict[int, list[int]] = defaultdict(list)
+                    for row in session.execute(assoc_stmt):
+                        hid, sid = row[0], row[1]
+                        if sid is not None:
+                            sessions_by_history[hid].append(sid)
+                    for hid, sids in sessions_by_history.items():
+                        self._history_owner_cache[hid] = (None, tuple(sids))
+
             while len(self._history_owner_cache) > OWNER_CACHE_MAX:
                 self._history_owner_cache.popitem(last=False)
         except Exception:

lib/galaxy/managers/sse.py

@@ -81,6 +81,7 @@ class SSEConnectionManager:
 
     def __init__(self, statsd_client: Optional[VanillaGalaxyStatsdClient] = None) -> None:
         self._connections: dict[int, set[asyncio.Queue]] = defaultdict(set)
+        self._session_connections: dict[int, set[asyncio.Queue]] = defaultdict(set)
         self._broadcast_connections: set[asyncio.Queue] = set()
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._statsd_client = statsd_client
@@ -92,27 +93,39 @@ def _ensure_loop(self) -> None:
 
     # -- Called from ASYNC context (uvicorn event loop thread) --
 
-    def connect(self, user_id: Optional[int]) -> asyncio.Queue:
+    def connect(self, user_id: Optional[int], galaxy_session_id: Optional[int] = None) -> asyncio.Queue:
         """Register a new SSE connection. Returns a queue to await events from.
 
         Called from the SSE endpoint handler (async context). A ``ready`` event is
         enqueued immediately so that clients (and tests) can synchronize on the
         server-side subscription rather than the underlying socket open event.
+
+        ``galaxy_session_id`` is the dispatch key for events that target a
+        specific browser session (e.g. history updates for anonymous users,
+        whose ``user_id`` is ``None``).
         """
         self._ensure_loop()
         queue: asyncio.Queue = asyncio.Queue(maxsize=64)
         if user_id is not None:
             self._connections[user_id].add(queue)
+        if galaxy_session_id is not None:
+            self._session_connections[galaxy_session_id].add(queue)
         self._broadcast_connections.add(queue)
         queue.put_nowait(SSEEvent(event="ready", data=""))
         log.debug(
-            "SSE connection opened for user_id=%s (total=%d)",
+            "SSE connection opened for user_id=%s session_id=%s (total=%d)",
             user_id,
+            galaxy_session_id,
             len(self._broadcast_connections),
         )
         return queue
 
-    def disconnect(self, user_id: Optional[int], queue: asyncio.Queue) -> None:
+    def disconnect(
+        self,
+        user_id: Optional[int],
+        queue: asyncio.Queue,
+        galaxy_session_id: Optional[int] = None,
+    ) -> None:
         """Unregister an SSE connection.
 
         Called from the SSE endpoint's ``finally`` block (async context).
@@ -121,10 +134,15 @@ def disconnect(self, user_id: Optional[int], queue: asyncio.Queue) -> None:
             self._connections[user_id].discard(queue)
             if not self._connections[user_id]:
                 del self._connections[user_id]
+        if galaxy_session_id is not None:
+            self._session_connections[galaxy_session_id].discard(queue)
+            if not self._session_connections[galaxy_session_id]:
+                del self._session_connections[galaxy_session_id]
         self._broadcast_connections.discard(queue)
         log.debug(
-            "SSE connection closed for user_id=%s (total=%d)",
+            "SSE connection closed for user_id=%s session_id=%s (total=%d)",
             user_id,
+            galaxy_session_id,
             len(self._broadcast_connections),
         )
 
@@ -135,6 +153,15 @@ def push_to_user(self, user_id: int, event: SSEEvent) -> None:
         for queue in list(self._connections.get(user_id, [])):
             self._safe_put(queue, event)
 
+    def push_to_session(self, galaxy_session_id: int, event: SSEEvent) -> None:
+        """Thread-safe. Push an event to all SSE connections for a specific galaxy_session.
+
+        Used to route per-browser events (e.g. history updates for anonymous
+        histories) when there is no registered ``user_id`` to key on.
+        """
+        for queue in list(self._session_connections.get(galaxy_session_id, [])):
+            self._safe_put(queue, event)
+
     def push_broadcast(self, event: SSEEvent) -> None:
         """Thread-safe. Push an event to ALL connected SSE clients."""
         for queue in list(self._broadcast_connections):
@@ -189,6 +216,7 @@ async def stream(
         user_id: Optional[int],
         catch_up: Optional[SSEEvent] = None,
         keepalive: float = 30.0,
+        galaxy_session_id: Optional[int] = None,
     ) -> AsyncIterator[str]:
         """Yield SSE-framed strings for one connected client.
 
@@ -198,7 +226,7 @@ async def stream(
         what the service passes in (typically ``request.is_disconnected`` from
         starlette) so the manager stays framework-agnostic.
         """
-        queue = self.connect(user_id)
+        queue = self.connect(user_id, galaxy_session_id)
         if catch_up is not None:
             await queue.put(catch_up)
         try:
@@ -211,4 +239,4 @@ async def stream(
                 except asyncio.TimeoutError:
                     yield ": keepalive\n\n"
         finally:
-            self.disconnect(user_id, queue)
+            self.disconnect(user_id, queue, galaxy_session_id)

lib/galaxy/managers/sse_dispatch.py

@@ -123,14 +123,21 @@ def notify_broadcast(self, payload: str, event_id: Optional[str] = None) -> None
             },
         )
 
-    def history_update(self, user_updates: dict[str, list[int]], event_id: Optional[str] = None) -> None:
-        self._send(
-            "history_update",
-            {
-                "user_updates": user_updates,
-                "event_id": event_id or make_event_id(),
-            },
-        )
+    def history_update(
+        self,
+        user_updates: dict[str, list[int]],
+        event_id: Optional[str] = None,
+        session_updates: Optional[dict[str, list[int]]] = None,
+    ) -> None:
+        kwargs: dict[str, Any] = {
+            "user_updates": user_updates,
+            "event_id": event_id or make_event_id(),
+        }
+        if session_updates:
+            # Only include when non-empty: anonymous histories are uncommon on
+            # most deployments, and an empty dict is wasted wire payload.
+            kwargs["session_updates"] = session_updates
+        self._send("history_update", kwargs)
 
     def entry_point_update(self, user_id: int, event_id: Optional[str] = None) -> None:
         """Fan out a wake-up ``entry_point_update`` event for one user.

lib/galaxy/queue_worker/__init__.py

@@ -69,10 +69,14 @@ class HistoryUpdatePayload(TypedDict, total=False):
     """Wire contract for the ``history_update`` control-task kwargs.
 
     ``user_updates`` maps stringified user IDs to lists of (unencoded) history IDs.
-    Stringified because AMQP JSON serialization coerces dict keys to strings.
+    ``session_updates`` is the parallel route for anonymous-owned histories,
+    keyed by stringified ``galaxy_session.id`` (the dispatch key never leaves
+    the server — browsers never see it). Stringified because AMQP JSON
+    serialization coerces dict keys to strings.
     """
 
     user_updates: dict[str, list[int]]
+    session_updates: dict[str, list[int]]
     event_id: Optional[str]
 
 
@@ -423,7 +427,9 @@ def history_update(app: "MinimalManagerApp", **kwargs) -> None:
     """Push SSE history update events to connected users on this worker process.
 
     Encodes integer history IDs here (not in the monitor) so the manager layer
-    stays free of presentation/security concerns.
+    stays free of presentation/security concerns. Handles both user-keyed
+    routing (registered users) and galaxy_session-keyed routing (anonymous
+    histories, which have ``user_id IS NULL``).
     """
     payload = cast(HistoryUpdatePayload, kwargs)
     sse_manager = app[SSEConnectionManager]
@@ -435,6 +441,12 @@ def history_update(app: "MinimalManagerApp", **kwargs) -> None:
         data = json.dumps({"history_ids": encoded_ids})
         event = SSEEvent(event="history_update", data=data, id=event_id)
         sse_manager.push_to_user(user_id, event)
+    for session_id_str, history_ids in payload.get("session_updates", {}).items():
+        session_id = int(session_id_str)
+        encoded_ids = [encode(hid) for hid in history_ids]
+        data = json.dumps({"history_ids": encoded_ids})
+        event = SSEEvent(event="history_update", data=data, id=event_id)
+        sse_manager.push_to_session(session_id, event)
 
 
 def entry_point_update(app: "MinimalManagerApp", **kwargs) -> None:

lib/galaxy/webapps/galaxy/services/events.py

@@ -32,7 +32,13 @@ def open_stream(
         last_event_id: Optional[str],
         is_disconnected: IsDisconnected,
     ) -> AsyncIterator[str]:
-        """Open an SSE events stream; anonymous users receive only broadcasts."""
+        """Open an SSE events stream.
+
+        Anonymous users still register under their ``galaxy_session.id`` so the
+        server can route per-session events (e.g. ``history_update`` for
+        anonymous-owned histories) even when ``user_id`` is ``None``.
+        """
         user_id = user_context.user.id if not user_context.anonymous else None
+        session_id = user_context.galaxy_session.id if user_context.galaxy_session else None
         catch_up = self.notifications.build_status_catchup(user_context, last_event_id)
-        return self.sse_manager.stream(is_disconnected, user_id, catch_up=catch_up)
+        return self.sse_manager.stream(is_disconnected, user_id, catch_up=catch_up, galaxy_session_id=session_id)