Address review feedback for SSE notifications

Backend

- Narrow SSEEventDispatcher's dependencies. Previous constructor took
  the whole app and reached for `queue_worker`, `application_stack`,
  `config.server_name` off it (via send_control_task) — a service-
  locator pattern where the `MinimalManagerApp` annotation didn't even
  cover the accessed fields. Inject `queue_worker: Optional[
  GalaxyQueueWorker]` and `application_stack: ApplicationStack`
  directly and publish via `ControlTask(qw).send_task(...)` so the
  dispatcher is unit-testable without a full app.

- Move user-id resolution, notifications-enabled guard, and catch-up
  wiring out of `api/events.py` and `api/notifications.py`.
  Controllers are now pure `StreamingResponse` wrappers around
  `EventsService.open_stream(...)` / `NotificationService.open_stream
  (...)`, matching the three-layer architecture.

- Drop the starlette `Request` leak from `managers/sse.py`.
  `SSEConnectionManager.stream` now takes an `IsDisconnected`
  (async callable) so the manager stays framework-agnostic; services
  pass `request.is_disconnected` in.

- Centralise SSE event-id generation. New `make_event_id()` /
  `parse_event_id()` helpers in `managers/sse.py` use
  `galaxy.model.orm.now` (timezone-naive UTC, matching the rest of
  Galaxy's DB timestamps) instead of `datetime.utcnow()` scattered
  across three callers — previous code risked a tz-aware/naïve mix
  that would silently drop Last-Event-ID catch-up.

- Drop dead `app.sse_connection_manager` attribute and
  `StructuredApp.sse_connection_manager` field. Nothing reads them;
  the manager is consumed via Lagom (`depends(SSEConnectionManager)` /
  `app[SSEConnectionManager]`).

- Type-annotate the public control-task surface: `send_control_task`,
  `ControlTask.send_task`, `all_control_queues_for_declare`, SSE
  control-task handlers (`notify_users` / `notify_broadcast` /
  `history_update`), and both new stream endpoints. Tighten
  `SSEEventDispatcher.history_update` to `dict[str, list[int]]`.

Tests

- Replace the substring `data` matches with `json.loads(data)["content"]
  ["subject"]` so a regression in the NotificationResponse envelope
  (renamed keys, missing fields) actually fails. Confirmed shape
  against `NotificationResponse.model_dump_json()`.

- Assert the *absence* of pre-Last-Event-ID subjects in the catch-up
  test. Without this, a server that replays everything on every
  reconnect silently passes.

- Delete smoke tests that only checked content-type + 200 (subsumed by
  the functional tests) and the two "existing polling API still works"
  tests that didn't exercise SSE at all.

- Add `window.__galaxy_sse_last_event_ts` to the `useSSE` composable
  and gate the selenium bell/notification tests on it advancing past
  a baseline. The previous 15s wait would silently become a polling
  test if the poll interval ever dropped — this makes "update came
  from SSE" a positive assertion.

- Vitest: add a shared `_testing/sseStoreSupport.ts` helper for the
  two store tests to share a `useSSE` mock that captures the
  `onEvent` callback. Use it to synthesize real `MessageEvent`s and
  assert store-state changes for `notification_update`,
  `notification_status`, and `history_update` — the most
  consequential handlers were previously uncovered. Also save/restore
  `document.visibilityState` so the patching doesn't leak across
  tests, drop the dead `mockSseConnected` ref, tighten the
  idempotency assertion to advance time by one interval, and collapse
  the repeated orchestration into a `primeStore(...)` helper.

Author: mvdbeek

client/src/composables/useNotificationSSE.ts

@@ -14,6 +14,15 @@ export const SSE_EVENT_TYPES = [
 
 export type SSEEventType = (typeof SSE_EVENT_TYPES)[number];
 
+interface SSEDebugGlobals {
+    __galaxy_sse_connected?: boolean;
+    __galaxy_sse_last_event_ts?: number;
+}
+
+function sseGlobals(): SSEDebugGlobals {
+    return window as unknown as SSEDebugGlobals;
+}
+
 /**
  * Composable for connecting to the unified SSE event stream.
  *
@@ -27,20 +36,28 @@ export function useSSE(onEvent: (event: MessageEvent) => void, eventTypes: reado
     const connected = ref(false);
     let eventSource: EventSource | null = null;
 
+    // Selenium tests watch __galaxy_sse_last_event_ts to prove that an
+    // observable state change came from an SSE push and not the polling
+    // fallback (where __galaxy_sse_last_event_ts would never advance).
+    const trackedOnEvent = (event: MessageEvent) => {
+        sseGlobals().__galaxy_sse_last_event_ts = Date.now();
+        onEvent(event);
+    };
+
     function connect() {
         disconnect();
         const url = withPrefix("/api/events/stream");
         eventSource = new EventSource(url);
 
         for (const eventType of eventTypes) {
-            eventSource.addEventListener(eventType, onEvent);
+            eventSource.addEventListener(eventType, trackedOnEvent);
         }
 
         eventSource.onopen = () => {
             connected.value = true;
             // 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;
+            sseGlobals().__galaxy_sse_connected = true;
         };
 
         eventSource.onerror = () => {
@@ -49,20 +66,20 @@ export function useSSE(onEvent: (event: MessageEvent) => void, eventTypes: reado
             // so we must not give up on transient errors here — doing so
             // would leave the client with no updates at all.
             connected.value = false;
-            (window as unknown as { __galaxy_sse_connected?: boolean }).__galaxy_sse_connected = false;
+            sseGlobals().__galaxy_sse_connected = false;
         };
     }
 
     function disconnect() {
         if (eventSource) {
             for (const eventType of eventTypes) {
-                eventSource.removeEventListener(eventType, onEvent);
+                eventSource.removeEventListener(eventType, trackedOnEvent);
             }
             eventSource.close();
             eventSource = null;
         }
         connected.value = false;
-        (window as unknown as { __galaxy_sse_connected?: boolean }).__galaxy_sse_connected = false;
+        sseGlobals().__galaxy_sse_connected = false;
     }
 
     onScopeDispose(() => {

client/src/stores/_testing/sseStoreSupport.ts

@@ -0,0 +1,67 @@
+/**
+ * Shared test helpers for the SSE-driven stores (historyStore, notificationsStore).
+ *
+ * Both stores consume the same `useSSE` composable and need:
+ *  - a mock that captures the onEvent callback so tests can synthesize SSE messages;
+ *  - visibility-state patching without leaking across tests (JSDOM's `document`
+ *    is shared by every test in the same worker, so an unrestored
+ *    `Object.defineProperty` causes silent bleed).
+ *
+ * Because ``vi.mock`` is hoisted above module-level variables, tests must
+ * construct the SSE-mock state via ``vi.hoisted`` and then hand it to
+ * ``sseMockFactory`` from inside the ``vi.mock`` factory. See the ``.test.ts``
+ * files in this directory for the pattern.
+ */
+
+import { vi } from "vitest";
+
+export interface SSEMockState {
+    onEvent: ((event: MessageEvent) => void) | null;
+    connect: ReturnType<typeof vi.fn>;
+    disconnect: ReturnType<typeof vi.fn>;
+}
+
+/** Build the factory used with ``vi.mock("@/composables/useNotificationSSE", ...)``. */
+export function sseMockFactory(state: SSEMockState) {
+    return {
+        useSSE: vi.fn((onEvent: (event: MessageEvent) => void) => {
+            state.onEvent = onEvent;
+            return { connect: state.connect, disconnect: state.disconnect };
+        }),
+    };
+}
+
+/** Synthesize an SSE message through the captured handler. */
+export function emitSse(state: SSEMockState, type: string, payload: unknown): void {
+    if (!state.onEvent) {
+        throw new Error("useSSE was not called by the store under test — cannot emit an SSE event");
+    }
+    state.onEvent(new MessageEvent(type, { data: JSON.stringify(payload) }));
+}
+
+/**
+ * Save the current ``document.visibilityState`` descriptor and return a restorer.
+ * Call the restorer in ``afterEach`` to prevent patching from leaking into later tests.
+ */
+export function useVisibilityPatch(): {
+    set: (state: "visible" | "hidden") => void;
+    restore: () => void;
+} {
+    const original = Object.getOwnPropertyDescriptor(document, "visibilityState");
+    return {
+        set(state: "visible" | "hidden") {
+            Object.defineProperty(document, "visibilityState", {
+                configurable: true,
+                get: () => state,
+            });
+            document.dispatchEvent(new Event("visibilitychange"));
+        },
+        restore() {
+            if (original) {
+                Object.defineProperty(document, "visibilityState", original);
+            } else {
+                delete (document as unknown as Record<string, unknown>).visibilityState;
+            }
+        },
+    };
+}

client/src/stores/historyStore.test.ts

@@ -1,26 +1,23 @@
 import flushPromises from "flush-promises";
 import { createPinia, setActivePinia } from "pinia";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { ref } from "vue";
 
 import { useServerMock } from "@/api/client/__mocks__";
 
+import { emitSse, sseMockFactory, useVisibilityPatch } from "./_testing/sseStoreSupport";
 import { useHistoryStore } from "./historyStore";
 
-// Capture SSE composable usage — neither test should rely on the real
-// EventSource. The `connected` ref stays false by default; the store must
-// NOT key polling behavior off it.
-const mockSseConnect = vi.fn();
-const mockSseDisconnect = vi.fn();
-const mockSseConnected = ref(false);
-
-vi.mock("@/composables/useNotificationSSE", () => ({
-    useSSE: vi.fn(() => ({
-        connect: mockSseConnect,
-        disconnect: mockSseDisconnect,
-        connected: mockSseConnected,
-    })),
-}));
+// ``vi.mock`` is hoisted above module-level ``const`` declarations, so the
+// capture-state has to be built via ``vi.hoisted`` to be visible to the factory.
+const sseState = vi.hoisted(() => {
+    return {
+        onEvent: null as ((event: MessageEvent) => void) | null,
+        connect: vi.fn(),
+        disconnect: vi.fn(),
+    };
+});
+
+vi.mock("@/composables/useNotificationSSE", () => sseMockFactory(sseState));
 
 // `watchHistory(app)` is the polling handler invoked on the short/long
 // interval. We mock it so each invocation is observable without pulling in
@@ -40,31 +37,39 @@ vi.mock("@/app", () => ({
 
 const { server, http } = useServerMock();
 
-function configResponse(enableSse: boolean) {
-    return { enable_sse_history_updates: enableSse };
-}
-
 function registerDefaultHandlers({ enableSse }: { enableSse: boolean }) {
     server.use(
         http.get("/api/configuration", ({ response }) => {
             // eslint-disable-next-line @typescript-eslint/no-explicit-any
-            return response(200).json(configResponse(enableSse) as any);
+            return response(200).json({ enable_sse_history_updates: enableSse } as any);
         }),
     );
 }
 
+async function primeStore(startFn: () => void): Promise<void> {
+    startFn();
+    // Config load is async; let the watch fire and the initial fetch complete.
+    await flushPromises();
+    await vi.runOnlyPendingTimersAsync();
+    await flushPromises();
+}
+
 describe("historyStore — config-driven SSE vs polling", () => {
+    let visibility: ReturnType<typeof useVisibilityPatch>;
+
     beforeEach(() => {
         setActivePinia(createPinia());
-        mockSseConnect.mockClear();
-        mockSseDisconnect.mockClear();
-        mockSseConnected.value = false;
+        sseState.connect.mockClear();
+        sseState.disconnect.mockClear();
+        sseState.onEvent = null;
         mockWatchHistory.mockClear();
         mockRefreshHistoryFromPush.mockClear();
         vi.useFakeTimers();
+        visibility = useVisibilityPatch();
     });
 
     afterEach(() => {
+        visibility.restore();
         vi.useRealTimers();
     });
 
@@ -75,17 +80,9 @@ describe("historyStore — config-driven SSE vs polling", () => {
 
         it("primes the store with one initial load, connects SSE, and does not keep polling", async () => {
             const store = useHistoryStore();
+            await primeStore(() => store.startWatchingHistory());
 
-            // `startWatchingHistory` is an exported alias for
-            // `startWatchingHistoryWithSSE` (see historyStore.ts).
-            store.startWatchingHistory();
-
-            // Config loads async; let the watch fire.
-            await flushPromises();
-            await vi.runOnlyPendingTimersAsync();
-            await flushPromises();
-
-            expect(mockSseConnect).toHaveBeenCalledTimes(1);
+            expect(sseState.connect).toHaveBeenCalledTimes(1);
             // One-shot initial fetch so the history panel isn't empty before
             // the first SSE event arrives.
             expect(mockWatchHistory).toHaveBeenCalledTimes(1);
@@ -99,33 +96,52 @@ describe("historyStore — config-driven SSE vs polling", () => {
 
         it("does not start polling when the tab regains visibility", async () => {
             const store = useHistoryStore();
-            store.startWatchingHistory();
-            await flushPromises();
-            await vi.runOnlyPendingTimersAsync();
-            await flushPromises();
+            await primeStore(() => store.startWatchingHistory());
             expect(mockWatchHistory).toHaveBeenCalledTimes(1);
 
             // Simulate a tab hide/show cycle. `useResourceWatcher` registers
             // a `visibilitychange` listener whose handler calls
             // `startWatchingResourceIfNeeded` — in SSE mode that would
             // silently resume polling. Because we never instantiated the
             // watcher, no listener should exist and no poll should fire.
-            Object.defineProperty(document, "visibilityState", {
-                configurable: true,
-                get: () => "hidden",
-            });
-            document.dispatchEvent(new Event("visibilitychange"));
-            Object.defineProperty(document, "visibilityState", {
-                configurable: true,
-                get: () => "visible",
-            });
-            document.dispatchEvent(new Event("visibilitychange"));
+            visibility.set("hidden");
+            visibility.set("visible");
 
             await flushPromises();
             vi.advanceTimersByTime(30_000);
             await flushPromises();
             expect(mockWatchHistory).toHaveBeenCalledTimes(1);
         });
+
+        it("triggers refreshHistoryFromPush when an SSE event names the current history", async () => {
+            const store = useHistoryStore();
+            await primeStore(() => store.startWatchingHistory());
+            // Drive the store to a known current-history id so the handler has
+            // something to match against. ``currentHistoryId`` is a computed
+            // that only returns the stored id when the history is present in
+            // ``storedHistories``, so the history has to be registered too.
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            store.setHistory({ id: "hist-1" } as any);
+            store.setCurrentHistoryId("hist-1");
+
+            mockRefreshHistoryFromPush.mockClear();
+            emitSse(sseState, "history_update", { history_ids: ["hist-1", "hist-2"] });
+            await flushPromises();
+
+            expect(mockRefreshHistoryFromPush).toHaveBeenCalledTimes(1);
+        });
+
+        it("ignores SSE history events that do not include the current history", async () => {
+            const store = useHistoryStore();
+            await primeStore(() => store.startWatchingHistory());
+            store.setCurrentHistoryId("hist-1");
+
+            mockRefreshHistoryFromPush.mockClear();
+            emitSse(sseState, "history_update", { history_ids: ["hist-2"] });
+            await flushPromises();
+
+            expect(mockRefreshHistoryFromPush).not.toHaveBeenCalled();
+        });
     });
 
     describe("when enable_sse_history_updates is false (polling scenario)", () => {
@@ -135,15 +151,9 @@ describe("historyStore — config-driven SSE vs polling", () => {
 
         it("does not connect SSE and polls on the configured interval", async () => {
             const store = useHistoryStore();
+            await primeStore(() => store.startWatchingHistory());
 
-            store.startWatchingHistory();
-
-            // Let the config load and the initial watch fire.
-            await flushPromises();
-            await vi.runOnlyPendingTimersAsync();
-            await flushPromises();
-
-            expect(mockSseConnect).not.toHaveBeenCalled();
+            expect(sseState.connect).not.toHaveBeenCalled();
 
             // The resource watcher invokes the handler immediately on start
             // and then re-schedules after each completion. Advance past the
@@ -156,24 +166,23 @@ describe("historyStore — config-driven SSE vs polling", () => {
             expect(mockWatchHistory.mock.calls.length).toBeGreaterThan(initialCalls);
         });
 
-        it("calling startWatchingHistory again is idempotent (no second SSE, polling already running)", async () => {
+        it("calling startWatchingHistory again is idempotent (no second SSE, polling tick count +1 only)", async () => {
             const store = useHistoryStore();
-
-            store.startWatchingHistory();
-            await flushPromises();
-            await vi.runOnlyPendingTimersAsync();
-            await flushPromises();
+            await primeStore(() => store.startWatchingHistory());
 
             const pollsAfterFirst = mockWatchHistory.mock.calls.length;
 
             store.startWatchingHistory();
             await flushPromises();
 
-            expect(mockSseConnect).not.toHaveBeenCalled();
-            // Calling again does not schedule an additional independent
-            // polling loop — the handler should not have been fired an
-            // extra time by the second call alone.
-            expect(mockWatchHistory.mock.calls.length).toBe(pollsAfterFirst);
+            expect(sseState.connect).not.toHaveBeenCalled();
+            // Calling again must not schedule a second independent polling loop.
+            // Advance past one interval and confirm only one handler tick fires,
+            // not two.
+            await vi.advanceTimersByTimeAsync(3000);
+            await flushPromises();
+            const deltaAfterSecond = mockWatchHistory.mock.calls.length - pollsAfterFirst;
+            expect(deltaAfterSecond).toBeLessThanOrEqual(1);
         });
     });
 });