feat: publish v1.0.0-rc.36-custom.1

This commit is contained in:
Codex committed 2026-09-11 05:07:48 +08:00
commit 3a7e8f3b54
2381 files changed
+551494

No files matched your search

@@ -0,0 +1,24 @@
/*
Copyright (C) 2023-2026 QuantumNous
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
For commercial licensing, please contact support@quantumnous.com
*/
export * from './use-playground-state'
export * from './use-stream-request'
export * from './use-chat-handler'
export * from './use-message-action-guard'
export * from './use-playground-conversation'
export * from './use-playground-options'
@@ -0,0 +1,391 @@
/*
Copyright (C) 2023-2026 QuantumNous
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
For commercial licensing, please contact support@quantumnous.com
*/
import { useCallback, useEffect, useRef, useState } from 'react'
import { useTranslation } from 'react-i18next'
import { toast } from 'sonner'
import { sendChatCompletion } from '../api'
import { ERROR_MESSAGES } from '../constants'
import {
applyStreamingChunk,
buildChatCompletionPayload,
updateAssistantMessageWithError,
updateLastAssistantMessage,
parseRequestErrorDetails,
applyChatCompletionResponse,
completeAssistantMessage,
hasChatCompletionChoice,
isAssistantMessageFinal,
isAssistantMessagePending,
} from '../lib'
import type { Message, PlaygroundConfig, ParameterEnabled } from '../types'
import { useStreamRequest } from './use-stream-request'
interface UseChatHandlerOptions {
config: PlaygroundConfig
parameterEnabled: ParameterEnabled
onMessageUpdate: (updater: (prev: Message[]) => Message[]) => void
}
const KNOWN_ERROR_MESSAGES = new Set<string>(Object.values(ERROR_MESSAGES))
const STREAM_UPDATE_FLUSH_MS = 50
type PendingStreamChunks = {
generation: number
content: string
reasoning: string
}
function mergePendingStreamChunk(
currentChunk: string,
nextChunk: string
): string {
if (!currentChunk || !nextChunk.startsWith(currentChunk)) {
return currentChunk + nextChunk
}
return nextChunk
}
/**
* Hook for handling chat message sending and receiving
*/
export function useChatHandler({
config,
parameterEnabled,
onMessageUpdate,
}: UseChatHandlerOptions) {
const { t } = useTranslation()
const { sendStreamRequest, stopStream, isStreaming } = useStreamRequest()
const [isRequesting, setIsRequesting] = useState(false)
const abortControllerRef = useRef<AbortController | null>(null)
const requestGenerationRef = useRef(0)
const pendingStreamChunksRef = useRef<PendingStreamChunks>({
generation: 0,
content: '',
reasoning: '',
})
const streamFlushTimerRef = useRef<number | null>(null)
const discardPendingStreamUpdates = useCallback((generation: number) => {
if (streamFlushTimerRef.current !== null) {
window.clearTimeout(streamFlushTimerRef.current)
streamFlushTimerRef.current = null
}
pendingStreamChunksRef.current = {
generation,
content: '',
reasoning: '',
}
}, [])
const flushStreamUpdates = useCallback(
(generation: number) => {
if (generation !== requestGenerationRef.current) return
if (streamFlushTimerRef.current !== null) {
window.clearTimeout(streamFlushTimerRef.current)
streamFlushTimerRef.current = null
}
const pendingChunks = pendingStreamChunksRef.current
if (pendingChunks.generation !== generation) return
if (!pendingChunks.reasoning && !pendingChunks.content) {
return
}
pendingStreamChunksRef.current = {
generation,
content: '',
reasoning: '',
}
onMessageUpdate((prev) => {
if (generation !== requestGenerationRef.current) return prev
return updateLastAssistantMessage(prev, (message) => {
let updatedMessage = message
if (pendingChunks.reasoning) {
updatedMessage = applyStreamingChunk(
updatedMessage,
'reasoning',
pendingChunks.reasoning
)
}
if (pendingChunks.content) {
updatedMessage = applyStreamingChunk(
updatedMessage,
'content',
pendingChunks.content
)
}
return updatedMessage
})
})
},
[onMessageUpdate]
)
const scheduleStreamFlush = useCallback(
(generation: number) => {
if (generation !== requestGenerationRef.current) return
if (streamFlushTimerRef.current !== null) {
return
}
streamFlushTimerRef.current = window.setTimeout(() => {
flushStreamUpdates(generation)
}, STREAM_UPDATE_FLUSH_MS)
},
[flushStreamUpdates]
)
useEffect(
() => () => {
requestGenerationRef.current += 1
if (streamFlushTimerRef.current !== null) {
window.clearTimeout(streamFlushTimerRef.current)
}
abortControllerRef.current?.abort()
abortControllerRef.current = null
},
[]
)
const getDisplayError = useCallback(
(error: string) => {
if (KNOWN_ERROR_MESSAGES.has(error)) {
return t(error)
}
const connectionClosedSuffix = `: ${ERROR_MESSAGES.CONNECTION_CLOSED}`
if (error.endsWith(connectionClosedSuffix)) {
return `${error.slice(0, -ERROR_MESSAGES.CONNECTION_CLOSED.length)}${t(
ERROR_MESSAGES.CONNECTION_CLOSED
)}`
}
return error
},
[t]
)
// Handle stream update
const handleStreamUpdate = useCallback(
(generation: number, type: 'reasoning' | 'content', chunk: string) => {
if (generation !== requestGenerationRef.current) return
if (pendingStreamChunksRef.current.generation !== generation) return
pendingStreamChunksRef.current[type] = mergePendingStreamChunk(
pendingStreamChunksRef.current[type],
chunk
)
scheduleStreamFlush(generation)
},
[scheduleStreamFlush]
)
// Handle stream complete
const handleStreamComplete = useCallback(
(generation: number) => {
if (generation !== requestGenerationRef.current) return
flushStreamUpdates(generation)
setIsRequesting(false)
onMessageUpdate((prev) => {
if (generation !== requestGenerationRef.current) return prev
return updateLastAssistantMessage(prev, (message) =>
isAssistantMessageFinal(message)
? message
: completeAssistantMessage(message)
)
})
},
[flushStreamUpdates, onMessageUpdate]
)
// Handle stream error
const handleStreamError = useCallback(
(generation: number, error: string, errorCode?: string) => {
if (generation !== requestGenerationRef.current) return
flushStreamUpdates(generation)
setIsRequesting(false)
const displayError = getDisplayError(error)
toast.error(displayError)
const errorTitle = t(ERROR_MESSAGES.API_REQUEST_ERROR)
onMessageUpdate((prev) => {
if (generation !== requestGenerationRef.current) return prev
return updateAssistantMessageWithError(
prev,
displayError,
errorCode,
errorTitle
)
})
},
[flushStreamUpdates, getDisplayError, onMessageUpdate, t]
)
// Send streaming chat request
const sendStreamingChat = useCallback(
(messages: Message[]) => {
const generation = requestGenerationRef.current + 1
requestGenerationRef.current = generation
abortControllerRef.current?.abort()
abortControllerRef.current = null
discardPendingStreamUpdates(generation)
setIsRequesting(true)
const payload = buildChatCompletionPayload(
messages,
config,
parameterEnabled
)
void sendStreamRequest(
payload,
(type, chunk) => handleStreamUpdate(generation, type, chunk),
() => handleStreamComplete(generation),
(error, errorCode) => handleStreamError(generation, error, errorCode)
)
},
[
config,
parameterEnabled,
sendStreamRequest,
discardPendingStreamUpdates,
handleStreamUpdate,
handleStreamComplete,
handleStreamError,
]
)
// Send non-streaming chat request
const sendNonStreamingChat = useCallback(
async (messages: Message[]) => {
const payload = buildChatCompletionPayload(
messages,
config,
parameterEnabled
)
const generation = requestGenerationRef.current + 1
const abortController = new AbortController()
requestGenerationRef.current = generation
stopStream()
discardPendingStreamUpdates(generation)
abortControllerRef.current?.abort()
abortControllerRef.current = abortController
try {
setIsRequesting(true)
const response = await sendChatCompletion(
payload,
abortController.signal
)
if (
abortController.signal.aborted ||
requestGenerationRef.current !== generation
) {
return
}
if (!hasChatCompletionChoice(response)) {
handleStreamError(generation, ERROR_MESSAGES.API_REQUEST_ERROR)
return
}
onMessageUpdate((prev) => {
if (requestGenerationRef.current !== generation) return prev
return updateLastAssistantMessage(prev, (message) => {
const updatedMessage = applyChatCompletionResponse(
message,
response
)
return updatedMessage ?? message
})
})
} catch (error: unknown) {
if (
abortController.signal.aborted ||
requestGenerationRef.current !== generation
) {
return
}
const { errorCode, errorMessage } = parseRequestErrorDetails(error)
handleStreamError(generation, errorMessage, errorCode)
} finally {
if (requestGenerationRef.current === generation) {
abortControllerRef.current = null
setIsRequesting(false)
}
}
},
[
config,
parameterEnabled,
stopStream,
discardPendingStreamUpdates,
onMessageUpdate,
handleStreamError,
]
)
// Send chat request (stream or non-stream based on config)
const sendChat = useCallback(
(messages: Message[]) => {
if (config.stream) {
sendStreamingChat(messages)
} else {
sendNonStreamingChat(messages)
}
},
[config.stream, sendStreamingChat, sendNonStreamingChat]
)
// Stop generation
const stopGeneration = useCallback(() => {
const stoppedGeneration = requestGenerationRef.current
flushStreamUpdates(stoppedGeneration)
const idleGeneration = stoppedGeneration + 1
requestGenerationRef.current = idleGeneration
discardPendingStreamUpdates(idleGeneration)
stopStream()
abortControllerRef.current?.abort()
abortControllerRef.current = null
setIsRequesting(false)
onMessageUpdate((prev) => {
if (requestGenerationRef.current !== idleGeneration) return prev
return updateLastAssistantMessage(prev, (message) =>
isAssistantMessagePending(message)
? completeAssistantMessage(message)
: message
)
})
}, [
stopStream,
flushStreamUpdates,
discardPendingStreamUpdates,
onMessageUpdate,
])
return {
sendChat,
stopGeneration,
isGenerating: isStreaming || isRequesting,
}
}
@@ -0,0 +1,45 @@
/*
Copyright (C) 2023-2026 QuantumNous
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
For commercial licensing, please contact support@quantumnous.com
*/
import { useCallback } from 'react'
import { useTranslation } from 'react-i18next'
import { toast } from 'sonner'
import { MESSAGE_ACTION_LABELS } from '../constants'
/**
* Hook to guard message actions when generation is in progress
* Provides a wrapper that checks if generation is active before executing
*/
export function useMessageActionGuard(isGenerating: boolean) {
const { t } = useTranslation()
const guardAction = useCallback(
(action: () => void) => {
return () => {
if (isGenerating) {
toast.warning(t(MESSAGE_ACTION_LABELS.WAIT_GENERATION))
return
}
action()
}
},
[isGenerating, t]
)
return { guardAction }
}
@@ -0,0 +1,116 @@
/*
Copyright (C) 2023-2026 QuantumNous
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
For commercial licensing, please contact support@quantumnous.com
*/
import { useCallback, useState } from 'react'
import {
appendUserMessagePair,
applyMessageEdit,
createRegeneratedMessages,
removeMessageByKey,
} from '../lib'
import type { Message } from '../types'
type UsePlaygroundConversationOptions = {
messages: Message[]
updateMessages: (
updater: Message[] | ((prev: Message[]) => Message[])
) => void
sendChat: (messages: Message[]) => void
}
export function usePlaygroundConversation({
messages,
updateMessages,
sendChat,
}: UsePlaygroundConversationOptions) {
const [editingMessageKey, setEditingMessageKey] = useState<string | null>(
null
)
const handleSendMessage = useCallback(
(text: string) => {
const nextMessages = appendUserMessagePair(messages, text)
updateMessages(nextMessages)
sendChat(nextMessages)
},
[messages, updateMessages, sendChat]
)
const handleRegenerateMessage = useCallback(
(message: Message) => {
const nextMessages = createRegeneratedMessages(messages, message.key)
if (!nextMessages) return
updateMessages(nextMessages)
sendChat(nextMessages)
},
[messages, updateMessages, sendChat]
)
const handleEditMessage = useCallback((message: Message) => {
setEditingMessageKey(message.key)
}, [])
const handleEditOpenChange = useCallback((open: boolean) => {
if (!open) {
setEditingMessageKey(null)
}
}, [])
const applyEdit = useCallback(
(newContent: string, shouldSubmit: boolean) => {
if (!editingMessageKey) return
const editResult = applyMessageEdit(
messages,
editingMessageKey,
newContent,
shouldSubmit
)
if (!editResult) return
setEditingMessageKey(null)
updateMessages(editResult.messages)
if (editResult.shouldSend) {
sendChat(editResult.messages)
}
},
[editingMessageKey, messages, updateMessages, sendChat]
)
const handleDeleteMessage = useCallback(
(message: Message) => {
updateMessages((previousMessages) =>
removeMessageByKey(previousMessages, message.key)
)
},
[updateMessages]
)
return {
editingMessageKey,
handleSendMessage,
handleRegenerateMessage,
handleEditMessage,
handleEditOpenChange,
applyEdit,
handleDeleteMessage,
}
}
@@ -0,0 +1,143 @@
/*
Copyright (C) 2023-2026 QuantumNous
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
For commercial licensing, please contact support@quantumnous.com
*/
import { useQuery } from '@tanstack/react-query'
/*
Copyright (C) 2023-2026 QuantumNous
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
For commercial licensing, please contact support@quantumnous.com
*/
import { useEffect } from 'react'
import { useTranslation } from 'react-i18next'
import { toast } from 'sonner'
import { getUserGroups, getUserModels } from '../api'
import {
getGroupFallback,
getModelFallback,
getOptionLoadErrorMessage,
shouldClearModelForGroup,
} from '../lib'
import type { GroupOption, ModelOption, PlaygroundConfig } from '../types'
type UsePlaygroundOptionsParams = {
currentGroup: string
currentModel: string
setGroups: (groups: GroupOption[]) => void
setModels: (models: ModelOption[]) => void
updateConfig: <K extends keyof PlaygroundConfig>(
key: K,
value: PlaygroundConfig[K]
) => void
}
export function usePlaygroundOptions({
currentGroup,
currentModel,
setGroups,
setModels,
updateConfig,
}: UsePlaygroundOptionsParams) {
const { t } = useTranslation()
const {
data: modelsData,
error: modelsError,
isError: isModelsError,
isLoading: isLoadingModels,
} = useQuery({
queryKey: ['playground-models', currentGroup],
queryFn: () => getUserModels(currentGroup),
enabled: currentGroup !== '',
})
const {
data: groupsData,
error: groupsError,
isError: isGroupsError,
} = useQuery({
queryKey: ['playground-groups'],
queryFn: getUserGroups,
})
useEffect(() => {
if (!isModelsError) return
toast.error(
getOptionLoadErrorMessage(
modelsError,
t('Failed to load playground models')
)
)
}, [isModelsError, modelsError, t])
useEffect(() => {
if (!isGroupsError) return
toast.error(
getOptionLoadErrorMessage(
groupsError,
t('Failed to load playground groups')
)
)
}, [isGroupsError, groupsError, t])
useEffect(() => {
if (!modelsData) return
setModels(modelsData)
const fallback = getModelFallback(modelsData, currentModel)
if (fallback) {
updateConfig('model', fallback)
return
}
if (shouldClearModelForGroup(modelsData, currentModel)) {
updateConfig('model', '')
}
}, [modelsData, currentModel, setModels, updateConfig])
useEffect(() => {
if (!groupsData) return
setGroups(groupsData)
const fallback = getGroupFallback(groupsData, currentGroup)
if (fallback) {
updateConfig('group', fallback)
}
}, [groupsData, currentGroup, setGroups, updateConfig])
return {
isLoadingModels,
}
}
@@ -0,0 +1,180 @@
/*
Copyright (C) 2023-2026 QuantumNous
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
For commercial licensing, please contact support@quantumnous.com
*/
import { useCallback, useEffect, useRef, useState } from 'react'
import { DEFAULT_CONFIG, DEFAULT_PARAMETER_ENABLED } from '../constants'
import {
saveConfig,
saveParameterEnabled,
saveMessages,
applyMessageStateUpdate,
getInitialParameterEnabled,
getInitialPlaygroundConfig,
loadMessages,
type MessageStateUpdater,
} from '../lib'
import type {
Message,
PlaygroundConfig,
ParameterEnabled,
ModelOption,
GroupOption,
} from '../types'
const MESSAGE_SAVE_DEBOUNCE_MS = 500
/**
* Main state management hook for playground
*/
export function usePlaygroundState() {
// Load initial state from localStorage
const [config, setConfig] = useState<PlaygroundConfig>(
getInitialPlaygroundConfig
)
const [parameterEnabled, setParameterEnabled] = useState<ParameterEnabled>(
getInitialParameterEnabled
)
const [messages, setMessages] = useState<Message[]>([])
const [isLoadingMessages, setIsLoadingMessages] = useState(true)
const messagesSaveTimerRef = useRef<number | null>(null)
const latestMessagesRef = useRef<Message[]>(messages)
const hasLoadedMessagesRef = useRef(false)
const [models, setModels] = useState<ModelOption[]>([])
const [groups, setGroups] = useState<GroupOption[]>([])
const persistMessages = useCallback((messagesToSave: Message[]) => {
latestMessagesRef.current = messagesToSave
if (!hasLoadedMessagesRef.current) {
return
}
if (messagesSaveTimerRef.current !== null) {
window.clearTimeout(messagesSaveTimerRef.current)
}
messagesSaveTimerRef.current = window.setTimeout(() => {
messagesSaveTimerRef.current = null
saveMessages(latestMessagesRef.current)
}, MESSAGE_SAVE_DEBOUNCE_MS)
}, [])
useEffect(() => {
let cancelled = false
window.setTimeout(() => {
const loadedMessages = loadMessages() ?? []
if (cancelled) {
return
}
latestMessagesRef.current = loadedMessages
hasLoadedMessagesRef.current = true
setMessages(loadedMessages)
setIsLoadingMessages(false)
}, 0)
return () => {
cancelled = true
}
}, [])
useEffect(
() => () => {
if (messagesSaveTimerRef.current !== null) {
window.clearTimeout(messagesSaveTimerRef.current)
saveMessages(latestMessagesRef.current)
}
},
[]
)
// Update config with automatic save
const updateConfig = useCallback(
<K extends keyof PlaygroundConfig>(key: K, value: PlaygroundConfig[K]) => {
setConfig((prev) => {
const updated = { ...prev, [key]: value }
saveConfig(updated)
return updated
})
},
[]
)
// Update parameter enabled with automatic save
const updateParameterEnabled = useCallback(
(key: keyof ParameterEnabled, value: boolean) => {
setParameterEnabled((prev) => {
const updated = { ...prev, [key]: value }
saveParameterEnabled(updated)
return updated
})
},
[]
)
// Update messages with automatic save
const updateMessages = useCallback(
(updater: MessageStateUpdater) => {
setMessages((prev) => {
const newMessages = applyMessageStateUpdate(prev, updater)
persistMessages(newMessages)
return newMessages
})
},
[persistMessages]
)
// Clear all messages
const clearMessages = useCallback(() => {
updateMessages([])
}, [updateMessages])
// Reset config to defaults
const resetConfig = useCallback(() => {
setConfig(DEFAULT_CONFIG)
setParameterEnabled(DEFAULT_PARAMETER_ENABLED)
saveConfig(DEFAULT_CONFIG)
saveParameterEnabled(DEFAULT_PARAMETER_ENABLED)
}, [])
return {
// State
config,
parameterEnabled,
messages,
isLoadingMessages,
models,
groups,
// Setters
setModels,
setGroups,
// Actions
updateConfig,
updateParameterEnabled,
updateMessages,
clearMessages,
resetConfig,
}
}
@@ -0,0 +1,199 @@
/*
Copyright (C) 2023-2026 QuantumNous
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
For commercial licensing, please contact support@quantumnous.com
*/
import { describe, expect, test } from 'vitest'
import type { ChatCompletionRequest } from '../types'
import { createStreamRequestController } from './use-stream-request'
function deferred<T>() {
let resolve!: (value: T) => void
const promise = new Promise<T>((promiseResolve) => {
resolve = promiseResolve
})
return { promise, resolve }
}
class FakeStreamSource {
readyState = 0
closed = false
streamed = false
private listeners = new Map<
string,
Array<(event: Event & { data?: string; readyState?: number }) => void>
>()
addEventListener(
type: string,
listener: (event: Event & { data?: string; readyState?: number }) => void
) {
const listeners = this.listeners.get(type) ?? []
listeners.push(listener)
this.listeners.set(type, listeners)
}
close() {
this.closed = true
}
stream() {
this.streamed = true
}
emit(type: string, data?: string) {
for (const listener of this.listeners.get(type) ?? []) {
listener({ data, readyState: this.readyState } as Event & {
data?: string
readyState?: number
})
}
}
}
const payload: ChatCompletionRequest = {
model: 'test-model',
messages: [{ role: 'user', content: 'hello' }],
stream: true,
}
const noopCallbacks = {
onUpdate: () => undefined,
onComplete: () => undefined,
onError: () => undefined,
}
describe('latest-wins stream request coordination', () => {
test('only creates a stream for the latest header request', async () => {
const firstHeaders = deferred<Record<string, string>>()
const secondHeaders = deferred<Record<string, string>>()
let headerRequest = 0
const sources: FakeStreamSource[] = []
const controller = createStreamRequestController({
getHeaders: () => {
headerRequest += 1
return headerRequest === 1
? firstHeaders.promise
: secondHeaders.promise
},
createSource: () => {
const source = new FakeStreamSource()
sources.push(source)
return source
},
setStreaming: () => undefined,
})
const first = controller.send(payload, noopCallbacks)
const second = controller.send(payload, noopCallbacks)
firstHeaders.resolve({ Authorization: 'Bearer stale' })
await first
expect(sources.length).toBe(0)
secondHeaders.resolve({ Authorization: 'Bearer current' })
await second
expect(sources.length).toBe(1)
expect(sources[0]?.streamed).toBe(true)
})
test('stop cancels a request that is still waiting for headers', async () => {
const headers = deferred<Record<string, string>>()
let sourceCount = 0
const controller = createStreamRequestController({
getHeaders: () => headers.promise,
createSource: () => {
sourceCount += 1
return new FakeStreamSource()
},
setStreaming: () => undefined,
})
const request = controller.send(payload, noopCallbacks)
controller.stop()
headers.resolve({ Authorization: 'Bearer ignored' })
await request
expect(sourceCount).toBe(0)
})
test('dispose cancels a pending header request without a state update', async () => {
const headers = deferred<Record<string, string>>()
const streamingStates: boolean[] = []
let sourceCount = 0
const controller = createStreamRequestController({
getHeaders: () => headers.promise,
createSource: () => {
sourceCount += 1
return new FakeStreamSource()
},
setStreaming: (streaming) => streamingStates.push(streaming),
})
const request = controller.send(payload, noopCallbacks)
controller.dispose()
headers.resolve({ Authorization: 'Bearer ignored' })
await request
expect(sourceCount).toBe(0)
expect(streamingStates).toEqual([false])
})
test('closes the previous source and ignores all of its later events', async () => {
const nextHeaders = deferred<Record<string, string>>()
let headerRequest = 0
const sources: FakeStreamSource[] = []
const updates: string[] = []
const controller = createStreamRequestController({
getHeaders: () => {
headerRequest += 1
if (headerRequest === 1) {
return Promise.resolve({ Authorization: 'Bearer first' })
}
return nextHeaders.promise
},
createSource: () => {
const source = new FakeStreamSource()
sources.push(source)
return source
},
setStreaming: () => undefined,
})
const callbacks = {
onUpdate: (_type: 'reasoning' | 'content', chunk: string) =>
updates.push(chunk),
onComplete: () => undefined,
onError: () => undefined,
}
await controller.send(payload, callbacks)
const second = controller.send(payload, callbacks)
expect(sources[0]?.closed).toBe(true)
sources[0]?.emit(
'message',
JSON.stringify({ choices: [{ delta: { content: 'stale' } }] })
)
nextHeaders.resolve({ Authorization: 'Bearer second' })
await second
sources[1]?.emit(
'message',
JSON.stringify({ choices: [{ delta: { content: 'current' } }] })
)
expect(updates).toEqual(['current'])
})
})
@@ -0,0 +1,235 @@
/*
Copyright (C) 2023-2026 QuantumNous
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
For commercial licensing, please contact support@quantumnous.com
*/
import { useCallback, useEffect, useRef, useState } from 'react'
import { SSE } from 'sse.js'
import { getFreshAuthHeaders } from '@/lib/api'
import { API_ENDPOINTS, ERROR_MESSAGES } from '../constants'
import {
getStreamReadyStateError,
isStreamClosedReadyState,
isStreamDoneMessage,
parseStreamErrorDetails,
parseStreamMessageUpdates,
} from '../lib'
import type { ChatCompletionRequest } from '../types'
interface StreamEventSource {
readyState?: number
addEventListener: (
type: string,
listener: (event: Event & { data?: string; readyState?: number }) => void
) => void
close: () => void
stream: () => void
}
interface StreamRequestCallbacks {
onUpdate: (type: 'reasoning' | 'content', chunk: string) => void
onComplete: () => void
onError: (error: string, errorCode?: string) => void
}
interface StreamRequestControllerRuntime {
getHeaders: () => Promise<Record<string, string>>
createSource: (
payload: ChatCompletionRequest,
headers: Record<string, string>
) => StreamEventSource
setStreaming: (streaming: boolean) => void
}
export function createStreamRequestController(
runtime: StreamRequestControllerRuntime
) {
let source: StreamEventSource | null = null
let generation = 0
const closeActiveSource = (target: StreamEventSource) => {
target.close()
if (source === target) {
source = null
runtime.setStreaming(false)
}
}
const send = async (
payload: ChatCompletionRequest,
callbacks: StreamRequestCallbacks
) => {
const requestGeneration = generation + 1
generation = requestGeneration
const previousSource = source
source = null
previousSource?.close()
runtime.setStreaming(false)
let headers: Record<string, string>
try {
headers = await runtime.getHeaders()
} catch (error: unknown) {
if (generation !== requestGeneration) return
callbacks.onError(
error instanceof Error
? error.message
: ERROR_MESSAGES.STREAM_START_ERROR
)
return
}
if (generation !== requestGeneration) return
const nextSource = runtime.createSource(payload, headers)
source = nextSource
runtime.setStreaming(true)
let completed = false
const isCurrent = () =>
generation === requestGeneration && source === nextSource
const handleError = (errorMessage: string, errorCode?: string) => {
if (!isCurrent() || completed) return
completed = true
callbacks.onError(errorMessage, errorCode)
closeActiveSource(nextSource)
}
nextSource.addEventListener('message', (event) => {
if (!isCurrent() || completed) return
const data = event.data ?? ''
if (isStreamDoneMessage(data)) {
completed = true
closeActiveSource(nextSource)
callbacks.onComplete()
return
}
try {
const updates = parseStreamMessageUpdates(data)
for (const update of updates) {
callbacks.onUpdate(update.type, update.chunk)
}
} catch (error) {
// eslint-disable-next-line no-console
console.error('Failed to parse SSE message:', error)
handleError(ERROR_MESSAGES.PARSE_ERROR)
}
})
nextSource.addEventListener('error', (event) => {
if (!isCurrent() || completed) return
if (!isStreamClosedReadyState(nextSource.readyState)) {
// eslint-disable-next-line no-console
console.error('SSE Error:', event)
const { errorCode, errorMessage } = parseStreamErrorDetails(event.data)
handleError(errorMessage, errorCode)
}
})
nextSource.addEventListener('readystatechange', (event) => {
if (!isCurrent() || completed) return
const errorMessage = getStreamReadyStateError(
event.readyState,
nextSource
)
if (errorMessage) {
handleError(errorMessage)
}
})
try {
if (!isCurrent()) return
nextSource.stream()
} catch (error: unknown) {
if (!isCurrent() || completed) return
// eslint-disable-next-line no-console
console.error('Failed to start SSE stream:', error)
handleError(ERROR_MESSAGES.STREAM_START_ERROR)
}
}
const cancel = (notify: boolean) => {
generation += 1
const activeSource = source
source = null
activeSource?.close()
if (notify) runtime.setStreaming(false)
}
const stop = () => cancel(true)
const dispose = () => cancel(false)
return { send, stop, dispose }
}
/**
* Hook for handling streaming chat completion requests
*/
export function useStreamRequest() {
const [isStreaming, setIsStreaming] = useState(false)
const controllerRef = useRef<ReturnType<
typeof createStreamRequestController
> | null>(null)
if (!controllerRef.current) {
controllerRef.current = createStreamRequestController({
getHeaders: getFreshAuthHeaders,
createSource: (payload, headers) =>
new SSE(API_ENDPOINTS.CHAT_COMPLETIONS, {
headers,
method: 'POST',
payload: JSON.stringify(payload),
}) as StreamEventSource,
setStreaming: setIsStreaming,
})
}
const sendStreamRequest = useCallback(
(
payload: ChatCompletionRequest,
onUpdate: (type: 'reasoning' | 'content', chunk: string) => void,
onComplete: () => void,
onError: (error: string, errorCode?: string) => void
) =>
controllerRef.current?.send(payload, {
onUpdate,
onComplete,
onError,
}),
[]
)
const stopStream = useCallback(() => {
controllerRef.current?.stop()
}, [])
useEffect(
() => () => {
controllerRef.current?.dispose()
},
[]
)
return {
sendStreamRequest,
stopStream,
isStreaming,
}
}