From 96465c8130ac677fa21a7bdb017e490fda2f2bc2 Mon Sep 17 00:00:00 2001 From: Columbina <140679517+0xSelenicDove@users.noreply.github.com> Date: Fri, 14 Aug 2026 16:24:56 +0800 Subject: [PATCH] fix(stage-tamagotchi): repair chat-list add, switch, and delete (#2086) Co-authored-by: RainbowBird --- apps/stage-tamagotchi/src/renderer/App.vue | 6 +- .../InteractiveArea.browser.test.ts | 320 +++++++++ .../renderer/components/InteractiveArea.vue | 37 +- apps/stage-tamagotchi/vitest.config.ts | 33 +- apps/stage-tamagotchi/vitest.node.config.ts | 20 + apps/stage-web/src/App.vue | 5 + package.json | 3 +- .../runtime/chat-orchestrator-runtime.test.ts | 157 +++-- .../src/runtime/chat-orchestrator-runtime.ts | 77 ++- .../components/Layouts/InteractiveArea.vue | 25 +- .../Layouts/MobileInteractiveArea.vue | 60 +- .../src/components/Widgets/ChatArea.vue | 37 +- .../sessions-dialog.browser.test.ts | 104 +++ .../chat/components/sessions-dialog.vue | 132 ++++ .../sessions-drawer.browser.test.ts | 149 +++++ .../chat/components/sessions-drawer.vue | 286 ++------- .../character/orchestrator/index.test.ts | 28 +- .../stage-ui/src/stores/chat.contract.test.ts | 156 ++++- packages/stage-ui/src/stores/chat.ts | 46 +- .../src/stores/chat/session-store.test.ts | 605 +++++++++++++++++- .../stage-ui/src/stores/chat/session-store.ts | 308 ++++++--- .../context-bridge.contract.browser.test.ts | 217 ++++++- .../src/stores/mods/api/context-bridge.ts | 189 ++++-- vitest.config.ts | 2 +- 24 files changed, 2379 insertions(+), 623 deletions(-) create mode 100644 apps/stage-tamagotchi/src/renderer/components/InteractiveArea.browser.test.ts create mode 100644 apps/stage-tamagotchi/vitest.node.config.ts create mode 100644 packages/stage-ui/src/components/scenarios/chat/components/sessions-dialog.browser.test.ts create mode 100644 packages/stage-ui/src/components/scenarios/chat/components/sessions-dialog.vue create mode 100644 packages/stage-ui/src/components/scenarios/chat/components/sessions-drawer.browser.test.ts diff --git a/apps/stage-tamagotchi/src/renderer/App.vue b/apps/stage-tamagotchi/src/renderer/App.vue index 550bf3ace..845388cc4 100644 --- a/apps/stage-tamagotchi/src/renderer/App.vue +++ b/apps/stage-tamagotchi/src/renderer/App.vue @@ -79,6 +79,7 @@ const builtinToolsStore = useTamagotchiBuiltinToolsStore() const mcpToolsStore = useTamagotchiMcpToolsStore() const pluginToolsStore = useTamagotchiPluginToolsStore() const syncedPinia = usePiniaSynced() +chatSessionStore.setCloudSyncOwnership(syncedPinia.isLeader()) const isSpotlightWindowRoute = initialWindowRoutePath === '/spotlight' const isSettingsWindowRoute = initialWindowRoutePath === '/settings' || initialWindowRoutePath.startsWith('/settings/') const isEditorWindowRoute = initialWindowRoutePath === '/editor' @@ -99,7 +100,8 @@ async function refreshPluginRuntimeTools() { // Every renderer creates the runtime tool stores because every renderer can // become the leader. Only the leader discovers tools and keeps executors. -const stopToolLeadershipListener = syncedPinia.onLeadershipChange((isLeader) => { +const stopLeadershipListener = syncedPinia.onLeadershipChange((isLeader) => { + chatSessionStore.setCloudSyncOwnership(isLeader) if (!isLeader) return @@ -313,7 +315,7 @@ watch(themeColorsHueDynamic, () => { }, { immediate: true }) onUnmounted(() => { - stopToolLeadershipListener?.() + stopLeadershipListener?.() fullStageRuntime?.dispose() }) diff --git a/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.browser.test.ts b/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.browser.test.ts new file mode 100644 index 000000000..f6c24e433 --- /dev/null +++ b/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.browser.test.ts @@ -0,0 +1,320 @@ +import type { ChatSessionMeta } from '@proj-airi/stage-ui/types/chat-session' +import type { Component } from 'vue' + +import SharedInteractiveArea from '@proj-airi/stage-layouts/components/Layouts/InteractiveArea' +import MobileInteractiveArea from '@proj-airi/stage-layouts/components/Layouts/MobileInteractiveArea' +import ChatArea from '@proj-airi/stage-layouts/components/Widgets/ChatArea' + +import { PiniaColada } from '@pinia/colada' +import { useChatStore } from '@proj-airi/stage-ui/stores/chat' +import { useChatSessionStore } from '@proj-airi/stage-ui/stores/chat/session-store' +import { useChatStreamStore } from '@proj-airi/stage-ui/stores/chat/stream-store' +import { createPinia } from 'pinia' +import { describe, expect, it, vi } from 'vitest' +import { render } from 'vitest-browser-vue' +import { userEvent } from 'vitest/browser' +import { nextTick } from 'vue' +import { createI18n } from 'vue-i18n' +import { createMemoryHistory, createRouter } from 'vue-router' + +import InteractiveArea from './InteractiveArea.vue' + +function createTestI18n() { + return createI18n({ + legacy: false, + locale: 'en', + missingWarn: false, + fallbackWarn: false, + messages: { en: {} }, + }) +} + +async function renderArea(component: Component = InteractiveArea) { + const sessionB: ChatSessionMeta = { + sessionId: 'session-b', + userId: 'local', + characterId: 'default', + createdAt: 1, + updatedAt: 1, + } + const sessionA: ChatSessionMeta = { + ...sessionB, + sessionId: 'session-a', + createdAt: 2, + updatedAt: 2, + } + const pinia = createPinia() + pinia.state.value = { + 'chat-session-selection': { activeSessionId: 'session-b' }, + 'chat-session': { + sessionMetas: { 'session-a': sessionA, 'session-b': sessionB }, + sessionMessages: { + 'session-a': [{ id: 'system-a', role: 'system', content: 'session A prompt' }], + 'session-b': [{ id: 'system', role: 'system', content: 'system prompt' }], + }, + }, + } + const router = createRouter({ + history: createMemoryHistory(), + routes: [{ path: '/', component: { template: '
' } }], + }) + await router.push('/') + await router.isReady() + + const screen = await render(component, { + global: { plugins: [pinia, PiniaColada, createTestI18n(), router] }, + }) + return { + chat: useChatStore(pinia), + chatSession: useChatSessionStore(pinia), + chatStream: useChatStreamStore(pinia), + screen, + } +} + +async function submitDraft(screen: Awaited>['screen'], draft: string) { + const input = screen.getByRole('textbox') + await userEvent.fill(input, draft) + await userEvent.click(input) + await userEvent.keyboard('{Enter}') + return input +} + +describe('interactive area synchronized state', () => { + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743121861 + it('renders the active synchronized stream through the real chat history for Issue #2085', async () => { + // ROOT CAUSE: + // + // A follower received the leader-owned active stream in the real chat + // store, but InteractiveArea passed its unrelated foreground stream to + // ChatHistory. Mocking either store or component hid that broken binding. + const { chat, chatStream, screen } = await renderArea() + chat.$patch({ + activeSendSessionId: 'session-b', + activeStreamingMessage: { + id: 'follower-b-stream', + role: 'assistant', + content: 'Follower B live response', + slices: [{ type: 'text', text: 'Follower B live response' }], + tool_results: [], + createdAt: 2, + }, + sending: true, + }) + chatStream.$patch({ + streamingMessage: { + id: 'leader-a-stream', + role: 'assistant', + content: 'Leader A foreground response', + slices: [{ type: 'text', text: 'Leader A foreground response' }], + tool_results: [], + createdAt: 3, + }, + }) + await nextTick() + + await expect.element(screen.getByText('Follower B live response')).toBeVisible() + await expect.element(screen.getByText('Leader A foreground response')).not.toBeInTheDocument() + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743309235 + it('scopes the mobile synchronized stream to its local session for Issue #2085', async () => { + // ROOT CAUSE: + // + // MobileInteractiveArea passed the synchronized global sending state and + // foreground stream directly to ChatHistory. A mobile window on session B + // therefore rendered the live response from a send targeting session A. + const { chat, chatStream, screen } = await renderArea(MobileInteractiveArea) + chat.$patch({ + activeSendSessionId: 'session-a', + activeStreamingMessage: { + id: 'session-a-stream', + role: 'assistant', + content: 'Session A live response', + slices: [{ type: 'text', text: 'Session A live response' }], + tool_results: [], + createdAt: 2, + }, + sending: true, + }) + chatStream.$patch({ + streamingMessage: { + id: 'session-a-foreground', + role: 'assistant', + content: 'Session A live response', + slices: [{ type: 'text', text: 'Session A live response' }], + tool_results: [], + createdAt: 2, + }, + }) + await nextTick() + await expect.element(screen.getByText('Session A live response')).not.toBeInTheDocument() + + chat.$patch({ + activeSendSessionId: 'session-b', + activeStreamingMessage: { + id: 'session-b-stream', + role: 'assistant', + content: 'Session B live response', + slices: [{ type: 'text', text: 'Session B live response' }], + tool_results: [], + createdAt: 3, + }, + }) + await nextTick() + await expect.element(screen.getByText('Session B live response')).toBeVisible() + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743366443 + it('scopes the stage-web desktop synchronized stream to its local session for Issue #2085', async () => { + // ROOT CAUSE: + // + // The shared desktop layout derived sending from the target session but + // still passed the leader foreground stream to ChatHistory. A web window + // on B could therefore append A's live response. + const { chat, chatStream, screen } = await renderArea(SharedInteractiveArea) + chat.$patch({ + activeSendSessionId: 'session-b', + activeStreamingMessage: { + id: 'session-b-web-stream', + role: 'assistant', + content: 'Session B web response', + slices: [{ type: 'text', text: 'Session B web response' }], + tool_results: [], + createdAt: 2, + }, + sending: true, + }) + chatStream.$patch({ + streamingMessage: { + id: 'session-a-web-foreground', + role: 'assistant', + content: 'Session A foreground response', + slices: [{ type: 'text', text: 'Session A foreground response' }], + tool_results: [], + createdAt: 3, + }, + }) + await nextTick() + + await expect.element(screen.getByText('Session B web response')).toBeVisible() + await expect.element(screen.getByText('Session A foreground response')).not.toBeInTheDocument() + }) + + it('routes a stage-web send through the synchronized chat action', async () => { + const { chat, screen } = await renderArea(SharedInteractiveArea) + const send = vi.spyOn(chat, 'send').mockResolvedValueOnce({ messages: [], sessionId: 'session-b' }) + + await submitDraft(screen, 'web follower message') + + await vi.waitFor(() => expect(send).toHaveBeenCalledWith({ + sessionId: 'session-b', + text: 'web follower message', + })) + }) + + it('routes a mobile send through the synchronized chat action', async () => { + const { chat, screen } = await renderArea(MobileInteractiveArea) + const send = vi.spyOn(chat, 'send').mockResolvedValueOnce({ messages: [], sessionId: 'session-b' }) + + await submitDraft(screen, 'mobile follower message') + + await vi.waitFor(() => expect(send).toHaveBeenCalledWith({ + sessionId: 'session-b', + text: 'mobile follower message', + })) + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3755530944 + it('keeps a failed mobile draft out of a newly selected session for Issue #2085', async () => { + // ROOT CAUSE: + // + // Shared layouts restored a rejected send into their component-wide input + // without checking whether the window still displayed the target session. + const { chat, chatSession, screen } = await renderArea(MobileInteractiveArea) + let rejectSend: ((error: Error) => void) | undefined + vi.spyOn(chat, 'send').mockImplementationOnce(() => new Promise((_resolve, reject) => { + rejectSend = reject + })) + + const input = await submitDraft(screen, 'mobile draft from B') + chatSession.activeSessionId = 'session-a' + rejectSend?.(new Error('send failed')) + + await expect.element(input).toHaveValue('') + }) + + it('does not restore a deleted-session draft in the shared chat widget', async () => { + const { chat, screen } = await renderArea(ChatArea) + let rejectSend: ((error: Error) => void) | undefined + vi.spyOn(chat, 'send').mockImplementationOnce(() => new Promise((_resolve, reject) => { + rejectSend = reject + })) + + const input = await submitDraft(screen, 'deleted web draft') + rejectSend?.(new Error('Chat session was removed before send completed')) + + await expect.element(input).toHaveValue('') + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3628804992 + it('does not restore a failed draft into a newly selected session for Issue #2085', async () => { + // ROOT CAUSE: + // + // Failure recovery used the reactive selection instead of the session + // captured by the send, so a late rejection could move a draft. + const { chat, chatSession, screen } = await renderArea() + let rejectSend: ((error: Error) => void) | undefined + vi.spyOn(chat, 'send').mockImplementationOnce(() => new Promise((_resolve, reject) => { + rejectSend = reject + })) + + const input = await submitDraft(screen, 'send from B') + await expect.element(input).toHaveValue('') + chatSession.activeSessionId = 'session-a' + rejectSend?.(new Error('hydrate failed')) + + await expect.element(input).toHaveValue('') + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3629004140 + it('restores a failed draft when its captured session is still active for Issue #2085', async () => { + const { chat, screen } = await renderArea() + vi.spyOn(chat, 'send').mockRejectedValueOnce(new Error('send failed')) + + const input = await submitDraft(screen, 'retry this draft') + await expect.element(input).toHaveValue('retry this draft') + }) + + it('keeps a newer draft when an earlier send fails', async () => { + // ROOT CAUSE: + // + // Failure recovery replaced the textarea unconditionally. Text entered + // while the request was pending was lost with its attachment previews. + const { chat, screen } = await renderArea() + let rejectSend: ((error: Error) => void) | undefined + vi.spyOn(chat, 'send').mockImplementationOnce(() => new Promise((_resolve, reject) => { + rejectSend = reject + })) + + const input = await submitDraft(screen, 'first draft') + await userEvent.fill(input, 'newer draft') + rejectSend?.(new Error('send failed')) + + await expect.element(input).toHaveValue('first draft\nnewer draft') + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743366446 + it('discards a queued draft when deletion cancels its send for Issue #2085', async () => { + const { chat, screen } = await renderArea() + let rejectSend: ((error: Error) => void) | undefined + vi.spyOn(chat, 'send').mockImplementationOnce(() => new Promise((_resolve, reject) => { + rejectSend = reject + })) + + const input = await submitDraft(screen, 'discard this deleted draft') + rejectSend?.(new Error('Chat session was reset before send could start')) + + await expect.element(input).toHaveValue('') + }) +}) diff --git a/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.vue b/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.vue index ba51f6a8c..62c045bae 100644 --- a/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.vue +++ b/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.vue @@ -2,6 +2,7 @@ import type { ChatToolCallRendererRegistry } from '@proj-airi/stage-ui/components' import type { ChatHistoryItem } from '@proj-airi/stage-ui/types/chat' +import { errorMessageFrom } from '@moeru/std' import { useStopSpeakingButton } from '@proj-airi/stage-layouts/composables/useStopSpeakingButton' import { ChatHistory, JournalPreviewModal } from '@proj-airi/stage-ui/components' import { useAnalytics } from '@proj-airi/stage-ui/composables/use-analytics' @@ -37,9 +38,9 @@ const backgroundStore = useBackgroundStore() const journalPreviewStore = useJournalPreviewStore() const airiCardStore = useAiriCardStore() -const { messages } = storeToRefs(chatSession) +const { activeSessionId, messages } = storeToRefs(chatSession) const { streamingMessage } = storeToRefs(chatStream) -const { sending } = storeToRefs(chatStore) +const { activeSendSessionId, activeStreamingMessage, sending } = storeToRefs(chatStore) const { activeCard, activeCardId } = storeToRefs(airiCardStore) const { t } = useI18n() const { openImagePreview } = journalPreviewStore @@ -88,6 +89,9 @@ async function handleSend() { const textToSend = messageInput.value const attachmentsToSend = attachments.value.map(att => ({ ...att })) + // The active session can change while the cross-window request is pending. + // Keep one correlation key for both the send and its failure recovery. + const targetSessionId = chatSession.activeSessionId // optimistic clear messageInput.value = '' @@ -95,7 +99,7 @@ async function handleSend() { try { await chatStore.send({ - sessionId: chatSession.activeSessionId, + sessionId: targetSessionId, text: textToSend, attachments: attachmentsToSend, tools: artistryToolReferences, @@ -103,10 +107,21 @@ async function handleSend() { attachmentsToSend.forEach(att => URL.revokeObjectURL(att.url)) } - catch { - // restore on failure - messageInput.value = textToSend - attachments.value = attachmentsToSend + catch (error) { + const errorMessage = errorMessageFrom(error) ?? String(error) + const wasCancelledForDeletedSession + = errorMessage.includes('Chat session was reset before send could start') + || errorMessage.includes('Chat session was removed before send completed') + if (!wasCancelledForDeletedSession && chatSession.activeSessionId === targetSessionId) { + const currentDraft = messageInput.value + messageInput.value = currentDraft ? `${textToSend}\n${currentDraft}` : textToSend + attachments.value = [...attachmentsToSend, ...attachments.value] + } + else { + // This window no longer owns a visible attachment preview, so its Blob + // URLs must be released instead of surviving until the window closes. + attachmentsToSend.forEach(attachment => URL.revokeObjectURL(attachment.url)) + } } } @@ -197,6 +212,10 @@ watch(sendMode, () => { const historyMessages = computed(() => messages.value as unknown as ChatHistoryItem[]) const assistantLabel = computed(() => activeCard.value?.name?.trim() || undefined) +const isActiveSessionSending = computed(() => sending.value && activeSendSessionId.value === activeSessionId.value) +const visibleStreamingMessage = computed(() => activeSendSessionId.value === activeSessionId.value + ? activeStreamingMessage.value + : streamingMessage.value) async function handleDeleteMessage(index: number) { const message = messages.value[index] @@ -252,8 +271,8 @@ async function handleCleanupMessages() { chatSessionStore.setCloudSyncOwnership(isLeader)) const serverChannelStore = useModsServerChannelStore() const characterOrchestratorStore = useCharacterOrchestratorStore() const settingsAudioDeviceStore = useSettingsAudioDevice() @@ -105,6 +109,7 @@ onMounted(async () => { }) onUnmounted(() => { + stopLeadershipListener() contextBridgeStore.dispose() }) diff --git a/package.json b/package.json index 2e1293f27..d29ec733f 100644 --- a/package.json +++ b/package.json @@ -34,7 +34,8 @@ "build:packages": "turbo run build -F=\"./packages/*\"", "build:engines": "turbo run build -F=\"./engines/*\"", "test": "vitest --coverage", - "test:run": "vitest run && pnpm run test-audio-pipelines-transcribe:run && pnpm run test-ui:run", + "test:run": "vitest run && pnpm run test-stage-tamagotchi:run && pnpm run test-audio-pipelines-transcribe:run && pnpm run test-ui:run", + "test-stage-tamagotchi:run": "vitest run --config apps/stage-tamagotchi/vitest.config.ts --project browser", "test-audio-pipelines-transcribe:run": "vitest run --config packages/audio-pipelines-transcribe/vitest.config.ts", "test-ui:run": "vitest run --config packages/stage-ui/vitest.config.ts", "lint": "moeru-lint .", diff --git a/packages/core-agent/src/runtime/chat-orchestrator-runtime.test.ts b/packages/core-agent/src/runtime/chat-orchestrator-runtime.test.ts index 765f63a96..05e106235 100644 --- a/packages/core-agent/src/runtime/chat-orchestrator-runtime.test.ts +++ b/packages/core-agent/src/runtime/chat-orchestrator-runtime.test.ts @@ -142,11 +142,6 @@ function createHarness() { } } -/** - * @example - * const runtime = createChatOrchestratorRuntime(deps) - * await runtime.ingest('hello', { model, chatProvider }) - */ describe('createChatOrchestratorRuntime', () => { // ROOT CAUSE: // @@ -290,10 +285,6 @@ describe('createChatOrchestratorRuntime', () => { }) }) - /** - * @example - * Hook order and prompt composition stay compatible with the stage-ui facade. - */ it('keeps hook order and appends context prompt to the latest user message', async () => { const harness = createHarness() harness.contextSnapshot['system:weather'] = [ @@ -447,11 +438,6 @@ describe('createChatOrchestratorRuntime', () => { expect(legacyUserMessage.createdAt).toBe(new Date(2026, 3, 25, 18, 47).getTime()) }) - /** - * @example - * deps.getSystemPromptSupplement() returns tool guidance. - * The runtime appends it to the existing provider system message. - */ it('appends system prompt supplement to the provider system message', async () => { const harness = createHarness() let composedMessages: Message[] = [] @@ -473,11 +459,6 @@ describe('createChatOrchestratorRuntime', () => { }) }) - /** - * @example - * A session has only user history. - * The runtime creates a provider system message for supplemental guidance. - */ it('creates a system message when only a system prompt supplement is available', async () => { const harness = createHarness() let composedMessages: Message[] = [] @@ -501,10 +482,6 @@ describe('createChatOrchestratorRuntime', () => { expect(composedMessages[1]).toMatchObject({ role: 'user' }) }) - /** - * @example - * Runtime telemetry callbacks expose client-visible latency milestones. - */ it('emits telemetry milestones for a successful voice-backed message round', async () => { const harness = createHarness() harness.monotonicNow.set([100, 150, 250, 400, 460]) @@ -626,10 +603,6 @@ describe('createChatOrchestratorRuntime', () => { expect(harness.telemetry.messageRound).toHaveLength(2) }) - /** - * @example - * await expect(runtime.ingest('hello', { model, chatProvider })).rejects.toThrow('provider rejected') - */ it('emits chat activation failure telemetry without raw provider messages', async () => { const harness = createHarness() harness.stream.mockRejectedValueOnce(new Error('provider rejected with sensitive details')) @@ -696,10 +669,6 @@ describe('createChatOrchestratorRuntime', () => { ]) }) - /** - * @example - * Cancelling a queued send rejects only pending work that has not started. - */ it('rejects cancelled queued sends before they start', async () => { const harness = createHarness() let releaseFirstSend: (() => void) | undefined @@ -731,10 +700,57 @@ describe('createChatOrchestratorRuntime', () => { await firstSend }) - /** - * @example - * A queued send rejects if its captured session generation becomes stale. - */ + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3714754876 + it('suppresses completion hooks when an active send session is deleted for Issue #2085', async () => { + // ROOT CAUSE: + // + // Generation checks protected message mutation during a stream, but the + // runtime still emitted completion hooks and success analytics after the + // provider returned for a deleted session. + const harness = createHarness() + const completionHook = vi.fn() + harness.runtime.hooks.onStreamEnd(completionHook) + harness.runtime.hooks.onAssistantResponseEnd(completionHook) + harness.runtime.hooks.onAfterSend(completionHook) + harness.runtime.hooks.onAssistantMessage(completionHook) + harness.runtime.hooks.onChatTurnComplete(completionHook) + + let finishStream: (() => void) | undefined + harness.stream.mockImplementationOnce(async (_model, _chatProvider, _messages, options) => { + await new Promise((resolve) => { + finishStream = resolve + }) + options?.onUsage?.({ + inputTokens: 1, + outputTokens: 1, + totalTokens: 2, + source: 'reported', + }) + await options?.onStreamEvent?.({ type: 'text-delta', text: 'deleted reply' }) + await options?.onStreamEvent?.({ type: 'finish', finishReason: 'stop' }) + }) + + const pendingSend = harness.runtime.ingest('delete this chat', { + model: 'gpt-test', + chatProvider: provider, + }) + + await vi.waitFor(() => { + expect(harness.stream).toHaveBeenCalledTimes(1) + }) + harness.generation.set(2) + finishStream?.() + await pendingSend + + expect(completionHook).not.toHaveBeenCalled() + expect(harness.assistantAppended).toEqual([]) + expect(harness.assistantTurns).toEqual([]) + expect(harness.telemetry.assistantResponseRendered).toEqual([]) + expect(harness.telemetry.llmGeneration).toEqual([]) + expect(harness.telemetry.messageRound).toEqual([]) + expect(harness.telemetry.chatActivationSucceeded).toEqual([]) + }) + it('rejects stale generation sends before they start', async () => { const harness = createHarness() let releaseFirstSend: (() => void) | undefined @@ -767,17 +783,14 @@ describe('createChatOrchestratorRuntime', () => { expect(harness.stream).toHaveBeenCalledTimes(1) }) - /** - * @example - * runtime.setSending(true) - * expect(runtime.getSending()).toBe(true) - */ it('keeps sending externally writable for UI facades', () => { const harness = createHarness() harness.runtime.setSending(true) expect(harness.runtime.getSending()).toBe(true) expect(harness.stateChanges.at(-1)).toEqual({ + activeSendSessionId: 'session-1', + activeStreamingMessage: undefined, sending: true, pendingQueuedSendCount: 0, }) @@ -785,16 +798,66 @@ describe('createChatOrchestratorRuntime', () => { harness.runtime.setSending(false) expect(harness.runtime.getSending()).toBe(false) expect(harness.stateChanges.at(-1)).toEqual({ + activeSendSessionId: undefined, + activeStreamingMessage: undefined, + sending: false, + pendingQueuedSendCount: 0, + }) + }) + + // https://github.com/moeru-ai/airi/issues/2085 + it('reports the queued send target while a background session is sending for Issue #2085', async () => { + // ROOT CAUSE: + // + // Runtime state exposed only a global sending boolean. A window-level sync + // layer therefore had to infer the owner from the authority's visible + // session, which is wrong when a follower targets a background session. + const harness = createHarness() + let finishSend: (() => void) | undefined + harness.stream.mockImplementationOnce(async (_model, _chatProvider, _messages, options) => { + await options?.onStreamEvent?.({ type: 'text-delta', text: 'background reply' }) + await new Promise((resolve) => { + finishSend = resolve + }) + }) + + const pendingSend = harness.runtime.ingest('background request', { + model: 'gpt-test', + chatProvider: provider, + }, 'session-2') + + await vi.waitFor(() => { + expect(harness.stateChanges).toContainEqual(expect.objectContaining({ + activeSendSessionId: 'session-2', + activeStreamingMessage: expect.objectContaining({ + role: 'assistant', + createdAt: expect.any(Number), + }), + sending: true, + pendingQueuedSendCount: 0, + })) + }) + await vi.waitFor(() => { + expect(harness.stream).toHaveBeenCalledTimes(1) + }) + await vi.waitFor(() => { + expect(harness.stateChanges).toContainEqual(expect.objectContaining({ + activeSendSessionId: 'session-2', + activeStreamingMessage: expect.objectContaining({ content: expect.stringContaining('background') }), + })) + }) + + finishSend?.() + await pendingSend + + expect(harness.stateChanges.at(-1)).toEqual({ + activeSendSessionId: undefined, + activeStreamingMessage: undefined, sending: false, pendingQueuedSendCount: 0, }) }) - /** - * @example - * const snapshot = runtime.getPendingQueuedSendSnapshot() - * expect(snapshot[0].inputType).toBe('input:text') - */ it('returns pending queued send snapshots with public fields', async () => { const harness = createHarness() let releaseFirstSend: (() => void) | undefined @@ -852,10 +915,6 @@ describe('createChatOrchestratorRuntime', () => { await firstSend }) - /** - * @example - * Attachments, reasoning deltas, and tool events update the assistant builder. - */ it('handles attachments, reasoning deltas, tool events, and assistant finalization', async () => { const harness = createHarness() let composedMessages: Message[] = [] diff --git a/packages/core-agent/src/runtime/chat-orchestrator-runtime.ts b/packages/core-agent/src/runtime/chat-orchestrator-runtime.ts index 2a71115f2..c4058a890 100644 --- a/packages/core-agent/src/runtime/chat-orchestrator-runtime.ts +++ b/packages/core-agent/src/runtime/chat-orchestrator-runtime.ts @@ -154,6 +154,10 @@ export interface ChatOrchestratorPromptProjection { export interface ChatOrchestratorRuntimeState { /** Whether the runtime currently owns an active send. */ sending: boolean + /** Session that owns the active send; undefined while the queue is idle. */ + activeSendSessionId?: string + /** Latest assistant stream snapshot owned by the active send session. */ + activeStreamingMessage?: StreamingAssistantMessage /** Number of sends waiting behind the active one. */ pendingQueuedSendCount: number } @@ -350,19 +354,29 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) const unwrapMessage = deps.unwrapMessage ?? ((message: T) => message) let sending = false + let activeSendSessionId: string | undefined + let activeStreamingMessage: StreamingAssistantMessage | undefined let pendingQueuedSends: QueuedSend[] = [] function emitStateChange() { deps.onStateChange?.({ sending, + activeSendSessionId, + activeStreamingMessage, pendingQueuedSendCount: pendingQueuedSends.length, }) } function setSending(next: boolean) { - if (sending === next) + const nextActiveSendSessionId = next + ? activeSendSessionId ?? deps.getActiveSessionId() + : undefined + if (sending === next && activeSendSessionId === nextActiveSendSessionId) return sending = next + activeSendSessionId = nextActiveSendSessionId + if (!next) + activeStreamingMessage = undefined emitStateChange() } @@ -370,7 +384,22 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) return sessionId === deps.getActiveSessionId() } - function patchForegroundStream(sessionId: string, message: StreamingAssistantMessage) { + function beginStream(sessionId: string, message: StreamingAssistantMessage) { + sending = true + activeSendSessionId = sessionId + activeStreamingMessage = cloneStreamingMessage(message) + emitStateChange() + + if (isForegroundSession(sessionId)) + deps.foregroundStream.patch(cloneStreamingMessage(message)) + } + + function updateStream(sessionId: string, message: StreamingAssistantMessage) { + if (sessionId === activeSendSessionId) { + activeStreamingMessage = cloneStreamingMessage(message) + emitStateChange() + } + if (isForegroundSession(sessionId)) deps.foregroundStream.patch(cloneStreamingMessage(message)) } @@ -481,8 +510,6 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) if (shouldAbort()) return - setSending(true) - const buildingMessage: StreamingAssistantMessage = { role: 'assistant', content: '', @@ -491,7 +518,7 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) createdAt: now(), id: assistantMessageId, } - patchForegroundStream(sessionId, buildingMessage) + beginStream(sessionId, buildingMessage) const sendSource = options.input ? 'voice' : 'text' const activeProvider = deps.getActiveProvider?.() ?? '' // The user message is the durable start of a round, so its ID also serves @@ -604,7 +631,7 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) text: speechOnly, }) } - patchForegroundStream(sessionId, buildingMessage) + updateStream(sessionId, buildingMessage) } }, onSpecial: async (special) => { @@ -624,7 +651,7 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) speech: finalCategorization.speech, reasoning: reasoningContentField || finalCategorization.reasoning, } - patchForegroundStream(sessionId, buildingMessage) + updateStream(sessionId, buildingMessage) }, // The parser keeps its own marker-safety tail. Emit each safe literal // chunk so slow providers update the chat before they reach 24 characters. @@ -638,13 +665,13 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) return if (ctx.data.type === 'tool-call') { buildingMessage.slices.push(ctx.data) - patchForegroundStream(sessionId, buildingMessage) + updateStream(sessionId, buildingMessage) return } if (ctx.data.type === 'tool-call-result') { buildingMessage.tool_results.push(ctx.data) - patchForegroundStream(sessionId, buildingMessage) + updateStream(sessionId, buildingMessage) } }, ], @@ -749,6 +776,9 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) providerTranscript = structuredClone(currentTurnMessages) }, onUsage: (usage) => { + if (shouldAbort()) + return + generationUsage = usage deps.onLlmGeneration?.({ ...correlation, @@ -761,6 +791,9 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) }) }, onStreamEvent: async (event: StreamEvent) => { + if (shouldAbort()) + return + switch (event.type) { case 'tool-call': toolCallQueue.enqueue({ @@ -812,7 +845,7 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) = Math.floor(nextReasoning.length / REASONING_UI_FLUSH_CHUNK_SIZE) > Math.floor(reasoning.length / REASONING_UI_FLUSH_CHUNK_SIZE) if (!reasoning || crossesBoundary) - patchForegroundStream(sessionId, buildingMessage) + updateStream(sessionId, buildingMessage) break } case 'finish': @@ -823,7 +856,16 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) }, }) + // Session generation is the lifecycle correlation key. Re-check it + // after every awaited completion boundary so deleting a session while a + // plugin hook runs cannot leak later hooks or success analytics. + if (shouldAbort()) + return + await parser.end() + if (shouldAbort()) + return + buildingMessage.providerTranscript = providerTranscript deps.onAssistantResponseRendered?.({ ...correlation, @@ -841,17 +883,29 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) }) } + if (shouldAbort()) + return await hooks.emitStreamEndHooks(streamingMessageContext) + if (shouldAbort()) + return await hooks.emitAssistantResponseEndHooks(fullText, streamingMessageContext) + if (shouldAbort()) + return await hooks.emitAfterSendHooks(sendingMessage, streamingMessageContext) + if (shouldAbort()) + return await hooks.emitAssistantMessageHooks({ ...buildingMessage }, fullText, streamingMessageContext) + if (shouldAbort()) + return await hooks.emitChatTurnCompleteHooks({ output: { ...buildingMessage }, outputText: fullText, toolCalls: sessionMessagesForSend.filter(msg => msg.role === 'tool') as ToolMessage[], }, streamingMessageContext) + if (shouldAbort()) + return deps.onAssistantTurnReady?.({ messageText: fullText, sessionMessages: sessionMessagesForSend, @@ -880,6 +934,9 @@ export function createChatOrchestratorRuntime(deps: ChatOrchestratorRuntimeDeps) } } catch (error) { + if (isStaleGeneration()) + return + console.error('Error sending message:', error) deps.onMessageRoundFailed?.({ ...correlation, diff --git a/packages/stage-layouts/src/components/Layouts/InteractiveArea.vue b/packages/stage-layouts/src/components/Layouts/InteractiveArea.vue index 66d66c74e..81ef5257c 100644 --- a/packages/stage-layouts/src/components/Layouts/InteractiveArea.vue +++ b/packages/stage-layouts/src/components/Layouts/InteractiveArea.vue @@ -6,6 +6,7 @@ import { useAnalytics } from '@proj-airi/stage-ui/composables/use-analytics' import { useChatStore } from '@proj-airi/stage-ui/stores/chat' import { useChatSessionStore } from '@proj-airi/stage-ui/stores/chat/session-store' import { useChatStreamStore } from '@proj-airi/stage-ui/stores/chat/stream-store' +import { useContextBridgeStore } from '@proj-airi/stage-ui/stores/mods/api/context-bridge' import { useDeferredMount } from '@proj-airi/ui' import { storeToRefs } from 'pinia' import { computed, ref } from 'vue' @@ -17,18 +18,30 @@ import ChatContainer from '../Widgets/ChatContainer.vue' import { useChatToolCallRerun } from '../../composables/useChatToolCallRerun' const { isReady } = useDeferredMount() -const { sending } = storeToRefs(useChatStore()) -const { messages } = storeToRefs(useChatSessionStore()) +const { activeSendSessionId, activeStreamingMessage, sending } = storeToRefs(useChatStore()) +const { activeSessionId, messages } = storeToRefs(useChatSessionStore()) const { streamingMessage } = storeToRefs(useChatStreamStore()) +const { isReceivingRemoteStream } = storeToRefs(useContextBridgeStore()) const isLoading = ref(true) const historyMessages = computed(() => messages.value as unknown as ChatHistoryItem[]) +const isActiveSessionSending = computed(() => ( + (sending.value && activeSendSessionId.value === activeSessionId.value) + || isReceivingRemoteStream.value +)) +const visibleStreamingMessage = computed(() => activeSendSessionId.value === activeSessionId.value + ? activeStreamingMessage.value + : streamingMessage.value) const { trackChatMessageDeleted } = useAnalytics() const { rerunToolCall } = useChatToolCallRerun() -function handleDeleteMessage(index: number) { +async function handleDeleteMessage(index: number) { const message = messages.value[index] - messages.value = messages.value.filter((_, messageIndex) => messageIndex !== index) + await useChatSessionStore().deleteMessage({ + sessionId: activeSessionId.value, + messageId: message?.id, + index, + }) trackChatMessageDeleted({ source: 'history', message_role: message?.role ?? 'unknown', @@ -51,8 +64,8 @@ function handleDeleteMessage(index: number) { import type { ChatHistoryItem } from '@proj-airi/stage-ui/types/chat' -import type { ChatProvider } from '@xsai-ext/providers/utils' import { errorMessageFrom } from '@moeru/std' import { isStageTamagotchi } from '@proj-airi/stage-shared' @@ -14,9 +13,7 @@ import { useChatMaintenanceStore } from '@proj-airi/stage-ui/stores/chat/mainten import { useChatSessionStore } from '@proj-airi/stage-ui/stores/chat/session-store' import { useChatStreamStore } from '@proj-airi/stage-ui/stores/chat/stream-store' import { useL2dViewControl } from '@proj-airi/stage-ui/stores/live2d' -import { useConsciousnessStore } from '@proj-airi/stage-ui/stores/modules/consciousness' -import { useProviderConfigStore } from '@proj-airi/stage-ui/stores/providers/config' -import { useProviderStore } from '@proj-airi/stage-ui/stores/providers/provider' +import { useContextBridgeStore } from '@proj-airi/stage-ui/stores/mods/api/context-bridge' import { useSettings, useSettingsAudioDevice } from '@proj-airi/stage-ui/stores/settings' import { BasicTextarea, useTheme } from '@proj-airi/ui' import { useResizeObserver, useScreenSafeArea } from '@vueuse/core' @@ -39,16 +36,28 @@ const chatOrchestrator = useChatStore() const chatSession = useChatSessionStore() const chatStream = useChatStreamStore() const { cleanupMessages } = useChatMaintenanceStore() -const { messages } = storeToRefs(chatSession) +const { activeSessionId, messages } = storeToRefs(chatSession) const { streamingMessage } = storeToRefs(chatStream) -const { sending } = storeToRefs(chatOrchestrator) +const { activeSendSessionId, activeStreamingMessage, sending } = storeToRefs(chatOrchestrator) +const { isReceivingRemoteStream } = storeToRefs(useContextBridgeStore()) const historyMessages = computed(() => messages.value as unknown as ChatHistoryItem[]) +const isActiveSessionSending = computed(() => ( + (sending.value && activeSendSessionId.value === activeSessionId.value) + || isReceivingRemoteStream.value +)) +const visibleStreamingMessage = computed(() => activeSendSessionId.value === activeSessionId.value + ? activeStreamingMessage.value + : streamingMessage.value) const { trackChatMessageDeleted, trackChatMessagesCleared } = useAnalytics() const { rerunToolCall } = useChatToolCallRerun() -function handleDeleteMessage(index: number) { +async function handleDeleteMessage(index: number) { const message = messages.value[index] - messages.value = messages.value.filter((_, messageIndex) => messageIndex !== index) + await chatSession.deleteMessage({ + sessionId: activeSessionId.value, + messageId: message?.id, + index, + }) trackChatMessageDeleted({ source: 'history', message_role: message?.role ?? 'unknown', @@ -70,17 +79,12 @@ const backgroundDialogOpen = ref(false) const sessionsDrawerOpen = ref(false) const screenSafeArea = useScreenSafeArea() -const providersStore = useProviderStore() -const providerStore = useProviderConfigStore() -const { activeProvider, activeModel } = storeToRefs(useConsciousnessStore()) - useResizeObserver(document.documentElement, () => screenSafeArea.update()) const { themeColorsHueDynamic } = storeToRefs(useSettings()) const { viewControlsEnabled: l2dViewCtrlEnabled } = useL2dViewControl() const { viewControlsEnabled: threeViewCtrlEnabled } = useThreeViewControl() const settingsAudioDevice = useSettingsAudioDevice() const { enabled, stream } = storeToRefs(settingsAudioDevice) -const { ingest, onAfterMessageComposed } = chatOrchestrator const { t } = useI18n() const { audioContext } = useAudioContext() const { startAnalyzer, stopAnalyzer } = useAudioAnalyzer() @@ -112,23 +116,24 @@ async function handleSend() { } const textToSend = messageInput.value + const targetSessionId = chatSession.activeSessionId messageInput.value = '' try { - const providerConfig = providerStore.getProviderConfig(activeProvider.value) - - await ingest(textToSend, { - chatProvider: await providersStore.getProviderInstance(activeProvider.value) as ChatProvider, - model: activeModel.value, - providerConfig, + await chatOrchestrator.send({ + sessionId: targetSessionId, + text: textToSend, }) } catch (error) { - messageInput.value = textToSend - chatSession.appendSessionMessage(chatSession.activeSessionId, { - role: 'error', - content: errorMessageFrom(error) ?? 'Failed to send message', - }) + const errorMessage = errorMessageFrom(error) ?? String(error) + const wasCancelledForDeletedSession + = errorMessage.includes('Chat session was reset before send could start') + || errorMessage.includes('Chat session was removed before send completed') + if (!wasCancelledForDeletedSession && chatSession.activeSessionId === targetSessionId) { + const currentDraft = messageInput.value + messageInput.value = currentDraft ? `${textToSend}\n${currentDraft}` : textToSend + } } } @@ -158,9 +163,6 @@ watch([enabled, stream], () => { setupAnalyzer() }, { immediate: true }) -onAfterMessageComposed(async () => { -}) - onUnmounted(() => { teardownAnalyzer() }) @@ -179,8 +181,8 @@ onMounted(() => { v-if="!threeViewCtrlEnabled && !l2dViewCtrlEnabled" variant="mobile" :messages="historyMessages" - :sending="sending" - :streaming-message="streamingMessage" + :sending="isActiveSessionSending" + :streaming-message="visibleStreamingMessage" max-w="[calc(100%-3.5rem)]" w-full self-start pb-3 pl-3 class="chat-history" diff --git a/packages/stage-layouts/src/components/Widgets/ChatArea.vue b/packages/stage-layouts/src/components/Widgets/ChatArea.vue index 05a121024..6f69f4351 100644 --- a/packages/stage-layouts/src/components/Widgets/ChatArea.vue +++ b/packages/stage-layouts/src/components/Widgets/ChatArea.vue @@ -1,6 +1,4 @@ + + diff --git a/packages/stage-ui/src/components/scenarios/chat/components/sessions-drawer.browser.test.ts b/packages/stage-ui/src/components/scenarios/chat/components/sessions-drawer.browser.test.ts new file mode 100644 index 000000000..301b399e1 --- /dev/null +++ b/packages/stage-ui/src/components/scenarios/chat/components/sessions-drawer.browser.test.ts @@ -0,0 +1,149 @@ +import type { ChatSessionMeta } from '../../../../types/chat-session' + +import { PiniaColada } from '@pinia/colada' +import { createPinia } from 'pinia' +import { describe, expect, it, vi } from 'vitest' +import { render } from 'vitest-browser-vue' +import { createI18n } from 'vue-i18n' + +import SessionsDrawer from './sessions-drawer.vue' + +import { useChatStore } from '../../../../stores/chat' +import { useChatSessionStore } from '../../../../stores/chat/session-store' + +function createTestI18n() { + return createI18n({ + legacy: false, + locale: 'en', + missingWarn: false, + fallbackWarn: false, + messages: { + en: { + stage: { + chat: { + sessions: { + 'title': 'Chats', + 'new': 'New chat', + 'empty': 'No chats', + 'delete': 'Delete', + 'cloud-badge': 'Cloud synced', + }, + }, + }, + }, + }, + }) +} + +function sessionMeta(sessionId: string, title: string, updatedAt: number): ChatSessionMeta { + return { + sessionId, + title, + characterId: 'default', + userId: 'local', + createdAt: updatedAt, + updatedAt, + } +} + +function createSessionsPinia() { + const pinia = createPinia() + pinia.state.value = { + 'chat-session-selection': { + activeSessionId: 'session-b', + }, + 'chat-session': { + sessionMetas: { + 'session-a': sessionMeta('session-a', 'Chat A', 1), + 'session-b': sessionMeta('session-b', 'Chat B', 3), + 'session-c': sessionMeta('session-c', 'Chat C', 2), + }, + sessionMessages: { + 'session-a': [], + 'session-b': [], + 'session-c': [], + }, + }, + } + return pinia +} + +describe('sessions drawer orchestration', () => { + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743073795 + it('preserves a newer selection while active-session deletion is pending for Issue #2085', async () => { + // ROOT CAUSE: + // + // Deletion captured that B was active, then unconditionally selected its + // fallback after the asynchronous leader action completed. A user choice + // of C made during that await was therefore overwritten by stale work. + const pinia = createSessionsPinia() + + const screen = await render(SessionsDrawer, { + props: { modelValue: false }, + global: { + plugins: [pinia, PiniaColada, createTestI18n()], + }, + }) + const chatSession = useChatSessionStore(pinia) + const chat = useChatStore(pinia) + let resolveDelete: (() => void) | undefined + vi.spyOn(chat, 'deleteSession').mockImplementationOnce(() => new Promise((resolve) => { + resolveDelete = resolve + })) + vi.spyOn(chatSession, 'loadSession').mockResolvedValue(true) + vi.spyOn(chatSession, 'setActiveSession').mockImplementation(async (sessionId) => { + chatSession.activeSessionId = sessionId + }) + await screen.rerender({ modelValue: true }) + + await screen.getByRole('button', { name: 'Delete: Chat B' }).click() + await vi.waitFor(() => expect(chat.deleteSession).toHaveBeenCalledWith('session-b')) + + await screen.getByRole('button', { name: /^Chat C / }).click() + expect(chatSession.activeSessionId).toBe('session-c') + + resolveDelete?.() + await vi.waitFor(() => expect(chatSession.setActiveSession).toHaveBeenCalledTimes(1)) + + expect(chatSession.setActiveSession).toHaveBeenLastCalledWith('session-c') + expect(chatSession.activeSessionId).toBe('session-c') + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743221030 + it('preserves a newer selection while session creation is pending for Issue #2085', async () => { + // ROOT CAUSE: + // + // New-session creation awaited leader persistence, then unconditionally + // selected its result. Session rows stay enabled during that wait, so a + // newer row selection was overwritten when creation eventually resolved. + const pinia = createSessionsPinia() + const screen = await render(SessionsDrawer, { + props: { modelValue: false }, + global: { + plugins: [pinia, PiniaColada, createTestI18n()], + }, + }) + const chatSession = useChatSessionStore(pinia) + let resolveCreate: ((sessionId: string) => void) | undefined + vi.spyOn(chatSession, 'createSession').mockImplementationOnce(() => new Promise((resolve) => { + resolveCreate = resolve + })) + vi.spyOn(chatSession, 'loadSession').mockResolvedValue(true) + vi.spyOn(chatSession, 'setActiveSession').mockImplementation(async (sessionId) => { + chatSession.activeSessionId = sessionId + }) + await screen.rerender({ modelValue: true }) + + await screen.getByRole('button', { name: 'New chat' }).click() + await vi.waitFor(() => expect(chatSession.createSession).toHaveBeenCalledWith('default', { setActive: false })) + + await screen.getByRole('button', { name: /^Chat C / }).click() + expect(chatSession.activeSessionId).toBe('session-c') + + resolveCreate?.('session-new') + await vi.waitFor(() => expect(chatSession.setActiveSession).toHaveBeenCalledTimes(1)) + + expect(chatSession.setActiveSession).toHaveBeenLastCalledWith('session-c') + expect(chatSession.activeSessionId).toBe('session-c') + }) +}) diff --git a/packages/stage-ui/src/components/scenarios/chat/components/sessions-drawer.vue b/packages/stage-ui/src/components/scenarios/chat/components/sessions-drawer.vue index 306a6c137..e0f107d4a 100644 --- a/packages/stage-ui/src/components/scenarios/chat/components/sessions-drawer.vue +++ b/packages/stage-ui/src/components/scenarios/chat/components/sessions-drawer.vue @@ -3,39 +3,20 @@ import type { ChatSessionMeta } from '../../../../types/chat-session' import { useResizeObserver, useScreenSafeArea } from '@vueuse/core' import { storeToRefs } from 'pinia' -import { DialogContent, DialogOverlay, DialogPortal, DialogRoot, DialogTitle } from 'reka-ui' -import { DrawerContent, DrawerHandle, DrawerOverlay, DrawerPortal, DrawerRoot, DrawerTitle } from 'vaul-vue' import { computed, onMounted, ref, watch } from 'vue' import { useI18n } from 'vue-i18n' +import SessionsDialog from './sessions-dialog.vue' + import { useAnalytics } from '../../../../composables/use-analytics' import { useBreakpoints } from '../../../../composables/use-breakpoints' import { extractMessageText } from '../../../../libs/chat-sync' import { useAuthStore } from '../../../../stores/auth' +import { useChatStore } from '../../../../stores/chat' import { useChatSessionStore } from '../../../../stores/chat/session-store' import { useAiriCardStore } from '../../../../stores/modules/airi-card' import { useConsciousnessStore } from '../../../../stores/modules/consciousness' -/** - * Bottom-sheet (mobile) / centered-modal (desktop) UI surface that lists every - * chat session belonging to the current user, lets the user switch between - * them, and start a fresh session for the active character. - * - * Use when: - * - The user is on a stage page and wants to browse / switch conversations. - * Mounted once near the global ChatArea so any input bar can flip the - * `v-model` open. - * - * Expects: - * - `useChatSessionStore` is initialized — `sessionMetas` and `activeSessionId` - * drive the list, and switching calls `setActiveSession` / `createSession`. - * - * Returns: - * - A scrollable list. List items render the session title (or first user - * message preview as a fallback), a cloud-sync badge, and a relative - * updatedAt timestamp. - */ - const showDialog = defineModel({ type: Boolean, default: false, required: false }) const { isDesktop } = useBreakpoints() @@ -43,18 +24,15 @@ const screenSafeArea = useScreenSafeArea() const { t } = useI18n() const chatSession = useChatSessionStore() +const chat = useChatStore() const { sessionMetas, sessionMessages, activeSessionId } = storeToRefs(chatSession) const { activeCardId } = storeToRefs(useAiriCardStore()) const { userId } = storeToRefs(useAuthStore()) const { activeModel } = storeToRefs(useConsciousnessStore()) const { trackChatSessionSelected, trackChatSessionStarted } = useAnalytics() -// Re-entry guard for the "new session" button. Without this, a rapid -// double-click would call `createSession` twice (creating two orphan -// sessions) and emit duplicate `chat_session_started` analytics events. -// The async `createSession` includes IndexedDB writes + a cloud reconcile -// kick-off, so even a single click can stay in flight long enough for a -// second click to slip through. +// Creating includes persistence and cloud reconciliation, so prevent a +// second click from creating an orphan session while the first is pending. const isCreatingSession = ref(false) useResizeObserver(document.documentElement, () => screenSafeArea.update()) @@ -67,30 +45,18 @@ interface SessionRow { updatedAtLabel: string } -/** - * Sessions visible in the drawer. Filters by the currently effective user - * (`userId.value || 'local'`) so: - * - Anonymous users see their local-only sessions (previously hidden by a - * blanket `userId !== 'local'` filter). - * - After an account swap, the previously signed-in user's sessions stay - * hidden until ensureActiveSessionForCharacter rehydrates the new tenant - * (the session-store also clears in-memory state on user change as a - * defense in depth). - */ +// Keep another account's sessions hidden while an account swap rehydrates. const ownedSessions = computed(() => { const effectiveUserId = userId.value || 'local' return Object.values(sessionMetas.value).filter(meta => meta.userId === effectiveUserId) }) /** - * Pull a 1-line preview from the first non-system message; falls back to the - * stored title or a generic placeholder when nothing readable is available. + * Normalizes a session into its one-line drawer preview. * - * Before: - * - messages: [system, { role: 'user', content: 'Tell me about the moon today' }, ...] - * - * After: - * - "Tell me about the moon today" + * @example + * previewFor({ title: 'Moon notes', ...meta }) + * // => 'Moon notes' */ function previewFor(meta: ChatSessionMeta): string { if (meta.title) @@ -118,13 +84,11 @@ const RELATIVE_UNITS: Array<[Intl.RelativeTimeFormatUnit, number]> = [ ] /** - * Format an epoch ms timestamp as a coarse relative label like "3 minutes ago". + * Normalizes an epoch timestamp into a coarse relative label. * - * Before: - * - Date.now() - 5 * 60 * 1000 - * - * After: - * - "5 minutes ago" + * @example + * formatUpdatedAt(Date.now() - 5 * 60 * 1000) + * // => '5 minutes ago' */ function formatUpdatedAt(ts: number): string { const formatter = new Intl.RelativeTimeFormat(undefined, { numeric: 'auto' }) @@ -147,12 +111,15 @@ const rows = computed(() => { isActive: meta.sessionId === activeSessionId.value, updatedAtLabel: formatUpdatedAt(meta.updatedAt), })) - // Most-recent first; the active session usually ends up at the top after a - // fresh send because `persistSession` bumps `updatedAt`. list.sort((a, b) => b.meta.updatedAt - a.meta.updatedAt) return list }) +const mobilePaddingBottom = computed(() => { + const safeAreaBottom = Number.parseFloat(screenSafeArea.bottom.value.replace('px', '')) + return `${Math.max(safeAreaBottom, 24)}px` +}) + async function selectSession(sessionId: string) { const selectedRow = rows.value.find(row => row.meta.sessionId === sessionId) if (sessionId !== activeSessionId.value && selectedRow) { @@ -162,7 +129,7 @@ async function selectSession(sessionId: string) { cloud_synced: !!selectedRow.meta.cloudChatId, }) } - chatSession.setActiveSession(sessionId) + await chatSession.setActiveSession(sessionId) showDialog.value = false } @@ -172,12 +139,16 @@ async function startNewSession() { isCreatingSession.value = true try { const characterId = activeCardId.value || 'default' - await chatSession.createSession(characterId, { setActive: true }) - // PostHog retention denominator. We pick this call site (UI new-session - // button) rather than `createSession` in the store because the store also - // creates sessions for cloud-reconcile / fork / restore flows that aren't - // user-initiated. Model id is informational; sessionIndex is omitted - // (PostHog can compute it from per-user event ordering). + const selectionBeforeCreation = activeSessionId.value + // Creation runs in the synchronized leader, while navigation belongs to + // this window. Activating inside createSession would navigate the leader. + const sessionId = await chatSession.createSession(characterId, { setActive: false }) + // Rows remain interactive while creation is persisted in the leader. Do + // not let that stale continuation replace a newer local user selection. + if (activeSessionId.value === selectionBeforeCreation) + await chatSession.setActiveSession(sessionId) + // Store-created sessions also include restore and fork flows; only this + // user action belongs in the retention denominator. trackChatSessionStarted(activeModel.value || 'unknown') showDialog.value = false } @@ -186,13 +157,6 @@ async function startNewSession() { } } -async function deleteRow(event: Event, sessionId: string) { - // Stop the parent button's click — otherwise we'd switch into the session - // we are about to remove and immediately need a fallback. - event.stopPropagation() - await chatSession.deleteSession(sessionId) -} - // Per-open generation counter. The batch loadSession loop checks this before // each batch so closing the drawer mid-load aborts cleanly instead of // continuing to hydrate sessions the user has navigated away from. Without @@ -200,17 +164,11 @@ async function deleteRow(event: Event, sessionId: string) { // re-added to `loadedSessions` as a phantom entry. let openGeneration = 0 -// Re-render relative timestamps + hydrate non-active session messages when -// the drawer opens so each row can show a real preview instead of the -// fallback. `loadSession` is idempotent (`loadedSessions` set), so reopening -// the drawer is cheap. watch(showDialog, async (open) => { if (!open) return openGeneration += 1 const myGeneration = openGeneration - // Touch `rows` first so reactive labels reflect a fresh `Date.now()`. - void rows.value const knownSessionIds = ownedSessions.value.map(meta => meta.sessionId) // Bounded concurrency keeps a long history list from spawning a hundred // simultaneous IndexedDB transactions; 4 in flight is plenty for a list @@ -225,172 +183,18 @@ watch(showDialog, async (open) => { diff --git a/packages/stage-ui/src/stores/character/orchestrator/index.test.ts b/packages/stage-ui/src/stores/character/orchestrator/index.test.ts index b19d7c9ec..97b69a1a6 100644 --- a/packages/stage-ui/src/stores/character/orchestrator/index.test.ts +++ b/packages/stage-ui/src/stores/character/orchestrator/index.test.ts @@ -239,11 +239,13 @@ describe('store character-orchestrator', () => { forceTextResponse: true, }) - const streamOptions = mockStream.mock.calls[0][3] - expect(streamOptions.supportsTools).toBe(false) - expect(streamOptions.waitForTools).toBe(false) - expect(streamOptions.tools).toEqual([]) - expect(streamOptions.toolChoice).toBeUndefined() + const streamOptions = mockStream.mock.lastCall?.[3] + expect(streamOptions).toMatchObject({ + supportsTools: false, + tools: [], + waitForTools: false, + }) + expect(streamOptions?.toolChoice).toBeUndefined() expect(onDelta).toHaveBeenCalled() expect(onEnd).toHaveBeenCalled() }) @@ -290,12 +292,14 @@ describe('store character-orchestrator', () => { forceSparkCommandResponse: true, }) - const streamOptions = mockStream.mock.calls[0][3] - expect(streamOptions.supportsTools).toBe(true) - expect(streamOptions.waitForTools).toBe(true) - expect(streamOptions.toolChoice).toEqual({ - type: 'function', - function: { name: 'builtIn_sparkCommand' }, + const streamOptions = mockStream.mock.lastCall?.[3] + expect(streamOptions).toMatchObject({ + supportsTools: true, + toolChoice: { + type: 'function', + function: { name: 'builtIn_sparkCommand' }, + }, + waitForTools: true, }) expect(result?.commands?.length).toBe(1) expect(sendSparkCommandMock).toHaveBeenCalledWith({ @@ -336,7 +340,7 @@ describe('store character-orchestrator', () => { }, }) - const renderedMessages = mockStream.mock.calls[0]?.[2] as Array<{ role: string, content: string }> | undefined + const renderedMessages = mockStream.mock.lastCall?.[2] as Array<{ role: string, content: string }> | undefined expect(String(renderedMessages?.[0]?.content)).toContain('Plugin-specific hint') expect(String(renderedMessages?.[1]?.content)).toContain('Rendered board snapshot') }) diff --git a/packages/stage-ui/src/stores/chat.contract.test.ts b/packages/stage-ui/src/stores/chat.contract.test.ts index 772c20109..577036cb9 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 { ChatProvider } from '@xsai-ext/providers/utils' import type { Message, Tool } from '@xsai/shared-chat' +import { errorMessageFrom } from '@moeru/std' import { IOAttributes, IOSpanNames } from '@proj-airi/stage-shared' import { createPinia, setActivePinia } from 'pinia' import { beforeEach, describe, expect, it, vi } from 'vitest' @@ -71,6 +72,8 @@ const createMinecraftContextMock = vi.fn() const persistSessionMessagesMock = vi.fn() const forkSessionMock = vi.fn() const ensureSessionMock = vi.fn() +const loadSessionMock = vi.fn() +const deleteSessionMock = vi.fn() const getProviderInstanceMock = vi.fn() const getToolsByNamesMock = vi.fn<(names: string[]) => Tool[]>() @@ -194,6 +197,9 @@ vi.mock('./chat/session-store', () => ({ sessionMessages[sessionId] = [] }, getSessionMessages: (sessionId: string) => sessionMessages[sessionId] ?? [], + getSessionMessagesIfLoaded: (sessionId: string) => sessionMessages[sessionId], + loadSession: loadSessionMock, + deleteSession: deleteSessionMock, persistSessionMessages: persistSessionMessagesMock, getSessionGeneration: () => currentGeneration, setSessionMessages: (sessionId: string, messages: any[]) => { @@ -285,6 +291,8 @@ describe('chat store contract', () => { persistSessionMessagesMock.mockReset() forkSessionMock.mockReset() ensureSessionMock.mockReset() + loadSessionMock.mockReset().mockResolvedValue(true) + deleteSessionMock.mockReset().mockResolvedValue(undefined) getProviderInstanceMock.mockReset().mockResolvedValue(provider) getToolsByNamesMock.mockReset().mockImplementation(names => names.map(name => ({ type: 'function', @@ -338,6 +346,51 @@ describe('chat store contract', () => { ]) }) + // https://github.com/moeru-ai/airi/issues/2085 + it('hydrates the target session before sending for Issue #2085', async () => { + // ROOT CAUSE: + // + // A synchronized follower could target a session known only by metadata. + // Reading through getSessionMessages before hydration created a fresh + // system-only history that could overwrite the persisted conversation. + delete sessionMessages['session-2'] + loadSessionMock.mockImplementationOnce(async () => { + sessionMessages['session-2'] = [ + { role: 'system', content: 'persisted system prompt', createdAt: 1, id: 'system-2' }, + ] + return true + }) + llmStreamMock.mockImplementationOnce(async (_model: string, _chatProvider: ChatProvider, _messages: Message[], options: any) => { + await options.onStreamEvent({ type: 'finish', finishReason: 'stop' }) + }) + + const store = useChatStore() + await store.send({ sessionId: 'session-2', text: 'continue persisted chat' }) + + expect(loadSessionMock).toHaveBeenCalledWith('session-2') + expect(loadSessionMock.mock.invocationCallOrder[0]).toBeLessThan(ensureSessionMock.mock.invocationCallOrder[0]) + expect(sessionMessages['session-2']?.[0]).toMatchObject({ + content: 'persisted system prompt', + id: 'system-2', + }) + }) + + // https://github.com/moeru-ai/airi/issues/2085 + it('does not create fallback history when target hydration fails for Issue #2085', async () => { + delete sessionMessages['session-2'] + loadSessionMock.mockResolvedValueOnce(false) + + const store = useChatStore() + await expect( + store.send({ sessionId: 'session-2', text: 'do not overwrite history' }), + ) + .rejects + .toThrow('Failed to load the target chat session') + + expect(ensureSessionMock).not.toHaveBeenCalledWith('session-2') + expect(sessionMessages['session-2']).toBeUndefined() + }) + it('forwards one correlation identity across every PostHog chat milestone', async () => { llmStreamMock.mockImplementation(async (_model: string, _chatProvider: ChatProvider, _messages: Message[], options: any) => { await options.onStreamEvent({ type: 'text-delta', text: 'ok' }) @@ -650,30 +703,31 @@ describe('chat store contract', () => { expect(specialHook.mock.calls[0]?.[1].turnId.length).toBeGreaterThan(0) }) - /** - * @example - * store.sending = true - * await nextTick() - * expect(store.sending).toBe(true) - */ - it('keeps sending writable for context bridge and chat sync consumers', async () => { + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743261505 + it('preserves a synchronized sending snapshot without replaying it through the follower runtime for Issue #2085', async () => { + // ROOT CAUSE: + // + // Applying `sending: true` invoked the follower's idle runtime, whose + // derived state cleared the synchronized stream payload and could target + // that follower's unrelated local selection. const store = useChatStore() - - expect(store.sending).toBe(false) - - store.sending = true + store.$patch({ + sending: true, + activeSendSessionId: 'session-b', + activeStreamingMessage: { + role: 'assistant', + content: 'authority stream', + slices: [], + tool_results: [], + }, + }) await nextTick() + expect(store.sending).toBe(true) - - store.sending = false - await nextTick() - expect(store.sending).toBe(false) + expect(store.activeSendSessionId).toBe('session-b') + expect(store.activeStreamingMessage?.content).toBe('authority stream') }) - /** - * @example - * store.sending = false while a local runtime send is still streaming. - */ it('does not end the owned IO turn span when external sending mirror is cleared mid-send', async () => { let releaseStream: (() => void) | undefined llmStreamMock.mockImplementationOnce(async () => { @@ -711,11 +765,6 @@ describe('chat store contract', () => { expect(ioTracerMocks.activeTurnSpan.value).toBeUndefined() }) - /** - * @example - * createMinecraftContext() returns a runtime context update. - * The facade passes it into the core runtime before prompt snapshots are read. - */ it('ingests runtime context providers before composing prompt snapshots', async () => { const minecraftContext = { id: 'minecraft-context', @@ -788,11 +837,60 @@ describe('chat store contract', () => { await firstSend }) - /** - * @example - * store.getPendingQueuedSendSnapshot() - * // => [{ sessionId, generation, cancelled, messagePreview, hasAttachments, inputType }] - */ + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3742939573 + it('does not recreate a deleted session when queued work is cancelled for Issue #2085', async () => { + // ROOT CAUSE: + // + // Cancelling queued work rejects the public send action. Its generic error + // handler used to recreate a system-plus-error history after deletion, + // leaving a ghost conversation that no longer had session metadata. + let releaseFirstSend: (() => void) | undefined + llmStreamMock.mockImplementationOnce(async () => { + await new Promise((resolve) => { + releaseFirstSend = resolve + }) + }) + deleteSessionMock.mockImplementationOnce(async () => { + currentGeneration += 1 + delete sessionMessages['session-1'] + }) + + const store = useChatStore() + const firstSend = store.send({ + sessionId: 'session-1', + text: 'active turn', + }) + const activeOutcome = firstSend.then( + () => 'resolved', + error => errorMessageFrom(error) ?? 'unknown error', + ) + const queuedSend = store.send({ + sessionId: 'session-1', + text: 'must be cancelled', + }) + const queuedOutcome = queuedSend.then( + () => 'resolved', + error => errorMessageFrom(error) ?? 'unknown error', + ) + + await vi.waitFor(() => { + expect(llmStreamMock).toHaveBeenCalledTimes(1) + }) + await vi.waitFor(() => { + expect(store.pendingQueuedSendCount).toBe(1) + }) + await store.deleteSession('session-1') + + expect(deleteSessionMock).toHaveBeenCalledWith('session-1') + expect(await queuedOutcome).toBe('Chat session was reset before send could start') + expect(sessionMessages['session-1']).toBeUndefined() + + releaseFirstSend?.() + expect(await activeOutcome).toBe('Chat session was removed before send completed') + expect(llmStreamMock).toHaveBeenCalledTimes(1) + expect(sessionMessages['session-1']).toBeUndefined() + }) + it('mirrors pending queued send snapshots from the core runtime', async () => { let releaseFirstSend: (() => void) | undefined llmStreamMock.mockImplementationOnce(async () => { diff --git a/packages/stage-ui/src/stores/chat.ts b/packages/stage-ui/src/stores/chat.ts index 2dc8a9b52..be592540b 100644 --- a/packages/stage-ui/src/stores/chat.ts +++ b/packages/stage-ui/src/stores/chat.ts @@ -4,7 +4,7 @@ import type { ChatProvider } from '@xsai-ext/providers/utils' import type { Message } from '@xsai/shared-chat' import type {} from 'pinia-plugin-synced' -import type { ChatHistoryItem, ChatToolReference } from '../types/chat' +import type { ChatHistoryItem, ChatToolReference, StreamingAssistantMessage } from '../types/chat' import type { ToolCallRerunPayload } from './tool-call-rerun' import { errorMessageFrom } from '@moeru/std' @@ -12,7 +12,7 @@ import { createChatOrchestratorRuntime } from '@proj-airi/core-agent' import { IOAttributes, IOEvents, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' import { nanoid } from 'nanoid' import { defineStore, storeToRefs } from 'pinia' -import { ref, toRaw, watch } from 'vue' +import { shallowRef, toRaw } from 'vue' import { getConversationAnalyticsSurface } from '../composables' import { activeTurnSpan, startSpan } from '../composables/use-io-tracer' @@ -157,8 +157,10 @@ export const useChatStore = defineStore('chat', () => { const { activeSessionId } = storeToRefs(chatSession) const { streamingMessage } = storeToRefs(chatStream) - const sending = ref(false) - const pendingQueuedSendCount = ref(0) + const sending = shallowRef(false) + const activeSendSessionId = shallowRef() + const activeStreamingMessage = shallowRef() + const pendingQueuedSendCount = shallowRef(0) let ownedActiveTurnSpan: typeof activeTurnSpan.value const analyticsHooks = createChatAnalyticsHooks({ getSessionMessages: sessionId => chatSession.getSessionMessages(sessionId), @@ -229,6 +231,8 @@ export const useChatStore = defineStore('chat', () => { function syncRuntimeState(state: ChatOrchestratorRuntimeState) { sending.value = state.sending + activeSendSessionId.value = state.activeSendSessionId + activeStreamingMessage.value = state.activeStreamingMessage pendingQueuedSendCount.value = state.pendingQueuedSendCount } @@ -317,11 +321,6 @@ export const useChatStore = defineStore('chat', () => { }, }) - watch(sending, (next) => { - if (runtime.getSending() !== next) - runtime.setSending(next) - }) - async function ingest( sendingMessage: string, options: ChatOrchestratorSendOptions, @@ -345,6 +344,9 @@ export const useChatStore = defineStore('chat', () => { } function appendSendError(sessionId: string, error: unknown) { + if (!chatSession.getSessionMessagesIfLoaded(sessionId)) + return + chatSession.appendSessionMessage(sessionId, { role: 'error', content: errorMessageFrom(error) ?? 'Unknown chat operation failure', @@ -357,6 +359,9 @@ export const useChatStore = defineStore('chat', () => { if (!providerId || !modelId) throw new Error('No active chat provider or model configured') + if (!await chatSession.loadSession(payload.sessionId)) + throw new Error('Failed to load the target chat session') + const messageCount = chatSession.getSessionMessages(payload.sessionId).length const chatProvider = await providerStore.getProviderInstance(providerId) if (!chatProvider) @@ -376,8 +381,12 @@ export const useChatStore = defineStore('chat', () => { }, }, payload.sessionId) + const completedMessages = chatSession.getSessionMessagesIfLoaded(payload.sessionId) + if (!completedMessages) + throw new Error('Chat session was removed before send completed') + return { - messages: chatSession.getSessionMessages(payload.sessionId) + messages: completedMessages .slice(messageCount) .map(message => structuredClone(toRaw(message))), sessionId: payload.sessionId, @@ -397,6 +406,9 @@ export const useChatStore = defineStore('chat', () => { /** Replaces one stored turn with a new execution of its user message. */ async function retry(payload: ChatRetryPayload): Promise { + if (!await chatSession.loadSession(payload.sessionId)) + throw new Error('Failed to load the target chat session') + const currentMessages = chatSession.getSessionMessages(payload.sessionId) const sourceIndex = retrySourceIndexFrom(currentMessages, payload.index) if (sourceIndex < 0) @@ -424,6 +436,9 @@ export const useChatStore = defineStore('chat', () => { /** Runs one stored tool call again and replaces its stored result. */ async function rerunToolCall(payload: ChatToolCallRerunPayload): Promise { + if (!await chatSession.loadSession(payload.sessionId)) + throw new Error('Failed to load the target chat session') + const nextMessages = await executeToolCallRerun({ messages: chatSession.getSessionMessages(payload.sessionId), payload, @@ -442,6 +457,12 @@ export const useChatStore = defineStore('chat', () => { chatStream.resetStream() } + /** Cancels queued work before permanently removing its owning session. */ + function deleteSession(sessionId: string): Promise { + runtime.cancelPendingSends(sessionId) + return chatSession.deleteSession(sessionId) + } + async function ingestOnFork( sendingMessage: string, options: ChatOrchestratorSendOptions, @@ -470,9 +491,12 @@ export const useChatStore = defineStore('chat', () => { return { sending, + activeSendSessionId, + activeStreamingMessage, pendingQueuedSendCount, cleanup, + deleteSession, ingest, ingestOnFork, rerunToolCall, @@ -507,7 +531,7 @@ export const useChatStore = defineStore('chat', () => { } }, { synced: { - actions: ['cleanup', 'rerunToolCall', 'retry', 'send'], + actions: ['cleanup', 'deleteSession', 'rerunToolCall', 'retry', 'send'], state: true, }, }) 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 c1600d774..c2e0a735d 100644 --- a/packages/stage-ui/src/stores/chat/session-store.test.ts +++ b/packages/stage-ui/src/stores/chat/session-store.test.ts @@ -1,7 +1,7 @@ import type { ChatSessionMeta, ChatSessionRecord, ChatSessionsIndex } from '../../types/chat-session' -import { createPinia, setActivePinia } from 'pinia' -import { beforeEach, describe, expect, it, vi } from 'vitest' +import { createPinia, disposePinia, setActivePinia } from 'pinia' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { nextTick, ref } from 'vue' // Refs the store reads through the mocked `useAuthStore` / `useAiriCardStore`. @@ -19,6 +19,14 @@ const getOutboxMock = vi.fn<(uid: string) => Promise>() const dropOutboxForSessionMock = vi.fn<(uid: string, id: string) => Promise>() const getTombstonesMock = vi.fn<(uid: string) => Promise>() const removeTombstonesMock = vi.fn<(uid: string, ids: string[]) => Promise>() +const addTombstoneMock = vi.fn<(uid: string, id: string) => Promise>() +const deleteCloudChatMock = vi.fn<(id: string) => Promise>() +const listChatsMock = vi.fn() +const pullMessagesMock = vi.fn() +const reconcileLocalAndRemoteMock = vi.fn() +const connectCloudWsMock = vi.fn() +let cloudWsStatus: 'idle' | 'open' = 'idle' +let cloudStatusListener: ((status: 'idle' | 'open') => void) | undefined vi.mock('pinia', async () => { const actual = await vi.importActual('pinia') @@ -52,7 +60,7 @@ vi.mock('../../database/repos/chat-sessions.repo', () => ({ updateOutboxEntries: vi.fn().mockResolvedValue(undefined), dropOutboxForSession: (uid: string, id: string) => dropOutboxForSessionMock(uid, id), getTombstones: (uid: string) => getTombstonesMock(uid), - addTombstone: vi.fn().mockResolvedValue(undefined), + addTombstone: (uid: string, id: string) => addTombstoneMock(uid, id), removeTombstones: (uid: string, ids: string[]) => removeTombstonesMock(uid, ids), }, })) @@ -74,20 +82,23 @@ vi.mock('../../libs/server', () => ({ // sufficient. We keep `extractMessageText` realistic so message previews work. vi.mock('../../libs/chat-sync', () => ({ applyCreateActions: vi.fn().mockResolvedValue([]), - reconcileLocalAndRemote: vi.fn().mockReturnValue({ adopt: [], claim: [], create: [] }), + reconcileLocalAndRemote: (...args: unknown[]) => reconcileLocalAndRemoteMock(...args), createCloudChatMapper: () => ({ - listChats: vi.fn().mockResolvedValue([]), - deleteChat: vi.fn().mockResolvedValue(undefined), + listChats: () => listChatsMock(), + deleteChat: (id: string) => deleteCloudChatMock(id), }), createChatWsClient: () => ({ - status: () => 'idle' as const, - connect: vi.fn(), + status: () => cloudWsStatus, + connect: connectCloudWsMock, disconnect: vi.fn(), destroy: vi.fn(), sendMessages: vi.fn().mockResolvedValue({ ok: true }), - pullMessages: vi.fn().mockResolvedValue({ messages: [], maxSeq: 0 }), + pullMessages: (...args: unknown[]) => pullMessagesMock(...args), onNewMessages: () => () => {}, - onStatusChange: () => () => {}, + onStatusChange: (listener: (status: 'idle' | 'open') => void) => { + cloudStatusListener = listener + return () => {} + }, }), extractMessageText: (m: any) => (typeof m?.content === 'string' ? m.content : ''), isCloudSyncableMessage: () => false, @@ -95,9 +106,11 @@ vi.mock('../../libs/chat-sync', () => ({ })) const { useChatSessionStore } = await import('./session-store') +let pinia: ReturnType beforeEach(() => { - setActivePinia(createPinia()) + pinia = createPinia() + setActivePinia(pinia) userIdRef.value = 'local' activeCardIdRef.value = 'default' systemPromptRef.value = '' @@ -111,6 +124,18 @@ beforeEach(() => { dropOutboxForSessionMock.mockReset().mockResolvedValue(undefined) getTombstonesMock.mockReset().mockResolvedValue([]) removeTombstonesMock.mockReset().mockResolvedValue(undefined) + addTombstoneMock.mockReset().mockResolvedValue(undefined) + deleteCloudChatMock.mockReset().mockResolvedValue(undefined) + listChatsMock.mockReset().mockResolvedValue([]) + pullMessagesMock.mockReset().mockResolvedValue({ messages: [], seq: 0 }) + reconcileLocalAndRemoteMock.mockReset().mockReturnValue({ adopt: [], claim: [], create: [] }) + connectCloudWsMock.mockReset() + cloudWsStatus = 'idle' + cloudStatusListener = undefined +}) + +afterEach(() => { + disposePinia(pinia) }) async function flushMicrotasks(rounds = 8) { @@ -295,6 +320,291 @@ describe('chat-session-store · loadSession vs concurrent deleteSession', () => }) }) +describe('chat-session-store · deletion and hydration failures', () => { + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743502031 + it('persists the fallback without changing the leader selection when a follower deletes its active session for Issue #2085', async () => { + // ROOT CAUSE: + // + // Session selection is window-local, so the synchronized leader can be on + // A while the persisted index still points to B. Deleting B returned A but + // left the persisted index empty, so the next startup created a blank chat. + const sessionA: ChatSessionMeta = { + sessionId: 'session-a', + userId: 'local', + characterId: 'default', + createdAt: 1, + updatedAt: 1, + } + const sessionB: ChatSessionMeta = { + sessionId: 'session-b', + userId: 'local', + characterId: 'default', + createdAt: 2, + updatedAt: 2, + } + const store = useChatSessionStore() + store.applyRemoteSnapshot({ + activeSessionId: 'session-a', + sessionMessages: { 'session-a': [], 'session-b': [] }, + sessionMetas: { 'session-a': sessionA, 'session-b': sessionB }, + index: { + userId: 'local', + characters: { + default: { + activeSessionId: 'session-b', + sessions: { 'session-a': sessionA, 'session-b': sessionB }, + }, + }, + }, + }) + + await store.deleteSession('session-b') + + expect(store.activeSessionId).toBe('session-a') + expect(store.getSnapshot().index?.characters.default?.activeSessionId).toBe('session-a') + expect(store.sessionMetas['session-b']).toBeUndefined() + expect(saveIndexMock).toHaveBeenLastCalledWith(expect.objectContaining({ + characters: expect.objectContaining({ + default: expect.objectContaining({ activeSessionId: 'session-a' }), + }), + })) + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743309237 + it('persists a replacement fallback without changing an unrelated leader selection for Issue #2085', async () => { + // ROOT CAUSE: + // + // When a follower deleted the final session for one character, the leader + // created a replacement with local activation disabled. The replacement + // was indexed but the persisted character active ID stayed empty, so the + // next initialization created another blank conversation. + const leaderSession: ChatSessionMeta = { + sessionId: 'leader-session', + userId: 'local', + characterId: 'other-character', + createdAt: 1, + updatedAt: 1, + } + const deletedSession: ChatSessionMeta = { + sessionId: 'deleted-session', + userId: 'local', + characterId: 'default', + createdAt: 2, + updatedAt: 2, + } + const store = useChatSessionStore() + store.applyRemoteSnapshot({ + activeSessionId: 'leader-session', + sessionMessages: { 'leader-session': [], 'deleted-session': [] }, + sessionMetas: { 'leader-session': leaderSession, 'deleted-session': deletedSession }, + index: { + userId: 'local', + characters: { + 'other-character': { + activeSessionId: 'leader-session', + sessions: { 'leader-session': leaderSession }, + }, + 'default': { + activeSessionId: 'deleted-session', + sessions: { 'deleted-session': deletedSession }, + }, + }, + }, + }) + + await store.deleteSession('deleted-session') + const snapshot = store.getSnapshot() + const [replacementSessionId] = Object.keys(snapshot.index?.characters.default?.sessions ?? {}) + expect(replacementSessionId).toBeDefined() + if (!replacementSessionId) + throw new Error('Expected deletion to create a replacement session') + + expect(store.activeSessionId).toBe('leader-session') + expect(snapshot.index?.characters.default?.activeSessionId).toBe(replacementSessionId) + expect(snapshot.index?.characters.default?.sessions[replacementSessionId]).toBeDefined() + expect(saveIndexMock).toHaveBeenLastCalledWith(expect.objectContaining({ + characters: expect.objectContaining({ + default: expect.objectContaining({ activeSessionId: replacementSessionId }), + }), + })) + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3628917803 + it('keeps deleted session generations invalid for Issue #2085', async () => { + // ROOT CAUSE: + // + // Deletion previously removed the generation entry. A send captured at + // generation zero could then read the deleted session as generation zero + // again and continue appending messages after the chat was gone. + const meta: ChatSessionMeta = { + sessionId: 'sess-1', + userId: 'local', + characterId: 'default', + createdAt: 1, + updatedAt: 1, + } + const store = useChatSessionStore() + store.applyRemoteSnapshot({ + activeSessionId: 'sess-1', + sessionMessages: { 'sess-1': [] }, + sessionMetas: { 'sess-1': meta }, + index: null, + }) + + expect(store.getSessionGeneration('sess-1')).toBe(0) + + await store.deleteSession('sess-1') + + expect(store.getSessionGeneration('sess-1')).toBe(1) + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3628003766 + it('reports hydration failure and permits a later retry for Issue #2085', async () => { + const meta: ChatSessionMeta = { + sessionId: 'sess-1', + userId: 'local', + characterId: 'default', + createdAt: 1, + updatedAt: 1, + } + vi.spyOn(console, 'warn').mockImplementation(() => {}) + getSessionMock + .mockRejectedValueOnce(new Error('IndexedDB read failed')) + .mockResolvedValueOnce(null) + + userIdRef.value = 'local' + const store = useChatSessionStore() + store.applyRemoteSnapshot({ + activeSessionId: '', + sessionMessages: {}, + sessionMetas: { 'sess-1': meta }, + index: null, + }) + + await expect(store.loadSession('sess-1')).resolves.toBe(false) + await expect(store.loadSession('sess-1')).resolves.toBe(true) + + expect(getSessionMock).toHaveBeenCalledTimes(2) + }) +}) + +describe('chat-session-store · cloud placeholder hydration', () => { + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743502032 + it('retries cloud hydration when the reconcile pull for an adopted placeholder fails for Issue #2085', async () => { + // ROOT CAUSE: + // + // Reconcile creates a system-only placeholder before its first cloud pull. + // If that pull fails, loadSession sees the message-map entry and marks the + // placeholder as loaded. Selecting the chat then skips every later pull. + const localMeta: ChatSessionMeta = { + sessionId: 'local-session', + userId: 'cloud-user', + characterId: 'default', + createdAt: 1, + updatedAt: 1, + } + const remoteChat = { + id: 'remote-session', + type: 'bot' as const, + title: null, + createdAt: '2026-01-01T00:00:00.000Z', + updatedAt: '2026-01-01T00:00:00.000Z', + } + userIdRef.value = 'cloud-user' + getIndexMock.mockResolvedValue({ + userId: 'cloud-user', + characters: { + default: { + activeSessionId: localMeta.sessionId, + sessions: { [localMeta.sessionId]: localMeta }, + }, + }, + }) + getSessionMock.mockImplementation((sessionId) => { + if (sessionId === remoteChat.id) { + return Promise.resolve({ + meta: { + sessionId: remoteChat.id, + userId: 'cloud-user', + characterId: 'default', + createdAt: Date.parse(remoteChat.createdAt), + updatedAt: Date.parse(remoteChat.updatedAt), + cloudChatId: remoteChat.id, + }, + messages: [], + }) + } + return Promise.resolve({ meta: localMeta, messages: [] }) + }) + listChatsMock.mockResolvedValue([remoteChat]) + reconcileLocalAndRemoteMock.mockReturnValue({ adopt: [remoteChat], claim: [], create: [] }) + pullMessagesMock + .mockRejectedValueOnce(new Error('temporary cloud failure')) + .mockResolvedValueOnce({ messages: [], seq: 0 }) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + + const store = useChatSessionStore() + await store.initialize() + expect(cloudStatusListener).toBeDefined() + + cloudWsStatus = 'open' + cloudStatusListener?.('open') + await vi.waitFor(() => { + expect(store.cloudSyncReady).toBe(true) + }) + expect(pullMessagesMock).toHaveBeenCalledTimes(1) + + await store.setActiveSession(remoteChat.id) + + expect(pullMessagesMock).toHaveBeenCalledTimes(2) + }) +}) + +describe('chat-session-store · cloud deletion', () => { + it('tombstones an unmapped cloud session before an in-flight create can finish', async () => { + // ROOT CAUSE: + // + // A newly created cloud session can be deleted before POST /chats binds + // its cloud id. Without a tombstone for the deterministic session id, the + // completed remote create is adopted again by the next reconcile. + userIdRef.value = 'cloud-user' + const deleted: ChatSessionMeta = { + sessionId: 'pending-cloud-session', + userId: 'cloud-user', + characterId: 'default', + createdAt: 1, + updatedAt: 1, + } + const survivor: ChatSessionMeta = { + sessionId: 'surviving-session', + userId: 'cloud-user', + characterId: 'default', + createdAt: 2, + updatedAt: 2, + } + const store = useChatSessionStore() + store.applyRemoteSnapshot({ + activeSessionId: 'surviving-session', + sessionMessages: { 'pending-cloud-session': [], 'surviving-session': [] }, + sessionMetas: { 'pending-cloud-session': deleted, 'surviving-session': survivor }, + index: { + userId: 'cloud-user', + characters: { + default: { + activeSessionId: 'pending-cloud-session', + sessions: { 'pending-cloud-session': deleted, 'surviving-session': survivor }, + }, + }, + }, + }) + + await store.deleteSession('pending-cloud-session') + + expect(addTombstoneMock).toHaveBeenCalledWith('cloud-user', 'pending-cloud-session') + expect(deleteCloudChatMock).not.toHaveBeenCalled() + }) +}) + describe('chat-session-store · active card prompt edits', () => { // ROOT CAUSE: // @@ -394,6 +704,257 @@ describe('chat-session-store · active card prompt edits', () => { }) describe('chat-session-store · synchronized data actions', () => { + // 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) + 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: + // + // `ready` was synchronized while selection was not. A joining window saw + // the leader's ready flag, skipped initialization, and remained on an + // empty local selection. + const session: ChatSessionMeta = { + sessionId: 'session-b', + userId: 'local', + characterId: 'default', + createdAt: 1, + updatedAt: 1, + } + const store = useChatSessionStore() + store.$patch({ + sessionMessages: { 'session-b': [{ id: 'system', role: 'system', content: 'prompt' }] }, + sessionMetas: { 'session-b': session }, + index: { + userId: 'local', + characters: { + default: { + activeSessionId: 'session-b', + sessions: { 'session-b': session }, + }, + }, + }, + }) + + expect(store.$state).not.toHaveProperty('ready') + expect(store.activeSessionId).toBe('') + + await store.initialize() + + expect(store.activeSessionId).toBe('session-b') + expect(store.isReady).toBe(true) + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743242529 + it('trusts synchronized messages instead of merging a stale follower IDB record for Issue #2085', async () => { + // ROOT CAUSE: + // + // Follower hydration read its own older IndexedDB record and mutated the + // fully synchronized store, allowing that stale snapshot to overwrite the + // leader's newer messages or resurrect a deleted session. + const session: ChatSessionMeta = { + sessionId: 'session-b', + userId: 'local', + characterId: 'default', + createdAt: 1, + updatedAt: 2, + } + getSessionMock.mockResolvedValue({ + meta: { ...session, updatedAt: 1 }, + messages: [{ id: 'stale', role: 'user', content: 'stale follower data' }], + }) + const store = useChatSessionStore() + store.$patch({ + sessionMessages: { + 'session-b': [{ id: 'current', role: 'assistant', content: 'leader data', slices: [], tool_results: [] }], + }, + sessionMetas: { 'session-b': session }, + }) + + await expect(store.loadSession('session-b')).resolves.toBe(true) + + expect(getSessionMock).not.toHaveBeenCalled() + expect(store.getSessionMessagesIfLoaded('session-b')?.map(message => message.id)).toEqual(['current']) + }) + + it('refreshes an already loaded session from IndexedDB for a completed remote stream', async () => { + const session: ChatSessionMeta = { + sessionId: 'session-b', + userId: 'local', + characterId: 'default', + createdAt: 1, + updatedAt: 2, + } + const store = useChatSessionStore() + store.applyRemoteSnapshot({ + activeSessionId: 'session-b', + sessionMessages: { 'session-b': [{ id: 'system', role: 'system', content: 'prompt' }] }, + sessionMetas: { 'session-b': session }, + index: { + userId: 'local', + characters: { + default: { + activeSessionId: 'session-b', + sessions: { 'session-b': session }, + }, + }, + }, + }) + getSessionMock.mockResolvedValue({ + meta: session, + messages: [ + { id: 'system', role: 'system', content: 'prompt' }, + { id: 'assistant', role: 'assistant', content: 'complete answer', slices: [], tool_results: [] }, + ], + }) + + await expect(store.refreshSession('session-b')).resolves.toBe(true) + + expect(getSessionMock).toHaveBeenCalledWith('session-b') + expect(store.getSessionMessages('session-b').map(message => message.id)).toEqual(['system', 'assistant']) + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743121862 + it('moves a follower away from a session removed by another window for Issue #2085', async () => { + // ROOT CAUSE: + // + // Synchronized deletion removed B's metadata, but activeSessionId is + // intentionally window-local. A follower that also selected B therefore + // kept an invalid selection until it manually chose another session. + const sessionA: ChatSessionMeta = { + sessionId: 'session-a', + userId: 'local', + characterId: 'default', + createdAt: 1, + updatedAt: 1, + } + const sessionB: ChatSessionMeta = { + sessionId: 'session-b', + userId: 'local', + characterId: 'default', + createdAt: 2, + updatedAt: 2, + } + const store = useChatSessionStore() + store.applyRemoteSnapshot({ + activeSessionId: 'session-b', + sessionMessages: { 'session-a': [], 'session-b': [] }, + sessionMetas: { 'session-a': sessionA, 'session-b': sessionB }, + index: { + userId: 'local', + characters: { + default: { + activeSessionId: 'session-a', + sessions: { 'session-a': sessionA, 'session-b': sessionB }, + }, + }, + }, + }) + await nextTick() + + store.applyRemoteSnapshot({ + activeSessionId: 'session-b', + sessionMessages: { 'session-a': [] }, + sessionMetas: { 'session-a': sessionA }, + index: { + userId: 'local', + characters: { + default: { + activeSessionId: 'session-a', + sessions: { 'session-a': sessionA }, + }, + }, + }, + }) + await nextTick() + + expect(store.activeSessionId).toBe('session-a') + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743221033 + it('waits for the leader replacement when every window loses its last session for Issue #2085', async () => { + // ROOT CAUSE: + // + // Every follower independently created a replacement when synchronized + // deletion temporarily left no metadata. Multiple windows could therefore + // turn one deletion into several empty chats before state converged. + const removedSession: ChatSessionMeta = { + sessionId: 'session-b', + userId: 'local', + characterId: 'default', + createdAt: 1, + updatedAt: 1, + } + const replacementSession: ChatSessionMeta = { + sessionId: 'session-c', + userId: 'local', + characterId: 'default', + createdAt: 2, + updatedAt: 2, + } + const store = useChatSessionStore() + store.applyRemoteSnapshot({ + activeSessionId: 'session-b', + sessionMessages: { 'session-b': [] }, + sessionMetas: { 'session-b': removedSession }, + index: { + userId: 'local', + characters: { + default: { + activeSessionId: 'session-b', + sessions: { 'session-b': removedSession }, + }, + }, + }, + }) + await nextTick() + saveSessionMock.mockClear() + + store.applyRemoteSnapshot({ + activeSessionId: 'session-b', + sessionMessages: {}, + sessionMetas: {}, + index: { userId: 'local', characters: {} }, + }) + await nextTick() + + expect(saveSessionMock).not.toHaveBeenCalled() + expect(store.activeSessionId).toBe('session-b') + + store.applyRemoteSnapshot({ + activeSessionId: 'session-b', + sessionMessages: { 'session-c': [] }, + sessionMetas: { 'session-c': replacementSession }, + index: { + userId: 'local', + characters: { + default: { + activeSessionId: 'session-c', + sessions: { 'session-c': replacementSession }, + }, + }, + }, + }) + await nextTick() + + expect(store.activeSessionId).toBe('session-c') + }) + it('deletes a message by its stable id from the specified session', async () => { const store = useChatSessionStore() store.applyRemoteSnapshot({ @@ -415,11 +976,29 @@ describe('chat-session-store · synchronized data actions', () => { expect(store.getSessionMessages('session-1').map(message => message.id)).toEqual(['keep']) }) - it('keeps the active session outside synchronized session state', () => { + it('keeps window-local selection out of synchronized and persisted session state', async () => { const store = useChatSessionStore() - store.activeSessionId = 'window-local-session' + store.applyRemoteSnapshot({ + activeSessionId: 'persisted-session', + index: { + userId: 'local', + characters: { + default: { + activeSessionId: 'persisted-session', + sessions: {}, + }, + }, + }, + sessionMessages: { + 'window-local-session': [{ id: 'system', role: 'system', content: 'prompt' }], + }, + sessionMetas: {}, + }) + + await store.setActiveSession('window-local-session') expect(store.activeSessionId).toBe('window-local-session') expect(store.$state).not.toHaveProperty('activeSessionId') + expect(store.getSnapshot().index?.characters.default?.activeSessionId).toBe('persisted-session') }) }) diff --git a/packages/stage-ui/src/stores/chat/session-store.ts b/packages/stage-ui/src/stores/chat/session-store.ts index 794fb8f3c..3eef6682c 100644 --- a/packages/stage-ui/src/stores/chat/session-store.ts +++ b/packages/stage-ui/src/stores/chat/session-store.ts @@ -76,6 +76,7 @@ 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. */ const index = ref(null) const ready = ref(false) @@ -92,7 +93,9 @@ export const useChatSessionStore = defineStore('chat-session', () => { let persistQueue = Promise.resolve() const loadedSessions = new Set() - const loadingSessions = new Map>() + const staleSessions = new Set() + const cloudHydratedSessions = new Set() + const loadingSessions = new Map>() // Cloud sync state. The WS client is constructed lazily so anonymous // (`userId === 'local'`) users never open a socket. `cloudSyncReady` is a @@ -110,6 +113,7 @@ 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. @@ -327,7 +331,10 @@ export const useChatSessionStore = defineStore('chat-session', () => { } /** Removes one message by stable id or by its current history index. */ - function deleteMessage(payload: DeleteChatMessagePayload) { + async function deleteMessage(payload: DeleteChatMessagePayload): Promise { + if (!await loadSession(payload.sessionId)) + throw new Error('Failed to load the target chat session') + const nextMessages = getSessionMessages(payload.sessionId).filter((message, messageIndex) => { if (payload.messageId) return message.id !== payload.messageId @@ -351,65 +358,83 @@ export const useChatSessionStore = defineStore('chat-session', () => { * - `sessionId` exists either in `sessionMetas` or in IDB. * * Returns: - * - Resolves once the session is in memory. On IDB error, removes the id + * - `true` when the session is in memory, or `false` when hydration failed. + * On IDB error, removes the id * from the loading map so subsequent calls can retry rather than wedge * on a stale promise. Errors are intentionally not rethrown — the * failing session is simply absent from local state and the next * loadSession call will retry. */ - async function loadSession(sessionId: string) { - if (loadedSessions.has(sessionId)) { - return + async function loadSession(sessionId: string): Promise { + const needsCloudHydration = () => { + const meta = sessionMetas.value[sessionId] + return !!meta?.cloudChatId && !cloudHydratedSessions.has(sessionId) + } + + if (loadedSessions.has(sessionId) && !staleSessions.has(sessionId) && !needsCloudHydration()) { + return true + } + // A synchronized snapshot already carries the authority's 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()) { + loadedSessions.add(sessionId) + return true } if (loadingSessions.has(sessionId)) { - await loadingSessions.get(sessionId) - return + return await loadingSessions.get(sessionId)! } const loadPromise = (async () => { try { - const stored = await chatSessionsRepo.getSession(sessionId) - // Re-check existence: `deleteSession` (or `clearInMemoryState` on a - // user swap) may have run while we were awaiting IDB. Without this - // guard, the post-await write resurrects the deleted entry and - // `loadedSessions.add` then short-circuits every future legitimate - // load — locking the resurrection in. The drawer's batch - // loadSession + per-row trash button hits this race in production. - if (!sessionMetas.value[sessionId]) - return - if (stored) { - const currentMessages = sessionMessages.value[sessionId] ?? [] - const mergedMessages = mergeLoadedSessionMessages(stored.messages, currentMessages) + if (!loadedSessions.has(sessionId) || staleSessions.has(sessionId)) { + const stored = await chatSessionsRepo.getSession(sessionId) + // Re-check existence after the IDB read. Deletion or an account + // change can remove this session while the read is pending. + if (!sessionMetas.value[sessionId]) + return false + if (staleSessions.has(sessionId) && !stored) + return false + if (stored) { + const currentMessages = sessionMessages.value[sessionId] ?? [] + const mergedMessages = mergeLoadedSessionMessages(stored.messages, currentMessages) - sessionMetas.value[sessionId] = stored.meta - replaceSessionMessages(sessionId, mergedMessages, { persist: false }) - ensureGeneration(sessionId) + sessionMetas.value[sessionId] = stored.meta + replaceSessionMessages(sessionId, mergedMessages, { persist: false }) + ensureGeneration(sessionId) - if (mergedMessages !== stored.messages) - await persistSession(sessionId) + if (mergedMessages !== stored.messages) + await persistSession(sessionId) + } + staleSessions.delete(sessionId) + loadedSessions.add(sessionId) + if (activeSessionId.value === sessionId) + refreshActiveSessionSystemMessage() } - loadedSessions.add(sessionId) - if (activeSessionId.value === sessionId) - refreshActiveSessionSystemMessage() - // Cloud gap fill: when the session is mapped to a cloud chat, ask - // the server for everything past our highest known seq. Best - // effort — failures are logged inside pullCloudMessages and the - // local view stays usable. - const meta = sessionMetas.value[sessionId] - if (meta?.cloudChatId) + // Local and cloud hydration are separate. A failed cloud pull leaves + // the local view usable and keeps the next selection eligible to retry. + if (needsCloudHydration()) await pullCloudMessages(sessionId) + + // Missing IDB payloads still need a valid canonical conversation. + // This action runs in the elected leader, so the initialized history + // is published once rather than independently by every follower. + ensureSession(sessionId) + + return true } catch (err) { // Do NOT add to loadedSessions on failure — the next call should // retry rather than fast-return on stale "already loaded" state. console.warn('[chat-session] loadSession failed for', sessionId, errorMessageFrom(err)) + return false } })() loadingSessions.set(sessionId, loadPromise) try { - await loadPromise + return await loadPromise } finally { // Always drain the loading map so a transient failure does not leave @@ -418,6 +443,13 @@ export const useChatSessionStore = defineStore('chat-session', () => { } } + /** Forces the next session load to merge the latest IndexedDB record into memory. */ + async function refreshSession(sessionId: string): Promise { + staleSessions.add(sessionId) + loadedSessions.delete(sessionId) + return loadSession(sessionId) + } + /** * Mint a new session for `characterId`, optionally seeding it with messages * and / or a title. Persists the new session and its index entry, then @@ -493,26 +525,17 @@ export const useChatSessionStore = defineStore('chat-session', () => { } /** - * Permanently remove a session from the local index + IDB and, when the - * session is cloud-mapped and the user is signed in, soft-delete the - * server chat via `DELETE /api/v1/chats/:id`. + * Permanently removes a session locally and tombstones its cloud identity. * - * Use when: - * - The user explicitly chooses "delete" from the sessions drawer. - * - * Expects: - * - The caller does not need to pre-confirm: this method is destructive. - * When the deleted session is the active one, the store falls back to - * another session for the same character or creates a fresh one. - * - * Returns: - * - Resolves once both local state and (if applicable) the remote DELETE - * call have settled. Cloud failures are swallowed with a console.warn — - * the local removal goes through either way so the user does not see - * a "ghost" session after the click. A tombstone is written so the - * reconcile `adopt` branch will not re-import the row on next login. + * Cloud failures are logged but do not restore the local session. A + * tombstone prevents the next reconcile from importing it again. */ async function deleteSession(sessionId: string) { + // Keep a monotonic tombstone in memory so queued and streaming sends that + // captured the previous generation cannot become current again after the + // session record is removed. + bumpSessionGeneration(sessionId) + const meta = sessionMetas.value[sessionId] if (!meta) return @@ -550,8 +573,9 @@ export const useChatSessionStore = defineStore('chat-session', () => { // fire-and-forget. Persistence races now read the post-deletion state. delete sessionMetas.value[sessionId] delete sessionMessages.value[sessionId] - delete sessionGenerations.value[sessionId] loadedSessions.delete(sessionId) + staleSessions.delete(sessionId) + cloudHydratedSessions.delete(sessionId) loadingSessions.delete(sessionId) if (index.value) { @@ -572,42 +596,55 @@ export const useChatSessionStore = defineStore('chat-session', () => { await persistIndex() await refreshOutboxPendingCount() - if (cloudChatId && isCloudUser) { + if (isCloudUser) { + const remoteChatId = cloudChatId ?? sessionId // Tombstone first: even if the cloud DELETE never reaches the server - // (offline, transient 5xx), the next reconcile will see the cloudChatId + // (offline, transient 5xx), the next reconcile will see the remote id // here and skip the adopt branch — preventing the ghost-session bug // where the server still has the row and re-creates the local mapping. // The reconcile-driven `drainTombstones` retries failed DELETEs. - await enqueuePersist(() => chatSessionsRepo.addTombstone(currentUserId, cloudChatId)) - getCloudMapper().deleteChat(cloudChatId).then( - async () => { - // Server confirmed the delete; reconcile will not see this id again, - // so we can drop the tombstone. - await enqueuePersist(() => chatSessionsRepo.removeTombstones(currentUserId, [cloudChatId])) - }, - (err) => { - console.warn('[chat-sync] DELETE /api/v1/chats failed for', sessionId, errorMessageFrom(err)) - }, - ) + await enqueuePersist(() => chatSessionsRepo.addTombstone(currentUserId, remoteChatId)) + if (cloudChatId) { + getCloudMapper().deleteChat(cloudChatId).then( + async () => { + // Server confirmed the delete; reconcile will not see this id again, + // so we can drop the tombstone. + await enqueuePersist(() => chatSessionsRepo.removeTombstones(currentUserId, [cloudChatId])) + }, + (err) => { + console.warn('[chat-sync] DELETE /api/v1/chats failed for', sessionId, errorMessageFrom(err)) + }, + ) + } } - // If the deleted session was active, pick another for the same - // character or mint a fresh one so the chat surface never lands on an - // empty void. - if (wasActive) { - const characterIndex = index.value?.characters[characterId] - const fallbackId = characterIndex - ? Object.keys(characterIndex.sessions).find(id => sessionMetas.value[id]) - : undefined - if (fallbackId) { + const characterIndex = index.value?.characters[characterId] + const fallbackId = characterIndex + ? Object.keys(characterIndex.sessions).find(id => sessionMetas.value[id]) + : undefined + + // Persisted character fallback is shared, but live selection is local to + // the window that was displaying the deleted session. + if (fallbackId && characterIndex) { + characterIndex.activeSessionId = fallbackId + if (wasActive) { activeSessionId.value = fallbackId - if (characterIndex) - characterIndex.activeSessionId = fallbackId await loadSession(fallbackId) - await persistIndex() } - else { - await createSession(characterId, { setActive: true }) + await persistIndex() + return + } + + const replacementSessionId = await createSession(characterId, { setActive: wasActive }) + if (!wasActive) { + // The synchronized leader may be displaying a different character, but + // this replacement is still the canonical fallback for the character + // whose final session was deleted. Persist that index choice without + // navigating the leader's window-local selection. + const replacementCharacterIndex = index.value?.characters[characterId] + if (replacementCharacterIndex) { + replacementCharacterIndex.activeSessionId = replacementSessionId + await persistIndex() } } } @@ -644,10 +681,9 @@ export const useChatSessionStore = defineStore('chat-session', () => { } activeSessionId.value = characterIndex.activeSessionId - await loadSession(characterIndex.activeSessionId) - if (isStaleEpoch()) - return - ensureSession(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) })() try { await ensureActivePromise @@ -719,6 +755,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { messages: result.messages, toSeq: result.seq, }) + cloudHydratedSessions.add(sessionId) } catch (err) { console.warn('[chat-sync] pullMessages failed for', sessionId, errorMessageFrom(err)) @@ -742,6 +779,8 @@ 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 @@ -750,7 +789,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { const myEpoch = reconcileEpoch const isStaleEpoch = () => myEpoch !== reconcileEpoch - cloudReconcileTask = (async () => { + const reconcileTask = (async () => { const currentUserId = getCurrentUserId() if (currentUserId === 'local') { console.info('[chat-sync] reconcile skipped: anonymous user') @@ -907,6 +946,8 @@ export const useChatSessionStore = defineStore('chat-session', () => { cloudSyncReady.value = true })().finally(() => { + if (cloudReconcileTask !== reconcileTask) + return cloudReconcileTask = undefined // A second 'open' event fired while we were running — schedule a // follow-up so its catch-up window is not lost. Skip if the epoch @@ -920,7 +961,8 @@ export const useChatSessionStore = defineStore('chat-session', () => { } }) - return cloudReconcileTask + cloudReconcileTask = reconcileTask + return reconcileTask } /** @@ -929,6 +971,8 @@ 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 @@ -985,6 +1029,20 @@ 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() + } + /** * Drop every in-memory session for the current user. Used when the auth * user changes — we must NOT keep account A's sessions visible (or @@ -1002,6 +1060,8 @@ export const useChatSessionStore = defineStore('chat-session', () => { sessionMetas.value = {} sessionGenerations.value = {} loadedSessions.clear() + staleSessions.clear() + cloudHydratedSessions.clear() loadingSessions.clear() index.value = null activeSessionId.value = '' @@ -1231,7 +1291,8 @@ export const useChatSessionStore = defineStore('chat-session', () => { // mid-send) before the WS even opens. The drain itself runs after // reconcile completes, but the count is observable immediately. await refreshOutboxPendingCount() - ensureCloudWsClient() + if (ownsCloudSync) + ensureCloudWsClient() })() try { @@ -1272,22 +1333,14 @@ export const useChatSessionStore = defineStore('chat-session', () => { }, }) - function setActiveSession(sessionId: string) { + /** Selects and hydrates one conversation only in the current window. */ + async function setActiveSession(sessionId: string) { activeSessionId.value = sessionId - const characterId = getCurrentCharacterId() - const characterIndex = index.value?.characters[characterId] - if (characterIndex) { - characterIndex.activeSessionId = sessionId - void persistIndex() - } - - if (ready.value) { - void loadSession(sessionId) - } - else if (!hasKnownSession(sessionId)) { + if (ready.value) + await useChatSessionStore().loadSession(sessionId) + else if (!hasKnownSession(sessionId)) ensureSession(sessionId) - } } function applyRemoteSnapshot(snapshot: { @@ -1306,6 +1359,8 @@ export const useChatSessionStore = defineStore('chat-session', () => { Object.keys(snapshot.sessionMessages).map(sessionId => [sessionId, sessionGenerations.value[sessionId] ?? 0]), ) loadedSessions.clear() + staleSessions.clear() + cloudHydratedSessions.clear() for (const sessionId of Object.keys(snapshot.sessionMessages)) { loadedSessions.add(sessionId) } @@ -1349,6 +1404,8 @@ export const useChatSessionStore = defineStore('chat-session', () => { sessionMetas.value = {} sessionGenerations.value = {} loadedSessions.clear() + staleSessions.clear() + cloudHydratedSessions.clear() loadingSessions.clear() index.value = { @@ -1364,6 +1421,11 @@ export const useChatSessionStore = defineStore('chat-session', () => { return sessionMessages.value[sessionId] ?? [] } + /** Returns persisted/in-memory messages without creating an unloaded session fallback. */ + function getSessionMessagesIfLoaded(sessionId: string) { + return sessionMessages.value[sessionId] + } + function getSessionGeneration(sessionId: string) { ensureGeneration(sessionId) return sessionGenerations.value[sessionId] ?? 0 @@ -1432,6 +1494,8 @@ export const useChatSessionStore = defineStore('chat-session', () => { sessionMetas.value = {} sessionGenerations.value = {} loadedSessions.clear() + staleSessions.clear() + cloudHydratedSessions.clear() loadingSessions.clear() await enqueuePersist(() => chatSessionsRepo.saveIndex(cloneDeep(payload.index))) @@ -1449,6 +1513,39 @@ export const useChatSessionStore = defineStore('chat-session', () => { await ensureActiveSessionForCharacter() } + let lastActiveSessionMeta: ChatSessionMeta | undefined + + // Session data is synchronized, but selection belongs to this window. If + // another window deletes the selected session, repair this window locally + // instead of leaving it pointed at an ID that can no longer hydrate. + watch([ + activeSessionId, + () => Object.values(sessionMetas.value), + () => hasKnownSession(activeSessionId.value), + ], ([sessionId, metas, isKnown]) => { + const meta = metas.find(candidate => candidate.sessionId === sessionId) + if (meta) { + lastActiveSessionMeta = meta + return + } + if (!sessionId || isKnown) + return + + const characterId = lastActiveSessionMeta?.sessionId === sessionId + ? lastActiveSessionMeta.characterId + : getCurrentCharacterId() + const fallbackSessionId = metas + .find(candidate => candidate.characterId === characterId && candidate.userId === getCurrentUserId()) + ?.sessionId + + if (fallbackSessionId) { + void setActiveSession(fallbackSessionId) + } + // If no fallback exists yet, wait for the synchronized delete action. + // Its elected leader creates the single replacement; creating here in + // every follower would fan one deletion out into several empty chats. + }) + watch([userId, activeCardId], () => { if (!ready.value) return @@ -1470,7 +1567,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { watch(userId, (next) => { teardownCloudWsClient() clearInMemoryState() - if (next && next !== 'local') { + if (ownsCloudSync && next && next !== 'local') { ensureCloudWsClient() } // Rehydrate for the new user. We trigger here (instead of relying on the @@ -1485,7 +1582,6 @@ export const useChatSessionStore = defineStore('chat-session', () => { }) return { - ready, isReady, initialize, @@ -1505,26 +1601,32 @@ export const useChatSessionStore = defineStore('chat-session', () => { appendSessionMessage, persistSessionMessages, getSessionMessages, + getSessionMessagesIfLoaded, sessionMessages, sessionMetas, getSessionGeneration, bumpSessionGeneration, getSessionGenerationValue, + // Pinia can synchronize only refs returned by a setup store. + index, forkSession, exportSessions, importSessions, createSession, loadSession, + refreshSession, deleteSession, + setCloudSyncOwnership, + cloudSyncReady, outboxPendingCount, pushMessageToCloud, } }, { synced: { - actions: ['deleteMessage', 'importSessions'], + actions: ['createSession', 'deleteMessage', 'importSessions', 'loadSession', 'refreshSession'], state: true, }, }) diff --git a/packages/stage-ui/src/stores/mods/api/context-bridge.contract.browser.test.ts b/packages/stage-ui/src/stores/mods/api/context-bridge.contract.browser.test.ts index c67b999ee..510b2967f 100644 --- a/packages/stage-ui/src/stores/mods/api/context-bridge.contract.browser.test.ts +++ b/packages/stage-ui/src/stores/mods/api/context-bridge.contract.browser.test.ts @@ -19,6 +19,7 @@ const beginStreamMock = vi.fn() const appendStreamLiteralMock = vi.fn() const finalizeStreamMock = vi.fn() const resetStreamMock = vi.fn() +const refreshSessionMock = vi.fn() const serverSendMock = vi.fn() const ensureConnectedMock = vi.fn().mockResolvedValue(undefined) const onReconnectedMock = vi.fn(() => () => {}) @@ -157,6 +158,7 @@ function createContextUpdateEvent(overrides: Record = {}) { } const chatOrchestratorMock = { + activeSendSessionId: undefined as string | undefined, sending: false, ingest: vi.fn(), @@ -237,6 +239,7 @@ vi.mock('../../chat/session-store', () => ({ return activeSessionIdRef.value }, getSessionGenerationValue: () => currentGeneration, + refreshSession: (sessionId: string) => refreshSessionMock(sessionId), }), })) @@ -291,6 +294,7 @@ describe('context bridge contract', () => { appendStreamLiteralMock.mockReset() finalizeStreamMock.mockReset() resetStreamMock.mockReset() + refreshSessionMock.mockReset().mockResolvedValue(true) serverSendMock.mockReset() ensureConnectedMock.mockClear() ensureConnectedMock.mockResolvedValue(undefined) @@ -304,6 +308,7 @@ describe('context bridge contract', () => { activeProviderRef.value = null activeModelRef.value = null activeSessionIdRef.value = 'session-1' + chatOrchestratorMock.activeSendSessionId = undefined currentGeneration = 7 chatOrchestratorMock.sending = false @@ -325,10 +330,6 @@ describe('context bridge contract', () => { closeTestChannels() }) - /** - * @example - * Broadcast context updates record store-ingested with core result fields. - */ it('records core ingest result for broadcast context updates', async () => { chatContextIngestMock.mockReturnValueOnce({ sourceKey: 'weather:station-1', @@ -361,10 +362,6 @@ describe('context bridge contract', () => { await store.dispose() }) - /** - * @example - * Server context updates record store-ingested before broadcast-posted. - */ it('records core ingest result for server context updates before broadcasting', async () => { chatContextIngestMock.mockReturnValueOnce({ sourceKey: 'weather:station-1', @@ -399,10 +396,6 @@ describe('context bridge contract', () => { await store.dispose() }) - /** - * @example - * Input context updates record store-ingested and stay in chat input payload. - */ it('records core ingest result for input context updates and forwards accepted updates', async () => { chatContextIngestMock.mockReturnValueOnce({ sourceKey: 'weather:station-1', @@ -453,10 +446,6 @@ describe('context bridge contract', () => { await store.dispose() }) - /** - * @example - * Broadcast context ingest failures record store-ingest-rejected instead of escaping. - */ it('records rejected lifecycle for broadcast ingest failures without interrupting the watcher', async () => { chatContextIngestMock.mockImplementationOnce(() => { throw new Error('Cannot clone broadcast context') @@ -485,10 +474,6 @@ describe('context bridge contract', () => { await store.dispose() }) - /** - * @example - * Server context ingest failures are not rebroadcast. - */ it('records rejected lifecycle and skips broadcast when server context ingest fails', async () => { chatContextIngestMock.mockImplementationOnce(() => { throw new Error('Cannot clone server context') @@ -520,10 +505,6 @@ describe('context bridge contract', () => { await store.dispose() }) - /** - * @example - * Input context ingest failures drop only the failed context update. - */ it('records rejected lifecycle and continues text ingestion when input context ingest fails', async () => { chatContextIngestMock.mockImplementationOnce(() => { throw new Error('Cannot clone input context') @@ -562,7 +543,14 @@ describe('context bridge contract', () => { await store.dispose() }) - it('replays remote stream lifecycle into sending and stream store APIs', async () => { + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3743366445 + it('keeps a remote stream visible locally without publishing chat authority state for Issue #2085', async () => { + // ROOT CAUSE: + // + // Stage Pocket uses plain Pinia, so remote stream tokens update only the + // local stream store. The history requires a sending flag, but writing + // that flag to the synchronized chat store would let a follower overwrite + // the elected authority's stream snapshot. const store = useContextBridgeStore() await store.initialize() const streamSender = createTestChannel(CHAT_STREAM_CHANNEL_NAME) @@ -574,18 +562,19 @@ describe('context bridge contract', () => { composedMessage: [], } satisfies ChatStreamEventContext - streamSender.postMessage({ type: 'before-send', message: 'ping', sessionId: 'remote-session', context }) + streamSender.postMessage({ type: 'before-send', message: 'ping', sessionId: 'session-1', context }) await vi.waitFor(() => { - expect(chatOrchestratorMock.sending).toBe(true) expect(beginStreamMock).toHaveBeenCalledWith('turn-1') }) + expect(chatOrchestratorMock.sending).toBe(false) + expect(store.isReceivingRemoteStream).toBe(true) - streamSender.postMessage({ type: 'token-literal', literal: 'hello', sessionId: 'remote-session', context }) + streamSender.postMessage({ type: 'token-literal', literal: 'hello', sessionId: 'session-1', context }) await vi.waitFor(() => { expect(appendStreamLiteralMock).toHaveBeenCalledWith('hello') }) - streamSender.postMessage({ type: 'assistant-end', message: 'final answer', sessionId: 'remote-session', context }) + streamSender.postMessage({ type: 'assistant-end', message: 'final answer', sessionId: 'session-1', context }) await vi.waitFor(() => { expect(resetStreamMock).toHaveBeenCalledTimes(1) }) @@ -594,6 +583,7 @@ describe('context bridge contract', () => { // to avoid corrupting history by persisting a duplicate assistant message. expect(finalizeStreamMock).not.toHaveBeenCalled() expect(chatOrchestratorMock.sending).toBe(false) + expect(store.isReceivingRemoteStream).toBe(false) await store.dispose() }) @@ -624,6 +614,166 @@ describe('context bridge contract', () => { await store.dispose() }) + it('labels outbound stream events with the session that owns the send', async () => { + const outgoingStreamMessages = collectChannelMessages<{ sessionId: string }>(CHAT_STREAM_CHANNEL_NAME) + const store = useContextBridgeStore() + await store.initialize() + const context = { + turnId: 'turn-1', + message: { role: 'user', content: 'ping' }, + contexts: {}, + composedMessage: [], + } satisfies ChatStreamEventContext + + chatOrchestratorMock.activeSendSessionId = 'session-a' + activeSessionIdRef.value = 'session-b' + await chatOrchestratorMock.emitTokenLiteralHooks('session A token', context) + await vi.waitFor(() => expect(outgoingStreamMessages).toHaveLength(1)) + + expect(outgoingStreamMessages[0]?.sessionId).toBe('session-a') + await store.dispose() + }) + + it('clears remote stream visibility when an end hook rejects', async () => { + const store = useContextBridgeStore() + await store.initialize() + const streamSender = createTestChannel(CHAT_STREAM_CHANNEL_NAME) + const context = { + turnId: 'turn-1', + message: { role: 'user', content: 'ping' }, + contexts: {}, + composedMessage: [], + } satisfies ChatStreamEventContext + chatOrchestratorMock.onStreamEnd(async () => { + throw new Error('end hook failed') + }) + vi.spyOn(console, 'error').mockImplementation(() => {}) + + streamSender.postMessage({ type: 'before-send', message: 'ping', sessionId: 'session-1', context }) + await vi.waitFor(() => expect(store.isReceivingRemoteStream).toBe(true)) + streamSender.postMessage({ type: 'stream-end', sessionId: 'session-1', context }) + await vi.waitFor(() => expect(store.isReceivingRemoteStream).toBe(false)) + + expect(resetStreamMock).toHaveBeenCalledTimes(1) + await store.dispose() + }) + + it('ignores stream events that do not match the active remote session', async () => { + const store = useContextBridgeStore() + await store.initialize() + const streamSender = createTestChannel(CHAT_STREAM_CHANNEL_NAME) + const context = { + turnId: 'turn-1', + message: { role: 'user', content: 'ping' }, + contexts: {}, + composedMessage: [], + } satisfies ChatStreamEventContext + + streamSender.postMessage({ type: 'before-send', message: 'ping', sessionId: 'session-1', context }) + await vi.waitFor(() => expect(store.isReceivingRemoteStream).toBe(true)) + streamSender.postMessage({ type: 'token-literal', literal: 'foreign token', sessionId: 'session-2', context }) + streamSender.postMessage({ type: 'stream-end', sessionId: 'session-2', context }) + await waitForBroadcastDelivery() + + expect(appendStreamLiteralMock).not.toHaveBeenCalledWith('foreign token') + expect(store.isReceivingRemoteStream).toBe(true) + streamSender.postMessage({ type: 'stream-end', sessionId: 'session-1', context }) + await vi.waitFor(() => expect(store.isReceivingRemoteStream).toBe(false)) + await store.dispose() + }) + + it('does not replace the foreground stream for an inactive remote session', async () => { + const store = useContextBridgeStore() + await store.initialize() + const streamSender = createTestChannel(CHAT_STREAM_CHANNEL_NAME) + const context = { + turnId: 'turn-2', + message: { role: 'user', content: 'background ping' }, + contexts: {}, + composedMessage: [], + } satisfies ChatStreamEventContext + + streamSender.postMessage({ type: 'before-send', message: 'background ping', sessionId: 'session-2', context }) + await waitForBroadcastDelivery() + expect(beginStreamMock).not.toHaveBeenCalled() + expect(store.isReceivingRemoteStream).toBe(false) + + streamSender.postMessage({ type: 'stream-end', sessionId: 'session-2', context }) + await waitForBroadcastDelivery() + expect(resetStreamMock).not.toHaveBeenCalled() + await store.dispose() + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3755585351 + it('keeps remote literals received before a mid-stream session switch for Issue #2085', async () => { + // ROOT CAUSE: + // + // The bridge discarded literals while their session was not selected. + // Selecting that session during the stream showed only later literals. + activeSessionIdRef.value = 'session-2' + const store = useContextBridgeStore() + await store.initialize() + const streamSender = createTestChannel(CHAT_STREAM_CHANNEL_NAME) + const context = { + turnId: 'turn-3', + message: { role: 'user', content: 'background ping' }, + contexts: {}, + composedMessage: [], + } satisfies ChatStreamEventContext + + streamSender.postMessage({ type: 'before-send', message: 'background ping', sessionId: 'session-1', context }) + streamSender.postMessage({ type: 'token-literal', literal: 'first half ', sessionId: 'session-1', context }) + await waitForBroadcastDelivery() + expect(beginStreamMock).not.toHaveBeenCalled() + expect(appendStreamLiteralMock).not.toHaveBeenCalled() + + activeSessionIdRef.value = 'session-1' + streamSender.postMessage({ type: 'token-literal', literal: 'second half', sessionId: 'session-1', context }) + + await vi.waitFor(() => expect(appendStreamLiteralMock).toHaveBeenCalledTimes(2)) + expect(beginStreamMock).toHaveBeenCalledWith('turn-3') + expect(appendStreamLiteralMock).toHaveBeenNthCalledWith(1, 'first half ') + expect(appendStreamLiteralMock).toHaveBeenNthCalledWith(2, 'second half') + + await store.dispose() + }) + + // https://github.com/moeru-ai/airi/pull/2086#discussion_r3755711154 + it('reloads a completed remote stream when its Pocket session becomes active for Issue #2085', async () => { + // ROOT CAUSE: + // + // A plain-Pinia Pocket tab discarded a completed background stream. + // Its loaded-session cache then prevented a later IndexedDB refresh. + activeSessionIdRef.value = 'session-1' + const store = useContextBridgeStore() + await store.initialize() + const streamSender = createTestChannel(CHAT_STREAM_CHANNEL_NAME) + const context = { + turnId: 'turn-4', + message: { role: 'user', content: 'background ping' }, + contexts: {}, + composedMessage: [], + } satisfies ChatStreamEventContext + + streamSender.postMessage({ type: 'before-send', message: 'background ping', sessionId: 'session-2', context }) + streamSender.postMessage({ type: 'token-literal', literal: 'complete answer', sessionId: 'session-2', context }) + streamSender.postMessage({ type: 'stream-end', sessionId: 'session-2', context }) + streamSender.postMessage({ type: 'assistant-end', message: 'complete answer', sessionId: 'session-2', context }) + await waitForBroadcastDelivery() + + expect(refreshSessionMock).not.toHaveBeenCalled() + expect(resetStreamMock).not.toHaveBeenCalled() + + activeSessionIdRef.value = 'session-2' + await vi.waitFor(() => expect(refreshSessionMock).toHaveBeenCalledWith('session-2')) + await vi.waitFor(() => expect(resetStreamMock).toHaveBeenCalledTimes(1)) + expect(beginStreamMock).toHaveBeenCalledWith('turn-4') + expect(appendStreamLiteralMock).toHaveBeenCalledWith('complete answer') + expect(store.isReceivingRemoteStream).toBe(false) + + await store.dispose() + }) + it('ignores remote literal and end events when generation guard is stale', async () => { const store = useContextBridgeStore() await store.initialize() @@ -636,21 +786,22 @@ describe('context bridge contract', () => { composedMessage: [], } satisfies ChatStreamEventContext - streamSender.postMessage({ type: 'before-send', message: 'ping', sessionId: 'remote-session', context }) + streamSender.postMessage({ type: 'before-send', message: 'ping', sessionId: 'session-1', context }) await vi.waitFor(() => { expect(beginStreamMock).toHaveBeenCalledWith('turn-1') }) currentGeneration = 8 - streamSender.postMessage({ type: 'token-literal', literal: 'stale-literal', sessionId: 'remote-session', context }) + streamSender.postMessage({ type: 'token-literal', literal: 'stale-literal', sessionId: 'session-1', context }) await waitForBroadcastDelivery() - streamSender.postMessage({ type: 'stream-end', sessionId: 'remote-session', context }) + streamSender.postMessage({ type: 'stream-end', sessionId: 'session-1', context }) await waitForBroadcastDelivery() expect(appendStreamLiteralMock).not.toHaveBeenCalledWith('stale-literal') expect(finalizeStreamMock).not.toHaveBeenCalled() - expect(chatOrchestratorMock.sending).toBe(true) + expect(chatOrchestratorMock.sending).toBe(false) + expect(store.isReceivingRemoteStream).toBe(false) await store.dispose() }) diff --git a/packages/stage-ui/src/stores/mods/api/context-bridge.ts b/packages/stage-ui/src/stores/mods/api/context-bridge.ts index f5d60ec4f..61b3546b2 100644 --- a/packages/stage-ui/src/stores/mods/api/context-bridge.ts +++ b/packages/stage-ui/src/stores/mods/api/context-bridge.ts @@ -12,7 +12,7 @@ import { useBroadcastChannel } from '@vueuse/core' import { Mutex } from 'es-toolkit' import { nanoid } from 'nanoid' import { defineStore, storeToRefs } from 'pinia' -import { ref, toRaw, watch } from 'vue' +import { computed, ref, shallowRef, toRaw, watch } from 'vue' import { getEventSourceKey, getMetadataSourceLabel } from '../../../utils/event-source' import { useLlmStreamingControlStore } from '../../ai/chat-llm/streaming-control' @@ -89,10 +89,61 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = const { post: postSparkNotifyBridgeMessage, data: incomingSparkNotifyBridgeMessage } = useBroadcastChannel({ name: SPARK_NOTIFY_BRIDGE_CHANNEL_NAME }) const disposeHookFns = ref void>>([]) - let remoteStreamGuard: { sessionId: string, generation: number } | null = null + // Remote stream data belongs to this renderer only. Keeping its visibility + // outside the synchronized chat store prevents a follower from publishing + // its local runtime state over the elected authority's snapshot. + const remoteStreamGuard = shallowRef<{ + sessionId: string + generation: number + turnId: string + started: boolean + completed: boolean + refreshing: boolean + pendingLiterals: string[] + }>() + const isReceivingRemoteStream = computed(() => remoteStreamGuard.value?.sessionId === chatSession.activeSessionId) let contextChannel: ReturnType | undefined let initialized = false + function presentRemoteStreamIfActive() { + const guard = remoteStreamGuard.value + if (!guard || guard.sessionId !== chatSession.activeSessionId) + return false + if (chatSession.getSessionGenerationValue(guard.sessionId) !== guard.generation) + return false + + if (!guard.started) { + guard.started = true + chatStream.beginStream(guard.turnId) + } + for (const literal of guard.pendingLiterals.splice(0)) + chatStream.appendStreamLiteral(literal) + if (guard.completed && !guard.refreshing) + void refreshCompletedRemoteStream(guard) + return true + } + + async function refreshCompletedRemoteStream(guard: NonNullable) { + guard.refreshing = true + let loaded = false + try { + loaded = await chatSession.refreshSession(guard.sessionId) + } + catch (error) { + console.warn('[context-bridge] Failed to refresh completed remote stream:', errorMessageFrom(error)) + } + if (remoteStreamGuard.value !== guard) + return + guard.refreshing = false + if (!loaded || guard.sessionId !== chatSession.activeSessionId) + return + if (chatSession.getSessionGenerationValue(guard.sessionId) !== guard.generation) + return + + chatStream.resetStream() + remoteStreamGuard.value = undefined + } + function recordContextIngestRejected(options: { channel: 'server' | 'broadcast' | 'input' contextMessage: ContextMessage @@ -699,49 +750,49 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = if (isProcessingRemoteStream) return - await contextChannel?.emitStream({ type: 'before-compose', message, sessionId: chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) + await contextChannel?.emitStream({ type: 'before-compose', message, sessionId: chatOrchestrator.activeSendSessionId ?? chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) }), chatOrchestrator.onAfterMessageComposed(async (message, context) => { if (isProcessingRemoteStream) return - await contextChannel?.emitStream({ type: 'after-compose', message, sessionId: chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) + await contextChannel?.emitStream({ type: 'after-compose', message, sessionId: chatOrchestrator.activeSendSessionId ?? chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) }), chatOrchestrator.onBeforeSend(async (message, context) => { if (isProcessingRemoteStream) return - await contextChannel?.emitStream({ type: 'before-send', message, sessionId: chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) + await contextChannel?.emitStream({ type: 'before-send', message, sessionId: chatOrchestrator.activeSendSessionId ?? chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) }), chatOrchestrator.onAfterSend(async (message, context) => { if (isProcessingRemoteStream) return - await contextChannel?.emitStream({ type: 'after-send', message, sessionId: chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) + await contextChannel?.emitStream({ type: 'after-send', message, sessionId: chatOrchestrator.activeSendSessionId ?? chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) }), chatOrchestrator.onTokenLiteral(async (literal, context) => { if (isProcessingRemoteStream) return - await contextChannel?.emitStream({ type: 'token-literal', literal, sessionId: chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) + await contextChannel?.emitStream({ type: 'token-literal', literal, sessionId: chatOrchestrator.activeSendSessionId ?? chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) }), chatOrchestrator.onTokenSpecial(async (special, context) => { if (isProcessingRemoteStream) return - await contextChannel?.emitStream({ type: 'token-special', special, sessionId: chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) + await contextChannel?.emitStream({ type: 'token-special', special, sessionId: chatOrchestrator.activeSendSessionId ?? chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) }), chatOrchestrator.onStreamEnd(async (context) => { if (isProcessingRemoteStream) return - await contextChannel?.emitStream({ type: 'stream-end', sessionId: chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) + await contextChannel?.emitStream({ type: 'stream-end', sessionId: chatOrchestrator.activeSendSessionId ?? chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) }), chatOrchestrator.onAssistantResponseEnd(async (message, context) => { if (isProcessingRemoteStream) return - await contextChannel?.emitStream({ type: 'assistant-end', message, sessionId: chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) + await contextChannel?.emitStream({ type: 'assistant-end', message, sessionId: chatOrchestrator.activeSendSessionId ?? chatSession.activeSessionId, context: structuredClone(normalizeContextSnapshot(context)) }) }), chatOrchestrator.onAssistantMessage(async (message, _messageText, context) => { @@ -794,7 +845,8 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = isProcessingRemoteStream = true try { - // Use the receiver's active session to avoid clobbering chat state when events come from other windows/devtools. + // Remote UI state is correlated by session and generation. The + // receiver never persists these mirrored stream events. switch (event.type) { case 'before-compose': await chatOrchestrator.emitBeforeMessageComposedHooks(event.message, event.context) @@ -804,58 +856,99 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = break case 'before-send': await chatOrchestrator.emitBeforeSendHooks(event.message, event.context) - remoteStreamGuard = { - sessionId: chatSession.activeSessionId, - generation: chatSession.getSessionGenerationValue(chatSession.activeSessionId), + remoteStreamGuard.value = { + sessionId: event.sessionId, + generation: chatSession.getSessionGenerationValue(event.sessionId), + turnId: event.context.turnId, + started: false, + completed: false, + refreshing: false, + pendingLiterals: [], } - chatOrchestrator.sending = true - chatStream.beginStream(event.context.turnId) + presentRemoteStreamIfActive() break case 'after-send': await chatOrchestrator.emitAfterSendHooks(event.message, event.context) break case 'token-literal': - if (!remoteStreamGuard) + if (!remoteStreamGuard.value) return - if (remoteStreamGuard.sessionId !== chatSession.activeSessionId) + if (event.sessionId !== remoteStreamGuard.value.sessionId) return - if (chatSession.getSessionGenerationValue(remoteStreamGuard.sessionId) !== remoteStreamGuard.generation) + if (chatSession.getSessionGenerationValue(remoteStreamGuard.value.sessionId) !== remoteStreamGuard.value.generation) + return + remoteStreamGuard.value.pendingLiterals.push(event.literal) + if (!presentRemoteStreamIfActive()) return - chatStream.appendStreamLiteral(event.literal) await chatOrchestrator.emitTokenLiteralHooks(event.literal, event.context) break case 'token-special': + if (!remoteStreamGuard.value || event.sessionId !== remoteStreamGuard.value.sessionId) + return + if (remoteStreamGuard.value.sessionId !== chatSession.activeSessionId) + return + if (chatSession.getSessionGenerationValue(remoteStreamGuard.value.sessionId) !== remoteStreamGuard.value.generation) + return await chatOrchestrator.emitTokenSpecialHooks(event.special, event.context) break case 'stream-end': - if (!remoteStreamGuard) + if (!remoteStreamGuard.value) break - if (remoteStreamGuard.sessionId !== chatSession.activeSessionId) - break - if (chatSession.getSessionGenerationValue(remoteStreamGuard.sessionId) !== remoteStreamGuard.generation) - break - await chatOrchestrator.emitStreamEndHooks(event.context) - // NOTICE: Remote stream events are mirrored across renderer windows for UI feedback only. - // Persisting them here would append assistant messages into the receiver's local session - // without the corresponding user message, corrupting IndexedDB history across windows. - chatStream.resetStream() - chatOrchestrator.sending = false - remoteStreamGuard = null + { + const guard = remoteStreamGuard.value + if (event.sessionId !== guard.sessionId) + break + if (guard.sessionId !== chatSession.activeSessionId + && chatSession.getSessionGenerationValue(guard.sessionId) === guard.generation) { + break + } + try { + if (guard.sessionId === chatSession.activeSessionId + && chatSession.getSessionGenerationValue(guard.sessionId) === guard.generation) { + await chatOrchestrator.emitStreamEndHooks(event.context) + } + } + finally { + if (remoteStreamGuard.value === guard) { + if (guard.started + && guard.sessionId === chatSession.activeSessionId + && chatSession.getSessionGenerationValue(guard.sessionId) === guard.generation) { + chatStream.resetStream() + } + remoteStreamGuard.value = undefined + } + } + } break case 'assistant-end': - if (!remoteStreamGuard) + if (!remoteStreamGuard.value) break - if (remoteStreamGuard.sessionId !== chatSession.activeSessionId) - break - if (chatSession.getSessionGenerationValue(remoteStreamGuard.sessionId) !== remoteStreamGuard.generation) - break - await chatOrchestrator.emitAssistantResponseEndHooks(event.message, event.context) - // NOTICE: The originating renderer already persists the final assistant message. - // Receiver windows must not write it again, or they can overwrite the same session - // with assistant-only history when their local session state is stale. - chatStream.resetStream() - chatOrchestrator.sending = false - remoteStreamGuard = null + { + const guard = remoteStreamGuard.value + if (event.sessionId !== guard.sessionId) + break + if (guard.sessionId !== chatSession.activeSessionId + && chatSession.getSessionGenerationValue(guard.sessionId) === guard.generation) { + guard.completed = true + break + } + try { + if (guard.sessionId === chatSession.activeSessionId + && chatSession.getSessionGenerationValue(guard.sessionId) === guard.generation) { + await chatOrchestrator.emitAssistantResponseEndHooks(event.message, event.context) + } + } + finally { + if (remoteStreamGuard.value === guard) { + if (guard.started + && guard.sessionId === chatSession.activeSessionId + && chatSession.getSessionGenerationValue(guard.sessionId) === guard.generation) { + chatStream.resetStream() + } + remoteStreamGuard.value = undefined + } + } + } break } } @@ -864,6 +957,11 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = } }) disposeHookFns.value.push(stopIncomingStreamWatch) + disposeHookFns.value.push(watch( + () => chatSession.activeSessionId, + () => presentRemoteStreamIfActive(), + { flush: 'sync' }, + )) initialized = true } catch (error) { @@ -905,7 +1003,7 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = contextChannel = undefined initialized = false - remoteStreamGuard = null + remoteStreamGuard.value = undefined for (const [requestId, waiter] of sparkNotifyBridgeWaiters) { if (waiter.timeout) @@ -925,6 +1023,7 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = dispose, dispatchSparkNotifyReaction, dispatchSparkNotifyPerformance, + isReceivingRemoteStream, setSparkNotifyHostRole, } }) diff --git a/vitest.config.ts b/vitest.config.ts index 112cd5a37..515becfc4 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -6,7 +6,7 @@ export default defineConfig({ 'server/apps/auth', 'server/apps/api', 'apps/ui-server-auth', - 'apps/stage-tamagotchi', + 'apps/stage-tamagotchi/vitest.node.config.ts', 'packages/cap-vite', 'packages/ccc', 'packages/core-agent',