fix(stage-tamagotchi): repair chat-list add, switch, and delete (#2086)

Co-authored-by: RainbowBird <git@luoling.moe>
This commit is contained in:
Columbina
2026-08-14 16:24:56 +08:00
committed by GitHub
parent cc791fab95
commit 96465c8130
24 changed files with 2379 additions and 623 deletions
+4 -2
View File
@@ -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()
})
</script>
@@ -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: '<div />' } }],
})
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<ReturnType<typeof renderArea>>['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('')
})
})
@@ -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() {
<ChatHistory
:messages="historyMessages"
:assistant-label="assistantLabel"
:sending="sending"
:streaming-message="streamingMessage"
:sending="isActiveSessionSending"
:streaming-message="visibleStreamingMessage"
:tool-call-renderers="toolCallRenderers"
@delete-message="handleDeleteMessage($event.index)"
@retry-message="handleRetryMessage($event.index)"
+29 -4
View File
@@ -3,19 +3,44 @@ import { cwd } from 'node:process'
import vue from '@vitejs/plugin-vue'
import Info from 'unplugin-info/vite'
import { playwright } from '@vitest/browser-playwright'
import { loadEnv } from 'vite'
import { defineConfig } from 'vitest/config'
export default defineConfig({
root: import.meta.dirname,
plugins: [
Info(),
vue(),
],
test: {
env: loadEnv('test', cwd(), ''),
include: ['src/**/*.test.ts', 'scripts/**/*.test.ts'],
exclude: ['**/node_modules/**', '**/.git/**'],
fileParallelism: false,
maxWorkers: 1,
projects: [
{
extends: true,
test: {
name: 'node',
include: ['src/**/*.test.ts', 'scripts/**/*.test.ts'],
exclude: ['src/**/*.browser.test.ts', '**/node_modules/**', '**/.git/**'],
fileParallelism: false,
maxWorkers: 1,
},
},
{
extends: true,
test: {
name: 'browser',
include: ['src/**/*.browser.test.ts'],
exclude: ['**/node_modules/**', '**/.git/**'],
browser: {
enabled: true,
provider: playwright(),
instances: [
{ browser: 'chromium' },
],
},
},
},
],
},
})
@@ -0,0 +1,20 @@
import { cwd } from 'node:process'
import vue from '@vitejs/plugin-vue'
import Info from 'unplugin-info/vite'
import { loadEnv } from 'vite'
import { defineProject } from 'vitest/config'
export default defineProject({
root: import.meta.dirname,
plugins: [Info(), vue()],
test: {
name: 'stage-tamagotchi:node',
env: loadEnv('test', cwd(), ''),
include: ['src/**/*.test.ts', 'scripts/**/*.test.ts'],
exclude: ['src/**/*.browser.test.ts', '**/node_modules/**', '**/.git/**'],
fileParallelism: false,
maxWorkers: 1,
},
})
+5
View File
@@ -3,6 +3,7 @@ import { OnboardingDialog, OnboardingStepAnalyticsNotice, ToasterRoot } from '@p
import { useInferencePreload } from '@proj-airi/stage-ui/composables'
import { useAuthProviderSync } from '@proj-airi/stage-ui/composables/use-auth-provider-sync'
import { initializeAnalytics, isAnalyticsAvailableInBuild } from '@proj-airi/stage-ui/libs/analytics'
import { usePiniaSynced } from '@proj-airi/stage-ui/libs/pinia'
import { useCharacterOrchestratorStore } from '@proj-airi/stage-ui/stores/character'
import { useChatSessionStore } from '@proj-airi/stage-ui/stores/chat/session-store'
import { useDisplayModelsStore } from '@proj-airi/stage-ui/stores/display-models'
@@ -33,6 +34,9 @@ const settingsStore = useSettings()
const settings = storeToRefs(settingsStore)
const onboardingStore = useOnboardingStore()
const chatSessionStore = useChatSessionStore()
const syncedPinia = usePiniaSynced()
chatSessionStore.setCloudSyncOwnership(syncedPinia.isLeader())
const stopLeadershipListener = syncedPinia.onLeadershipChange(isLeader => chatSessionStore.setCloudSyncOwnership(isLeader))
const serverChannelStore = useModsServerChannelStore()
const characterOrchestratorStore = useCharacterOrchestratorStore()
const settingsAudioDeviceStore = useSettingsAudioDevice()
@@ -105,6 +109,7 @@ onMounted(async () => {
})
onUnmounted(() => {
stopLeadershipListener()
contextBridgeStore.dispose()
})
+2 -1
View File
@@ -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 .",
@@ -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<void>((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<void>((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[] = []
@@ -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 ?? (<T>(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,
@@ -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) {
<ChatHistory
v-if="isReady"
:messages="historyMessages"
:sending="sending"
:streaming-message="streamingMessage"
:sending="isActiveSessionSending"
:streaming-message="visibleStreamingMessage"
h-full
variant="desktop"
@delete-message="handleDeleteMessage($event.index)"
@@ -1,6 +1,5 @@
<script setup lang="ts">
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"
@@ -1,6 +1,4 @@
<script setup lang="ts">
import type { ChatProvider } from '@xsai-ext/providers/utils'
import { errorMessageFrom } from '@moeru/std'
import { isStageTamagotchi } from '@proj-airi/stage-shared'
import { HearingConfig } from '@proj-airi/stage-ui/components/scenarios/dialogs/audio-input/index'
@@ -8,9 +6,6 @@ import { useAudioAnalyzer } from '@proj-airi/stage-ui/composables'
import { useAudioContext } from '@proj-airi/stage-ui/stores/audio'
import { useChatStore } from '@proj-airi/stage-ui/stores/chat'
import { useChatSessionStore } from '@proj-airi/stage-ui/stores/chat/session-store'
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 { useSettings, useSettingsAudioDevice } from '@proj-airi/stage-ui/stores/settings'
import { BasicTextarea } from '@proj-airi/ui'
import { useLocalStorage } from '@vueuse/core'
@@ -34,17 +29,12 @@ type SendMode = (typeof SEND_MODES)[number]
const sendMode = useLocalStorage<SendMode>('ui/chat/settings/send-mode', 'enter')
const lastEnterTime = ref(0)
const providersStore = useProviderStore()
const providerStore = useProviderConfigStore()
const { activeProvider, activeModel } = storeToRefs(useConsciousnessStore())
const { themeColorsHueDynamic } = storeToRefs(useSettings())
const { askPermission } = useSettingsAudioDevice()
const { enabled, stream } = storeToRefs(useSettingsAudioDevice())
const chatOrchestrator = useChatStore()
const chatSession = useChatSessionStore()
const { ingest, onAfterMessageComposed } = chatOrchestrator
const { audioContext } = useAudioContext()
const { t } = useI18n()
const sendModeLabels = computed<Record<SendMode, string>>(() => ({
@@ -68,24 +58,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) {
// preserve any user input when failed to send the message
messageInput.value = [textToSend, messageInput.value.trim()].filter(Boolean).join(' ')
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
}
}
}
@@ -135,9 +125,6 @@ watch(hearingPopoverOpen, async (value) => {
}
})
onAfterMessageComposed(async () => {
})
const { startAnalyzer, stopAnalyzer } = useAudioAnalyzer()
let analyzerSource: MediaStreamAudioSourceNode | undefined
@@ -0,0 +1,104 @@
import type { ChatSessionMeta } from '../../../../types/chat-session'
import { describe, expect, it } from 'vitest'
import { render } from 'vitest-browser-vue'
import { defineComponent, ref } from 'vue'
import { createI18n } from 'vue-i18n'
import SessionsDialog from './sessions-dialog.vue'
function createTestI18n() {
return createI18n({
legacy: false,
locale: 'en',
messages: {
en: {
stage: {
chat: {
sessions: {
'title': 'Chats',
'new': 'New chat',
'empty': 'No chats',
'delete': 'Delete',
'cloud-badge': 'Cloud synced',
},
},
},
},
},
})
}
function sessionMeta(sessionId: string, updatedAt: number): ChatSessionMeta {
return {
sessionId,
characterId: 'default',
userId: 'local',
createdAt: updatedAt,
updatedAt,
}
}
function createHarness() {
return defineComponent({
name: 'SessionsDialogHarness',
components: { SessionsDialog },
setup() {
const created = ref(0)
const selected = ref('none')
const deleted = ref('none')
return {
created,
deleted,
selected,
rows: [
{ meta: sessionMeta('session-one', 2), preview: 'First chat', isActive: true, updatedAtLabel: 'now' },
{ meta: sessionMeta('session-two', 1), preview: 'Second chat', isActive: false, updatedAtLabel: 'yesterday' },
],
}
},
template: `
<SessionsDialog
:open="true"
:rows="rows"
:is-desktop="false"
:is-creating-session="false"
mobile-padding-bottom="24px"
@new-session="created += 1"
@select-session="selected = $event"
@delete-session="deleted = $event"
/>
<output aria-label="created-session-count">{{ created }}</output>
<output aria-label="selected-session-id">{{ selected }}</output>
<output aria-label="deleted-session-id">{{ deleted }}</output>
`,
})
}
describe('sessions dialog actions', () => {
// https://github.com/moeru-ai/airi/issues/2085
it('keeps add, switch, and delete actions independent for Issue #2085', async () => {
// ROOT CAUSE:
//
// Vaul handled every pointer release on DrawerContent, including releases
// from its action buttons, and unmounted the sheet before `click` ran.
// The replacement uses a Reka dialog surface whose buttons emit one action
// each without a competing gesture-release lifecycle.
const screen = await render(createHarness(), {
global: {
plugins: [createTestI18n()],
},
})
await screen.getByRole('button', { name: 'New chat' }).click()
await expect.element(screen.getByLabelText('created-session-count')).toHaveTextContent('1')
await screen.getByRole('button', { name: 'Delete: Second chat' }).click()
await expect.element(screen.getByLabelText('deleted-session-id')).toHaveTextContent('session-two')
await expect.element(screen.getByLabelText('selected-session-id')).toHaveTextContent('none')
await screen.getByRole('button', { name: 'First chat now' }).click()
await expect.element(screen.getByLabelText('selected-session-id')).toHaveTextContent('session-one')
})
})
@@ -0,0 +1,132 @@
<script setup lang="ts">
import type { ChatSessionMeta } from '../../../../types/chat-session'
import { DialogContent, DialogOverlay, DialogPortal, DialogRoot, DialogTitle } from 'reka-ui'
import { useI18n } from 'vue-i18n'
interface SessionRow {
meta: ChatSessionMeta
preview: string
isActive: boolean
updatedAtLabel: string
}
interface Props {
open: boolean
rows: SessionRow[]
isDesktop: boolean
isCreatingSession: boolean
mobilePaddingBottom: string
}
defineProps<Props>()
const emit = defineEmits<{
'deleteSession': [sessionId: string]
'newSession': []
'selectSession': [sessionId: string]
'update:open': [open: boolean]
}>()
const { t } = useI18n()
</script>
<template>
<DialogRoot :open="open" @update:open="value => emit('update:open', value)">
<slot name="trigger" />
<DialogPortal>
<DialogOverlay
:class="[
'fixed inset-0 z-[9999] bg-black/50 backdrop-blur-sm',
'data-[state=closed]:animate-fadeOut data-[state=open]:animate-fadeIn',
]"
/>
<DialogContent
:aria-describedby="undefined"
:class="[
'fixed z-[9999] flex flex-col overflow-hidden bg-white/95 shadow-xl outline-none backdrop-blur-md dark:bg-neutral-900/95',
isDesktop
? 'left-1/2 top-1/2 max-h-[80dvh] max-w-md w-[92dvw] rounded-2xl -translate-x-1/2 -translate-y-1/2 data-[state=closed]:animate-contentHide data-[state=open]:animate-contentShow'
: 'bottom-0 left-0 right-0 max-h-[85dvh] rounded-t-[32px]',
]"
:style="isDesktop ? undefined : { paddingBottom: mobilePaddingBottom }"
>
<div
v-if="!isDesktop"
:class="['mx-auto mt-3 h-1.5 w-12 shrink-0 rounded-full bg-neutral-400 dark:bg-neutral-600']"
aria-hidden="true"
/>
<div :class="['flex min-h-0 flex-1 flex-col']">
<div :class="['flex items-center justify-between px-5 pt-5 pb-3']">
<DialogTitle :class="['text-base font-medium text-neutral-700 dark:text-neutral-200']">
{{ t('stage.chat.sessions.title') }}
</DialogTitle>
<button
type="button"
:class="[
'rounded-lg px-3 py-1.5 text-xs font-medium',
'bg-primary-100/60 text-primary-700 dark:bg-primary-900/40 dark:text-primary-200',
'hover:bg-primary-200/70 dark:hover:bg-primary-800/50',
'transition-colors',
]"
:disabled="isCreatingSession"
@click="emit('newSession')"
>
{{ t('stage.chat.sessions.new') }}
</button>
</div>
<div :class="['flex-1 overflow-y-auto px-2 pb-4']">
<div v-if="rows.length === 0" :class="['p-6 text-center text-sm text-neutral-500 dark:text-neutral-400']">
{{ t('stage.chat.sessions.empty') }}
</div>
<div
v-for="row in rows"
:key="row.meta.sessionId"
:class="[
'group relative mb-1 w-full rounded-xl transition-colors',
row.isActive
? 'bg-primary-100/70 dark:bg-primary-900/40'
: 'hover:bg-neutral-100/80 dark:hover:bg-neutral-800/60',
]"
>
<button
type="button"
:class="['w-full flex flex-col gap-1 px-3 py-3 text-left outline-none']"
@click="emit('selectSession', row.meta.sessionId)"
>
<div :class="['flex items-center gap-2 text-sm font-medium text-neutral-700 dark:text-neutral-200']">
<span :class="['flex-1 truncate']">{{ row.preview }}</span>
<span
v-if="row.meta.cloudChatId"
:class="['shrink-0 rounded px-1.5 py-0.5 text-[10px] uppercase tracking-wide', 'bg-emerald-500/15 text-emerald-700 dark:text-emerald-300']"
:title="t('stage.chat.sessions.cloud-badge')"
>
cloud
</span>
<span :class="['w-7']" />
</div>
<div :class="['text-[11px] text-neutral-500 dark:text-neutral-400']">
{{ row.updatedAtLabel }}
</div>
</button>
<button
type="button"
:class="[
'absolute right-2 top-2 z-10 h-7 w-7 flex items-center justify-center rounded-md',
'opacity-100 md:opacity-0 md:group-hover:opacity-100 focus:opacity-100',
'text-neutral-400 hover:bg-red-500/10 hover:text-red-500',
'transition-opacity duration-150',
]"
:aria-label="`${t('stage.chat.sessions.delete')}: ${row.preview}`"
:title="t('stage.chat.sessions.delete')"
@click.stop="emit('deleteSession', row.meta.sessionId)"
>
<div class="i-solar:trash-bin-trash-bold-duotone h-4 w-4" />
</button>
</div>
</div>
</div>
</DialogContent>
</DialogPortal>
</DialogRoot>
</template>
@@ -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')
})
})
@@ -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<SessionRow[]>(() => {
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) => {
</script>
<template>
<DialogRoot v-if="isDesktop" :open="showDialog" @update:open="value => showDialog = value">
<slot name="trigger" />
<DialogPortal>
<DialogOverlay
:class="[
'fixed inset-0 z-[9999] bg-black/50 backdrop-blur-sm',
'data-[state=closed]:animate-fadeOut data-[state=open]:animate-fadeIn',
]"
/>
<DialogContent
:class="[
'fixed left-1/2 top-1/2 z-[9999] max-h-[80dvh] max-w-md w-[92dvw] transform overflow-hidden rounded-2xl bg-white/95 shadow-xl outline-none backdrop-blur-md scrollbar-none -translate-x-1/2 -translate-y-1/2 data-[state=closed]:animate-contentHide data-[state=open]:animate-contentShow dark:bg-neutral-900/90',
]"
>
<div :class="['flex flex-col h-full max-h-[80dvh]']">
<div :class="['flex items-center justify-between px-5 pt-5 pb-3']">
<DialogTitle :class="['text-base font-medium text-neutral-700 dark:text-neutral-200']">
{{ t('stage.chat.sessions.title') }}
</DialogTitle>
<button
:class="[
'rounded-lg px-3 py-1.5 text-xs font-medium',
'bg-primary-100/60 text-primary-700 dark:bg-primary-900/40 dark:text-primary-200',
'hover:bg-primary-200/70 dark:hover:bg-primary-800/50',
'transition-colors',
]"
:disabled="isCreatingSession"
@click="startNewSession"
>
{{ t('stage.chat.sessions.new') }}
</button>
</div>
<div :class="['flex-1 overflow-y-auto px-2 pb-4']">
<div v-if="rows.length === 0" :class="['p-6 text-center text-sm text-neutral-500 dark:text-neutral-400']">
{{ t('stage.chat.sessions.empty') }}
</div>
<div
v-for="row in rows"
:key="row.meta.sessionId"
:class="[
'group relative w-full rounded-xl mb-1',
'transition-colors',
row.isActive
? 'bg-primary-100/70 dark:bg-primary-900/40'
: 'hover:bg-neutral-100/80 dark:hover:bg-neutral-800/60',
]"
>
<button
:class="['w-full text-left px-3 py-2.5 outline-none flex flex-col gap-1']"
@click="selectSession(row.meta.sessionId)"
>
<div :class="['flex items-center gap-2 text-sm font-medium text-neutral-700 dark:text-neutral-200']">
<span :class="['truncate flex-1']">{{ row.preview }}</span>
<span
v-if="row.meta.cloudChatId"
:class="['shrink-0 text-[10px] uppercase tracking-wide rounded px-1.5 py-0.5', 'bg-emerald-500/15 text-emerald-700 dark:text-emerald-300']"
:title="t('stage.chat.sessions.cloud-badge')"
>
cloud
</span>
<!-- placeholder for trash icon to reserve hit space -->
<span :class="['w-7']" />
</div>
<div :class="['text-[11px] text-neutral-500 dark:text-neutral-400']">
{{ row.updatedAtLabel }}
</div>
</button>
<button
:class="[
'absolute right-2 top-2 h-7 w-7 flex items-center justify-center rounded-md',
'opacity-0 group-hover:opacity-100 focus:opacity-100',
'text-neutral-400 hover:text-red-500 hover:bg-red-500/10',
'transition-opacity duration-150',
]"
:title="t('stage.chat.sessions.delete')"
@click="deleteRow($event, row.meta.sessionId)"
>
<div class="i-solar:trash-bin-trash-bold-duotone h-4 w-4" />
</button>
</div>
</div>
</div>
</DialogContent>
</DialogPortal>
</DialogRoot>
<DrawerRoot v-else :open="showDialog" should-scale-background @update:open="value => showDialog = value">
<DrawerPortal>
<DrawerOverlay :class="['fixed inset-0']" />
<DrawerContent
:class="[
'fixed bottom-0 left-0 right-0 z-1000',
'mt-20 px-2 pt-3',
'flex flex-col',
'h-full max-h-[85%]',
'rounded-t-[32px] outline-none backdrop-blur-md',
'bg-neutral-50/95 dark:bg-neutral-900/95',
]"
:style="{ paddingBottom: `${Math.max(Number.parseFloat(screenSafeArea.bottom.value.replace('px', '')), 24)}px` }"
>
<DrawerHandle :class="['[div&]:bg-neutral-400 [div&]:dark:bg-neutral-600']" />
<div :class="['flex items-center justify-between px-4 pt-3 pb-2']">
<DrawerTitle :class="['text-base font-medium text-neutral-700 dark:text-neutral-200']">
{{ t('stage.chat.sessions.title') }}
</DrawerTitle>
<button
:class="[
'rounded-lg px-3 py-1.5 text-xs font-medium',
'bg-primary-100/60 text-primary-700 dark:bg-primary-900/40 dark:text-primary-200',
'hover:bg-primary-200/70 dark:hover:bg-primary-800/50',
'transition-colors',
]"
:disabled="isCreatingSession"
@click="startNewSession"
>
{{ t('stage.chat.sessions.new') }}
</button>
</div>
<div :class="['flex-1 overflow-y-auto px-2 pb-2']">
<div v-if="rows.length === 0" :class="['p-6 text-center text-sm text-neutral-500 dark:text-neutral-400']">
{{ t('stage.chat.sessions.empty') }}
</div>
<div
v-for="row in rows"
:key="row.meta.sessionId"
:class="[
'group relative w-full rounded-xl mb-1',
'transition-colors',
row.isActive
? 'bg-primary-100/70 dark:bg-primary-900/40'
: 'hover:bg-neutral-100/80 dark:hover:bg-neutral-800/60',
]"
>
<button
:class="['w-full text-left px-3 py-3 outline-none flex flex-col gap-1']"
@click="selectSession(row.meta.sessionId)"
>
<div :class="['flex items-center gap-2 text-sm font-medium text-neutral-700 dark:text-neutral-200']">
<span :class="['truncate flex-1']">{{ row.preview }}</span>
<span
v-if="row.meta.cloudChatId"
:class="['shrink-0 text-[10px] uppercase tracking-wide rounded px-1.5 py-0.5', 'bg-emerald-500/15 text-emerald-700 dark:text-emerald-300']"
:title="t('stage.chat.sessions.cloud-badge')"
>
cloud
</span>
<span :class="['w-7']" />
</div>
<div :class="['text-[11px] text-neutral-500 dark:text-neutral-400']">
{{ row.updatedAtLabel }}
</div>
</button>
<button
:class="[
'absolute right-2 top-2 h-7 w-7 flex items-center justify-center rounded-md',
'opacity-100 md:opacity-0 md:group-hover:opacity-100 focus:opacity-100',
'text-neutral-400 hover:text-red-500 hover:bg-red-500/10',
'transition-opacity duration-150',
]"
:title="t('stage.chat.sessions.delete')"
@click="deleteRow($event, row.meta.sessionId)"
>
<div class="i-solar:trash-bin-trash-bold-duotone h-4 w-4" />
</button>
</div>
</div>
</DrawerContent>
</DrawerPortal>
</DrawerRoot>
<SessionsDialog
v-model:open="showDialog"
:rows="rows"
:is-desktop="isDesktop"
:is-creating-session="isCreatingSession"
:mobile-padding-bottom="mobilePaddingBottom"
@new-session="startNewSession"
@select-session="selectSession"
@delete-session="chat.deleteSession"
>
<template #trigger>
<slot name="trigger" />
</template>
</SessionsDialog>
</template>
@@ -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')
})
@@ -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<void>((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 () => {
+35 -11
View File
@@ -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<string>()
const activeStreamingMessage = shallowRef<StreamingAssistantMessage>()
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<ChatProvider>(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<ChatSendResult> {
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<void> {
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<void> {
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,
},
})
@@ -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<any[]>>()
const dropOutboxForSessionMock = vi.fn<(uid: string, id: string) => Promise<void>>()
const getTombstonesMock = vi.fn<(uid: string) => Promise<string[]>>()
const removeTombstonesMock = vi.fn<(uid: string, ids: string[]) => Promise<void>>()
const addTombstoneMock = vi.fn<(uid: string, id: string) => Promise<void>>()
const deleteCloudChatMock = vi.fn<(id: string) => Promise<void>>()
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<typeof import('pinia')>('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<typeof createPinia>
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')
})
})
+205 -103
View File
@@ -76,6 +76,7 @@ export const useChatSessionStore = defineStore('chat-session', () => {
const sessionMessages = ref<Record<string, ChatHistoryItem[]>>({})
const sessionMetas = ref<Record<string, ChatSessionMeta>>({})
const sessionGenerations = ref<Record<string, number>>({})
/** Authority-owned session index replicated so each window can derive its local selection. */
const index = ref<ChatSessionsIndex | null>(null)
const ready = ref(false)
@@ -92,7 +93,9 @@ export const useChatSessionStore = defineStore('chat-session', () => {
let persistQueue = Promise.resolve()
const loadedSessions = new Set<string>()
const loadingSessions = new Map<string, Promise<void>>()
const staleSessions = new Set<string>()
const cloudHydratedSessions = new Set<string>()
const loadingSessions = new Map<string, Promise<boolean>>()
// 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<void> | 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<void> {
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<boolean> {
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<boolean> {
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<void> {
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,
},
})
@@ -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<string, unknown> = {}) {
}
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()
})
@@ -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<SparkNotifyBridgeMessage, SparkNotifyBridgeMessage>({ name: SPARK_NOTIFY_BRIDGE_CHANNEL_NAME })
const disposeHookFns = ref<Array<() => 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<typeof createContextChannel> | 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<typeof remoteStreamGuard.value>) {
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,
}
})
+1 -1
View File
@@ -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',