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 @@
+
+
+ emit('update:open', value)">
+
+
+
+
+
+
+
+
+ {{ t('stage.chat.sessions.title') }}
+
+
+
+
+
+ {{ t('stage.chat.sessions.empty') }}
+
+
+
+
+
+
+
+
+
+
+
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) => {
- showDialog = value">
-
-
-
-
-
-
-
- {{ t('stage.chat.sessions.title') }}
-
-
-
-
-
- {{ t('stage.chat.sessions.empty') }}
-
-
-
-
-
-
-
-
-
-
- showDialog = value">
-
-
-
-
-
-
- {{ t('stage.chat.sessions.title') }}
-
-
-
-
-
- {{ t('stage.chat.sessions.empty') }}
-
-
-
-
-
-
-
-
-
+
+
+
+
+
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',