Replace notification polling with Server-Sent Events (SSE)

Add real-time notification delivery via SSE to replace the 30-second
polling interval. The SSE endpoint streams notification_update,
broadcast_update, and notification_status events to connected clients.

Backend:
- New SSEConnectionManager (lib/galaxy/managers/sse.py) maps user IDs
  to asyncio queues with thread-safe push via call_soon_threadsafe
- SSE streaming endpoint at GET /api/notifications/stream with
  Last-Event-ID catch-up support and 30s keepalive
- Kombu control tasks (notify_users, notify_broadcast) fan out events
  across all Galaxy worker processes
- Existing polling API unchanged for backward compatibility

Frontend:
- New useNotificationSSE composable using EventSource with auto-reconnect
- notificationsStore tries SSE first, falls back to polling after 5+
  consecutive errors

Tests:
- API integration tests for SSE event delivery, broadcasts, and reconnect
- Selenium E2E tests for notification appearance and bell indicator

Add SSE-based real-time history update notifications

Replace aggressive 3-second history polling with Server-Sent Events
driven by database change detection, configurable via admin setting.

Backend:
- Add pg_notify() to PostgreSQL audit triggers for instant LISTEN/NOTIFY
- New HistoryAuditMonitor: PG LISTEN/NOTIFY with SQLite polling fallback
- New /api/events/stream SSE endpoint (uses StructuredApp, not MinimalManagerApp)
- Kombu control task "history_update" with message TTL (expiration=10s)
- Config: enable_sse_history_updates, history_audit_monitor_poll_interval

Frontend:
- Generalize useNotificationSSE → useSSE composable with event type filtering
- historyStore connects SSE for history_update events, triggers immediate refresh
- notificationsStore updated to use /api/events/stream and useSSE
- Polling kept as fallback at existing intervals

Tests:
- 5 integration tests: endpoint, dataset upload event, history ID in payload,
  cross-user isolation, polling backward compatibility

client/src/api/schema/schema.ts

