From 3269d04bb9ead66c8551e9954f18c399c6df907a Mon Sep 17 00:00:00 2001 From: Lincoln Stein Date: Thu, 15 Oct 2026 21:15:40 -0400 Subject: [PATCH] feat(gallery): refresh when other clients complete generations Completed queue items owned by the signed-in user and enqueued outside the web UI now refresh the gallery and board counts. Move the session user id helper into identity so app runtimes share it. Co-Authored-By: Claude Opus 5.5 --- .../webv2/src/app/GalleryRealtimeRuntime.tsx | 9 +- .../webv2/src/app/RecallParametersRuntime.tsx | 8 +- .../gallery/data/realtimeRuntime.test.ts | 94 +++++++++++++++++-- .../features/gallery/data/realtimeRuntime.ts | 35 ++++++- .../webv2/src/features/identity/index.ts | 1 + .../webv2/src/features/identity/session.ts | 9 ++ .../webv2/src/features/queue/contracts.ts | 1 + .../src/features/queue/data/events.test.ts | 16 ++++ .../webv2/src/features/queue/data/events.ts | 4 + 9 files changed, 160 insertions(+), 17 deletions(-) create mode 100644 invokeai/frontend/webv2/src/features/queue/data/events.test.ts diff --git a/invokeai/frontend/webv2/src/app/GalleryRealtimeRuntime.tsx b/invokeai/frontend/webv2/src/app/GalleryRealtimeRuntime.tsx index 1f76463cb6c..72289a5fb6f 100644 --- a/invokeai/frontend/webv2/src/app/GalleryRealtimeRuntime.tsx +++ b/invokeai/frontend/webv2/src/app/GalleryRealtimeRuntime.tsx @@ -1,16 +1,23 @@ import { createGalleryRealtimeRuntime, invalidateGallery } from '@features/gallery/queries'; +import { getSessionUserId } from '@features/identity'; +import { isWebv2QueueItemOrigin } from '@features/queue/contracts'; import { useMountEffect } from '@platform/react/useMountEffect'; import { socketHub } from '@platform/transport/socketHub'; import { useQueryClient } from '@tanstack/react-query'; -/** App-owned composition: uploads the backend announces over the socket refresh the Gallery read model. */ +/** + * App-owned composition: uploads and other clients' completed generations the backend announces over the socket + * refresh the Gallery read model. + */ export const GalleryRealtimeRuntime = () => { const queryClient = useQueryClient(); useMountEffect(() => { const runtime = createGalleryRealtimeRuntime({ backend: socketHub, + getSessionUserId, invalidate: () => invalidateGallery(queryClient), + isClientQueueItemOrigin: isWebv2QueueItemOrigin, }); runtime.start(); diff --git a/invokeai/frontend/webv2/src/app/RecallParametersRuntime.tsx b/invokeai/frontend/webv2/src/app/RecallParametersRuntime.tsx index 3145bca0185..901f060e373 100644 --- a/invokeai/frontend/webv2/src/app/RecallParametersRuntime.tsx +++ b/invokeai/frontend/webv2/src/app/RecallParametersRuntime.tsx @@ -1,4 +1,4 @@ -import { getAuthSession } from '@features/identity'; +import { getSessionUserId } from '@features/identity'; import { useMountEffect } from '@platform/react/useMountEffect'; import { socketHub } from '@platform/transport/socketHub'; import { isEditableHotkeyTarget } from '@workbench/hotkeys/keys'; @@ -10,12 +10,6 @@ import { useWorkbenchInternalStore } from '@workbench/WorkbenchContext'; import { useWorkbenchWidgetRegistry } from '@workbench/WorkbenchWidgetRegistryContext'; import { useTranslation } from 'react-i18next'; -/** Single-user servers tag events with a system user id, so only multi-user sessions can be fenced. */ -const getSessionUserId = (): string | null => { - const session = getAuthSession(); - return session.multiuserEnabled ? (session.user?.user_id ?? null) : null; -}; - /** * App-owned composition: external `POST /api/v1/recall` updates land in the active project's Generate panel, and * `POST /api/v1/recall/video` updates in its Video panel; each brings its panel to the front. diff --git a/invokeai/frontend/webv2/src/features/gallery/data/realtimeRuntime.test.ts b/invokeai/frontend/webv2/src/features/gallery/data/realtimeRuntime.test.ts index ab1557e2fcd..c0183893830 100644 --- a/invokeai/frontend/webv2/src/features/gallery/data/realtimeRuntime.test.ts +++ b/invokeai/frontend/webv2/src/features/gallery/data/realtimeRuntime.test.ts @@ -28,6 +28,23 @@ describe('gallery realtime runtime', () => { }), }; const invalidate = vi.fn(); + let sessionUserId: string | null = 'alice'; + const getSessionUserId = () => sessionUserId; + const CLIENT_ORIGIN = 'this-client'; + const createRuntime = () => + createGalleryRealtimeRuntime({ + backend, + getSessionUserId, + invalidate, + isClientQueueItemOrigin: (origin) => origin === CLIENT_ORIGIN, + }); + const emitStatusChanged = (event: { origin?: string | null; status?: string; user_id?: string }): void => + handlers.get('queue_item_status_changed')?.({ + origin: 'chatbot', + status: 'completed', + user_id: 'alice', + ...event, + } as never); beforeEach(() => { vi.useFakeTimers(); @@ -35,6 +52,7 @@ describe('gallery realtime runtime', () => { detachers.length = 0; connectionHandler = null; initialStatus = 'connected'; + sessionUserId = 'alice'; backend.on.mockClear(); backend.onConnectionChange.mockClear(); invalidate.mockReset(); @@ -43,7 +61,7 @@ describe('gallery realtime runtime', () => { afterEach(() => vi.useRealTimers()); it('coalesces a burst of upload events into one invalidation and detaches on dispose', () => { - const runtime = createGalleryRealtimeRuntime({ backend, invalidate }); + const runtime = createRuntime(); runtime.start(); runtime.start(); @@ -51,7 +69,7 @@ describe('gallery realtime runtime', () => { handlers.get('image_uploaded')?.({ image_name: 'b.png' } as never); handlers.get('video_uploaded')?.({ video_name: 'c.mp4' } as never); - expect(backend.on).toHaveBeenCalledTimes(2); + expect(backend.on).toHaveBeenCalledTimes(3); expect(backend.onConnectionChange).toHaveBeenCalledTimes(1); expect(invalidate).not.toHaveBeenCalled(); @@ -68,7 +86,7 @@ describe('gallery realtime runtime', () => { }); it('holds passes a minimum interval apart under a stream of uploads and still runs a trailing pass', () => { - const runtime = createGalleryRealtimeRuntime({ backend, invalidate }); + const runtime = createRuntime(); runtime.start(); @@ -93,7 +111,7 @@ describe('gallery realtime runtime', () => { it('refetches after an outage but not on the boot-time connect', () => { initialStatus = 'connecting'; - const runtime = createGalleryRealtimeRuntime({ backend, invalidate }); + const runtime = createRuntime(); runtime.start(); connectionHandler?.('connected'); @@ -111,7 +129,7 @@ describe('gallery realtime runtime', () => { it('treats a connect after mounting on a downed backend as a reconnect', () => { initialStatus = 'disconnected'; - const runtime = createGalleryRealtimeRuntime({ backend, invalidate }); + const runtime = createRuntime(); runtime.start(); connectionHandler?.('connected'); @@ -122,7 +140,7 @@ describe('gallery realtime runtime', () => { }); it('drops a pending invalidation on dispose', () => { - const runtime = createGalleryRealtimeRuntime({ backend, invalidate }); + const runtime = createRuntime(); runtime.start(); handlers.get('image_uploaded')?.({ image_name: 'a.png' } as never); @@ -138,7 +156,7 @@ describe('gallery realtime runtime', () => { }); it('ignores socket callbacks after its account scope expires', () => { - const runtime = createGalleryRealtimeRuntime({ backend, invalidate }); + const runtime = createRuntime(); runtime.start(); accountLifecycle.invalidate(); @@ -148,4 +166,66 @@ describe('gallery realtime runtime', () => { expect(invalidate).not.toHaveBeenCalled(); runtime.dispose(); }); + + it("refreshes once for a burst of the session user's completions enqueued by other clients", () => { + const runtime = createRuntime(); + + runtime.start(); + emitStatusChanged({}); + emitStatusChanged({ origin: null }); + vi.advanceTimersByTime(250); + + expect(invalidate).toHaveBeenCalledTimes(1); + runtime.dispose(); + }); + + it('ignores completions that cannot add to this gallery or that Queue already refreshes for', () => { + const runtime = createRuntime(); + + runtime.start(); + // Another user's item, as an admin receives it, and as everyone else receives it. + emitStatusChanged({ user_id: 'bob' }); + emitStatusChanged({ origin: null, user_id: 'redacted' }); + // An item this client enqueued, which Queue refreshes for in the tab tracking it. + emitStatusChanged({ origin: CLIENT_ORIGIN }); + + for (const status of ['pending', 'in_progress', 'waiting', 'failed', 'canceled'] as const) { + emitStatusChanged({ status }); + } + + vi.advanceTimersByTime(1000); + + expect(invalidate).not.toHaveBeenCalled(); + runtime.dispose(); + }); + + it("refreshes for the server's system user when a single-user session cannot fence by user", () => { + sessionUserId = null; + const runtime = createRuntime(); + + runtime.start(); + emitStatusChanged({ user_id: 'system' }); + vi.advanceTimersByTime(250); + + expect(invalidate).toHaveBeenCalledTimes(1); + runtime.dispose(); + }); + + it('attributes each completion to the session user at the time it arrives', () => { + sessionUserId = 'alice'; + const runtime = createRuntime(); + + runtime.start(); + sessionUserId = 'bob'; + emitStatusChanged({ user_id: 'alice' }); + vi.advanceTimersByTime(1000); + + expect(invalidate).not.toHaveBeenCalled(); + + emitStatusChanged({ user_id: 'bob' }); + vi.advanceTimersByTime(250); + + expect(invalidate).toHaveBeenCalledTimes(1); + runtime.dispose(); + }); }); diff --git a/invokeai/frontend/webv2/src/features/gallery/data/realtimeRuntime.ts b/invokeai/frontend/webv2/src/features/gallery/data/realtimeRuntime.ts index 9f6d34bbedf..103e40bb638 100644 --- a/invokeai/frontend/webv2/src/features/gallery/data/realtimeRuntime.ts +++ b/invokeai/frontend/webv2/src/features/gallery/data/realtimeRuntime.ts @@ -7,24 +7,38 @@ export interface GalleryRealtimeBackend { onConnectionChange(handler: (status: BackendConnectionStatus) => void): () => void; } +/** The fields of the backend's `queue_item_status_changed` payload that Gallery reads. */ +interface QueueItemStatusChangedPayload { + origin: string | null; + status: string; + user_id: string; +} + export interface GalleryRealtimeRuntime { dispose(): void; start(): void; } /** - * Refresh external uploads and reconnects at least minIntervalMs apart. Coalesce pending events and guarantee a - * trailing pass without repeatedly cancelling page fetches. + * Refresh external uploads, the session user's completions enqueued by other clients, and reconnects at least + * minIntervalMs apart. Coalesce pending events and guarantee a trailing pass without repeatedly cancelling page + * fetches. */ export const createGalleryRealtimeRuntime = ({ backend, coalesceMs = 250, + getSessionUserId, invalidate, + isClientQueueItemOrigin, minIntervalMs = 1000, }: { backend: GalleryRealtimeBackend; coalesceMs?: number; + /** Null when events cannot be attributed to a user (single-user servers tag every event `system`). */ + getSessionUserId: () => string | null; invalidate: () => void | Promise; + /** True for origins of items this client enqueues; Queue refreshes for those in the tab that tracks them. */ + isClientQueueItemOrigin: (origin: string | null) => boolean; minIntervalMs?: number; }): GalleryRealtimeRuntime => { const owner = captureAccountScope(); @@ -53,6 +67,22 @@ export const createGalleryRealtimeRuntime = ({ }, delay); }; + const handleQueueItemStatusChanged = (event: QueueItemStatusChangedPayload): void => { + // Admins also receive other users' full events and everyone else's arrive redacted, so only the session user's + // completions can add to this gallery. Items this client enqueued are left to Queue, which refreshes only in the + // tab tracking the item: another tab or device of the same user is not refreshed for them. Known limitation, + // left intentionally: refreshing here too would double-refresh the enqueuing tab. + const sessionUserId = getSessionUserId(); + + if ( + event.status === 'completed' && + !isClientQueueItemOrigin(event.origin) && + (sessionUserId === null || event.user_id === sessionUserId) + ) { + scheduleInvalidation(); + } + }; + const start = (): void => { if (isStarted || !isAccountScopeCurrent(owner)) { return; @@ -67,6 +97,7 @@ export const createGalleryRealtimeRuntime = ({ detachers.push( backend.on('image_uploaded', scheduleInvalidation), backend.on('video_uploaded', scheduleInvalidation), + backend.on('queue_item_status_changed', handleQueueItemStatusChanged), backend.onConnectionChange((status) => { const isReconnect = previousStatus === 'disconnected' && status === 'connected'; diff --git a/invokeai/frontend/webv2/src/features/identity/index.ts b/invokeai/frontend/webv2/src/features/identity/index.ts index 51e45d91bd1..d1728f719b9 100644 --- a/invokeai/frontend/webv2/src/features/identity/index.ts +++ b/invokeai/frontend/webv2/src/features/identity/index.ts @@ -6,6 +6,7 @@ export { ensureAuthSession, ensureReadyAuthSession, getAuthSession, + getSessionUserId, getUserStorageScope, identityTransportAuthAdapter, isLoginAttemptSupersededError, diff --git a/invokeai/frontend/webv2/src/features/identity/session.ts b/invokeai/frontend/webv2/src/features/identity/session.ts index 60e387457e8..b203b1d986c 100644 --- a/invokeai/frontend/webv2/src/features/identity/session.ts +++ b/invokeai/frontend/webv2/src/features/identity/session.ts @@ -80,6 +80,15 @@ export const useAuthSession = (): AuthSession => store.useSnapshot(); /** Workbench imperative reads follow route-guard resolution; changing users remounts the route. */ export const getAuthSession = (): AuthSession => store.getSnapshot(); +/** + * The user id that owns this session's socket events. Single-user servers tag events with a system user id, so + * only multi-user sessions can be fenced. + */ +export const getSessionUserId = (): string | null => { + const session = getAuthSession(); + return session.multiuserEnabled ? (session.user?.user_id ?? null) : null; +}; + /** Stable subscription for App-composed capability read ports. */ export const subscribeAuthSession = (listener: () => void): (() => void) => store.subscribe(listener); diff --git a/invokeai/frontend/webv2/src/features/queue/contracts.ts b/invokeai/frontend/webv2/src/features/queue/contracts.ts index e858b6d8b08..d9374a3a87c 100644 --- a/invokeai/frontend/webv2/src/features/queue/contracts.ts +++ b/invokeai/frontend/webv2/src/features/queue/contracts.ts @@ -63,6 +63,7 @@ export { buildUtilityQueueItemOrigin, isTerminalBackendStatus, isUtilityQueueItemOrigin, + isWebv2QueueItemOrigin, parseQueueItemOrigin, parseQueueItemOriginProjectId, type BackendSocketEvents, diff --git a/invokeai/frontend/webv2/src/features/queue/data/events.test.ts b/invokeai/frontend/webv2/src/features/queue/data/events.test.ts new file mode 100644 index 00000000000..36ae1c0acb9 --- /dev/null +++ b/invokeai/frontend/webv2/src/features/queue/data/events.test.ts @@ -0,0 +1,16 @@ +import { describe, expect, it } from 'vitest'; + +import { buildQueueItemOrigin, buildUtilityQueueItemOrigin, isWebv2QueueItemOrigin } from './events'; + +describe('isWebv2QueueItemOrigin', () => { + it('recognizes every origin this client builds and no other client', () => { + // Gallery skips these completions; an unrecognized form would double-refresh, a false match would miss one. + expect(isWebv2QueueItemOrigin(buildQueueItemOrigin('local-1'))).toBe(true); + expect(isWebv2QueueItemOrigin(buildQueueItemOrigin('local-1', 'project-1'))).toBe(true); + expect(isWebv2QueueItemOrigin(buildUtilityQueueItemOrigin('filter-1'))).toBe(true); + + expect(isWebv2QueueItemOrigin('canvas')).toBe(false); + expect(isWebv2QueueItemOrigin('chatbot:webv2:1')).toBe(false); + expect(isWebv2QueueItemOrigin(null)).toBe(false); + }); +}); diff --git a/invokeai/frontend/webv2/src/features/queue/data/events.ts b/invokeai/frontend/webv2/src/features/queue/data/events.ts index 1554fc30cbc..46decc98839 100644 --- a/invokeai/frontend/webv2/src/features/queue/data/events.ts +++ b/invokeai/frontend/webv2/src/features/queue/data/events.ts @@ -150,6 +150,10 @@ export const buildQueueItemOrigin = (localQueueItemId: string, projectId?: strin export const buildUtilityQueueItemOrigin = (utilityId: string): string => `${UTILITY_QUEUE_ITEM_ORIGIN_PREFIX}${utilityId}`; +/** True for every origin this client builds: local, project and utility queue items. */ +export const isWebv2QueueItemOrigin = (origin: string | null | undefined): boolean => + origin?.startsWith(QUEUE_ITEM_ORIGIN_PREFIX) ?? false; + /** True when `origin` belongs to the utility queue (never a project/local queue item). */ export const isUtilityQueueItemOrigin = (origin: string | null | undefined): boolean => origin?.startsWith(UTILITY_QUEUE_ITEM_ORIGIN_PREFIX) ?? false;