diff --git a/packages/stage-ui/src/stores/character/orchestrator/store.ts b/packages/stage-ui/src/stores/character/orchestrator/store.ts index a2116d228..d87109e71 100644 --- a/packages/stage-ui/src/stores/character/orchestrator/store.ts +++ b/packages/stage-ui/src/stores/character/orchestrator/store.ts @@ -38,6 +38,8 @@ export const useCharacterOrchestratorStore = defineStore('character-orchestrator maxAttempts: 3, }) let tickTimer: ReturnType | undefined + let initialized = false + const eventUnsubscribes: Array<() => void> = [] const sparkNotifyAgent = setupAgentSparkNotifyHandler({ stream, getActiveProvider: () => activeProvider.value, @@ -76,17 +78,17 @@ export const useCharacterOrchestratorStore = defineStore('character-orchestrator function enqueueSparkNotify(event: WebSocketEventOf<'spark:notify'>, options?: { reason?: string, nextRunAt?: number, maxAttempts?: number }) { if (!pendingNotifies.value.some(item => item.data.id === event.data.id)) { - pendingNotifies.value = [...pendingNotifies.value, event] + pendingNotifies.value.push(event) } - scheduledNotifies.value = [...scheduledNotifies.value, { + scheduledNotifies.value.push({ event, enqueuedAt: Date.now(), nextRunAt: options?.nextRunAt ?? computeNextRunAt(event, 0), attempts: 0, maxAttempts: options?.maxAttempts ?? attentionConfig.value.maxAttempts, reason: options?.reason, - }] + }) } async function processSparkNotify(event: WebSocketEventOf<'spark:notify'>) { @@ -198,27 +200,47 @@ export const useCharacterOrchestratorStore = defineStore('character-orchestrator } function initialize() { - modsServerChannelStore.onEvent('spark:notify', async (event) => { - try { - await handleIncomingSparkNotify(event) - } - catch (error) { - console.warn('Failed to handle spark:notify event:', error) - } - }) + if (initialized) + return - modsServerChannelStore.onEvent('spark:emit', async (event) => { - try { - await handleSparkEmit(event) - } - catch (error) { - console.warn('Failed to handle spark:emit event:', error) - } - }) + initialized = true + + eventUnsubscribes.push( + modsServerChannelStore.onEvent('spark:notify', async (event) => { + try { + await handleIncomingSparkNotify(event) + } + catch (error) { + console.warn('Failed to handle spark:notify event:', error) + } + }), + ) + + eventUnsubscribes.push( + modsServerChannelStore.onEvent('spark:emit', async (event) => { + try { + await handleSparkEmit(event) + } + catch (error) { + console.warn('Failed to handle spark:emit event:', error) + } + }), + ) startTicker() } + function dispose() { + stopTicker() + + for (const unsubscribe of eventUnsubscribes) { + unsubscribe() + } + + eventUnsubscribes.length = 0 + initialized = false + } + return { processing, pendingNotifies, @@ -228,6 +250,7 @@ export const useCharacterOrchestratorStore = defineStore('character-orchestrator initialize, startTicker, stopTicker, + dispose, handleSparkNotify: handleIncomingSparkNotify, handleSparkEmit, diff --git a/packages/stage-ui/src/stores/chat.ts b/packages/stage-ui/src/stores/chat.ts index 5b2ed956d..6f1f7a0df 100644 --- a/packages/stage-ui/src/stores/chat.ts +++ b/packages/stage-ui/src/stores/chat.ts @@ -8,7 +8,7 @@ import type { StreamEvent, StreamOptions } from './llm' import { createQueue } from '@proj-airi/stream-kit' import { nanoid } from 'nanoid' import { defineStore, storeToRefs } from 'pinia' -import { ref, toRaw } from 'vue' +import { computed, ref, toRaw } from 'vue' import { useAnalytics } from '../composables' import { useLlmmarkerParser } from '../composables/llm-marker-parser' @@ -23,6 +23,15 @@ import { useContextObservabilityStore } from './devtools/context-observability' import { useLLM } from './llm' import { useConsciousnessStore } from './modules/consciousness' +function cloneStreamingMessage(message: StreamingAssistantMessage): StreamingAssistantMessage { + try { + return structuredClone(message) + } + catch { + return JSON.parse(JSON.stringify(message)) as StreamingAssistantMessage + } +} + interface SendOptions { model: string chatProvider: ChatProvider @@ -75,7 +84,7 @@ export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => { const sending = ref(false) const pendingQueuedSends = ref([]) - const pendingQueuedSendCount = ref(0) + const pendingQueuedSendCount = computed(() => pendingQueuedSends.value.length) const hooks = createChatHooks() const sendQueue = createQueue({ @@ -103,13 +112,11 @@ export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => { }) sendQueue.on('enqueue', (queuedSend) => { - pendingQueuedSends.value = [...pendingQueuedSends.value, queuedSend] - pendingQueuedSendCount.value = pendingQueuedSends.value.length + pendingQueuedSends.value.push(queuedSend) }) sendQueue.on('dequeue', (queuedSend) => { pendingQueuedSends.value = pendingQueuedSends.value.filter(item => item !== queuedSend) - pendingQueuedSendCount.value = pendingQueuedSends.value.length }) async function performSend( @@ -162,7 +169,7 @@ export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => { const updateUI = () => { if (isForegroundSession()) { - streamingMessage.value = JSON.parse(JSON.stringify(buildingMessage)) + streamingMessage.value = cloneStreamingMessage(buildingMessage) } } @@ -476,7 +483,6 @@ export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => { pendingQueuedSends.value = sessionId ? pendingQueuedSends.value.filter(item => item.sessionId !== sessionId) : [] - pendingQueuedSendCount.value = pendingQueuedSends.value.length } function getPendingQueuedSendSnapshot() { diff --git a/packages/stage-ui/src/stores/chat/data-store.ts b/packages/stage-ui/src/stores/chat/data-store.ts index 1a9114013..f1486f001 100644 --- a/packages/stage-ui/src/stores/chat/data-store.ts +++ b/packages/stage-ui/src/stores/chat/data-store.ts @@ -28,6 +28,15 @@ export interface ChatDataStore { } export function createChatDataStore(access: ChatDataAccess): ChatDataStore { + function cloneDeep(value: T): T { + try { + return structuredClone(value) + } + catch { + return JSON.parse(JSON.stringify(value)) as T + } + } + function ensureGeneration(sessionId: string) { const generations = access.getGenerations() if (generations[sessionId] === undefined) @@ -118,7 +127,7 @@ export function createChatDataStore(access: ChatDataAccess): ChatDataStore { } function getAllSessions() { - return JSON.parse(JSON.stringify(access.getSessions())) as Record + return cloneDeep(access.getSessions()) } function getSessionGenerationValue(sessionId?: string) { diff --git a/packages/stage-ui/src/stores/chat/session-store.ts b/packages/stage-ui/src/stores/chat/session-store.ts index 97c6a2f29..e5e796e07 100644 --- a/packages/stage-ui/src/stores/chat/session-store.ts +++ b/packages/stage-ui/src/stores/chat/session-store.ts @@ -46,8 +46,17 @@ export const useChatSessionStore = defineStore('chat-session', () => { return persistQueue } + function cloneDeep(value: T): T { + try { + return structuredClone(value) + } + catch { + return JSON.parse(JSON.stringify(value)) as T + } + } + function snapshotMessages(messages: ChatHistoryItem[]) { - return JSON.parse(JSON.stringify(messages)) as ChatHistoryItem[] + return cloneDeep(messages) } function ensureSessionMessageIds(sessionId: string) { @@ -106,7 +115,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { async function persistIndex() { if (!index.value) return - const snapshot = JSON.parse(JSON.stringify(index.value)) as ChatSessionsIndex + const snapshot = cloneDeep(index.value) await enqueuePersist(() => chatSessionsRepo.saveIndex(snapshot)) } @@ -136,7 +145,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { await chatSessionsRepo.saveSession(sessionId, record) if (index.value) { - const snapshot = JSON.parse(JSON.stringify(index.value)) as ChatSessionsIndex + const snapshot = cloneDeep(index.value) await chatSessionsRepo.saveIndex(snapshot) } }) @@ -208,7 +217,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { updatedAt: now, } - const initialMessages = options?.messages?.length ? options.messages : [generateInitialMessage()] + const initialMessages = options?.messages?.length ? cloneDeep(options.messages) : [generateInitialMessage()] sessionMetas.value[sessionId] = meta replaceSessionMessages(sessionId, initialMessages, { persist: false }) @@ -302,7 +311,6 @@ export const useChatSessionStore = defineStore('chat-session', () => { if (!loadedSessions.has(activeSessionId.value) && !sessionMessages.value[activeSessionId.value] && hasKnownSession(activeSessionId.value)) { return [] } - ensureSession(activeSessionId.value) return sessionMessages.value[activeSessionId.value] ?? [] }, set: (value) => { @@ -334,10 +342,14 @@ export const useChatSessionStore = defineStore('chat-session', () => { activeSessionId: string sessionMessages: Record sessionMetas: Record + index?: ChatSessionsIndex | null }) { activeSessionId.value = snapshot.activeSessionId - sessionMessages.value = snapshot.sessionMessages - sessionMetas.value = snapshot.sessionMetas + sessionMessages.value = cloneDeep(snapshot.sessionMessages) + sessionMetas.value = cloneDeep(snapshot.sessionMetas) + if (snapshot.index !== undefined) { + index.value = cloneDeep(snapshot.index) + } sessionGenerations.value = Object.fromEntries( Object.keys(snapshot.sessionMessages).map(sessionId => [sessionId, sessionGenerations.value[sessionId] ?? 0]), ) @@ -350,8 +362,9 @@ export const useChatSessionStore = defineStore('chat-session', () => { function getSnapshot() { return { activeSessionId: activeSessionId.value, - sessionMessages: JSON.parse(JSON.stringify(sessionMessages.value)) as Record, - sessionMetas: JSON.parse(JSON.stringify(sessionMetas.value)) as Record, + sessionMessages: cloneDeep(sessionMessages.value), + sessionMetas: cloneDeep(sessionMetas.value), + index: cloneDeep(index.value), } } @@ -362,7 +375,7 @@ export const useChatSessionStore = defineStore('chat-session', () => { } function getAllSessions() { - return JSON.parse(JSON.stringify(sessionMessages.value)) as Record + return cloneDeep(sessionMessages.value) } async function resetAllSessions() { @@ -453,8 +466,8 @@ export const useChatSessionStore = defineStore('chat-session', () => { return { format: 'chat-sessions-index:v1', - index: index.value, - sessions, + index: cloneDeep(index.value), + sessions: cloneDeep(sessions), } } @@ -462,20 +475,23 @@ export const useChatSessionStore = defineStore('chat-session', () => { if (payload.format !== 'chat-sessions-index:v1') return - index.value = payload.index + index.value = cloneDeep(payload.index) sessionMessages.value = {} sessionMetas.value = {} sessionGenerations.value = {} loadedSessions.clear() loadingSessions.clear() - await enqueuePersist(() => chatSessionsRepo.saveIndex(payload.index)) + await enqueuePersist(() => chatSessionsRepo.saveIndex(cloneDeep(payload.index))) for (const [sessionId, record] of Object.entries(payload.sessions)) { - sessionMetas.value[sessionId] = record.meta - sessionMessages.value[sessionId] = record.messages + sessionMetas.value[sessionId] = cloneDeep(record.meta) + sessionMessages.value[sessionId] = cloneDeep(record.messages) ensureGeneration(sessionId) - await enqueuePersist(() => chatSessionsRepo.saveSession(sessionId, record)) + await enqueuePersist(() => chatSessionsRepo.saveSession(sessionId, { + meta: cloneDeep(record.meta), + messages: cloneDeep(record.messages), + })) } await ensureActiveSessionForCharacter() diff --git a/packages/stage-ui/src/stores/chat/stream-store.ts b/packages/stage-ui/src/stores/chat/stream-store.ts index 9aef1203f..24952e6c3 100644 --- a/packages/stage-ui/src/stores/chat/stream-store.ts +++ b/packages/stage-ui/src/stores/chat/stream-store.ts @@ -1,7 +1,7 @@ import type { StreamingAssistantMessage } from '../../types/chat' import { defineStore } from 'pinia' -import { ref } from 'vue' +import { ref, toRaw } from 'vue' import { useChatSessionStore } from './session-store' @@ -31,7 +31,7 @@ export const useChatStreamStore = defineStore('chat-stream', () => { function finalizeStream(fullText?: string) { const sessionId = chatSession.activeSessionId if (streamingMessage.value.slices.length > 0) - chatSession.appendSessionMessage(sessionId, streamingMessage.value) + chatSession.appendSessionMessage(sessionId, toRaw(streamingMessage.value)) streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [] } if (fullText) streamingMessage.value.content = fullText diff --git a/packages/stage-ui/src/stores/devtools/context-observability.ts b/packages/stage-ui/src/stores/devtools/context-observability.ts index a17486eca..a69cc1e81 100644 --- a/packages/stage-ui/src/stores/devtools/context-observability.ts +++ b/packages/stage-ui/src/stores/devtools/context-observability.ts @@ -55,7 +55,12 @@ function truncateText(value: string, limit = 220) { } function cloneValue(value: T): T { - return JSON.parse(JSON.stringify(value)) as T + try { + return structuredClone(value) + } + catch { + return JSON.parse(JSON.stringify(value)) as T + } } export const useContextObservabilityStore = defineStore('devtools:context-observability', () => { diff --git a/packages/stage-ui/src/stores/mods/api/context-bridge.ts b/packages/stage-ui/src/stores/mods/api/context-bridge.ts index 081f222c6..264c6ec54 100644 --- a/packages/stage-ui/src/stores/mods/api/context-bridge.ts +++ b/packages/stage-ui/src/stores/mods/api/context-bridge.ts @@ -58,11 +58,15 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = const disposeHookFns = ref void>>([]) let remoteStreamGuard: { sessionId: string, generation: number } | null = null + let initialized = false async function initialize() { await mutex.acquire() try { + if (initialized) + return + const registerConsumers = () => { for (const consumerEvent of consumerRegistrationEvents) { serverChannelStore.send({ @@ -173,6 +177,13 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = }) })) + function withContextBridgeLock(key: string, callback: () => Promise) { + if (typeof navigator !== 'undefined' && 'locks' in navigator && typeof navigator.locks.request === 'function') { + return navigator.locks.request(key, callback) + } + return callback() + } + disposeHookFns.value.push(serverChannelStore.onEvent('input:text', async (event) => { const { text, @@ -275,7 +286,7 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = // - https://chromestatus.com/feature/6265472244514816 // - https://developer.mozilla.org/en-US/docs/Web/API/SharedWorker // - https://developer.mozilla.org/en-US/docs/Web/API/Web_Locks_API - navigator.locks.request('context-bridge:event:input:text', async () => { + await withContextBridgeLock('context-bridge:event:input:text', async () => { try { await chatOrchestrator.ingest(messageText, { model: activeModel.value, @@ -472,6 +483,7 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = } }) disposeHookFns.value.push(stopIncomingStreamWatch) + initialized = true } finally { mutex.release() @@ -482,6 +494,9 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = await mutex.acquire() try { + if (!initialized) + return + for (const consumerEvent of consumerRegistrationEvents) { serverChannelStore.send({ type: 'module:consumer:unregister', @@ -496,6 +511,9 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () = for (const fn of disposeHookFns.value) { fn() } + + initialized = false + remoteStreamGuard = null } finally { mutex.release()