@@ -1292,6 +1292,33 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/events/stream": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /**
+         * Server-Sent Events stream for real-time updates.
+         * @description Opens a Server-Sent Events (SSE) connection that pushes real-time
+         *     updates for notifications, history changes, and other events.
+         *
+         *     On reconnect, the browser sends the ``Last-Event-ID`` header automatically.
+         *     If the notification system is enabled, any notifications created since that
+         *     timestamp are delivered as a catch-up ``notification_status`` event.
+         *
+         *     Anonymous users receive only broadcast events.
+         */
+        get: operations["stream_events_api_events_stream_get"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/exports": {
         parameters: {
             query?: never;
@@ -3968,6 +3995,32 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/notifications/stream": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /**
+         * Server-Sent Events stream for real-time notification updates.
+         * @description Opens a Server-Sent Events (SSE) connection that pushes notification updates in real-time.
+         *
+         *     On reconnect, the browser sends the ``Last-Event-ID`` header automatically.
+         *     Any notifications created since that timestamp are delivered as a catch-up
+         *     ``notification_status`` event before the stream begins.
+         *
+         *     Anonymous users receive only broadcast events.
+         */
+        get: operations["stream_notifications_api_notifications_stream_get"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/notifications/{notification_id}": {
         parameters: {
             query?: never;
@@ -33209,6 +33262,46 @@ export interface operations {
             };
         };
     };
+    stream_events_api_events_stream_get: {
+        parameters: {
+            query?: never;
+            header?: {
+                "Last-Event-ID"?: string | null;
+                /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+                "run-as"?: string | null;
+            };
+            path?: never;
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            /** @description Request Error */
+            "4XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+            /** @description Server Error */
+            "5XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+        };
+    };
     index_api_exports_get: {
         parameters: {
             query?: {
@@ -42407,6 +42500,46 @@ export interface operations {
             };
         };
     };
+    stream_notifications_api_notifications_stream_get: {
+        parameters: {
+            query?: never;
+            header?: {
+                "Last-Event-ID"?: string | null;
+                /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+                "run-as"?: string | null;
+            };
+            path?: never;
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            /** @description Request Error */
+            "4XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+            /** @description Server Error */
+            "5XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+        };
+    };
     show_notification_api_notifications__notification_id__get: {
         parameters: {
             query?: never;

client/src/composables/useNotificationSSE.ts

@@ -0,0 +1,84 @@
+import { onScopeDispose, ref } from "vue";
+
+import { withPrefix } from "@/utils/redirect";
+
+/**
+ * All SSE event types the server may emit.
+ */
+export const SSE_EVENT_TYPES = [
+    "notification_update",
+    "broadcast_update",
+    "notification_status",
+    "history_update",
+] as const;
+
+export type SSEEventType = (typeof SSE_EVENT_TYPES)[number];
+
+/**
+ * Composable for connecting to the unified SSE event stream.
+ *
+ * The browser's EventSource handles reconnection automatically and
+ * sends the Last-Event-ID header so the server can catch up on missed events.
+ *
+ * @param onEvent - callback invoked for every SSE event
+ * @param eventTypes - subset of event types to listen to (defaults to all)
+ */
+export function useSSE(onEvent: (event: MessageEvent) => void, eventTypes: readonly SSEEventType[] = SSE_EVENT_TYPES) {
+    const connected = ref(false);
+    let eventSource: EventSource | null = null;
+    let consecutiveErrors = 0;
+
+    function connect() {
+        disconnect();
+        consecutiveErrors = 0;
+        const url = withPrefix("/api/events/stream");
+        eventSource = new EventSource(url);
+
+        for (const eventType of eventTypes) {
+            eventSource.addEventListener(eventType, onEvent);
+        }
+
+        eventSource.onopen = () => {
+            connected.value = true;
+            consecutiveErrors = 0;
+            // Expose a global readiness flag so Selenium tests can distinguish
+            // a working SSE pipeline from the polling fallback.
+            (window as unknown as { __galaxy_sse_connected?: boolean }).__galaxy_sse_connected = true;
+        };
+
+        eventSource.onerror = () => {
+            connected.value = false;
+            (window as unknown as { __galaxy_sse_connected?: boolean }).__galaxy_sse_connected = false;
+            consecutiveErrors++;
+            // EventSource auto-reconnects, but if we get too many errors
+            // in a row, the server likely doesn't support SSE — give up
+            // and let the caller fall back to polling.
+            if (consecutiveErrors > 5) {
+                disconnect();
+            }
+        };
+    }
+
+    function disconnect() {
+        if (eventSource) {
+            for (const eventType of eventTypes) {
+                eventSource.removeEventListener(eventType, onEvent);
+            }
+            eventSource.close();
+            eventSource = null;
+        }
+        connected.value = false;
+        (window as unknown as { __galaxy_sse_connected?: boolean }).__galaxy_sse_connected = false;
+    }
+
+    onScopeDispose(() => {
+        disconnect();
+    });
+
+    return { connect, disconnect, connected };
+}
+
+/**
+ * @deprecated Use `useSSE` instead. This alias exists for backward compatibility.
+ */
+export const useNotificationSSE = useSSE;

client/src/stores/historyStore.ts

@@ -15,6 +15,7 @@ import type { ArchivedHistoryDetailed } from "@/api/histories.archived";
 import { getGalaxyInstance } from "@/app";
 import { HistoryFilters } from "@/components/History/HistoryFilters";
 import { useResourceWatcher } from "@/composables/resourceWatcher";
+import { useSSE } from "@/composables/useNotificationSSE";
 import { useUserLocalStorage } from "@/composables/userLocalStorage";
 import {
     createAndSelectNewHistory,
@@ -31,6 +32,7 @@ import {
     ACTIVE_POLLING_INTERVAL,
     INACTIVE_POLLING_INTERVAL,
     watchHistory as watchHistorySuppliedApp,
+    watchHistoryOnce as watchHistoryOnceSuppliedApp,
 } from "@/watch/watchHistory";
 
 const PAGINATION_LIMIT = 10;
@@ -391,6 +393,26 @@ export const useHistoryStore = defineStore("historyStore", () => {
         return watchHistorySuppliedApp(app);
     }
 
+    // SSE-driven history updates: when we receive a history_update event,
+    // immediately trigger a refresh of the current history
+    const SSE_HISTORY_EVENT_TYPES = ["history_update"] as const;
+    const { connect: sseHistoryConnect } = useSSE(handleHistorySSEEvent, SSE_HISTORY_EVENT_TYPES);
+
+    function handleHistorySSEEvent(event: MessageEvent) {
+        try {
+            const data = JSON.parse(event.data);
+            const changedHistoryIds: string[] = data.history_ids ?? [];
+            // If the current history was updated, trigger a refresh
+            if (currentHistoryId.value && changedHistoryIds.includes(currentHistoryId.value)) {
+                const app = getGalaxyInstance();
+                watchHistoryOnceSuppliedApp(app);
+            }
+        } catch (e) {
+            console.error("Error handling history SSE event:", e);
+        }
+    }
+
+    // Polling fallback — keeps running as a safety net even when SSE is connected
     const {
         startWatchingResource: startWatchingHistory,
         stopWatchingResource: stopWatchingHistory,
@@ -400,6 +422,13 @@ export const useHistoryStore = defineStore("historyStore", () => {
         longPollingInterval: INACTIVE_POLLING_INTERVAL,
     });
 
+    function startWatchingHistoryWithSSE() {
+        // Always start polling as a baseline
+        startWatchingHistory();
+        // Also connect SSE for instant updates
+        sseHistoryConnect();
+    }
+
     async function loadHistoryById(historyId: string) {
         if (!isLoadingHistory.has(historyId)) {
             isLoadingHistory.add(historyId);
@@ -525,7 +554,7 @@ export const useHistoryStore = defineStore("historyStore", () => {
         restoreHistory,
         restoreHistories,
         handleTotalCountChange,
-        startWatchingHistory,
+        startWatchingHistory: startWatchingHistoryWithSSE,
         stopWatchingHistory,
         isWatchingHistory,
         loadCurrentHistory,

client/src/stores/notificationsStore.ts

@@ -1,9 +1,10 @@
 import { defineStore } from "pinia";
-import { computed, ref } from "vue";
+import { computed, ref, watch } from "vue";
 
 import { GalaxyApi } from "@/api";
 import type { NotificationChanges, UserNotification, UserNotificationsBatchUpdateRequest } from "@/api/notifications";
 import { useResourceWatcher } from "@/composables/resourceWatcher";
+import { useSSE } from "@/composables/useNotificationSSE";
 import { rethrowSimple } from "@/utils/simple-error";
 import { mergeObjectListsById } from "@/utils/utils";
 
@@ -13,11 +14,6 @@ const ACTIVE_POLLING_INTERVAL = 30000; // 30 seconds
 const INACTIVE_POLLING_INTERVAL = ACTIVE_POLLING_INTERVAL * 20; // 10 minutes
 
 export const useNotificationsStore = defineStore("notificationsStore", () => {
-    const { startWatchingResource: startWatchingNotifications, stopWatchingResource: stopWatchingNotifications } =
-        useResourceWatcher(getNotificationStatus, {
-            shortPollingInterval: ACTIVE_POLLING_INTERVAL,
-            longPollingInterval: INACTIVE_POLLING_INTERVAL,
-        });
     const broadcastsStore = useBroadcastsStore();
 
     const totalUnreadCount = ref<number>(0);
@@ -25,9 +21,82 @@ export const useNotificationsStore = defineStore("notificationsStore", () => {
 
     const loadingNotifications = ref<boolean>(false);
     const lastNotificationUpdate = ref<Date | null>(null);
+    const wantSSE = ref(true);
 
     const unreadNotifications = computed(() => notifications.value.filter((n) => !n.seen_time));
 
+    // --- SSE setup (listen only for notification event types) ---
+    const NOTIFICATION_EVENT_TYPES = ["notification_update", "broadcast_update", "notification_status"] as const;
+    const {
+        connect: sseConnect,
+        disconnect: sseDisconnect,
+        connected: sseConnected,
+    } = useSSE(handleSSEEvent, NOTIFICATION_EVENT_TYPES);
+
+    // --- Polling fallback ---
+    const { startWatchingResource: startPolling, stopWatchingResource: stopPolling } = useResourceWatcher(
+        getNotificationStatus,
+        {
+            shortPollingInterval: ACTIVE_POLLING_INTERVAL,
+            longPollingInterval: INACTIVE_POLLING_INTERVAL,
+        },
+    );
+
+    function stopWatchingNotifications() {
+        sseDisconnect();
+        stopPolling();
+    }
+
+    // When SSE connection drops and doesn't recover, fall back to polling
+    watch(sseConnected, (isConnected) => {
+        if (!isConnected && wantSSE.value) {
+            // SSE disconnected but we still want updates — don't start polling
+            // immediately, EventSource will auto-reconnect. Only if useSSE is
+            // set to false (after too many errors) do we fall back.
+        }
+    });
+
+    watch(wantSSE, (wantSSE) => {
+        if (!wantSSE) {
+            sseDisconnect();
+            startPolling();
+        }
+    });
+
+    function handleSSEEvent(event: MessageEvent) {
+        try {
+            const data = JSON.parse(event.data);
+            switch (event.type) {
+                case "notification_update":
+                    notifications.value = mergeObjectListsById(
+                        notifications.value,
+                        [data as UserNotification],
+                        "create_time",
+                        "desc",
+                    );
+                    updateUnreadCount();
+                    break;
+                case "broadcast_update":
+                    broadcastsStore.updateBroadcasts([data]);
+                    break;
+                case "notification_status":
+                    // Full catch-up on reconnect (same shape as GET /api/notifications/status)
+                    totalUnreadCount.value = data.total_unread_count;
+                    notifications.value = mergeObjectListsById(
+                        notifications.value,
+                        data.notifications as UserNotification[],
+                        "create_time",
+                        "desc",
+                    );
+                    broadcastsStore.updateBroadcasts(data.broadcasts);
+                    break;
+            }
+            lastNotificationUpdate.value = new Date();
+        } catch (e) {
+            console.error("Error handling SSE event:", e);
+        }
+    }
+
     async function loadNotifications() {
         const { data, error } = await GalaxyApi().GET("/api/notifications");
 
@@ -76,6 +145,29 @@ export const useNotificationsStore = defineStore("notificationsStore", () => {
         }
     }
 
+    async function startWatchingNotifications() {
+        // Always do an initial load first
+        if (!lastNotificationUpdate.value) {
+            try {
+                loadingNotifications.value = true;
+                await broadcastsStore.loadBroadcasts();
+                await loadNotifications();
+                updateUnreadCount();
+                lastNotificationUpdate.value = new Date();
+            } catch (e) {
+                console.error(e);
+            } finally {
+                loadingNotifications.value = false;
+            }
+        }
+
+        if (wantSSE.value) {
+            sseConnect();
+        } else {
+            startPolling();
+        }
+    }
+
     async function updateBatchNotification(request: UserNotificationsBatchUpdateRequest) {
         const { error } = await GalaxyApi().PUT("/api/notifications", {
             body: request,
@@ -88,7 +180,10 @@ export const useNotificationsStore = defineStore("notificationsStore", () => {
         if (request.changes.deleted) {
             notifications.value = notifications.value.filter((n) => !request.notification_ids.includes(n.id));
         }
-        startWatchingNotifications();
+        // If not using SSE, trigger a poll to refresh state
+        if (!sseConnected.value) {
+            startWatchingNotifications();
+        }
     }
 
     async function updateNotification(notification: UserNotification, changes: NotificationChanges) {

doc/source/admin/galaxy_options.rst

@@ -5804,6 +5804,34 @@
 :Type: str
 
 
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+``enable_sse_history_updates``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+:Description:
+    Enables real-time history update notifications via Server-Sent
+    Events (SSE). When enabled, a background monitor watches for
+    history changes (via PostgreSQL LISTEN/NOTIFY or audit table
+    polling as a fallback for SQLite) and pushes update signals to
+    connected browsers, replacing aggressive 3-second polling.
+:Default: ``false``
+:Type: bool
+
+
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+``history_audit_monitor_poll_interval``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+:Description:
+    The interval in seconds between history audit table polls when
+    using the polling fallback (SQLite or when PostgreSQL
+    LISTEN/NOTIFY is unavailable). Only used when
+    enable_sse_history_updates is true. Lower values mean faster
+    updates but more database queries. Recommended range: 1-5 seconds.
+:Default: ``2``
+:Type: int
+
+
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 ``enable_notification_system``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

lib/galaxy/app/__init__.py

@@ -77,6 +77,10 @@ from galaxy.managers.notification import NotificationManager
 from galaxy.managers.object_store_instances import UserObjectStoreResolverImpl
 from galaxy.managers.roles import RoleManager
 from galaxy.managers.session import GalaxySessionManager
+from galaxy.managers.sse import (
+    SSEConnectionManager,
+    SSEEventDispatcher,
+)
 from galaxy.managers.tasks import (
     AsyncTasksManager,
     CeleryAsyncTasksManager,
@@ -150,8 +154,12 @@ from galaxy.structured_app import (
 )
 from galaxy.tool_shed.cache import ToolShedRepositoryCache
 from galaxy.tool_shed.galaxy_install.client import InstallationTarget
-from galaxy.tool_shed.galaxy_install.installed_repository_manager import InstalledRepositoryManager
-from galaxy.tool_shed.galaxy_install.update_repository_manager import UpdateRepositoryManager
+from galaxy.tool_shed.galaxy_install.installed_repository_manager import (
+    InstalledRepositoryManager,
+)
+from galaxy.tool_shed.galaxy_install.update_repository_manager import (
+    UpdateRepositoryManager,
+)
 from galaxy.tool_util.data import ToolDataTableManager as BaseToolDataTableManager
 from galaxy.tool_util.deps import containers
 from galaxy.tool_util.deps.dependencies import AppInfo
@@ -250,7 +258,11 @@ class SentryClientMixin:
                         #  "cannot find 'file_name' while searching for 'species_chromosomes.file_name'"]
                         # If we don't do this issues are never properly grouped since by default the calling stack is inspected,
                         # and that is always unique in cheetah as it is dynamically generated.
-                        event["fingerprint"] = [str(exc_value), str(exc_value.tool_version), str(exc_value.__cause__)]
+                        event["fingerprint"] = [
+                            str(exc_value),
+                            str(exc_value.tool_version),
+                            str(exc_value.__cause__),
+                        ]
                         event.setdefault("tags", {}).update(
                             {
                                 "tool_is_latest": exc_value.is_latest,
@@ -307,7 +319,10 @@ class MinimalGalaxyApplication(BasicSharedApp, HaltableContainer, SentryClientMi
         config_file = kwargs.get("global_conf", {}).get("__file__", None)
         if config_file:
             log.debug('Using "galaxy.ini" config file: %s', config_file)
-        self._configure_models(check_migrate_databases=self.config.check_migrate_databases, config_file=config_file)
+        self._configure_models(
+            check_migrate_databases=self.config.check_migrate_databases,
+            config_file=config_file,
+        )
         # Security helper
         self._configure_security()
         self._register_singleton(IdEncodingHelper, self.security)
@@ -411,7 +426,11 @@ class MinimalGalaxyApplication(BasicSharedApp, HaltableContainer, SentryClientMi
         index_help = getattr(self.config, "index_tool_help", True)
         self.toolbox_search = self._register_singleton(
             ToolBoxSearch,
-            ToolBoxSearch(self.toolbox, index_dir=self.config.tool_search_index_dir, index_help=index_help),
+            ToolBoxSearch(
+                self.toolbox,
+                index_dir=self.config.tool_search_index_dir,
+                index_help=index_help,
+            ),
         )
 
     @property
@@ -496,8 +515,9 @@ class MinimalGalaxyApplication(BasicSharedApp, HaltableContainer, SentryClientMi
         templates = ConfiguredObjectStoreTemplates.from_app_config(self.config, vault_configured=vault_configured)
         self.object_store_templates = self._register_singleton(ConfiguredObjectStoreTemplates, templates)
         user_object_store_resolver = self._register_abstract_singleton(
-            UserObjectStoreResolver, UserObjectStoreResolverImpl  # type: ignore[type-abstract]
-        )  # Ignored because of https://github.com/python/mypy/issues/4717
+            UserObjectStoreResolver,  # type: ignore[type-abstract]  # https://github.com/python/mypy/issues/4717
+            UserObjectStoreResolverImpl,
+        )
         kwds["user_object_store_resolver"] = user_object_store_resolver
         self.object_store = build_object_store_from_config(self.config, **kwds)
 
@@ -600,7 +620,13 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication):
 
     model: GalaxyModelMapping
 
-    def __init__(self, configure_logging=True, use_converters=True, use_display_applications=True, **kwargs):
+    def __init__(
+        self,
+        configure_logging=True,
+        use_converters=True,
+        use_display_applications=True,
+        **kwargs,
+    ):
         super().__init__(**kwargs)
         self._register_singleton(MinimalManagerApp, self)  # type: ignore[type-abstract]
         self.execution_timer_factory = self._register_singleton(
@@ -617,7 +643,8 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication):
         # Initialize job metrics manager, needs to be in place before
         # config so per-destination modifications can be made.
         self.job_metrics = self._register_singleton(
-            JobMetrics, JobMetrics(self.config.job_metrics_config_file, self.config.job_metrics, app=self)
+            JobMetrics,
+            JobMetrics(self.config.job_metrics_config_file, self.config.job_metrics, app=self),
         )
         # Initialize the job management configuration
         self.job_config = self._register_singleton(jobs.JobConfiguration)
@@ -655,11 +682,15 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication):
         self.role_manager = self._register_singleton(RoleManager)
         self.job_manager = self._register_singleton(JobManager)
 
+        # SSE dispatcher must be registered before NotificationManager so Lagom
+        # can auto-inject the Optional[SSEEventDispatcher] constructor arg.
+        self._register_singleton(SSEEventDispatcher, SSEEventDispatcher(self))
         self.notification_manager = self._register_singleton(NotificationManager)
         self.interactivetool_manager = InteractiveToolManager(self)
 
         self.task_manager = self._register_abstract_singleton(
-            AsyncTasksManager, CeleryAsyncTasksManager  # type: ignore[type-abstract]  # https://github.com/python/mypy/issues/4717
+            AsyncTasksManager,  # type: ignore[type-abstract]  # https://github.com/python/mypy/issues/4717
+            CeleryAsyncTasksManager,
         )
 
         # ConfiguredFileSources
@@ -671,10 +702,12 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication):
         self._register_singleton(FileSourcePluginLoader, file_source_plugin_loader)
         self.file_source_templates = self._register_singleton(ConfiguredFileSourceTemplates, templates)
         self._register_singleton(
-            UserDefinedFileSourcesConfig, UserDefinedFileSourcesConfig.from_app_config(self.config)
+            UserDefinedFileSourcesConfig,
+            UserDefinedFileSourcesConfig.from_app_config(self.config),
         )
         user_defined_file_sources = self._register_abstract_singleton(
-            UserDefinedFileSources, UserDefinedFileSourcesImpl  # type: ignore[type-abstract]  # https://github.com/python/mypy/issues/4717
+            UserDefinedFileSources,  # type: ignore[type-abstract]  # https://github.com/python/mypy/issues/4717
+            UserDefinedFileSourcesImpl,
         )
         configured_file_source_conf: ConfiguredFileSourcesConf = ConfiguredFileSourcesConf.from_app_config(self.config)
         file_sources = ConfiguredFileSources(
@@ -690,7 +723,8 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication):
         # Load security policy.
         self.security_agent = self.model.security_agent
         self.host_security_agent = galaxy.model.security.HostAgent(
-            self.security_agent.sa_session, permitted_actions=self.security_agent.permitted_actions
+            self.security_agent.sa_session,
+            permitted_actions=self.security_agent.permitted_actions,
         )
 
         # We need the datatype registry for running certain tasks that modify HDAs, and to build the registry we need
@@ -791,6 +825,7 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT
             ("queue worker", self._shutdown_queue_worker),
             ("file watcher", self._shutdown_watcher),
             ("database heartbeat", self._shutdown_database_heartbeat),
+            ("history audit monitor", self._shutdown_history_audit_monitor),
             ("workflow scheduler", self._shutdown_scheduling_manager),
             ("object store", self._shutdown_object_store),
             ("job manager", self._shutdown_job_manager),
@@ -809,17 +844,23 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT
         # queue_worker *can* be initialized with a queue, but here we don't
         # want to and we'll allow postfork to bind and start it.
         self.queue_worker = self._register_singleton(GalaxyQueueWorker, GalaxyQueueWorker(self))
+        # SSE connection manager for real-time notification push
+        self.sse_connection_manager = self._register_singleton(SSEConnectionManager)
 
         # AI agent registry and service
         agent_registry = build_agent_registry(self.config)
         self._register_singleton(AgentRegistry, agent_registry)
-        self._register_singleton(AgentService, AgentService(self.config, JobQueryManager(self), agent_registry))
+        self._register_singleton(
+            AgentService,
+            AgentService(self.config, JobQueryManager(self), agent_registry),
+        )
 
         self.dependency_resolvers_view = self._register_singleton(
             DependencyResolversView, DependencyResolversView(self)
         )
         self.test_data_resolver = self._register_singleton(
-            TestDataResolver, TestDataResolver(file_dirs=self.config.tool_test_data_directories)
+            TestDataResolver,
+            TestDataResolver(file_dirs=self.config.tool_test_data_directories),
         )
         self.api_keys_manager = self._register_singleton(ApiKeyManager)
 
@@ -876,7 +917,9 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT
         # Start the heartbeat process if configured and available
         if self.config.use_heartbeat:
             self.heartbeat = heartbeat.Heartbeat(
-                self.config, period=self.config.heartbeat_interval, fname=self.config.heartbeat_log
+                self.config,
+                period=self.config.heartbeat_interval,
+                fname=self.config.heartbeat_log,
             )
             self.heartbeat.daemon = True
             self.application_stack.register_postfork_function(self.heartbeat.start)
@@ -886,7 +929,9 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT
             from galaxy.authnz import managers
 
             self.authnz_manager = managers.AuthnzManager(
-                self, self.config.oidc_config_file, self.config.oidc_backends_config_file
+                self,
+                self.config.oidc_config_file,
+                self.config.oidc_backends_config_file,
             )
 
             # If there is only a single external authentication provider in use
@@ -923,7 +968,12 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT
                 self.workflow_completion_hook_registry,
             )
             self.application_stack.register_postfork_function(self.workflow_completion_monitor.start)
-            self.haltables.append(("WorkflowCompletionMonitor", self.workflow_completion_monitor.shutdown_monitor))
+            self.haltables.append(
+                (
+                    "WorkflowCompletionMonitor",
+                    self.workflow_completion_monitor.shutdown_monitor,
+                )
+            )
 
         # Start the job manager
         self.application_stack.register_postfork_function(self.job_manager.start)
@@ -942,12 +992,20 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT
         self.database_heartbeat.add_change_callback(self.watchers.change_state)
         self.application_stack.register_postfork_function(self.database_heartbeat.start)
 
+        # History audit monitor for SSE-based history updates
+        if self.config.enable_sse_history_updates:
+            from galaxy.managers.history_audit_monitor import HistoryAuditMonitor
+
+            self._history_audit_monitor = self._register_singleton(HistoryAuditMonitor)
+            self.application_stack.register_postfork_function(self._history_audit_monitor.start)
+
         # Start web stack message handling
         self.application_stack.register_postfork_function(self.application_stack.start)
         self.application_stack.register_postfork_function(self.queue_worker.bind_and_start)
         # Reload toolbox to pick up changes to toolbox made after master was ready
         self.application_stack.register_postfork_function(
-            lambda: reload_toolbox(self, save_integrated_tool_panel=False), post_fork_only=True
+            lambda: reload_toolbox(self, save_integrated_tool_panel=False),
+            post_fork_only=True,
         )
         # Delay toolbox index until after startup
         self.application_stack.register_postfork_function(
@@ -975,6 +1033,11 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT
     def _shutdown_database_heartbeat(self):
         self.database_heartbeat.shutdown()
 
+    def _shutdown_history_audit_monitor(self):
+        monitor = getattr(self, "_history_audit_monitor", None)
+        if monitor:
+            monitor.shutdown()
+
     def _shutdown_scheduling_manager(self):
         self.workflow_scheduling_manager.shutdown()
 

lib/galaxy/config/sample/galaxy.yml.sample

@@ -3129,6 +3129,20 @@ galaxy:
   # Message to display on the export citations tool page
   #citations_export_message_html: When writing up your analysis, remember to include all references that should be cited in order to completely describe your work. Also, please remember to <a href="https://galaxyproject.org/citing-galaxy">cite Galaxy</a>.
 
+  # Enables real-time history update notifications via Server-Sent
+  # Events (SSE). When enabled, a background monitor watches for history
+  # changes (via PostgreSQL LISTEN/NOTIFY or audit table polling as a
+  # fallback for SQLite) and pushes update signals to connected
+  # browsers, replacing aggressive 3-second polling.
+  #enable_sse_history_updates: false
+
+  # The interval in seconds between history audit table polls when using
+  # the polling fallback (SQLite or when PostgreSQL LISTEN/NOTIFY is
+  # unavailable). Only used when enable_sse_history_updates is true.
+  # Lower values mean faster updates but more database queries.
+  # Recommended range: 1-5 seconds.
+  #history_audit_monitor_poll_interval: 2
+
   # Enables the Notification System integrated in Galaxy.
   # Users can receive automatic notifications when a certain resource is
   # shared with them or when some long running operations have finished,

lib/galaxy/config/schemas/config_schema.yml

@@ -4293,6 +4293,26 @@ mapping:
         desc: |
           Message to display on the export citations tool page
 
+      enable_sse_history_updates:
+        type: bool
+        default: false
+        required: false
+        desc: |
+          Enables real-time history update notifications via Server-Sent Events (SSE).
+          When enabled, a background monitor watches for history changes (via PostgreSQL
+          LISTEN/NOTIFY or audit table polling as a fallback for SQLite) and pushes update
+          signals to connected browsers, replacing aggressive 3-second polling.
+
+      history_audit_monitor_poll_interval:
+        type: int
+        default: 2
+        required: false
+        desc: |
+          The interval in seconds between history audit table polls when using the polling
+          fallback (SQLite or when PostgreSQL LISTEN/NOTIFY is unavailable). Only used when
+          enable_sse_history_updates is true. Lower values mean faster updates but more
+          database queries. Recommended range: 1-5 seconds.
+
       enable_notification_system:
         type: bool
         default: false

lib/galaxy/managers/history_audit_monitor.py

@@ -0,0 +1,263 @@
+"""Monitor for history audit table changes.
+
+Detects history changes via PostgreSQL LISTEN/NOTIFY (instant) or by polling
+the history_audit table (SQLite fallback). Dispatches SSE events to connected
+users via Kombu control tasks.
+
+Only active when ``enable_sse_history_updates`` is True in the Galaxy config.
+"""
+
+import logging
+import select
+import threading
+import time
+from collections import (
+    defaultdict,
+    OrderedDict,
+)
+from datetime import (
+    datetime,
+    timedelta,
+)
+from typing import (
+    Any,
+    Iterator,
+    Optional,
+)
+
+from sqlalchemy import select as sa_select
+from sqlalchemy.engine import Engine
+
+from galaxy.config import GalaxyAppConfiguration
+from galaxy.managers.sse import SSEEventDispatcher
+from galaxy.model import (
+    History,
+    HistoryAudit,
+)
+from galaxy.model.mapping import GalaxyModelMapping
+
+log = logging.getLogger(__name__)
+
+CHANNEL_NAME = "galaxy_history_update"
+OWNER_CACHE_MAX = 10_000
+DEBOUNCE_SECONDS = 0.2
+
+
+class _PgListenAdapter:
+    """Thin DBAPI-level adapter for PostgreSQL LISTEN/NOTIFY.
+
+    Hides the receiving-API differences between psycopg2 (``conn.poll()`` +
+    ``conn.notifies`` list, driven by ``select.select``) and psycopg3
+    (``conn.notifies(timeout=...)`` generator). The SA URL is used to inherit
+    DSN / SSL / auth config, but the connection itself is opened directly with
+    the DBAPI driver so it stays outside the SA pool — LISTEN connections must
+    live for the lifetime of the monitor and never be returned to the pool.
+    """
+
+    def __init__(self, engine: Engine) -> None:
+        # Strip the SA ``+driver`` suffix so the raw DBAPI libraries accept the URL.
+        dsn = engine.url.set(drivername="postgresql").render_as_string(hide_password=False)
+        driver = engine.dialect.driver
+        if driver == "psycopg":
+            import psycopg  # conditional: psycopg3 driver
+
+            self._conn: Any = psycopg.connect(dsn, autocommit=True)
+            self.driver = "psycopg3"
+        else:
+            import psycopg2  # conditional: psycopg2 driver
+
+            self._conn = psycopg2.connect(dsn)
+            self._conn.autocommit = True  # same effect as set_isolation_level(AUTOCOMMIT)
+            self.driver = "psycopg2"
+
+    def listen(self, channel: str) -> None:
+        with self._conn.cursor() as cursor:
+            cursor.execute(f"LISTEN {channel};")
+
+    def close(self) -> None:
+        try:
+            self._conn.close()
+        except Exception:
+            log.debug("Error closing LISTEN connection", exc_info=True)
+
+    def poll(self, timeout: float) -> Iterator[str]:
+        """Block up to ``timeout`` seconds and yield notification payloads.
+
+        Returns an empty iterator on timeout so callers can uniformly treat
+        "nothing received in this tick" regardless of driver.
+        """
+        if self.driver == "psycopg3":
+            # psycopg3: notifies() is a blocking generator bounded by ``timeout``.
+            yield from (n.payload for n in self._conn.notifies(timeout=timeout))
+            return
+        # psycopg2: block on the socket via select(), then drain notifies list.
+        if select.select([self._conn], [], [], timeout) == ([], [], []):
+            return
+        self._conn.poll()
+        while self._conn.notifies:
+            yield self._conn.notifies.pop(0).payload
+
+
+class HistoryAuditMonitor:
+    """Background thread that monitors history_audit for changes and dispatches SSE events.
+
+    On PostgreSQL: uses LISTEN/NOTIFY for instant notification.
+    On SQLite: polls history_audit table at a configurable interval.
+    """
+
+    def __init__(
+        self,
+        config: GalaxyAppConfiguration,
+        model: GalaxyModelMapping,
+        sse_dispatcher: SSEEventDispatcher,
+    ) -> None:
+        self._config = config
+        self._model = model
+        self._dispatcher = sse_dispatcher
+        self.poll_interval: int = config.history_audit_monitor_poll_interval
+        self._is_postgres: bool = "postgres" in model.engine.name
+        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()
+
+    def start(self) -> None:
+        if self._active:
+            return
+        self._active = True
+        target = self._listen_postgres if self._is_postgres else self._poll_audit_table
+        self._thread = threading.Thread(
+            target=target,
+            name="history_audit_monitor",
+            daemon=True,
+        )
+        self._thread.start()
+        log.info(
+            "HistoryAuditMonitor started (mode=%s, interval=%ds)",
+            "pg_listen" if self._is_postgres else "poll",
+            self.poll_interval,
+        )
+
+    def shutdown(self) -> None:
+        self._active = False
+        self._exit.set()
+        if self._thread:
+            self._thread.join(timeout=5)
+
+    # --- PostgreSQL LISTEN/NOTIFY mode ---
+
+    def _listen_postgres(self) -> None:
+        """LISTEN for history update notifications.
+
+        Works against both psycopg2 and psycopg3 — whichever driver the SA
+        engine was built with. Falls back to the SQLite polling path if the
+        DBAPI driver can't be imported or the initial LISTEN fails.
+        """
+        try:
+            adapter = _PgListenAdapter(self._model.engine)
+            adapter.listen(CHANNEL_NAME)
+        except Exception:
+            log.warning(
+                "Failed to establish PostgreSQL LISTEN connection, falling back to polling",
+                exc_info=True,
+            )
+            self._poll_audit_table()
+            return
+
+        log.debug("LISTEN %s established (driver=%s)", CHANNEL_NAME, adapter.driver)
+        pending: dict[int, float] = {}  # history_id -> first_seen_time
+
+        try:
+            while not self._exit.is_set():
+                received_any = False
+                for payload in adapter.poll(self.poll_interval):
+                    received_any = True
+                    try:
+                        history_id = int(payload)
+                    except (ValueError, TypeError):
+                        continue
+                    pending.setdefault(history_id, time.monotonic())
+
+                if not received_any:
+                    # Timeout — flush anything that's been pending since last tick
+                    if pending:
+                        self._dispatch_history_updates(set(pending.keys()))
+                        pending.clear()
+                    continue
+
+                # Debounce: dispatch events that have been pending long enough
+                now = time.monotonic()
+                ready = {hid for hid, ts in pending.items() if now - ts >= DEBOUNCE_SECONDS}
+                if ready:
+                    self._dispatch_history_updates(ready)
+                    for hid in ready:
+                        del pending[hid]
+        except Exception:
+            log.exception("HistoryAuditMonitor LISTEN loop error")
+        finally:
+            adapter.close()
+
+    # --- SQLite polling fallback ---
+
+    def _poll_audit_table(self) -> None:
+        """Poll history_audit for recent changes."""
+        last_check = datetime.utcnow() - timedelta(seconds=self.poll_interval)
+
+        while not self._exit.is_set():
+            try:
+                check_time = datetime.utcnow()
+                stmt = (
+                    sa_select(HistoryAudit.history_id)
+                    .where(HistoryAudit.update_time > last_check)
+                    .group_by(HistoryAudit.history_id)
+                )
+                with self._model.new_session() as session:
+                    changed_ids = set(session.scalars(stmt).all())
+
+                if changed_ids:
+                    self._dispatch_history_updates(changed_ids)
+
+                last_check = check_time
+            except Exception:
+                log.exception("HistoryAuditMonitor poll error")
+
+            self._exit.wait(self.poll_interval)
+
+    # --- Common dispatch logic ---
+
+    def _dispatch_history_updates(self, history_ids: set[int]) -> None:
+        """Map history_ids to user_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,
+        keeping this manager free of presentation concerns.
+        """
+        # Resolve owners for unknown history_ids
+        unknown = history_ids - self._history_owner_cache.keys()
+        if unknown:
+            self._refresh_owner_cache(unknown)
+
+        user_updates: dict[str, list[int]] = defaultdict(list)
+        for history_id in history_ids:
+            user_id = self._history_owner_cache.get(history_id)
+            if user_id is not None:
+                user_updates[str(user_id)].append(history_id)
+
+        if not user_updates:
+            return
+
+        self._dispatcher.history_update(user_updates=dict(user_updates))
+
+    def _refresh_owner_cache(self, history_ids: set[int]) -> None:
+        """Look up user_id for given history_ids and update the bounded cache."""
+        try:
+            stmt = sa_select(History.id, History.user_id).where(History.id.in_(history_ids))
+            with self._model.new_session() as session:
+                for row in session.execute(stmt):
+                    self._history_owner_cache[row[0]] = row[1]
+                    self._history_owner_cache.move_to_end(row[0])
+            while len(self._history_owner_cache) > OWNER_CACHE_MAX:
+                self._history_owner_cache.popitem(last=False)
+        except Exception:
+            log.debug("Failed to refresh history owner cache", exc_info=True)

lib/galaxy/managers/notification.py

@@ -37,6 +37,7 @@ from galaxy.exceptions import (
     ObjectNotFound,
 )
 from galaxy.managers.markdown_util import to_html
+from galaxy.managers.sse import SSEEventDispatcher
 from galaxy.model import (
     GroupRoleAssociation,
     Notification,
@@ -49,6 +50,7 @@ from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.schema.notifications import (
     AnyNotificationContent,
     BroadcastNotificationCreateRequest,
+    BroadcastNotificationResponse,
     MandatoryNotificationCategory,
     MessageNotificationContent,
     NewSharedItemNotificationContent,
@@ -58,6 +60,7 @@ from galaxy.schema.notifications import (
     NotificationCreateData,
     NotificationCreateRequest,
     NotificationRecipients,
+    NotificationResponse,
     NotificationVariant,
     PersonalNotificationCategory,
     UpdateUserNotificationPreferencesRequest,
@@ -94,9 +97,15 @@ class NotificationChannelPlugin(Protocol):
 class NotificationManager:
     """Manager class to interact with the database models related with Notifications."""
 
-    def __init__(self, sa_session: galaxy_scoped_session, config: GalaxyAppConfiguration):
+    def __init__(
+        self,
+        sa_session: galaxy_scoped_session,
+        config: GalaxyAppConfiguration,
+        sse_dispatcher: Optional[SSEEventDispatcher] = None,
+    ):
         self.sa_session = sa_session
         self.config = config
+        self.sse_dispatcher = sse_dispatcher
         self.recipient_resolver = NotificationRecipientResolver(strategy=DefaultStrategy(sa_session))
         self.user_notification_columns: list[InstrumentedAttribute] = [
             Notification.id,
@@ -164,6 +173,10 @@ class NotificationManager:
         notifications_sent = self._create_associations(notification, recipient_users)
         self.sa_session.commit()
 
+        # Push SSE events to connected users via control queue
+        user_ids = [user.id for user in recipient_users]
+        self._notify_users_via_sse(user_ids, notification)
+
         return notification, notifications_sent
 
     def _create_associations(self, notification: Notification, users: list[User]) -> int:
@@ -179,6 +192,26 @@ class NotificationManager:
                 continue
         return success_count
 
+    def _notify_users_via_sse(self, user_ids: list[int], notification: Notification) -> None:
+        """Broadcast a control task to all workers to push SSE events to connected users."""
+        if not self.sse_dispatcher or not user_ids:
+            return
+        try:
+            payload = NotificationResponse.model_validate(notification).model_dump_json()
+            self.sse_dispatcher.notify_users(user_ids, payload)
+        except Exception:
+            log.warning("Failed to send SSE notification event", exc_info=True)
+
+    def _notify_broadcast_via_sse(self, notification: Notification) -> None:
+        """Broadcast a control task to all workers to push SSE broadcast events."""
+        if not self.sse_dispatcher:
+            return
+        try:
+            payload = BroadcastNotificationResponse.model_validate(notification).model_dump_json()
+            self.sse_dispatcher.notify_broadcast(payload)
+        except Exception:
+            log.warning("Failed to send SSE broadcast event", exc_info=True)
+
     def dispatch_pending_notifications_via_channels(self) -> int:
         """
         Dispatches all pending notifications to the users depending on the configured channels.
@@ -273,6 +306,7 @@ class NotificationManager:
         notification = self._create_notification_model(request)
         self.sa_session.add(notification)
         self.sa_session.commit()
+        self._notify_broadcast_via_sse(notification)
         return notification
 
     def get_user_notification(self, user: User, notification_id: int, active_only: Optional[bool] = True):
@@ -353,7 +387,10 @@ class NotificationManager:
         return result
 
     def update_user_notifications(
-        self, user: User, notification_ids: set[int], request: UserNotificationUpdateRequest
+        self,
+        user: User,
+        notification_ids: set[int],
+        request: UserNotificationUpdateRequest,
     ) -> int:
         """Updates a batch of notifications associated with the user using the requested values."""
         updated_row_count = 0
@@ -447,7 +484,8 @@ class NotificationManager:
             UserNotificationAssociation.notification_id.in_(expired_notifications_stmt)
         )
         result = cast(
-            CursorResult, self.sa_session.execute(delete_stmt, execution_options={"synchronize_session": False})
+            CursorResult,
+            self.sa_session.execute(delete_stmt, execution_options={"synchronize_session": False}),
         )
         deleted_associations_count = result.rowcount
 
@@ -474,7 +512,10 @@ class NotificationManager:
         return notification
 
     def _user_notifications_query(
-        self, user: User, since: Optional[datetime] = None, active_only: Optional[bool] = True
+        self,
+        user: User,
+        since: Optional[datetime] = None,
+        active_only: Optional[bool] = True,
     ):
         stmt = (
             select(*self.user_notification_columns)
@@ -552,7 +593,7 @@ class DefaultStrategy(NotificationRecipientResolverStrategy):
         user_ids_from_roles_stmt = self._get_all_user_ids_from_roles_query(all_role_ids)
 
         union_stmt = union(user_ids_from_groups_stmt, user_ids_from_roles_stmt)
-        user_ids_from_groups_and_roles = {id for id, in self.sa_session.execute(union_stmt)}
+        user_ids_from_groups_and_roles = {id for (id,) in self.sa_session.execute(union_stmt)}
         unique_user_ids.update(user_ids_from_groups_and_roles)
 
         stmt = select(User).where(User.id.in_(unique_user_ids))
@@ -591,7 +632,7 @@ class DefaultStrategy(NotificationRecipientResolverStrategy):
                 .where(GroupRoleAssociation.role_id.in_(role_ids))
                 .distinct()
             )
-            group_ids_from_roles = {id for id, in self.sa_session.execute(stmt) if id is not None}
+            group_ids_from_roles = {id for (id,) in self.sa_session.execute(stmt) if id is not None}
             new_group_ids = group_ids_from_roles - processed_group_ids
 
             # Get role IDs associated with any of the given group IDs
@@ -601,7 +642,7 @@ class DefaultStrategy(NotificationRecipientResolverStrategy):
                 .where(GroupRoleAssociation.group_id.in_(group_ids))
                 .distinct()
             )
-            role_ids_from_groups = {id for id, in self.sa_session.execute(stmt) if id is not None}
+            role_ids_from_groups = {id for (id,) in self.sa_session.execute(stmt) if id is not None}
             new_role_ids = role_ids_from_groups - processed_role_ids
 
             # Stop if there are no new group or role IDs to process
@@ -713,7 +754,6 @@ class EmailNotificationTemplateBuilder(Protocol):
 
 
 class MessageEmailNotificationTemplateBuilder(EmailNotificationTemplateBuilder):
-
     markdown_to = {
         TemplateFormats.HTML: to_html,
         TemplateFormats.TXT: lambda x: x,  # TODO: strip markdown?
@@ -730,9 +770,10 @@ class MessageEmailNotificationTemplateBuilder(EmailNotificationTemplateBuilder):
 
 
 class NewSharedItemEmailNotificationTemplateBuilder(EmailNotificationTemplateBuilder):
-
     def get_content(self, template_format: TemplateFormats) -> AnyNotificationContent:
-        content = NewSharedItemNotificationContent.model_construct(**self.notification.content)  # type: ignore[arg-type]
+        content = NewSharedItemNotificationContent.model_construct(
+            **self.notification.content
+        )  # type: ignore[arg-type]
         return content
 
     def get_subject(self) -> str:
@@ -741,7 +782,6 @@ class NewSharedItemEmailNotificationTemplateBuilder(EmailNotificationTemplateBui
 
 
 class EmailNotificationChannelPlugin(NotificationChannelPlugin):
-
     # Register the supported email templates here
     email_templates_by_category: dict[PersonalNotificationCategory, type[EmailNotificationTemplateBuilder]] = {
         PersonalNotificationCategory.message: MessageEmailNotificationTemplateBuilder,

lib/galaxy/managers/sse.py

@@ -0,0 +1,227 @@
+"""Server-Sent Events (SSE) connection manager for real-time notifications.
+
+Manages per-worker in-memory mapping of user IDs to asyncio.Queue instances,
+enabling push of events from any thread (e.g. Kombu control queue worker)
+to async SSE endpoint handlers running in the uvicorn event loop.
+"""
+
+import asyncio
+import logging
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime
+from typing import (
+    AsyncIterator,
+    Optional,
+    TYPE_CHECKING,
+)
+
+if TYPE_CHECKING:
+    from starlette.requests import Request
+
+    from galaxy.structured_app import MinimalManagerApp
+
+log = logging.getLogger(__name__)
+
+
+@dataclass
+class SSEEvent:
+    """An event to be sent to an SSE client."""
+
+    event: str  # e.g. "notification_update", "broadcast_update", "notification_status"
+    data: str  # JSON payload
+    id: Optional[str] = None  # ISO timestamp, used by EventSource as Last-Event-ID on reconnect
+
+    def to_wire(self) -> str:
+        """Serialize this event to the SSE wire format (``event:…\\ndata:…\\n[id:…\\n]\\n``)."""
+        frame = f"event: {self.event}\ndata: {self.data}\n"
+        if self.id:
+            frame += f"id: {self.id}\n"
+        return frame + "\n"
+
+
+class SSEConnectionManager:
+    """Per-worker manager for SSE connections.
+
+    Maps user_ids to sets of asyncio.Queue instances. Each SSE connection
+    gets its own queue. The manager is thread-safe for push operations
+    via ``loop.call_soon_threadsafe``.
+
+    Lifecycle:
+    - Instantiated once per Galaxy worker process (on app object).
+    - ``connect()`` is called from the SSE async endpoint (event loop thread).
+    - ``disconnect()`` is called from the SSE endpoint's ``finally`` block.
+    - ``push_to_user()`` / ``push_broadcast()`` are called from ANY thread
+      (typically the Kombu daemon thread via control task handlers).
+    """
+
+    def __init__(self) -> None:
+        self._connections: dict[int, set[asyncio.Queue]] = defaultdict(set)
+        self._broadcast_connections: set[asyncio.Queue] = set()
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+
+    def _ensure_loop(self) -> None:
+        """Capture the running asyncio event loop. Must be called from async context."""
+        if self._loop is None or self._loop.is_closed():
+            self._loop = asyncio.get_running_loop()
+
+    # -- Called from ASYNC context (uvicorn event loop thread) --
+
+    def connect(self, user_id: Optional[int]) -> 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.
+        """
+        self._ensure_loop()
+        queue: asyncio.Queue = asyncio.Queue(maxsize=64)
+        if user_id is not None:
+            self._connections[user_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)",
+            user_id,
+            len(self._broadcast_connections),
+        )
+        return queue
+
+    def disconnect(self, user_id: Optional[int], queue: asyncio.Queue) -> None:
+        """Unregister an SSE connection.
+
+        Called from the SSE endpoint's ``finally`` block (async context).
+        """
+        if user_id is not None:
+            self._connections[user_id].discard(queue)
+            if not self._connections[user_id]:
+                del self._connections[user_id]
+        self._broadcast_connections.discard(queue)
+        log.debug(
+            "SSE connection closed for user_id=%s (total=%d)",
+            user_id,
+            len(self._broadcast_connections),
+        )
+
+    # -- Called from ANY thread (Kombu thread or async) --
+
+    def push_to_user(self, user_id: int, event: SSEEvent) -> None:
+        """Thread-safe. Push an event to all SSE connections for a specific user."""
+        for queue in list(self._connections.get(user_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):
+            self._safe_put(queue, event)
+
+    def _safe_put(self, queue: asyncio.Queue, event: SSEEvent) -> None:
+        """Cross the thread boundary safely using ``call_soon_threadsafe``."""
+        if self._loop is None or self._loop.is_closed():
+            return
+        try:
+            self._loop.call_soon_threadsafe(self._do_put, queue, event)
+        except RuntimeError:
+            # Event loop is closed or shutting down
+            pass
+
+    @staticmethod
+    def _do_put(queue: asyncio.Queue, event: SSEEvent) -> None:
+        """Runs ON the event loop thread. Safe to touch asyncio.Queue here."""
+        try:
+            queue.put_nowait(event)
+        except asyncio.QueueFull:
+            log.warning("SSE queue full, dropping event: %s", event.event)
+
+    @property
+    def connected_user_ids(self) -> set[int]:
+        return set(self._connections.keys())
+
+    @property
+    def total_connections(self) -> int:
+        return len(self._broadcast_connections)
+
+    # -- High-level streaming helper --
+
+    async def stream(
+        self,
+        request: "Request",
+        user_id: Optional[int],
+        catch_up: Optional[SSEEvent] = None,
+        keepalive: float = 30.0,
+    ) -> AsyncIterator[str]:
+        """Yield SSE-framed strings for one connected client.
+
+        Handles ``connect``, optional catch-up event priming, the main event
+        loop with a keepalive comment on timeout, disconnect detection, and
+        ``disconnect`` in ``finally``. Controllers should call this and return
+        the iterator wrapped in a ``StreamingResponse``.
+        """
+        queue = self.connect(user_id)
+        if catch_up is not None:
+            await queue.put(catch_up)
+        try:
+            while True:
+                if await request.is_disconnected():
+                    break
+                try:
+                    event: SSEEvent = await asyncio.wait_for(queue.get(), timeout=keepalive)
+                    yield event.to_wire()
+                except asyncio.TimeoutError:
+                    yield ": keepalive\n\n"
+        finally:
+            self.disconnect(user_id, queue)
+
+
+class SSEEventDispatcher:
+    """Fans out SSE events across all Galaxy worker processes via the control queue.
+
+    This is a thin wrapper around ``send_control_task`` so that managers can
+    depend on a narrow, injectable collaborator instead of importing the
+    queue-worker module directly.
+
+    In Celery / background-task processes the app object has no ``queue_worker``
+    (it's only built in ``UniverseApplication``), so dispatch is silently a
+    no-op there — notifications created from Celery tasks will still be
+    delivered the next time a client polls, just not pushed in real time.
+    """
+
+    def __init__(self, app: "MinimalManagerApp") -> None:
+        self._app = app
+
+    def _send(self, task: str, kwargs: dict) -> None:
+        if getattr(self._app, "queue_worker", None) is None:
+            # No control-queue publisher available (e.g. Celery worker context).
+            log.debug("SSE dispatch skipped: app has no queue_worker (task=%s)", task)
+            return
+        from galaxy.queue_worker import send_control_task  # circular: queue_worker -> app -> managers
+
+        send_control_task(self._app, task, kwargs=kwargs, expiration=10)
+
+    def notify_users(self, user_ids: list[int], payload: str, event_id: Optional[str] = None) -> None:
+        self._send(
+            "notify_users",
+            {
+                "user_ids": user_ids,
+                "payload": payload,
+                "event_id": event_id or datetime.utcnow().isoformat(),
+            },
+        )
+
+    def notify_broadcast(self, payload: str, event_id: Optional[str] = None) -> None:
+        self._send(
+            "notify_broadcast",
+            {
+                "payload": payload,
+                "event_id": event_id or datetime.utcnow().isoformat(),
+            },
+        )
+
+    def history_update(self, user_updates: dict[str, list], event_id: Optional[str] = None) -> None:
+        self._send(
+            "history_update",
+            {
+                "user_updates": user_updates,
+                "event_id": event_id or datetime.utcnow().isoformat(),
+            },
+        )

lib/galaxy/model/triggers/update_audit_table.py

@@ -54,12 +54,18 @@ def _postgres_install(engine):
                 RETURNS TRIGGER
                 LANGUAGE 'plpgsql'
             AS $BODY$
+                DECLARE
+                    _history_id integer;
                 BEGIN
                     INSERT INTO history_audit (history_id, update_time)
                     SELECT DISTINCT {id_field}, clock_timestamp() AT TIME ZONE 'UTC'
                     FROM new_table
                     WHERE {id_field} IS NOT NULL
                     ON CONFLICT DO NOTHING;
+                    FOR _history_id IN SELECT DISTINCT {id_field} FROM new_table WHERE {id_field} IS NOT NULL
+                    LOOP
+                        PERFORM pg_notify('galaxy_history_update', _history_id::text);
+                    END LOOP;
                     RETURN NULL;
                 END;
             $BODY$
@@ -77,6 +83,7 @@ def _postgres_install(engine):
                     INSERT INTO history_audit (history_id, update_time)
                     VALUES (NEW.{id_field}, clock_timestamp() AT TIME ZONE 'UTC')
                     ON CONFLICT DO NOTHING;
+                    PERFORM pg_notify('galaxy_history_update', NEW.{id_field}::text);
                     RETURN NULL;
                 END;
             $BODY$

lib/galaxy/queue_worker/__init__.py

@@ -4,6 +4,7 @@ reloading the toolbox, etc., across multiple processes.
 """
 
 import importlib
+import json
 import logging
 import math
 import socket
@@ -27,6 +28,10 @@ from kombu.pools import producers
 import galaxy.queues
 from galaxy import util
 from galaxy.config import reload_config_options
+from galaxy.managers.sse import (
+    SSEConnectionManager,
+    SSEEvent,
+)
 from galaxy.model import User
 from galaxy.tools import ToolBox
 from galaxy.tools.data_manager.manager import DataManagers
@@ -43,7 +48,12 @@ if TYPE_CHECKING:
     )
 
 
-def send_local_control_task(app: "StructuredApp", task: str, get_response: bool = False, kwargs: Optional[dict] = None):
+def send_local_control_task(
+    app: "StructuredApp",
+    task: str,
+    get_response: bool = False,
+    kwargs: Optional[dict] = None,
+):
     """
     This sends a message to the process-local control worker, which is useful
     for one-time asynchronous tasks like recalculating user disk usage.
@@ -57,7 +67,15 @@ def send_local_control_task(app: "StructuredApp", task: str, get_response: bool
     return control_task.send_task(payload, routing_key, local=True, get_response=get_response)
 
 
-def send_control_task(app, task, noop_self=False, get_response=False, routing_key="control.*", kwargs=None):
+def send_control_task(
+    app,
+    task,
+    noop_self=False,
+    get_response=False,
+    routing_key="control.*",
+    kwargs=None,
+    expiration: Optional[int] = None,
+):
     """
     This sends a control task out to all processes, useful for things like
     reloading a data table, which needs to happen individually in all
@@ -65,6 +83,7 @@ def send_control_task(app, task, noop_self=False, get_response=False, routing_ke
     Set noop_self to True to not run task for current process.
     Set get_response to True to wait for and return the task results
     as a list.
+    Set expiration to a number of seconds for message TTL.
     """
     if kwargs is None:
         kwargs = {}
@@ -73,7 +92,9 @@ def send_control_task(app, task, noop_self=False, get_response=False, routing_ke
     if noop_self:
         payload["noop"] = app.config.server_name
     control_task = ControlTask(app.queue_worker)
-    return control_task.send_task(payload=payload, routing_key=routing_key, get_response=get_response)
+    return control_task.send_task(
+        payload=payload, routing_key=routing_key, get_response=get_response, expiration=expiration
+    )
 
 
 class ControlTask:
@@ -107,7 +128,15 @@ class ControlTask:
         if message.properties["correlation_id"] == self.correlation_id:
             self.response = message.payload["result"]
 
-    def send_task(self, payload, routing_key, local=False, get_response=False, timeout=10):
+    def send_task(
+        self,
+        payload,
+        routing_key,
+        local=False,
+        get_response=False,
+        timeout=10,
+        expiration: Optional[int] = None,
+    ):
         if local:
             declare_queues = self.control_queues
         else:
@@ -129,14 +158,24 @@ class ControlTask:
                     correlation_id=self.correlation_id,
                     retry=True,
                     headers={"epoch": time.time()},
+                    expiration=expiration,
                 )
             if get_response:
-                with Consumer(self.connection, on_message=self.on_response, queues=callback_queue, no_ack=True):
+                with Consumer(
+                    self.connection,
+                    on_message=self.on_response,
+                    queues=callback_queue,
+                    no_ack=True,
+                ):
                     while self.response is self._response:
                         self.connection.drain_events(timeout=timeout)
                 return self.response
         except TimeoutError:
-            log.exception("Error waiting for task: '%s' sent with routing key '%s'", payload, routing_key)
+            log.exception(
+                "Error waiting for task: '%s' sent with routing key '%s'",
+                payload,
+                routing_key,
+            )
         except Exception:
             log.exception("Error queueing async task: '%s'. for %s", payload, routing_key)
 
@@ -189,7 +228,10 @@ def _get_new_toolbox(app: "UniverseApplication", save_integrated_tool_panel: boo
     tool_configs = app.config.tool_configs
 
     new_toolbox = ToolBox(
-        tool_configs, app.config.tool_path, app, save_integrated_tool_panel=save_integrated_tool_panel
+        tool_configs,
+        app.config.tool_path,
+        app,
+        save_integrated_tool_panel=save_integrated_tool_panel,
     )
     new_toolbox.data_manager_tools = app.toolbox.data_manager_tools
     app.datatypes_registry.load_datatype_converters(new_toolbox, use_cached=True)
@@ -309,6 +351,44 @@ def admin_job_lock(app, **kwargs):
     log.info(f"Administrative Job Lock is now set to {job_lock}. Jobs will {'not' if job_lock else 'now'} dispatch.")
 
 
+def notify_users(app, **kwargs):
+    """Push SSE events to connected users on this worker process."""
+    sse_manager = app[SSEConnectionManager]
+    user_ids = kwargs.get("user_ids", [])
+    payload = kwargs.get("payload", "{}")
+    event_id = kwargs.get("event_id")
+    event = SSEEvent(event="notification_update", data=payload, id=event_id)
+    for user_id in user_ids:
+        sse_manager.push_to_user(user_id, event)
+
+
+def notify_broadcast(app, **kwargs):
+    """Push SSE broadcast events to all connected clients on this worker process."""
+    sse_manager = app[SSEConnectionManager]
+    payload = kwargs.get("payload", "{}")
+    event_id = kwargs.get("event_id")
+    event = SSEEvent(event="broadcast_update", data=payload, id=event_id)
+    sse_manager.push_broadcast(event)
+
+
+def history_update(app, **kwargs):
+    """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.
+    """
+    sse_manager = app[SSEConnectionManager]
+    user_updates = kwargs.get("user_updates", {})
+    event_id = kwargs.get("event_id")
+    encode = app.security.encode_id
+    for user_id_str, history_ids in user_updates.items():
+        user_id = int(user_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_user(user_id, event)
+
+
 control_message_to_task = {
     "create_panel_section": create_panel_section,
     "reload_tool": reload_tool,
@@ -324,6 +404,9 @@ control_message_to_task = {
     "reconfigure_watcher": reconfigure_watcher,
     "reload_tour": reload_tour,
     "reload_core_config": reload_core_config,
+    "notify_users": notify_users,
+    "notify_broadcast": notify_broadcast,
+    "history_update": history_update,
 }
 
 
@@ -354,7 +437,14 @@ class GalaxyQueueWorker(ConsumerProducerMixin, threading.Thread):
         self.control_queues = []
         self.epoch = 0
 
-    def send_control_task(self, task, noop_self=False, get_response=False, routing_key="control.*", kwargs=None):
+    def send_control_task(
+        self,
+        task,
+        noop_self=False,
+        get_response=False,
+        routing_key="control.*",
+        kwargs=None,
+    ):
         return send_control_task(
             app=self.app,
             task=task,
@@ -374,7 +464,10 @@ class GalaxyQueueWorker(ConsumerProducerMixin, threading.Thread):
 
     def bind_and_start(self):
         # This is post-forking, so we got the correct sever name
-        log.info("Binding and starting galaxy control worker for %s", self.app.config.server_name)
+        log.info(
+            "Binding and starting galaxy control worker for %s",
+            self.app.config.server_name,
+        )
         self.exchange_queue, self.direct_queue = galaxy.queues.control_queues_from_config(self.app.config)
         self.control_queues = [self.exchange_queue, self.direct_queue]
         self.epoch = time.time()

lib/galaxy/structured_app/__init__.py

@@ -50,6 +50,7 @@ if TYPE_CHECKING:
     from galaxy.managers.histories import HistoryManager
     from galaxy.managers.interactivetool import InteractiveToolManager
     from galaxy.managers.jobs import JobSearch
+    from galaxy.managers.sse import SSEConnectionManager
     from galaxy.managers.tools import DynamicToolManager
     from galaxy.managers.users import UserManager
     from galaxy.managers.workflows import (
@@ -57,7 +58,9 @@ if TYPE_CHECKING:
         WorkflowsManager,
     )
     from galaxy.tool_shed.galaxy_install.client import DataManagersInterface
-    from galaxy.tool_shed.galaxy_install.installed_repository_manager import InstalledRepositoryManager
+    from galaxy.tool_shed.galaxy_install.installed_repository_manager import (
+        InstalledRepositoryManager,
+    )
     from galaxy.tool_util.data import ToolDataTableManager
     from galaxy.tools import ToolBox
     from galaxy.tools.cache import ToolCache
@@ -174,6 +177,7 @@ class StructuredApp(MinimalManagerApp):
     vault: Vault
     webhooks_registry: WebhooksRegistry
     queue_worker: Any  # 'galaxy.queue_worker.GalaxyQueueWorker'
+    sse_connection_manager: "SSEConnectionManager"
     data_provider_registry: Any  # 'galaxy.visualization.data_providers.registry.DataProviderRegistry'
     tool_cache: "ToolCache"
     tool_shed_repository_cache: Optional[ToolShedRepositoryCache]

lib/galaxy/webapps/galaxy/api/events.py

@@ -0,0 +1,66 @@
+"""
+API endpoint for Server-Sent Events (SSE) stream.
+
+Provides a unified event stream for all real-time push events (notifications,
+history updates, etc.) independent of the notification system configuration.
+"""
+
+import logging
+from typing import Optional
+
+from fastapi import (
+    Header,
+    Request,
+)
+from starlette.responses import StreamingResponse
+
+from galaxy.managers.context import ProvidesUserContext
+from galaxy.managers.sse import SSEConnectionManager
+from galaxy.webapps.galaxy.services.notifications import NotificationService
+from . import (
+    depends,
+    DependsOnTrans,
+    Router,
+)
+
+log = logging.getLogger(__name__)
+
+router = Router(tags=["events"])
+
+
+@router.cbv
+class FastAPIEvents:
+    sse_manager: SSEConnectionManager = depends(SSEConnectionManager)
+    notifications: NotificationService = depends(NotificationService)
+
+    @router.get(
+        "/api/events/stream",
+        summary="Server-Sent Events stream for real-time updates.",
+        response_class=StreamingResponse,
+    )
+    async def stream_events(
+        self,
+        request: Request,
+        trans: ProvidesUserContext = DependsOnTrans,
+        last_event_id: Optional[str] = Header(None, alias="Last-Event-ID"),
+    ):
+        """Opens a Server-Sent Events (SSE) connection that pushes real-time
+        updates for notifications, history changes, and other events.
+
+        On reconnect, the browser sends the ``Last-Event-ID`` header automatically.
+        If the notification system is enabled, any notifications created since that
+        timestamp are delivered as a catch-up ``notification_status`` event.
+
+        Anonymous users receive only broadcast events.
+        """
+        user_id = trans.user.id if not trans.anonymous else None
+        catch_up = self.notifications.build_status_catchup(trans, last_event_id)
+        return StreamingResponse(
+            self.sse_manager.stream(request, user_id, catch_up=catch_up),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+            },
+        )

lib/galaxy/webapps/galaxy/api/notifications.py

@@ -10,12 +10,16 @@ from typing import (
 
 from fastapi import (
     Body,
+    Header,
     Query,
+    Request,
     Response,
     status,
 )
+from starlette.responses import StreamingResponse
 
 from galaxy.managers.context import ProvidesUserContext
+from galaxy.managers.sse import SSEConnectionManager
 from galaxy.schema.notifications import (
     BroadcastNotificationCreateRequest,
     BroadcastNotificationListResponse,
@@ -51,6 +55,39 @@ router = Router(tags=["notifications"])
 @router.cbv
 class FastAPINotifications:
     service: NotificationService = depends(NotificationService)
+    sse_manager: SSEConnectionManager = depends(SSEConnectionManager)
+
+    @router.get(
+        "/api/notifications/stream",
+        summary="Server-Sent Events stream for real-time notification updates.",
+        response_class=StreamingResponse,
+    )
+    async def stream_notifications(
+        self,
+        request: Request,
+        trans: ProvidesUserContext = DependsOnTrans,
+        last_event_id: Optional[str] = Header(None, alias="Last-Event-ID"),
+    ):
+        """Opens a Server-Sent Events (SSE) connection that pushes notification updates in real-time.
+
+        On reconnect, the browser sends the ``Last-Event-ID`` header automatically.
+        Any notifications created since that timestamp are delivered as a catch-up
+        ``notification_status`` event before the stream begins.
+
+        Anonymous users receive only broadcast events.
+        """
+        self.service.notification_manager.ensure_notifications_enabled()
+        user_id = trans.user.id if not trans.anonymous else None
+        catch_up = self.service.build_status_catchup(trans, last_event_id)
+        return StreamingResponse(
+            self.sse_manager.stream(request, user_id, catch_up=catch_up),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+            },
+        )
 
     @router.get(
         "/api/notifications/status",

lib/galaxy/webapps/galaxy/services/notifications.py

@@ -14,6 +14,7 @@ from galaxy.exceptions import (
 )
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.notification import NotificationManager
+from galaxy.managers.sse import SSEEvent
 from galaxy.model import User
 from galaxy.schema.fields import Security
 from galaxy.schema.notifications import (
@@ -99,6 +100,28 @@ class NotificationService(ServiceBase):
             total_notifications_sent=1, notification=NotificationResponse.model_validate(notification)
         )
 
+    def build_status_catchup(
+        self, user_context: ProvidesUserContext, last_event_id: Optional[str]
+    ) -> Optional[SSEEvent]:
+        """Build a ``notification_status`` SSE event covering everything since ``last_event_id``.
+
+        Returns ``None`` when catch-up isn't possible (no ``Last-Event-ID``,
+        unparseable timestamp, or notifications disabled) so callers can simply
+        pass the result to ``SSEConnectionManager.stream`` without extra guards.
+        """
+        if not last_event_id or not self.notification_manager.notifications_enabled:
+            return None
+        try:
+            since = datetime.fromisoformat(last_event_id)
+        except (ValueError, TypeError):
+            return None
+        catchup = self.get_notifications_status(user_context, since)
+        return SSEEvent(
+            event="notification_status",
+            data=catchup.model_dump_json(),
+            id=datetime.utcnow().isoformat(),
+        )
+
     def get_notifications_status(self, user_context: ProvidesUserContext, since: datetime) -> NotificationStatusSummary:
         """Returns the status of (unread or updated) notifications received by the user **since** a particular date and time.
 

lib/galaxy_test/base/sse.py

@@ -0,0 +1,136 @@
+"""Shared helpers for SSE integration tests.
+
+The stream layer emits a ``ready`` event as the first frame on every connection
+so tests can synchronize on the server-side subscription rather than the
+underlying TCP socket. ``SSELineListener`` waits for that ``ready`` event before
+``start()`` returns, and propagates listener-thread exceptions back to the main
+thread instead of silently swallowing them.
+"""
+
+import queue
+import threading
+from typing import (
+    Optional,
+)
+
+import requests
+
+from galaxy.util.wait import wait_on
+
+CONNECT_TIMEOUT = 15
+DEFAULT_WAIT_TIMEOUT = 15
+
+
+def parse_sse_events(raw: str) -> list[dict]:
+    """Parse raw SSE text into a list of event dicts with ``event``, ``data``, and ``id`` keys."""
+    events: list[dict] = []
+    current: dict[str, str] = {}
+    for line in raw.split("\n"):
+        if line.startswith(":"):
+            continue  # comment / keepalive
+        if line == "":
+            if current:
+                events.append(current)
+                current = {}
+            continue
+        if ": " in line:
+            field, _, value = line.partition(": ")
+        else:
+            field, value = line.rstrip(":"), ""
+        if field in ("event", "data", "id"):
+            current[field] = value
+    if current:
+        events.append(current)
+    return events
+
+
+class SSEListenerError(Exception):
+    """Wraps an exception raised inside the listener thread."""
+
+
+class SSELineListener:
+    """Runs an SSE connection on a background thread and collects raw chunks.
+
+    ``start()`` blocks until the server-side ``ready`` event has been received,
+    guaranteeing that any event *posted after* ``start()`` returns will be seen
+    by this listener. Failures in the background thread are surfaced via
+    ``wait_for_event`` instead of silently timing out.
+    """
+
+    def __init__(
+        self,
+        url: str,
+        api_key: str,
+        headers: Optional[dict] = None,
+        timeout: int = 30,
+    ):
+        self.url = url
+        self.api_key = api_key
+        self.headers = headers or {}
+        self.timeout = timeout
+        self._collected: list[str] = []
+        self._stop = threading.Event()
+        self._ready = threading.Event()
+        self._errors: "queue.Queue[BaseException]" = queue.Queue()
+        self._thread = threading.Thread(target=self._listen, daemon=True)
+
+    def start(self) -> None:
+        self._thread.start()
+        wait_on(
+            lambda: True if self._ready.is_set() else None,
+            "SSE `ready` event",
+            timeout=CONNECT_TIMEOUT,
+        )
+        self._raise_if_errored()
+
+    def stop(self) -> None:
+        self._stop.set()
+        self._thread.join(timeout=5)
+
+    def wait_for_event(self, event_type: str, timeout: int = DEFAULT_WAIT_TIMEOUT) -> list[dict]:
+        """Block until at least one event of ``event_type`` has been observed, then return all such events."""
+
+        def _check():
+            self._raise_if_errored()
+            events = self.get_events(event_type)
+            return events if events else None
+
+        return wait_on(_check, f"SSE {event_type} event", timeout=timeout)
+
+    def get_events(self, event_type: Optional[str] = None) -> list[dict]:
+        """Return all collected events so far, optionally filtered by type."""
+        all_events = parse_sse_events("".join(self._collected))
+        if event_type is None:
+            return all_events
+        return [e for e in all_events if e.get("event") == event_type]
+
+    def _raise_if_errored(self) -> None:
+        try:
+            err = self._errors.get_nowait()
+        except queue.Empty:
+            return
+        raise SSEListenerError(f"SSE listener thread failed: {err!r}") from err
+
+    def _listen(self) -> None:
+        try:
+            resp = requests.get(
+                self.url,
+                params={"key": self.api_key},
+                headers=self.headers,
+                stream=True,
+                timeout=self.timeout,
+            )
+            if resp.status_code != 200:
+                raise RuntimeError(f"SSE connect returned HTTP {resp.status_code}: {resp.text[:200]}")
+            for chunk in resp.iter_content(chunk_size=None, decode_unicode=True):
+                if chunk:
+                    self._collected.append(chunk)
+                    if not self._ready.is_set() and "event: ready" in "".join(self._collected):
+                        self._ready.set()
+                if self._stop.is_set():
+                    break
+            resp.close()
+        except BaseException as exc:
+            self._errors.put(exc)
+            # Ensure start() doesn't hang forever on connection failure.
+            self._ready.set()

test/integration/test_history_sse.py

@@ -0,0 +1,131 @@
+"""Integration tests for SSE-based history update notifications."""
+
+import json
+from urllib.parse import urljoin
+from uuid import uuid4
+
+import requests
+
+from galaxy_test.base.populators import DatasetPopulator
+from galaxy_test.base.sse import SSELineListener
+from galaxy_test.driver.integration_util import IntegrationTestCase
+
+
+class TestHistorySSEIntegration(IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        super().handle_galaxy_config_kwds(config)
+        config["enable_celery_tasks"] = False
+        config["enable_sse_history_updates"] = True
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    def _events_stream_url(self) -> str:
+        return urljoin(self.url, "api/events/stream")
+
+    def _create_history(self, name=None) -> str:
+        name = name or f"test_history_{uuid4()}"
+        response = self._post("histories", data={"name": name}, json=True)
+        self._assert_status_code_is_ok(response)
+        return response.json()["id"]
+
+    def test_sse_events_endpoint_returns_event_stream(self):
+        """The /api/events/stream endpoint should return content-type text/event-stream."""
+        response = requests.get(
+            self._events_stream_url(),
+            params={"key": self.galaxy_interactor.api_key},
+            stream=True,
+            timeout=5,
+        )
+        assert response.status_code == 200
+        assert "text/event-stream" in response.headers.get("content-type", "")
+        response.close()
+
+    def test_sse_receives_history_update_on_dataset_upload(self):
+        """When a dataset is uploaded, a history_update SSE event should be received."""
+        history_id = self._create_history()
+
+        listener = SSELineListener(self._events_stream_url(), self.galaxy_interactor.api_key)
+        listener.start()
+        try:
+            self.dataset_populator.new_dataset(history_id, wait=False)
+            history_events = listener.wait_for_event("history_update")
+            assert len(history_events) > 0
+        finally:
+            listener.stop()
+
+    def test_history_update_contains_current_history_id(self):
+        """The history_update event should contain the history's encoded ID."""
+        history_id = self._create_history()
+
+        listener = SSELineListener(self._events_stream_url(), self.galaxy_interactor.api_key)
+        listener.start()
+        try:
+            self.dataset_populator.new_dataset(history_id, wait=False)
+            history_events = listener.wait_for_event("history_update")
+            found = any(history_id in json.loads(e["data"]).get("history_ids", []) for e in history_events)
+            assert found, f"Expected history_id '{history_id}' in history_update events, got: {history_events}"
+        finally:
+            listener.stop()
+
+    def test_history_update_is_scoped_to_owning_user(self):
+        """User A must only see history_update events for their own histories.
+
+        Inverted positive assertion: after user B's upload, user A uploads to
+        their own history and we assert A's stream contains A's encoded id and
+        not B's. This avoids a sleep-based "no events" test that was prone to
+        flaking under slow CI.
+        """
+        user_b = self._setup_user(f"{uuid4()}@galaxy.test")
+        _, user_b_api_key = self._setup_user_get_key(user_b["email"])
+
+        user_a_history_id = self._create_history()
+
+        listener = SSELineListener(self._events_stream_url(), self.galaxy_interactor.api_key)
+        listener.start()
+        try:
+            # User B creates a history and uploads to it. User A must NOT see this.
+            create_resp = requests.post(
+                urljoin(self.url, "api/histories"),
+                params={"key": user_b_api_key},
+                json={"name": "User B History"},
+            )
+            assert create_resp.status_code == 200
+            user_b_history_id = create_resp.json()["id"]
+
+            requests.post(
+                urljoin(self.url, f"api/histories/{user_b_history_id}/contents"),
+                params={"key": user_b_api_key},
+                json={"from_hda_id": None, "source": "pasted", "content": "user b content"},
+            )
+
+            # User A uploads to their own history — this is what A's stream must observe.
+            self.dataset_populator.new_dataset(user_a_history_id, wait=False)
+            history_events = listener.wait_for_event("history_update")
+        finally:
+            listener.stop()
+
+        seen_ids: set[str] = set()
+        for event in history_events:
+            seen_ids.update(json.loads(event["data"]).get("history_ids", []))
+        assert user_a_history_id in seen_ids, f"User A missed its own history_update: {history_events}"
+        assert (
+            user_b_history_id not in seen_ids
+        ), f"User A received history_update for user B's history ({user_b_history_id}): {history_events}"
+
+    def test_existing_polling_api_still_works(self):
+        """The existing current_history_json endpoint should continue to work."""
+        url = urljoin(self.url, "history/current_history_json")
+        response = requests.get(
+            url,
+            params={"key": self.galaxy_interactor.api_key},
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert "id" in data
+        assert "update_time" in data

test/integration/test_notification_sse.py

@@ -0,0 +1,186 @@
+"""Integration tests for the notification SSE (Server-Sent Events) endpoint."""
+
+from datetime import datetime
+from typing import Optional
+from urllib.parse import urljoin
+from uuid import uuid4
+
+import requests
+
+from galaxy_test.base.populators import DatasetPopulator
+from galaxy_test.base.sse import SSELineListener
+from galaxy_test.driver.integration_util import IntegrationTestCase
+
+
+def notification_test_data(subject: Optional[str] = None, message: Optional[str] = None) -> dict:
+    return {
+        "source": "integration_tests",
+        "variant": "info",
+        "category": "message",
+        "content": {
+            "category": "message",
+            "subject": subject or "Testing Subject",
+            "message": message or "Testing Message",
+        },
+    }
+
+
+def notification_broadcast_test_data(subject: Optional[str] = None, message: Optional[str] = None) -> dict:
+    return {
+        "source": "integration_tests",
+        "variant": "info",
+        "category": "broadcast",
+        "content": {
+            "category": "broadcast",
+            "subject": subject or "Testing Broadcast Subject",
+            "message": message or "Testing Broadcast Message",
+        },
+    }
+
+
+class TestNotificationSSEIntegration(IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+    framework_tool_and_types = False
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        super().handle_galaxy_config_kwds(config)
+        config["enable_celery_tasks"] = False
+        config["enable_notification_system"] = True
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    def _stream_url(self) -> str:
+        return urljoin(self.url, "api/notifications/stream")
+
+    def test_sse_endpoint_returns_event_stream(self):
+        """The SSE endpoint should return content-type text/event-stream."""
+        response = requests.get(
+            self._stream_url(),
+            params={"key": self.galaxy_interactor.api_key},
+            stream=True,
+            timeout=5,
+        )
+        assert response.status_code == 200
+        assert "text/event-stream" in response.headers.get("content-type", "")
+        response.close()
+
+    def test_sse_receives_notification_events(self):
+        """When a notification is created, the SSE stream should receive it."""
+        user = self._setup_user(f"{uuid4()}@galaxy.test")
+        _, user_api_key = self._setup_user_get_key(user["email"])
+
+        listener = SSELineListener(self._stream_url(), user_api_key)
+        listener.start()
+        try:
+            subject = f"sse_test_{uuid4()}"
+            request = {
+                "recipients": {"user_ids": [user["id"]]},
+                "notification": notification_test_data(subject=subject, message="SSE test notification"),
+            }
+            response = self._post("notifications", data=request, admin=True, json=True)
+            self._assert_status_code_is_ok(response)
+
+            notification_events = listener.wait_for_event("notification_update")
+        finally:
+            listener.stop()
+
+        assert any(
+            subject in e.get("data", "") for e in notification_events
+        ), f"Expected subject '{subject}' in SSE events, got: {notification_events}"
+
+    def test_sse_receives_broadcast_events(self):
+        """When a broadcast is created, the SSE stream should receive it."""
+        listener = SSELineListener(self._stream_url(), self.galaxy_interactor.api_key)
+        listener.start()
+        try:
+            subject = f"broadcast_sse_test_{uuid4()}"
+            payload = notification_broadcast_test_data(subject=subject)
+            response = self._post("notifications/broadcast", data=payload, admin=True, json=True)
+            self._assert_status_code_is_ok(response)
+
+            broadcast_events = listener.wait_for_event("broadcast_update")
+        finally:
+            listener.stop()
+
+        assert any(
+            subject in e.get("data", "") for e in broadcast_events
+        ), f"Expected subject '{subject}' in broadcast SSE events, got: {broadcast_events}"
+
+    def test_sse_catchup_on_reconnect(self):
+        """Reconnecting with Last-Event-ID should replay a catch-up notification_status event.
+
+        The ``Last-Event-ID`` value is the server-issued ID from a prior event,
+        not a client-side ``datetime.utcnow()``. This avoids clock-skew flake
+        between the test runner and the app in containerized CI.
+        """
+        user = self._setup_user(f"{uuid4()}@galaxy.test")
+        _, user_api_key = self._setup_user_get_key(user["email"])
+
+        # First connection: capture the server-issued event id of the first notification.
+        listener_1 = SSELineListener(self._stream_url(), user_api_key)
+        listener_1.start()
+        try:
+            subject_1 = f"first_{uuid4()}"
+            request = {
+                "recipients": {"user_ids": [user["id"]]},
+                "notification": notification_test_data(subject=subject_1),
+            }
+            response = self._post("notifications", data=request, admin=True, json=True)
+            self._assert_status_code_is_ok(response)
+            first_events = listener_1.wait_for_event("notification_update")
+        finally:
+            listener_1.stop()
+
+        last_event_id = next((e["id"] for e in first_events if e.get("id")), None)
+        assert last_event_id, f"No server-issued id on first notification event: {first_events}"
+
+        # Emit a second notification while disconnected; it should appear in the catch-up.
+        subject_2 = f"catchup_{uuid4()}"
+        request = {
+            "recipients": {"user_ids": [user["id"]]},
+            "notification": notification_test_data(subject=subject_2, message="Catch-up test"),
+        }
+        response = self._post("notifications", data=request, admin=True, json=True)
+        self._assert_status_code_is_ok(response)
+
+        # Reconnect with Last-Event-ID = the captured id. The server catch-up runs before
+        # the `ready` event and must include the missed notification.
+        listener_2 = SSELineListener(
+            self._stream_url(),
+            user_api_key,
+            headers={"Last-Event-ID": last_event_id},
+        )
+        listener_2.start()
+        try:
+            status_events = listener_2.wait_for_event("notification_status")
+        finally:
+            listener_2.stop()
+
+        assert any(
+            subject_2 in e.get("data", "") for e in status_events
+        ), f"Expected subject '{subject_2}' in catch-up event, got: {status_events}"
+
+    def test_existing_polling_api_still_works(self):
+        """The existing polling endpoint should continue to work alongside SSE."""
+        user = self._setup_user(f"{uuid4()}@galaxy.test")
+
+        before = datetime.utcnow()
+
+        subject = f"polling_test_{uuid4()}"
+        request = {
+            "recipients": {"user_ids": [user["id"]]},
+            "notification": notification_test_data(subject=subject),
+        }
+        response = self._post("notifications", data=request, admin=True, json=True)
+        self._assert_status_code_is_ok(response)
+
+        with self._different_user(user["email"]):
+            status_response = self._get(f"notifications/status?since={before.isoformat()}")
+            self._assert_status_code_is_ok(status_response)
+            status = status_response.json()
+            assert status["total_unread_count"] == 1
+            assert len(status["notifications"]) == 1
+            assert status["notifications"][0]["content"]["subject"] == subject

test/integration_selenium/test_notification_sse.py

@@ -0,0 +1,110 @@
+"""Playwright E2E test for the notification SSE pipeline.
+
+Verifies that when an admin creates a notification via the API,
+a logged-in user sees it appear in the UI in real-time (within seconds)
+without a page refresh, proving the SSE push pipeline works end-to-end.
+"""
+
+from uuid import uuid4
+
+from galaxy.util.wait import wait_on
+from galaxy_test.selenium.framework import (
+    managed_history,
+    selenium_test,
+)
+from .framework import SeleniumIntegrationTestCase
+
+SSE_CONNECT_TIMEOUT_SECONDS = 15
+
+
+class TestNotificationSSESeleniumIntegration(SeleniumIntegrationTestCase):
+    ensure_registered = True
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        super().handle_galaxy_config_kwds(config)
+        config["enable_notification_system"] = True
+        config["enable_celery_tasks"] = False
+
+    def _wait_for_sse_connected(self) -> None:
+        """Block until the frontend confirms the SSE pipeline is live.
+
+        Without this gate, the 30 s polling fallback silently masks a broken
+        SSE pipeline — the UI would still update, but via polling, and the
+        test would falsely pass.
+        """
+        wait_on(
+            lambda: True if self.driver.execute_script("return window.__galaxy_sse_connected === true") else None,
+            "window.__galaxy_sse_connected === true",
+            timeout=SSE_CONNECT_TIMEOUT_SECONDS,
+        )
+
+    @selenium_test
+    @managed_history
+    def test_notification_appears_via_sse(self):
+        """Send a notification via the API and verify it appears in the UI without refresh."""
+        # Get the logged-in user's info so we can send a notification to them
+        user_info = self._get("users/current").json()
+        user_id = user_info["id"]
+
+        # Navigate to notifications page so the store is watching
+        self.driver.get(f"{self.target_url_from_selenium}/user/notifications")
+        self._wait_for_sse_connected()
+        self.screenshot("notification_sse_before")
+
+        # Send a notification to this user via the admin API
+        subject = f"SSE E2E Test {uuid4()}"
+        notification_request = {
+            "recipients": {"user_ids": [user_id]},
+            "notification": {
+                "source": "integration_tests",
+                "variant": "info",
+                "category": "message",
+                "content": {
+                    "category": "message",
+                    "subject": subject,
+                    "message": "This notification was pushed via SSE",
+                },
+            },
+        }
+        response = self._post("notifications", data=notification_request, admin=True, json=True)
+        self._assert_status_code_is_ok(response)
+
+        # Wait for the notification to appear in the UI — SSE should push it
+        # within a few seconds, without needing a page refresh.
+        # We wait up to 15 seconds checking for the subject text to appear.
+        self.driver.wait_for_selector_visible(f"text={subject}", timeout=15000)
+        self.screenshot("notification_sse_after")
+
+    @selenium_test
+    @managed_history
+    def test_notification_bell_updates_via_sse(self):
+        """The notification bell indicator should update when a new notification arrives via SSE."""
+        user_info = self._get("users/current").json()
+        user_id = user_info["id"]
+
+        # Go to home page (bell is in masthead)
+        self.home()
+        self._wait_for_sse_connected()
+
+        # Send a notification
+        subject = f"Bell Test {uuid4()}"
+        notification_request = {
+            "recipients": {"user_ids": [user_id]},
+            "notification": {
+                "source": "integration_tests",
+                "variant": "info",
+                "category": "message",
+                "content": {
+                    "category": "message",
+                    "subject": subject,
+                    "message": "Testing bell indicator update via SSE",
+                },
+            },
+        }
+        response = self._post("notifications", data=notification_request, admin=True, json=True)
+        self._assert_status_code_is_ok(response)
+
+        # The indicator dot should appear on the bell (within the #activity-notifications element)
+        self.driver.wait_for_selector_visible("#activity-notifications .indicator", timeout=15000)
+        self.screenshot("notification_bell_indicator")