nanakura/moeka-project
1
0
Fork 0
Code Issues Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

fix(chat): serialize sends and reset on cleanup (#827)

Browse Source
This commit is contained in:
s3d-i
2025-12-25 02:16:09 +08:00
committed by GitHub
parent d68289a532
commit 26fadcb2fb
7 changed files with 216 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/stage-ui/src/components/scenarios/chat/ChatHistory.vue
+20 -13
View File
@@ -11,7 +11,7 @@ import ChatUserItem from './ChatUserItem.vue'
const props = withDefaults(defineProps<{
messages: ChatHistoryMessage[]
streamingMessage?: ChatAssistantMessage
streamingMessage?: ChatAssistantMessage & { context?: { ts?: number } }
sending?: boolean
assistantLabel?: string
userLabel?: string
@@ -46,18 +46,33 @@ watch([() => props.messages, () => props.streamingMessage], scrollToBottom, { de
watch(() => props.sending, scrollToBottom, { flush: 'post' })
onMounted(scrollToBottom)
const streaming = computed<ChatAssistantMessage>(() => props.streamingMessage ?? { role: 'assistant', content: '', slices: [], tool_results: [] })
const streaming = computed<ChatAssistantMessage & { context?: { ts?: number } }>(() => props.streamingMessage ?? { role: 'assistant', content: '', slices: [], tool_results: [] })
const showStreamingPlaceholder = computed(() => (streaming.value.slices?.length ?? 0) === 0 && !streaming.value.content)
const streamingTs = computed(() => streaming.value.context?.ts)
const renderMessages = computed<ChatHistoryMessage[]>(() => {
if (!props.sending)
return props.messages
const streamTs = streamingTs.value
if (!streamTs)
return props.messages
const hasStreamAlready = streamTs && props.messages.some(msg => msg.context?.ts === streamTs)
if (hasStreamAlready)
return props.messages
return [...props.messages, streaming.value]
})
</script>
<template>
<div ref="chatHistoryRef" v-auto-animate flex="~ col" relative h-full w-full overflow-y-auto rounded-xl px="<sm:2" py="<sm:2" :class="variant === 'mobile' ? 'gap-1' : 'gap-2'">
<template v-for="(message, index) in messages" :key="index">
<template v-for="(message, index) in renderMessages" :key="message.context?.ts ?? index">
<div v-if="message.role === 'error'">
<ChatErrorItem
:message="message"
:label="labels.error"
:show-placeholder="sending && index === messages.length - 1"
:show-placeholder="sending && index === renderMessages.length - 1"
:variant="variant"
/>
</div>
@@ -66,6 +81,7 @@ const showStreamingPlaceholder = computed(() => (streaming.value.slices?.length
<ChatAssistantItem
:message="message"
:label="labels.assistant"
:show-placeholder="message.context?.ts === streamingTs ? showStreamingPlaceholder : false"
:variant="variant"
/>
</div>
@@ -78,14 +94,5 @@ const showStreamingPlaceholder = computed(() => (streaming.value.slices?.length
/>
</div>
</template>
<div v-if="sending">
<ChatAssistantItem
:message="streaming"
:label="labels.assistant"
:show-placeholder="showStreamingPlaceholder"
:variant="variant"
/>
</div>
</div>
</template>
packages/stage-ui/src/components/scenarios/chat/types.ts
+4
View File
@@ -7,4 +7,8 @@ export interface ChatErrorMessage {
export type ChatHistoryMessage = (ChatMessage | ChatErrorMessage) & {
slices?: ChatSlices[]
context?: {
ts?: number
[key: string]: unknown
}
}
packages/stage-ui/src/stores/chat.ts
+148 -25
View File
@@ -27,8 +27,7 @@ interface MessageContext {
meta?: Record<string, unknown>
}
type ChatEntry = (ChatMessage | ErrorMessage) & { context?: MessageContext }
export type { ChatEntry }
export type ChatEntry = (ChatMessage | ErrorMessage) & { context?: MessageContext }
export interface ContextPayload {
content?: unknown
@@ -52,6 +51,8 @@ const ACTIVE_SESSION_STORAGE_KEY = 'chat/active-session'
export const CONTEXT_CHANNEL_NAME = 'airi-context-update'
export const CHAT_STREAM_CHANNEL_NAME = 'airi-chat-stream'
type StreamingAssistantMessage = ChatAssistantMessage & { context?: MessageContext }
export const useChatStore = defineStore('chat', () => {
const { stream, discoverToolsCompatibility } = useLLM()
const { systemPrompt } = storeToRefs(useAiriCardStore())
@@ -60,6 +61,62 @@ export const useChatStore = defineStore('chat', () => {
const sessionMessages = useLocalStorage<Record<string, ChatEntry[]>>(CHAT_STORAGE_KEY, {})
const sending = ref(false)
const streamingMessage = ref<StreamingAssistantMessage>({ role: 'assistant', content: '', slices: [], tool_results: [] })
const sessionGenerations = ref<Record<string, number>>({})
interface SendOptions {
model: string
chatProvider: ChatProvider
providerConfig?: Record<string, unknown>
attachments?: { type: 'image', data: string, mimeType: string }[]
tools?: StreamOptions['tools']
}
interface QueuedSend {
sendingMessage: string
options: SendOptions
generation: number
sessionId: string
cancelled?: boolean
deferred: {
resolve: () => void
reject: (error: unknown) => void
}
}
const pendingQueuedSends = ref<QueuedSend[]>([])
const sendQueue = createQueue<QueuedSend>({
handlers: [
async ({ data }) => {
const { sendingMessage, options, generation, deferred, sessionId, cancelled } = data
if (cancelled)
return
if (getSessionGeneration(sessionId) !== generation) {
deferred.reject(new Error('Chat session was reset before send could start'))
return
}
try {
await performSend(sendingMessage, options, generation, sessionId)
deferred.resolve()
}
catch (error) {
deferred.reject(error)
}
},
],
})
sendQueue.on('enqueue', (queuedSend) => {
pendingQueuedSends.value = [...pendingQueuedSends.value, queuedSend]
})
sendQueue.on('dequeue', (queuedSend) => {
pendingQueuedSends.value = pendingQueuedSends.value.filter(item => item !== queuedSend)
})
// ----- Hooks (UI callbacks) -----
const onBeforeMessageComposedHooks = ref<Array<(message: string) => Promise<void>>>([])
@@ -177,6 +234,22 @@ export const useChatStore = defineStore('chat', () => {
const codeBlockSystemPrompt = '- For any programming code block, always specify the programming language that supported on @shikijs/rehype on the rendered markdown, eg. ```python ... ```\n'
const mathSyntaxSystemPrompt = '- For any math equation, use LaTeX format, eg: $ x^3 $, always escape dollar sign outside math equation\n'
function ensureSessionGeneration(sessionId: string) {
if (sessionGenerations.value[sessionId] === undefined)
sessionGenerations.value = { ...sessionGenerations.value, [sessionId]: 0 }
}
function getSessionGeneration(sessionId: string) {
ensureSessionGeneration(sessionId)
return sessionGenerations.value[sessionId] ?? 0
}
function bumpSessionGeneration(sessionId: string) {
const nextGeneration = getSessionGeneration(sessionId) + 1
sessionGenerations.value = { ...sessionGenerations.value, [sessionId]: nextGeneration }
return nextGeneration
}
function generateInitialMessage() {
// TODO: compose, replace {{ user }} tag, etc
return {
@@ -186,6 +259,7 @@ export const useChatStore = defineStore('chat', () => {
}
function ensureSession(sessionId: string) {
ensureSessionGeneration(sessionId)
if (!sessionMessages.value[sessionId] || sessionMessages.value[sessionId].length === 0) {
sessionMessages.value[sessionId] = [{
...generateInitialMessage(),
@@ -200,6 +274,11 @@ export const useChatStore = defineStore('chat', () => {
ensureSession(activeSessionId.value)
function getSessionMessagesById(sessionId: string) {
ensureSession(sessionId)
return sessionMessages.value[sessionId]!
}
const messages = computed<ChatEntry[]>({
get: () => {
ensureSession(activeSessionId.value)
@@ -216,6 +295,7 @@ export const useChatStore = defineStore('chat', () => {
}
function cleanupMessages(sessionId = activeSessionId.value) {
bumpSessionGeneration(sessionId)
sessionMessages.value[sessionId] = [{
...generateInitialMessage(),
context: {
@@ -224,6 +304,17 @@ export const useChatStore = defineStore('chat', () => {
ts: Date.now(),
},
}]
// Reject pending sends for this session so callers don't hang after cleanup
for (const queued of pendingQueuedSends.value) {
if (queued.sessionId !== sessionId)
continue
queued.cancelled = true
queued.deferred.reject(new Error('Chat session was reset before send could start'))
}
pendingQueuedSends.value = pendingQueuedSends.value.filter(item => item.sessionId !== sessionId)
sending.value = false
streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [] }
}
function getAllSessions() {
@@ -232,6 +323,7 @@ export const useChatStore = defineStore('chat', () => {
function replaceSessions(sessions: Record<string, ChatEntry[]>) {
sessionMessages.value = sessions
sessionGenerations.value = Object.fromEntries(Object.keys(sessions).map(sessionId => [sessionId, 0]))
const [firstSessionId] = Object.keys(sessions)
if (!sessionMessages.value[activeSessionId.value] && firstSessionId)
activeSessionId.value = firstSessionId
@@ -241,6 +333,7 @@ export const useChatStore = defineStore('chat', () => {
function resetAllSessions() {
sessionMessages.value = {}
sessionGenerations.value = {}
activeSessionId.value = 'default'
ensureSession(activeSessionId.value)
}
@@ -319,21 +412,28 @@ export const useChatStore = defineStore('chat', () => {
}
// ----- Send flow (user -> LLM -> assistant) -----
const streamingMessage = ref<ChatAssistantMessage>({ role: 'assistant', content: '', slices: [], tool_results: [] })
async function send(
async function performSend(
sendingMessage: string,
options: {
model: string
chatProvider: ChatProvider
providerConfig?: Record<string, unknown>
attachments?: { type: 'image', data: string, mimeType: string }[]
tools?: StreamOptions['tools']
},
options: SendOptions,
generation: number,
sessionId: string,
) {
if (!sendingMessage && !options.attachments?.length)
return
ensureSession(sessionId)
const isStaleGeneration = () => getSessionGeneration(sessionId) !== generation
const shouldAbort = () => isStaleGeneration()
if (shouldAbort())
return
sending.value = true
const assistantContext: MessageContext = {
sessionId,
source: 'llm',
ts: Date.now(),
}
streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [], context: assistantContext }
try {
await emitBeforeMessageComposedHooks(sendingMessage)
@@ -355,8 +455,12 @@ export const useChatStore = defineStore('chat', () => {
const finalContent = contentParts.length > 1 ? contentParts : sendingMessage
const userContext: MessageContext = { sessionId: activeSessionId.value, source: 'text', ts: Date.now() }
messages.value.push({ role: 'user', content: finalContent, context: userContext })
if (shouldAbort())
return
const sessionMessagesForSend = getSessionMessagesById(sessionId)
const userContext: MessageContext = { sessionId, source: 'text', ts: Date.now() }
sessionMessagesForSend.push({ role: 'user', content: finalContent, context: userContext })
publishContextMessage({
sessionId: userContext.sessionId,
@@ -368,6 +472,8 @@ export const useChatStore = defineStore('chat', () => {
const parser = useLlmmarkerParser({
onLiteral: async (literal) => {
if (shouldAbort())
return
await emitTokenLiteralHooks(literal)
streamingMessage.value.content += literal
@@ -385,6 +491,8 @@ export const useChatStore = defineStore('chat', () => {
})
},
onSpecial: async (special) => {
if (shouldAbort())
return
await emitTokenSpecialHooks(special)
},
minLiteralEmitLength: 24, // Avoid emitting literals too fast. This is a magic number and can be changed later.
@@ -393,6 +501,8 @@ export const useChatStore = defineStore('chat', () => {
const toolCallQueue = createQueue<ChatSlices>({
handlers: [
async (ctx) => {
if (shouldAbort())
return
if (ctx.data.type === 'tool-call') {
streamingMessage.value.slices.push(ctx.data)
return
@@ -405,9 +515,7 @@ export const useChatStore = defineStore('chat', () => {
],
})
streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [] }
const newMessages = messages.value.map((msg) => {
const newMessages = sessionMessagesForSend.map((msg) => {
const { context: _context, ...withoutContext } = msg
const rawMessage = toRaw(withoutContext)
if (rawMessage.role === 'assistant') {
@@ -427,6 +535,9 @@ export const useChatStore = defineStore('chat', () => {
let fullText = ''
const headers = (options.providerConfig?.headers || {}) as Record<string, string>
if (shouldAbort())
return
await stream(options.model, options.chatProvider, newMessages as Message[], {
headers,
tools: options.tools,
@@ -461,19 +572,13 @@ export const useChatStore = defineStore('chat', () => {
await parser.end()
// Add the completed message to the history only if it has content
if (streamingMessage.value.slices.length > 0) {
const assistantContext: MessageContext = {
sessionId: activeSessionId.value,
source: 'llm',
ts: Date.now(),
}
if (!isStaleGeneration() && streamingMessage.value.slices.length > 0) {
const assistantMessage: ChatEntry = {
...(toRaw(streamingMessage.value) as ChatAssistantMessage),
context: assistantContext,
}
messages.value.push(assistantMessage)
sessionMessagesForSend.push(assistantMessage)
publishContextMessage({
sessionId: assistantContext.sessionId,
@@ -515,6 +620,24 @@ export const useChatStore = defineStore('chat', () => {
}
}
async function send(
sendingMessage: string,
options: SendOptions,
) {
const sessionId = activeSessionId.value
const generation = getSessionGeneration(sessionId)
return new Promise<void>((resolve, reject) => {
sendQueue.enqueue({
sendingMessage,
options,
generation,
sessionId,
deferred: { resolve, reject },
})
})
}
return {
sending,
activeSessionId,