Address arch-review polish items

- Move the HistoryAuditMonitor import to the top of lib/galaxy/app —
  it's a pure Python class with no import-time side effects, so the
  conditional import added nothing. Drop the self._history_audit_monitor
  attribute and resolve the monitor via self[HistoryAuditMonitor] in
  the shutdown path so it stays off the app surface.
- Inject NotificationManager into ShareableService directly instead of
  reaching self.notification_service.notification_manager from inside
  sharable.py. Thread the new param through HistoriesService,
  PagesService, VisualizationsService and WorkflowsService (all
  Lagom-resolved, so there are no manual call sites to update).
- Add missing return annotations: EventsService.__init__ -> None,
  SSELineListener.__init__ -> None. Declare _PgListenAdapter._conn /
  .driver once at the class level so the psycopg / psycopg2 branches
  can assign without re-annotating the same attribute.
- Tighten the historyStore idempotency test: assert
  expect(deltaAfterSecond).toBe(1) instead of toBeLessThanOrEqual(1)
  so a no-op second startWatchingHistory fails the test rather than
  trivially passing. Add a comment to the SSE-event test noting that
  the mocked refreshHistoryFromPush asserts dispatch only; the actual
  refresh is covered by the Selenium SSE integration tests.

Author: mvdbeek

client/src/stores/historyStore.test.ts

@@ -114,6 +114,10 @@ describe("historyStore — config-driven SSE vs polling", () => {
         });
 
         it("triggers refreshHistoryFromPush when an SSE event names the current history", async () => {
+            // This test asserts the store's *decision* to refresh, not the refresh
+            // itself — ``refreshHistoryFromPush`` is mocked so we can observe the
+            // dispatch. The real refresh is covered end-to-end in the Selenium
+            // SSE integration tests (see test/integration_selenium/test_history_sse.py).
             const store = useHistoryStore();
             await primeStore(() => store.startWatchingHistory());
             // Drive the store to a known current-history id so the handler has
@@ -181,8 +185,10 @@ describe("historyStore — config-driven SSE vs polling", () => {
             // not two.
             await vi.advanceTimersByTimeAsync(3000);
             await flushPromises();
+            // Exactly one additional poll after the 3000ms advance — anything
+            // else means a second independent polling loop was scheduled.
             const deltaAfterSecond = mockWatchHistory.mock.calls.length - pollsAfterFirst;
-            expect(deltaAfterSecond).toBeLessThanOrEqual(1);
+            expect(deltaAfterSecond).toBe(1);
         });
     });
 });

lib/galaxy/app/__init__.py

@@ -65,6 +65,7 @@
 from galaxy.managers.folders import FolderManager
 from galaxy.managers.hdas import HDAManager
 from galaxy.managers.histories import HistoryManager
