From 434f8de2be95941e4cc3c4debadf6ad57d8e9d8f Mon Sep 17 00:00:00 2001 From: Neko Date: Sat, 29 Aug 2026 03:35:31 +0800 Subject: [PATCH] fix(stage-ui): remove the chat session initialization race (#2394) --- apps/stage-tamagotchi/src/renderer/App.vue | 9 +- apps/stage-web/src/App.vue | 10 +- .../stage-ui/src/stores/chat.contract.test.ts | 56 +++++++ packages/stage-ui/src/stores/chat.ts | 31 +++- .../stores/chat/session-store.browser.test.ts | 147 ++++++++++++++++-- .../src/stores/chat/session-store.test.ts | 20 --- .../stage-ui/src/stores/chat/session-store.ts | 125 ++++++++------- 7 files changed, 296 insertions(+), 102 deletions(-) diff --git a/apps/stage-tamagotchi/src/renderer/App.vue b/apps/stage-tamagotchi/src/renderer/App.vue index 2585fb38f..f5a06d604 100644 --- a/apps/stage-tamagotchi/src/renderer/App.vue +++ b/apps/stage-tamagotchi/src/renderer/App.vue @@ -12,7 +12,6 @@ import { usePiniaSynced } from '@proj-airi/stage-ui/libs/pinia' import { useAuthStore } from '@proj-airi/stage-ui/stores/auth' import { useCharacterOrchestratorStore } from '@proj-airi/stage-ui/stores/character' import { useChatStore } from '@proj-airi/stage-ui/stores/chat' -import { useChatSessionStore } from '@proj-airi/stage-ui/stores/chat/session-store' import { usePluginHostInspectorStore } from '@proj-airi/stage-ui/stores/devtools/plugin-host-debug' import { useDisplayModelsStore } from '@proj-airi/stage-ui/stores/display-models' import { useModsServerChannelStore } from '@proj-airi/stage-ui/stores/mods/api/channel-server' @@ -79,18 +78,16 @@ const settingsStore = useSettings() const { language, themeColorsHue, themeColorsHueDynamic } = storeToRefs(settingsStore) const router = useRouter() const route = useRoute() -const chatSessionStore = useChatSessionStore() const context = useElectronEventaContext() const getMainLocale = useElectronEventaInvoke(i18nGetLocale) const setLocale = useElectronEventaInvoke(i18nSetLocale) const windowContext = resolveRendererWindowContext() const initialRoutePath = resolveInitialRendererRoutePath(route.path) -useChatStore() +const chatStore = useChatStore() const builtinToolsStore = useTamagotchiBuiltinToolsStore() const mcpToolsStore = useTamagotchiMcpToolsStore() const pluginToolsStore = useTamagotchiPluginToolsStore() const syncedPinia = usePiniaSynced() -chatSessionStore.setCloudSyncOwnership(syncedPinia.isLeader()) const isSpotlightWindow = initialRoutePath === '/spotlight' const isSettingsWindow = initialRoutePath === '/settings' || initialRoutePath.startsWith('/settings/') @@ -106,7 +103,6 @@ async function refreshPluginRuntimeTools() { // Every renderer creates the runtime tool stores for synchronized state. Only // the main Stage renderer discovers tools and keeps executors. const stopLeadershipListener = syncedPinia.onLeadershipChange((isLeader) => { - chatSessionStore.setCloudSyncOwnership(isLeader) if (!isLeader) return @@ -351,7 +347,7 @@ onMounted(async () => { // https://github.com/moeru-ai/airi/issues/1658 await restoreLocale() - await chatSessionStore.initialize() + await chatStore.initialize(syncedPinia) await fullStageRuntime?.initialize() }) @@ -366,6 +362,7 @@ watch(themeColorsHueDynamic, () => { onUnmounted(() => { stopLeadershipListener?.() + chatStore.dispose() fullStageRuntime?.dispose() }) diff --git a/apps/stage-web/src/App.vue b/apps/stage-web/src/App.vue index 75d4a652e..bd35153ff 100644 --- a/apps/stage-web/src/App.vue +++ b/apps/stage-web/src/App.vue @@ -5,7 +5,7 @@ import { initializeAnalytics, isAnalyticsAvailableInBuild } from '@proj-airi/sta import { usePiniaSynced } from '@proj-airi/stage-ui/libs/pinia' import { useAuthStore } from '@proj-airi/stage-ui/stores/auth' import { useCharacterOrchestratorStore } from '@proj-airi/stage-ui/stores/character' -import { useChatSessionStore } from '@proj-airi/stage-ui/stores/chat/session-store' +import { useChatStore } from '@proj-airi/stage-ui/stores/chat' import { useDisplayModelsStore } from '@proj-airi/stage-ui/stores/display-models' import { useModsServerChannelStore } from '@proj-airi/stage-ui/stores/mods/api/channel-server' import { useContextBridgeStore } from '@proj-airi/stage-ui/stores/mods/api/context-bridge' @@ -40,10 +40,8 @@ const displayModelsStore = useDisplayModelsStore() const settingsStore = useSettings() const settings = storeToRefs(settingsStore) const onboardingStore = useOnboardingStore() -const chatSessionStore = useChatSessionStore() +const chatStore = useChatStore() const syncedPinia = usePiniaSynced() -chatSessionStore.setCloudSyncOwnership(syncedPinia.isLeader()) -const stopLeadershipListener = syncedPinia.onLeadershipChange(isLeader => chatSessionStore.setCloudSyncOwnership(isLeader)) const serverChannelStore = useModsServerChannelStore() const characterOrchestratorStore = useCharacterOrchestratorStore() const settingsAudioDeviceStore = useSettingsAudioDevice() @@ -136,7 +134,7 @@ onMounted(async () => { onboardingStore.showingSetup = true } - await chatSessionStore.initialize() + await chatStore.initialize(syncedPinia) await serverChannelStore.initialize({ possibleEvents: ['ui:configure'] }).catch(err => console.error('Failed to initialize Mods Server Channel in App.vue:', err)) contextBridgeStore.initialize() characterOrchestratorStore.initialize() @@ -152,7 +150,7 @@ onMounted(async () => { onUnmounted(() => { stopAuthenticatedSetup?.() stopLoggedOutSetup?.() - stopLeadershipListener() + chatStore.dispose() contextBridgeStore.dispose() }) diff --git a/packages/stage-ui/src/stores/chat.contract.test.ts b/packages/stage-ui/src/stores/chat.contract.test.ts index 7e2e801a7..3a1f3d96c 100644 --- a/packages/stage-ui/src/stores/chat.contract.test.ts +++ b/packages/stage-ui/src/stores/chat.contract.test.ts @@ -1,6 +1,7 @@ import type { StreamOptions } from '@proj-airi/core-agent' import type { ChatProvider } from '@xsai-ext/providers/utils' import type { Message, Tool } from '@xsai/shared-chat' +import type { SyncedPiniaRuntime } from 'pinia-plugin-synced' import { errorMessageFrom } from '@moeru/std' import { IOAttributes, IOSpanNames } from '@proj-airi/stage-shared' @@ -67,6 +68,9 @@ const forkSessionMock = vi.fn() const ensureSessionMock = vi.fn() const loadSessionMock = vi.fn() const deleteSessionMock = vi.fn() +const initializeSessionMock = vi.fn() +const disposeSessionMock = vi.fn() +const ensureCurrentSessionMock = vi.fn() const getChatProviderInstanceMock = vi.fn() const getToolsByNamesMock = vi.fn<(names: string[]) => Tool[]>() @@ -150,6 +154,9 @@ vi.mock('./chat/session-store', () => ({ getSessionMessagesIfLoaded: (sessionId: string) => sessionMessages[sessionId], loadSession: loadSessionMock, deleteSession: deleteSessionMock, + initialize: initializeSessionMock, + dispose: disposeSessionMock, + ensureCurrentSession: ensureCurrentSessionMock, persistSessionMessages: persistSessionMessagesMock, getSessionGeneration: () => currentGeneration, setSessionMessages: (sessionId: string, messages: any[]) => { @@ -240,6 +247,9 @@ describe('chat store contract', () => { ensureSessionMock.mockReset() loadSessionMock.mockReset().mockResolvedValue(true) deleteSessionMock.mockReset().mockResolvedValue(undefined) + initializeSessionMock.mockReset().mockResolvedValue(undefined) + disposeSessionMock.mockReset() + ensureCurrentSessionMock.mockReset().mockResolvedValue('session-1') getChatProviderInstanceMock.mockReset().mockResolvedValue(provider) getToolsByNamesMock.mockReset().mockImplementation(names => names.map(name => ({ type: 'function', @@ -293,6 +303,52 @@ describe('chat store contract', () => { ]) }) + // https://github.com/moeru-ai/airi/pull/2394#discussion_r3883162024 + it('restarts chat consumers when this renderer becomes the leader', async () => { + // ROOT CAUSE: + // + // The application stopped observing chat leadership changes. If the Web + // leader closed, the promoted renderer kept the replicated session state + // but did not start a new cloud WebSocket. + // + // The chat store now owns the leadership subscription. It starts the + // session consumers after promotion and stops local consumers after + // demotion or disposal. + let leadershipListener: ((isLeader: boolean) => void) | undefined + const stopLeadershipListener = vi.fn() + const syncedPinia: SyncedPiniaRuntime = { + dispose: vi.fn(), + getLeaderId: vi.fn(), + getParticipantCount: vi.fn(() => 1), + isLeader: vi.fn(() => false), + onCoordinationChange: vi.fn(() => vi.fn()), + onLeadershipChange: vi.fn((listener) => { + leadershipListener = listener + listener(false) + return stopLeadershipListener + }), + participantId: 'chat-test', + plugin: vi.fn(), + } + const store = useChatStore() + + await store.initialize(syncedPinia) + + expect(initializeSessionMock).toHaveBeenCalledOnce() + expect(disposeSessionMock).toHaveBeenCalledOnce() + expect(ensureCurrentSessionMock).not.toHaveBeenCalled() + + leadershipListener?.(true) + await vi.waitFor(() => expect(ensureCurrentSessionMock).toHaveBeenCalledOnce()) + + leadershipListener?.(false) + expect(disposeSessionMock).toHaveBeenCalledTimes(2) + + store.dispose() + expect(stopLeadershipListener).toHaveBeenCalledOnce() + expect(disposeSessionMock).toHaveBeenCalledTimes(3) + }) + it('passes the current consciousness reasoning option to the chat provider', async () => { const settings = useConsciousnessSettingsStore() await settings.setReasoning(true) diff --git a/packages/stage-ui/src/stores/chat.ts b/packages/stage-ui/src/stores/chat.ts index 58004fddc..28e976a9b 100644 --- a/packages/stage-ui/src/stores/chat.ts +++ b/packages/stage-ui/src/stores/chat.ts @@ -2,7 +2,7 @@ import type { ChatOrchestratorRuntimeState, ChatOrchestratorSendOptions, StreamE import type { WebSocketEventInputs } from '@proj-airi/server-sdk' import type { ChatProvider } from '@xsai-ext/providers/utils' import type { Message } from '@xsai/shared-chat' -import type {} from 'pinia-plugin-synced' +import type { SyncedPiniaRuntime } from 'pinia-plugin-synced' import type { ChatHistoryItem, ChatToolReference, StreamingAssistantMessage } from '../types/chat' import type { ToolCallRerunPayload } from './tool-call-rerun' @@ -160,10 +160,37 @@ export const useChatStore = defineStore('chat', () => { const activeStreamingMessage = shallowRef() const pendingQueuedSendCount = shallowRef(0) let ownedActiveTurnSpan: typeof activeTurnSpan.value + let stopLeadershipListener: (() => void) | undefined const analyticsHooks = createChatAnalyticsHooks({ getSessionMessages: sessionId => chatSession.getSessionMessages(sessionId), }) + /** + * Initializes chat state and binds local consumers to synchronized leadership. + * A promoted renderer restarts the leader-owned cloud consumer. + */ + async function initialize(syncedPinia: SyncedPiniaRuntime) { + stopLeadershipListener ??= syncedPinia.onLeadershipChange((isLeader) => { + if (!isLeader) { + chatSession.dispose() + return + } + + void chatSession.ensureCurrentSession().catch((error) => { + console.error('[chat] Failed to start chat consumers after leader promotion:', error) + }) + }) + + await chatSession.initialize() + } + + /** Stops chat consumers that belong to this window. */ + function dispose() { + stopLeadershipListener?.() + stopLeadershipListener = undefined + chatSession.dispose() + } + async function streamWithStageAdapters( model: string, chatProvider: ChatProvider, @@ -493,6 +520,8 @@ export const useChatStore = defineStore('chat', () => { activeStreamingMessage, pendingQueuedSendCount, + initialize, + dispose, cleanup, deleteSession, ingest, diff --git a/packages/stage-ui/src/stores/chat/session-store.browser.test.ts b/packages/stage-ui/src/stores/chat/session-store.browser.test.ts index f139b5fa0..adf03d97f 100644 --- a/packages/stage-ui/src/stores/chat/session-store.browser.test.ts +++ b/packages/stage-ui/src/stores/chat/session-store.browser.test.ts @@ -63,22 +63,33 @@ vi.mock('../../libs/server', () => ({ SERVER_URL: 'http://test', })) +const chatSyncMocks = vi.hoisted(() => ({ + clients: [] as Array<{ + connect: ReturnType + destroy: ReturnType + }>, +})) + vi.mock('../../libs/chat-sync', () => ({ applyCreateActions: vi.fn().mockResolvedValue([]), createCloudChatMapper: () => ({ deleteChat: vi.fn().mockResolvedValue(undefined), listChats: vi.fn().mockResolvedValue([]), }), - createChatWsClient: () => ({ - connect: vi.fn(), - destroy: vi.fn(), - disconnect: vi.fn(), - onNewMessages: () => () => {}, - onStatusChange: () => () => {}, - pullMessages: vi.fn().mockResolvedValue({ messages: [], seq: 0 }), - sendMessages: vi.fn().mockResolvedValue({ ok: true }), - status: () => 'idle', - }), + createChatWsClient: () => { + const client = { + connect: vi.fn(), + destroy: vi.fn(), + disconnect: vi.fn(), + onNewMessages: () => () => {}, + onStatusChange: () => () => {}, + pullMessages: vi.fn().mockResolvedValue({ messages: [], seq: 0 }), + sendMessages: vi.fn().mockResolvedValue({ ok: true }), + status: () => 'idle', + } + chatSyncMocks.clients.push(client) + return client + }, extractMessageText: () => '', isCloudSyncableMessage: () => false, mergeCloudMessagesIntoLocal: () => ({ dirty: false, messages: [], maxSeq: 0 }), @@ -110,9 +121,41 @@ afterEach(() => { context.runtime.dispose() disposePinia(context.pinia) } + chatSyncMocks.clients.length = 0 }) describe('chat session synchronization', () => { + it('initializes a follower through the canonical session action', async () => { + // ROOT CAUSE: + // + // Chat initialization used the local leadership value before the Web Lock + // election finished. A renderer that started as a follower skipped both + // session loading and session creation. A later leadership update only + // started cloud sync, so anonymous chat kept an empty session id. + // + // Initialization now calls a synchronized action. The plugin routes the + // stateful work to the leader and returns the canonical session id. Each + // window stores that id as its local selection. + const namespace = `chat-session:${crypto.randomUUID()}` + const leaderContext = createSyncedContext(namespace, 'leader-only') + await vi.waitFor(() => expect(leaderContext.runtime.isLeader()).toBe(true)) + + setActivePinia(leaderContext.pinia) + const leaderChatStore = useChatSessionStore() + + const followerContext = createSyncedContext(namespace, 'follower-only') + setActivePinia(followerContext.pinia) + const followerChatStore = useChatSessionStore() + await vi.waitFor(() => expect(followerContext.runtime.getLeaderId()).toBe(leaderContext.runtime.participantId)) + + await followerChatStore.initialize() + + expect(followerChatStore.activeSessionId).not.toBe('') + expect(followerChatStore.activeSessionId).toBe(leaderChatStore.index?.characters.default?.activeSessionId) + expect(followerChatStore.sessionMetas[followerChatStore.activeSessionId]).toBeDefined() + expect(Object.keys(leaderChatStore.index?.characters.default?.sessions ?? {})).toHaveLength(1) + }) + it('keeps the leader chat snapshot when new followers receive the auth identity', async () => { // ROOT CAUSE: // @@ -187,4 +230,88 @@ describe('chat session synchronization', () => { expect(leaderIdentityActions).toBe(2) expect(leaderMutations).toBe(0) }) + + // https://github.com/moeru-ai/airi/pull/2394#discussion_r3883360315 + it('keeps synchronized state unchanged when a follower disposes local consumers', async () => { + // ROOT CAUSE: + // + // Follower disposal used the cloud teardown path, which changed the + // synchronized cloudSyncReady ref. The synchronization plugin then sent + // the follower's full, potentially stale snapshot to the leader. + // + // Follower disposal now destroys only its window-local cloud runtime. + // Leader-owned actions remain responsible for synchronized state changes. + const namespace = `chat-session:${crypto.randomUUID()}` + const leaderContext = createSyncedContext(namespace, 'leader-only') + await vi.waitFor(() => expect(leaderContext.runtime.isLeader()).toBe(true)) + + setActivePinia(leaderContext.pinia) + const leaderChatStore = useChatSessionStore() + leaderChatStore.$patch({ cloudSyncReady: true }) + + const followerContext = createSyncedContext(namespace, 'follower-only') + setActivePinia(followerContext.pinia) + const followerChatStore = useChatSessionStore() + await vi.waitFor(() => expect(followerContext.runtime.getLeaderId()).toBe(leaderContext.runtime.participantId)) + await vi.waitFor(() => expect(followerChatStore.cloudSyncReady).toBe(true)) + + let leaderMutations = 0 + leaderChatStore.$subscribe(() => leaderMutations++) + + followerChatStore.dispose() + await Promise.resolve() + + expect(followerChatStore.cloudSyncReady).toBe(true) + expect(leaderChatStore.cloudSyncReady).toBe(true) + expect(leaderMutations).toBe(0) + }) + + // https://github.com/moeru-ai/airi/pull/2394#discussion_r3883162024 + it('starts a new cloud consumer after leader failover', async () => { + // ROOT CAUSE: + // + // The cloud WebSocket belongs to the elected renderer. If that renderer + // closed, the next leader received the synchronized state but no action + // restarted its local WebSocket. + // + // The chat lifecycle now observes leader promotion and calls the routed + // session action. The new leader then starts its local cloud consumer. + const namespace = `chat-session:${crypto.randomUUID()}` + const leaderContext = createSyncedContext(namespace, 'follower-preferred') + await vi.waitFor(() => expect(leaderContext.runtime.isLeader()).toBe(true)) + + setActivePinia(leaderContext.pinia) + const leaderAuthStore = useTestAuthStore() + leaderAuthStore.userId = 'cloud-user' + leaderAuthStore.token = 'cloud-token' + const leaderChatStore = useChatSessionStore() + await leaderChatStore.initialize() + expect(chatSyncMocks.clients).toHaveLength(1) + + const followerContext = createSyncedContext(namespace, 'follower-preferred') + setActivePinia(followerContext.pinia) + const followerAuthStore = useTestAuthStore() + const followerChatStore = useChatSessionStore() + await vi.waitFor(() => expect(followerContext.runtime.getLeaderId()).toBe(leaderContext.runtime.participantId)) + await vi.waitFor(() => expect(followerAuthStore.userId).toBe('cloud-user')) + await followerChatStore.initialize() + + const stopLeadershipListener = followerContext.runtime.onLeadershipChange((isLeader) => { + if (isLeader) + void followerChatStore.ensureCurrentSession() + else + followerChatStore.dispose() + }) + + leaderChatStore.dispose() + leaderContext.runtime.dispose() + disposePinia(leaderContext.pinia) + + await vi.waitFor(() => expect(followerContext.runtime.isLeader()).toBe(true)) + await vi.waitFor(() => expect(chatSyncMocks.clients).toHaveLength(2)) + + expect(chatSyncMocks.clients[0]?.destroy).toHaveBeenCalledOnce() + expect(chatSyncMocks.clients[1]?.connect).toHaveBeenCalledOnce() + stopLeadershipListener() + }) }) diff --git a/packages/stage-ui/src/stores/chat/session-store.test.ts b/packages/stage-ui/src/stores/chat/session-store.test.ts index 12e1c4235..b1d4a507f 100644 --- a/packages/stage-ui/src/stores/chat/session-store.test.ts +++ b/packages/stage-ui/src/stores/chat/session-store.test.ts @@ -730,7 +730,6 @@ describe('chat-session-store · synchronized data actions', () => { updatedAt: 1, } const store = useChatSessionStore() - store.setCloudSyncOwnership(false) store.applyRemoteSnapshot({ activeSessionId: 'session-a', sessionMessages: { @@ -762,24 +761,6 @@ describe('chat-session-store · synchronized data actions', () => { expect(store.index?.userId).toBe('cloud-user') }) - // https://github.com/moeru-ai/airi/pull/2086#discussion_r3755711151 - it('keeps cloud synchronization in the elected leader for Issue #2085', async () => { - // ROOT CAUSE: - // - // Window-local initialization opened a cloud WebSocket in every window. - // Follower callbacks then proposed direct full-state mutations. - const store = useChatSessionStore() - store.setCloudSyncOwnership(false) - await store.initialize() - - userIdRef.value = 'cloud-user' - await nextTick() - expect(connectCloudWsMock).not.toHaveBeenCalled() - - store.setCloudSyncOwnership(true) - await vi.waitFor(() => expect(connectCloudWsMock).toHaveBeenCalledTimes(1)) - }) - // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743242525 it('initializes a new window selection from the synchronized index for Issue #2085', async () => { // ROOT CAUSE: @@ -795,7 +776,6 @@ describe('chat-session-store · synchronized data actions', () => { updatedAt: 1, } const store = useChatSessionStore() - store.setCloudSyncOwnership(false) store.$patch({ sessionMessages: { 'session-b': [{ id: 'system', role: 'system', content: 'prompt' }] }, sessionMetas: { 'session-b': session }, diff --git a/packages/stage-ui/src/stores/chat/session-store.ts b/packages/stage-ui/src/stores/chat/session-store.ts index ca45d76d9..75669fd32 100644 --- a/packages/stage-ui/src/stores/chat/session-store.ts +++ b/packages/stage-ui/src/stores/chat/session-store.ts @@ -76,14 +76,14 @@ export const useChatSessionStore = defineStore('chat-session', () => { const sessionMessages = ref>({}) const sessionMetas = ref>({}) const sessionGenerations = ref>({}) - /** Authority-owned session index replicated so each window can derive its local selection. */ + /** Canonical session index replicated so each window can derive its local selection. */ const index = ref(null) const ready = ref(false) const isReady = computed(() => ready.value) const initializing = ref(false) let initializePromise: Promise | null = null - let ensureActivePromise: Promise | null = null + let ensureActivePromise: Promise | null = null // Bumped by `clearInMemoryState` (user swap / teardown). The // `ensureActiveSessionForCharacter` IIFE captures this at call time and // bails after every await once it changes, so a stale hydrate from the @@ -113,7 +113,6 @@ export const useChatSessionStore = defineStore('chat-session', () => { let cloudMapper: CloudChatMapper | undefined let cloudReconcileTask: Promise | undefined let pendingReconcile = false - let ownsCloudSync = true // Incremented on every teardown / user swap. Long-running reconcile IIFEs // capture the epoch at start and bail after every await once it changes, // so account-A mutations cannot land on account-B state after a sign-out. @@ -380,7 +379,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { if (loadedSessions.has(sessionId) && !staleSessions.has(sessionId) && !needsCloudHydration()) { return true } - // A synchronized snapshot already carries the authority's hydrated + // A synchronized snapshot already carries the canonical hydrated // messages. Trust it instead of letting this follower merge an older IDB // record and publish that stale full-store proposal back to the leader. if (Object.hasOwn(sessionMessages.value, sessionId) && hasKnownSession(sessionId) && !staleSessions.has(sessionId) && !needsCloudHydration()) { @@ -661,7 +660,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { * callers share a single in-flight promise so a rapid `[userId, characterId]` * change burst does not produce duplicate sessions. */ - async function ensureActiveSessionForCharacter(): Promise { + async function ensureActiveSessionForCharacter(): Promise { if (ensureActivePromise) return ensureActivePromise const myEpoch = ensureActiveEpoch @@ -673,26 +672,23 @@ export const useChatSessionStore = defineStore('chat-session', () => { if (!index.value || index.value.userId !== currentUserId) await loadIndexForUser(currentUserId) if (isStaleEpoch()) - return + return '' const characterIndex = getCharacterIndex(characterId) - if (!characterIndex) { - await createSession(characterId) - return - } + if (!characterIndex) + return createSession(characterId) - if (!characterIndex.activeSessionId) { - await createSession(characterId) - return - } + if (!characterIndex.activeSessionId) + return createSession(characterId) activeSessionId.value = characterIndex.activeSessionId // Use the public action so follower hydration is routed to the elected // leader instead of becoming a stale full-state proposal. await useChatSessionStore().loadSession(characterIndex.activeSessionId) + return characterIndex.activeSessionId })() try { - await ensureActivePromise + return await ensureActivePromise } finally { // Only release the slot if we still own it. A user swap mid-flight @@ -785,8 +781,6 @@ export const useChatSessionStore = defineStore('chat-session', () => { * pass is scheduled in `finally` so catch-up pulls do not get lost. */ async function reconcileCloudSessions(): Promise { - if (!ownsCloudSync) - return if (cloudReconcileTask) { pendingReconcile = true return cloudReconcileTask @@ -977,8 +971,6 @@ export const useChatSessionStore = defineStore('chat-session', () => { * from the auth `watch`. */ function ensureCloudWsClient() { - if (!ownsCloudSync) - return if (getCurrentUserId() === 'local') { console.info('[chat-sync] WS skipped: anonymous user') return @@ -1021,8 +1013,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { wsClient.connect() } - function teardownCloudWsClient() { - cloudSyncReady.value = false + function disposeCloudWsClient() { cloudReconcileTask = undefined pendingReconcile = false // Invalidate any in-flight reconcile IIFE so its post-await mutations @@ -1035,18 +1026,9 @@ export const useChatSessionStore = defineStore('chat-session', () => { cloudMapper = undefined } - /** Starts or stops cloud synchronization when this window gains or loses synchronized-store leadership. */ - function setCloudSyncOwnership(owns: boolean) { - if (ownsCloudSync === owns) - return - - ownsCloudSync = owns - if (!ready.value) - return - if (owns) - ensureCloudWsClient() - else if (wsClient) - teardownCloudWsClient() + function teardownCloudWsClient() { + cloudSyncReady.value = false + disposeCloudWsClient() } /** @@ -1089,19 +1071,23 @@ export const useChatSessionStore = defineStore('chat-session', () => { * The synchronization plugin routes this action to the elected renderer. */ async function activateCurrentUser() { - if (sessionStateMatchesCurrentUser()) { - ensureCloudWsClient() - return + if (!sessionStateMatchesCurrentUser()) { + teardownCloudWsClient() + clearInMemoryState() } - teardownCloudWsClient() - clearInMemoryState() - if (!ready.value && !initializing.value) - return + await ensureCurrentSession() + } - await ensureActiveSessionForCharacter() + /** + * Resolves the canonical session and starts its persistence consumers. + * The synchronization plugin routes this action to one renderer. + */ + async function ensureCurrentSession(): Promise { + const sessionId = await ensureActiveSessionForCharacter() await refreshOutboxPendingCount() ensureCloudWsClient() + return sessionId } /** @@ -1318,19 +1304,13 @@ export const useChatSessionStore = defineStore('chat-session', () => { } initializing.value = true initializePromise = (async () => { - if (ownsCloudSync) - await ensureActiveSessionForCharacter() + const sessionId = await useChatSessionStore().ensureCurrentSession() + if (sessionId) + activeSessionId.value = sessionId else selectWindowSessionFromIndex() ready.value = true - // Surface any outbox left over from a previous session (closed tab - // mid-send) before the WS even opens. The drain itself runs after - // reconcile completes, but the count is observable immediately. - if (ownsCloudSync) - await refreshOutboxPendingCount() - if (ownsCloudSync) - ensureCloudWsClient() })() try { @@ -1365,6 +1345,11 @@ export const useChatSessionStore = defineStore('chat-session', () => { activeSessionId.value = getCharacterIndex(getCurrentCharacterId())?.activeSessionId ?? '' } + /** Stops local runtime consumers without changing synchronized session data. */ + function dispose() { + disposeCloudWsClient() + } + const messages = computed({ get: () => { if (!activeSessionId.value) { @@ -1595,16 +1580,25 @@ export const useChatSessionStore = defineStore('chat-session', () => { // every follower would fan one deletion out into several empty chats. }) - watch([activeCardId, index], () => { + watch(index, () => { if (!ready.value) return - if (!ownsCloudSync) { - selectWindowSessionFromIndex() - return - } + selectWindowSessionFromIndex() + }) - void ensureActiveSessionForCharacter() + watch(activeCardId, async () => { + if (!ready.value) + return + + try { + const sessionId = await useChatSessionStore().ensureCurrentSession() + if (sessionId) + activeSessionId.value = sessionId + } + catch (error) { + console.error('[chat-session] Failed to select a session for the current character:', error) + } }) // Each renderer observes the synchronized identity. Route the transition to @@ -1626,6 +1620,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { return { isReady, initialize, + dispose, activeSessionId, messages, @@ -1660,8 +1655,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { refreshSession, deleteSession, activateCurrentUser, - - setCloudSyncOwnership, + ensureCurrentSession, cloudSyncReady, outboxPendingCount, @@ -1669,7 +1663,20 @@ export const useChatSessionStore = defineStore('chat-session', () => { } }, { synced: { - actions: ['activateCurrentUser', 'createSession', 'deleteMessage', 'importSessions', 'loadSession', 'refreshSession'], + actions: [ + 'activateCurrentUser', + 'createSession', + 'deleteMessage', + 'deleteSession', + 'ensureCurrentSession', + 'exportSessions', + 'forkSession', + 'importSessions', + 'loadSession', + 'pushMessageToCloud', + 'refreshSession', + 'resetAllSessions', + ], state: true, }, })