You've already forked agentic-coding-workflow
主流程
This commit is contained in:
@@ -0,0 +1,169 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
|
||||
|
||||
import type { ChatStreamEvent } from './useChatStream'
|
||||
|
||||
const chatApiMock = vi.hoisted(() => ({
|
||||
getConversation: vi.fn(),
|
||||
listMessages: vi.fn(),
|
||||
sendMessage: vi.fn(),
|
||||
cancelMessage: vi.fn(),
|
||||
retryMessage: vi.fn(),
|
||||
}))
|
||||
|
||||
vi.mock('@/api/chat', () => ({ chatApi: chatApiMock }))
|
||||
|
||||
// startChatStream 的 mock：记录每次调用，暴露 emit 让测试驱动事件。
|
||||
interface StreamInstance {
|
||||
url: string
|
||||
since: number
|
||||
emit: (ev: ChatStreamEvent) => void
|
||||
stopped: boolean
|
||||
}
|
||||
const streamInstances = vi.hoisted(() => [] as StreamInstance[])
|
||||
|
||||
vi.mock('@/composables/useChatStream', () => ({
|
||||
startChatStream: vi.fn(
|
||||
(
|
||||
url: string,
|
||||
since: number,
|
||||
handlers: { onEvent: (ev: ChatStreamEvent) => void; onError: (e: Error) => void },
|
||||
) => {
|
||||
const inst: StreamInstance = {
|
||||
url,
|
||||
since,
|
||||
emit: (ev) => handlers.onEvent(ev),
|
||||
stopped: false,
|
||||
}
|
||||
streamInstances.push(inst)
|
||||
return {
|
||||
stop: () => {
|
||||
inst.stopped = true
|
||||
},
|
||||
}
|
||||
},
|
||||
),
|
||||
}))
|
||||
|
||||
import { useConversationSession } from './useConversationSession'
|
||||
|
||||
function conv(id: string) {
|
||||
return { id, user_id: 'u1', model_id: 'm1', system_prompt: '', created_at: '', updated_at: '' }
|
||||
}
|
||||
|
||||
function msg(id: number, status = 'ok', role = 'assistant') {
|
||||
return {
|
||||
id,
|
||||
conversation_id: 'c1',
|
||||
role,
|
||||
content: `msg-${id}`,
|
||||
status,
|
||||
created_at: '',
|
||||
updated_at: '',
|
||||
}
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
streamInstances.length = 0
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
describe('useConversationSession', () => {
|
||||
it('openConversation 将 DESC 消息排为 id 升序', async () => {
|
||||
chatApiMock.getConversation.mockResolvedValue({
|
||||
conversation: conv('c1'),
|
||||
messages: [msg(3), msg(2), msg(1)],
|
||||
})
|
||||
const s = useConversationSession()
|
||||
await s.openConversation('c1')
|
||||
expect(s.messages.value.map((m) => m.id)).toEqual([1, 2, 3])
|
||||
expect(s.isStreaming.value).toBe(false)
|
||||
})
|
||||
|
||||
it('openConversation 发现 pending 消息时续接流', async () => {
|
||||
chatApiMock.getConversation.mockResolvedValue({
|
||||
conversation: conv('c1'),
|
||||
messages: [msg(2, 'pending'), msg(1)],
|
||||
})
|
||||
const s = useConversationSession()
|
||||
await s.openConversation('c1')
|
||||
expect(s.streamingMessageId.value).toBe(2)
|
||||
expect(streamInstances).toHaveLength(1)
|
||||
expect(streamInstances[0].since).toBe(1)
|
||||
})
|
||||
|
||||
it('sendMessage 追加两条占位消息并累积 text_delta 至 done', async () => {
|
||||
chatApiMock.getConversation.mockResolvedValue({ conversation: conv('c1'), messages: [] })
|
||||
chatApiMock.sendMessage.mockResolvedValue({
|
||||
user_message_id: 10,
|
||||
assistant_message_id: 11,
|
||||
stream_url: '/stream',
|
||||
})
|
||||
const s = useConversationSession()
|
||||
await s.openConversation('c1')
|
||||
await s.sendMessage('hello')
|
||||
|
||||
expect(s.messages.value.map((m) => [m.id, m.role, m.status])).toEqual([
|
||||
[10, 'user', 'ok'],
|
||||
[11, 'assistant', 'pending'],
|
||||
])
|
||||
expect(s.isStreaming.value).toBe(true)
|
||||
|
||||
const stream = streamInstances[0]
|
||||
stream.emit({ id: 1, event: 'text_delta', data: { text: 'foo' } })
|
||||
stream.emit({ id: 2, event: 'text_delta', data: { text: 'bar' } })
|
||||
stream.emit({ id: 3, event: 'done', data: {} })
|
||||
|
||||
const assistant = s.messages.value[1]
|
||||
expect(assistant.content).toBe('foobar')
|
||||
expect(assistant.status).toBe('ok')
|
||||
expect(s.isStreaming.value).toBe(false)
|
||||
})
|
||||
|
||||
it('error 事件标记消息失败并结束流式', async () => {
|
||||
chatApiMock.getConversation.mockResolvedValue({ conversation: conv('c1'), messages: [] })
|
||||
chatApiMock.sendMessage.mockResolvedValue({
|
||||
user_message_id: 1,
|
||||
assistant_message_id: 2,
|
||||
stream_url: '/stream',
|
||||
})
|
||||
const s = useConversationSession()
|
||||
await s.openConversation('c1')
|
||||
await s.sendMessage('x')
|
||||
streamInstances[0].emit({ id: 1, event: 'error', data: { message: 'boom' } })
|
||||
expect(s.messages.value[1].status).toBe('error')
|
||||
expect(s.messages.value[1].error_message).toBe('boom')
|
||||
expect(s.isStreaming.value).toBe(false)
|
||||
})
|
||||
|
||||
it('closeCurrent 终止流并清空状态', async () => {
|
||||
chatApiMock.getConversation.mockResolvedValue({
|
||||
conversation: conv('c1'),
|
||||
messages: [msg(2, 'pending')],
|
||||
})
|
||||
const s = useConversationSession()
|
||||
await s.openConversation('c1')
|
||||
expect(streamInstances[0].stopped).toBe(false)
|
||||
s.closeCurrent()
|
||||
expect(streamInstances[0].stopped).toBe(true)
|
||||
expect(s.currentConversation.value).toBeNull()
|
||||
expect(s.messages.value).toEqual([])
|
||||
})
|
||||
|
||||
it('双实例互不干扰', async () => {
|
||||
chatApiMock.getConversation
|
||||
.mockResolvedValueOnce({ conversation: conv('c1'), messages: [msg(1)] })
|
||||
.mockResolvedValueOnce({ conversation: conv('c2'), messages: [msg(5), msg(4)] })
|
||||
const a = useConversationSession()
|
||||
const b = useConversationSession()
|
||||
await a.openConversation('c1')
|
||||
await b.openConversation('c2')
|
||||
expect(a.messages.value.map((m) => m.id)).toEqual([1])
|
||||
expect(b.messages.value.map((m) => m.id)).toEqual([4, 5])
|
||||
b.closeCurrent()
|
||||
expect(a.currentConversation.value?.id).toBe('c1')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,223 @@
|
||||
import { computed, ref } from 'vue'
|
||||
|
||||
import { chatApi, type ChatMessage, type Conversation } from '@/api/chat'
|
||||
import {
|
||||
startChatStream,
|
||||
type ChatStreamEvent,
|
||||
type ChatStreamHandle,
|
||||
} from '@/composables/useChatStream'
|
||||
|
||||
interface UsagePayload {
|
||||
PromptTokens?: number
|
||||
CompletionTokens?: number
|
||||
ThinkingTokens?: number
|
||||
}
|
||||
|
||||
interface TextDeltaPayload {
|
||||
text?: string
|
||||
}
|
||||
|
||||
interface ErrorPayload {
|
||||
code?: string
|
||||
message?: string
|
||||
}
|
||||
|
||||
interface MetaPayload {
|
||||
id?: number
|
||||
role?: string
|
||||
status?: ChatMessage['status']
|
||||
}
|
||||
|
||||
// useConversationSession 承载"单个会话的消息流"全部状态与操作（打开/分页/发送/
|
||||
// 取消/重试/SSE 流式接收）。每次调用返回一个独立实例，互不串扰——全局 /chat 页
|
||||
// 经 chat store 持有一份，需求详情页内嵌面板各自持有一份。
|
||||
// 调用方负责在组件卸载时调用 closeCurrent() 终止流。
|
||||
export function useConversationSession() {
|
||||
const currentConversation = ref<Conversation | null>(null)
|
||||
const messages = ref<ChatMessage[]>([])
|
||||
const streamingMessageId = ref<number | null>(null)
|
||||
const streamHandle = ref<ChatStreamHandle | null>(null)
|
||||
|
||||
const isStreaming = computed(() => streamingMessageId.value !== null)
|
||||
|
||||
async function openConversation(id: string) {
|
||||
detachStream()
|
||||
const detail = await chatApi.getConversation(id)
|
||||
currentConversation.value = detail.conversation
|
||||
// 后端按 id DESC 返回（最新在前）；视图按升序渲染（最新在底部），故排为升序。
|
||||
messages.value = [...detail.messages].sort((a, b) => a.id - b.id)
|
||||
const pending = detail.messages.find((m) => m.status === 'pending')
|
||||
if (pending) {
|
||||
streamingMessageId.value = pending.id
|
||||
attachStream(`/api/v1/conversations/${id}/stream`, pending.id - 1)
|
||||
}
|
||||
}
|
||||
|
||||
async function loadOlderMessages() {
|
||||
const conv = currentConversation.value
|
||||
if (!conv) return
|
||||
if (messages.value.length === 0) return
|
||||
const beforeId = messages.value.reduce(
|
||||
(min, m) => (m.id < min ? m.id : min),
|
||||
messages.value[0].id,
|
||||
)
|
||||
const older = await chatApi.listMessages(conv.id, { before_id: beforeId })
|
||||
if (older.length === 0) return
|
||||
const existingIds = new Set(messages.value.map((m) => m.id))
|
||||
const fresh = older.filter((m) => !existingIds.has(m.id))
|
||||
if (fresh.length === 0) return
|
||||
// 合并后统一按 id 升序，避免后端 DESC 批次与现有升序列表错位。
|
||||
messages.value = [...fresh, ...messages.value].sort((a, b) => a.id - b.id)
|
||||
}
|
||||
|
||||
async function sendMessage(content: string, attachmentIds: string[] = []) {
|
||||
if (!currentConversation.value) return
|
||||
if (streamingMessageId.value !== null) return
|
||||
const conv = currentConversation.value
|
||||
const resp = await chatApi.sendMessage(conv.id, {
|
||||
content,
|
||||
attachment_ids: attachmentIds,
|
||||
})
|
||||
const now = new Date().toISOString()
|
||||
messages.value.push({
|
||||
id: resp.user_message_id,
|
||||
conversation_id: conv.id,
|
||||
role: 'user',
|
||||
content,
|
||||
status: 'ok',
|
||||
created_at: now,
|
||||
updated_at: now,
|
||||
})
|
||||
messages.value.push({
|
||||
id: resp.assistant_message_id,
|
||||
conversation_id: conv.id,
|
||||
role: 'assistant',
|
||||
content: '',
|
||||
status: 'pending',
|
||||
created_at: now,
|
||||
updated_at: now,
|
||||
})
|
||||
streamingMessageId.value = resp.assistant_message_id
|
||||
attachStream(resp.stream_url, resp.user_message_id)
|
||||
}
|
||||
|
||||
async function cancelMessage() {
|
||||
const id = streamingMessageId.value
|
||||
if (id == null) return
|
||||
await chatApi.cancelMessage(id)
|
||||
detachStream()
|
||||
}
|
||||
|
||||
async function retryMessage(messageId: number) {
|
||||
if (!currentConversation.value) return
|
||||
const resp = await chatApi.retryMessage(messageId)
|
||||
const idx = messages.value.findIndex((m) => m.id === messageId)
|
||||
if (idx >= 0) {
|
||||
const old = messages.value[idx]
|
||||
messages.value[idx] = {
|
||||
...old,
|
||||
id: resp.assistant_message_id,
|
||||
content: '',
|
||||
thinking: null,
|
||||
status: 'pending',
|
||||
error_message: null,
|
||||
}
|
||||
}
|
||||
streamingMessageId.value = resp.assistant_message_id
|
||||
attachStream(resp.stream_url, resp.assistant_message_id - 1)
|
||||
}
|
||||
|
||||
function closeCurrent() {
|
||||
detachStream()
|
||||
currentConversation.value = null
|
||||
messages.value = []
|
||||
streamingMessageId.value = null
|
||||
}
|
||||
|
||||
function attachStream(streamUrl: string, sinceMessageID: number) {
|
||||
detachStream()
|
||||
streamHandle.value = startChatStream(streamUrl, sinceMessageID, {
|
||||
onEvent: (ev) => applyStreamEvent(ev),
|
||||
onClose: () => {
|
||||
streamHandle.value = null
|
||||
},
|
||||
onError: (err) => {
|
||||
console.error('chat stream error', err)
|
||||
const id = streamingMessageId.value
|
||||
if (id != null) {
|
||||
const msg = messages.value.find((m) => m.id === id)
|
||||
if (msg && msg.status === 'pending') {
|
||||
msg.status = 'error'
|
||||
msg.error_message = err.message
|
||||
}
|
||||
}
|
||||
streamingMessageId.value = null
|
||||
streamHandle.value = null
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
function detachStream() {
|
||||
streamHandle.value?.stop()
|
||||
streamHandle.value = null
|
||||
}
|
||||
|
||||
function applyStreamEvent(ev: ChatStreamEvent) {
|
||||
const targetId = streamingMessageId.value
|
||||
if (targetId == null) return
|
||||
const msg = messages.value.find((m) => m.id === targetId)
|
||||
if (!msg) return
|
||||
switch (ev.event) {
|
||||
case 'message_meta': {
|
||||
const d = ev.data as MetaPayload
|
||||
if (d.status) msg.status = d.status
|
||||
break
|
||||
}
|
||||
case 'text_delta': {
|
||||
const d = ev.data as TextDeltaPayload
|
||||
if (d.text) msg.content += d.text
|
||||
break
|
||||
}
|
||||
case 'thinking_delta': {
|
||||
const d = ev.data as TextDeltaPayload
|
||||
if (d.text) msg.thinking = (msg.thinking ?? '') + d.text
|
||||
break
|
||||
}
|
||||
case 'usage': {
|
||||
const u = ev.data as UsagePayload
|
||||
msg.prompt_tokens = u.PromptTokens ?? msg.prompt_tokens
|
||||
msg.completion_tokens = u.CompletionTokens ?? msg.completion_tokens
|
||||
msg.thinking_tokens = u.ThinkingTokens ?? msg.thinking_tokens
|
||||
break
|
||||
}
|
||||
case 'done':
|
||||
// 后端事件顺序：EventUsage → EventDone（参见 internal/chat/streamer.go）。
|
||||
// tokens 已在 'usage' 分支写入；某些 provider 不发 usage 时此处不会兜底。
|
||||
msg.status = 'ok'
|
||||
streamingMessageId.value = null
|
||||
break
|
||||
case 'error': {
|
||||
const d = ev.data as ErrorPayload
|
||||
msg.status = 'error'
|
||||
msg.error_message = d.message ?? d.code ?? 'stream error'
|
||||
streamingMessageId.value = null
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
currentConversation,
|
||||
messages,
|
||||
streamingMessageId,
|
||||
isStreaming,
|
||||
openConversation,
|
||||
loadOlderMessages,
|
||||
sendMessage,
|
||||
cancelMessage,
|
||||
retryMessage,
|
||||
closeCurrent,
|
||||
}
|
||||
}
|
||||
|
||||
export type ConversationSession = ReturnType<typeof useConversationSession>
|
||||
Reference in New Issue
Block a user