+from galaxy.managers.history_audit_monitor import HistoryAuditMonitor
 from galaxy.managers.interactivetool import InteractiveToolManager
 from galaxy.managers.jobs import (
     JobManager as JobQueryManager,
@@ -1018,10 +1019,8 @@ def __init__(self, **kwargs) -> None:
         # up. start/stop are driven by heartbeat role transitions rather than
         # postfork, so the monitor cleanly migrates when the leader dies.
         if self.config.enable_sse_history_updates:
-            from galaxy.managers.history_audit_monitor import HistoryAuditMonitor
-
-            self._history_audit_monitor = self._register_singleton(HistoryAuditMonitor)
-            self.database_heartbeat.add_audit_monitor_change_callback(self._history_audit_monitor.on_role_change)
+            monitor = self._register_singleton(HistoryAuditMonitor)
+            self.database_heartbeat.add_audit_monitor_change_callback(monitor.on_role_change)
 
         # Start web stack message handling
         self.application_stack.register_postfork_function(self.application_stack.start)
@@ -1058,9 +1057,9 @@ 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()
+        if not self.config.enable_sse_history_updates:
+            return
+        self[HistoryAuditMonitor].shutdown()
 
     def _shutdown_scheduling_manager(self):
         self.workflow_scheduling_manager.shutdown()

lib/galaxy/managers/history_audit_monitor.py

@@ -55,14 +55,19 @@ class _PgListenAdapter:
     live for the lifetime of the monitor and never be returned to the pool.
     """
 
+    # Typed once at the class level so both driver branches can assign without
+    # re-annotating the same attribute.
+    _conn: Any
+    driver: str
+
     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._conn = psycopg.connect(dsn, autocommit=True)
             self.driver = "psycopg3"
         else:
             import psycopg2  # conditional: psycopg2 driver

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

@@ -22,7 +22,7 @@
 
 
 class EventsService(ServiceBase):
-    def __init__(self, sse_manager: SSEConnectionManager, notifications: NotificationService):
+    def __init__(self, sse_manager: SSEConnectionManager, notifications: NotificationService) -> None:
         self.sse_manager = sse_manager
         self.notifications = notifications
 

lib/galaxy/webapps/galaxy/services/histories.py

@@ -94,6 +94,7 @@
     ServiceBase,
     tool_request_to_model,
 )
+from galaxy.managers.notification import NotificationManager
 from galaxy.webapps.galaxy.services.notifications import NotificationService
 from galaxy.webapps.galaxy.services.sharable import ShareableService
 from galaxy.workflow.extract import summarize
@@ -129,6 +130,7 @@ def __init__(
         filters: HistoryFilters,
         short_term_storage_allocator: ShortTermStorageAllocator,
         notification_service: NotificationService,
+        notification_manager: NotificationManager,
     ):
         super().__init__(security)
         self.manager = manager
@@ -138,7 +140,9 @@ def __init__(
         self.citations_manager = citations_manager
         self.history_export_manager = history_export_manager
         self.filters = filters
-        self.shareable_service = ShareableHistoryService(self.manager, self.serializer, notification_service)
+        self.shareable_service = ShareableHistoryService(
+            self.manager, self.serializer, notification_service, notification_manager
+        )
         self.short_term_storage_allocator = short_term_storage_allocator
 
     def index(

lib/galaxy/webapps/galaxy/services/pages.py

@@ -38,6 +38,7 @@
     ensure_celery_tasks_enabled,
     ServiceBase,
 )
+from galaxy.managers.notification import NotificationManager
 from galaxy.webapps.galaxy.services.notifications import NotificationService
 from galaxy.webapps.galaxy.services.sharable import ShareableService
 
@@ -58,11 +59,14 @@ def __init__(
         serializer: PageSerializer,
         short_term_storage_allocator: ShortTermStorageAllocator,
         notification_service: NotificationService,
+        notification_manager: NotificationManager,
     ):
         super().__init__(security)
         self.manager = manager
         self.serializer = serializer
-        self.shareable_service = ShareableService(self.manager, self.serializer, notification_service)
+        self.shareable_service = ShareableService(
+            self.manager, self.serializer, notification_service, notification_manager
+        )
         self.short_term_storage_allocator = short_term_storage_allocator
 
     def index(

lib/galaxy/webapps/galaxy/services/sharable.py

@@ -5,6 +5,7 @@
 )
 
 from galaxy.managers import base
+from galaxy.managers.notification import NotificationManager
 from galaxy.managers.sharable import (
     SharableModelManager,
     SharableModelSerializer,
@@ -62,10 +63,12 @@ def __init__(
         manager: SharableModelManager,
         serializer: SharableModelSerializer,
         notification_service: NotificationService,
+        notification_manager: NotificationManager,
     ) -> None:
         self.manager = manager
         self.serializer = serializer
         self.notification_service = notification_service
+        self.notification_manager = notification_manager
 
     def set_slug(self, trans, id: DecodedDatabaseIdField, payload: SetSlugPayload):
         item = self._get_item_by_id(trans, id)
@@ -177,17 +180,13 @@ def _get_users(self, trans, emails_or_ids: list[UserIdentifier]) -> tuple[set[Us
     def _send_notification_to_users(
         self, users_to_notify: set[User], item: SharableItem, status: ShareWithStatus, galaxy_url: Optional[str] = None
     ):
-        if (
-            self.notification_service.notification_manager.notifications_enabled
-            and not status.errors
-            and users_to_notify
-        ):
+        if self.notification_manager.notifications_enabled and not status.errors and users_to_notify:
             request = SharedItemNotificationFactory.build_notification_request(
                 item, users_to_notify, status, galaxy_url
             )
             # We can set force_sync=True here because we already have the set of users to notify
             # and there is no need to resolve them asynchronously as no groups or roles are involved.
-            self.notification_service.notification_manager.send_notification_internal(request, force_sync=True)
+            self.notification_manager.send_notification_internal(request, force_sync=True)
 
 
 class SharedItemNotificationFactory:

lib/galaxy/webapps/galaxy/services/visualizations.py

@@ -9,6 +9,7 @@
 from galaxy import exceptions
 from galaxy.managers.base import security_check
 from galaxy.managers.context import ProvidesUserContext
+from galaxy.managers.notification import NotificationManager
 from galaxy.managers.sharable import (
     slug_exists,
     SlugBuilder,
@@ -62,11 +63,14 @@ def __init__(
         manager: VisualizationManager,
         serializer: VisualizationSerializer,
         notification_service: NotificationService,
+        notification_manager: NotificationManager,
     ):
         super().__init__(security)
         self.manager = manager
         self.serializer = serializer
-        self.shareable_service = ShareableService(self.manager, self.serializer, notification_service)
+        self.shareable_service = ShareableService(
+            self.manager, self.serializer, notification_service, notification_manager
+        )
 
     # TODO: add the rest of the API actions here and call them directly from the API controller
 

lib/galaxy/webapps/galaxy/services/workflows.py

@@ -12,6 +12,7 @@
     web,
 )
 from galaxy.managers.context import ProvidesUserContext
+from galaxy.managers.notification import NotificationManager
 from galaxy.managers.workflows import (
     RefactorRequest,
     RefactorResponse,
@@ -57,11 +58,14 @@ def __init__(
         serializer: WorkflowSerializer,
         tool_shed_registry: Registry,
         notification_service: NotificationService,
+        notification_manager: NotificationManager,
     ):
         self._workflows_manager = workflows_manager
         self._workflow_contents_manager = workflow_contents_manager
         self._serializer = serializer
-        self.shareable_service = ShareableService(workflows_manager, serializer, notification_service)
+        self.shareable_service = ShareableService(
+            workflows_manager, serializer, notification_service, notification_manager
+        )
         self._tool_shed_registry = tool_shed_registry
 
     def index(

lib/galaxy_test/base/sse.py

@@ -64,7 +64,7 @@ def __init__(
         api_key: str,
         headers: Optional[dict] = None,
         timeout: int = 30,
-    ):
+    ) -> None:
         self.url = url
         self.api_key = api_key
         self.headers = headers or {}