diff --git a/web/src/App.tsx b/web/src/App.tsx index 465ee4b6..5560596e 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -1708,12 +1708,19 @@ export default function App() { // 驱动一次真实后端 SSE 流式对话尝试:气泡状态机 pending → streaming_status → completed/failed/cancelled。 // 闲聊路径在无 client_turn_id 幂等前禁止自动重放 POST;失败仅展示手动「重试」。 + type StreamAttemptOpts = { + userMsgId?: string; + clientTurnId?: string; + quote?: QuotedMessage; + }; + const runStreamAttempt = ( userText: string, asstId: string, attemptIndex: number, - quote?: QuotedMessage + opts?: StreamAttemptOpts ) => { + const quote = opts?.quote; // 同一时刻只允许一个活跃流:新尝试前中止上一个 (若存在) streamAbortRef.current?.abort(); const controller = new AbortController(); @@ -1734,7 +1741,7 @@ export default function App() { setIsThinking(false); return; } - runStreamAttempt(userText, asstId, attemptIndex + 1, quote); + runStreamAttempt(userText, asstId, attemptIndex + 1, opts); }, delayMs); }; @@ -1771,6 +1778,14 @@ export default function App() { }); setIsThinking(true); + const bindUserServerId = (serverId: number) => { + if (!opts?.userMsgId) return; + chatWindow.replaceMessageServerId(opts.userMsgId, serverId); + setMessages((prev) => + prev.map((m) => (m.id === opts.userMsgId ? { ...m, serverId } : m)) + ); + }; + streamChat( { session_id: activeSessionId, @@ -1782,12 +1797,20 @@ export default function App() { user_name: currentUser.name, user_role: currentDocRole, message: userText, + client_turn_id: opts?.clientTurnId, quote: quote, }, { onEvent: (ev) => { if (controller.signal.aborted || settled) return; switch (ev.type) { + case 'turn_ack': { + const userServerId = ev.turn?.user_message_id ?? ev.user_message_id; + if (typeof userServerId === 'number' && userServerId > 0) { + bindUserServerId(userServerId); + } + break; + } case 'phase': case 'status': { const label = ev.phase?.message || ev.status?.message; @@ -1805,7 +1828,11 @@ export default function App() { } case 'done': { settled = true; - const serverMsgId = ev.message_id || (ev as any).messageId || ev.done?.message_id; + const userServerId = ev.done?.user_message_id ?? ev.user_message_id; + const serverMsgId = ev.done?.message_id ?? ev.message_id ?? (ev as { messageId?: number }).messageId; + if (typeof userServerId === 'number' && userServerId > 0) { + bindUserServerId(userServerId); + } if (typeof serverMsgId === 'number' && serverMsgId > 0) { chatWindow.replaceMessageServerId(asstId, serverMsgId); } @@ -1824,7 +1851,8 @@ export default function App() { } }, }, - controller.signal + controller.signal, + opts?.clientTurnId ? { idempotencyKey: opts.clientTurnId } : undefined ).catch((_err) => { if (settled) return; if (controller.signal.aborted) { @@ -2104,8 +2132,11 @@ export default function App() { } } + const clientTurnId = crypto.randomUUID(); const userMsg: ChatMessage = { id: 'msg-' + Date.now(), + clientTurnId, + clientMessageId: `${clientTurnId}:user`, sessionId: activeSessionId, role: 'user', authorName: currentUser.name, @@ -2211,6 +2242,8 @@ export default function App() { const asstId = 'ast-' + Date.now(); const placeholder: ChatMessage = { id: asstId, + clientTurnId, + clientMessageId: `${clientTurnId}:assistant`, sessionId: activeSessionId, role: 'assistant', authorName: '项目助手', @@ -2221,7 +2254,11 @@ export default function App() { sourceUserMessageId: userMsg.id, }; setMessages((prev) => [...prev, placeholder]); - runStreamAttempt(text, asstId, 0, quotes && quotes.length > 0 ? quotes[0] : undefined); + runStreamAttempt(text, asstId, 0, { + userMsgId: userMsg.id, + clientTurnId, + quote: quotes && quotes.length > 0 ? quotes[0] : undefined, + }); return; } diff --git a/web/src/api/chatStream.ts b/web/src/api/chatStream.ts index 4bebb2b9..165debad 100644 --- a/web/src/api/chatStream.ts +++ b/web/src/api/chatStream.ts @@ -33,6 +33,12 @@ export type StreamEvent = { workspace_id?: string; actions?: string[]; }; + turn?: { + client_turn_id?: string; + user_message_id?: number; + status?: string; + duplicate?: boolean; + }; message_id?: number; user_message_id?: number; content_revision?: number; @@ -131,10 +137,18 @@ export async function streamChat( onRequestId?: (id: string) => void; }, signal?: AbortSignal, + options?: { idempotencyKey?: string }, ): Promise { + const headers: Record = { + 'Content-Type': 'application/json', + Accept: 'text/event-stream', + }; + if (options?.idempotencyKey) { + headers['Idempotency-Key'] = options.idempotencyKey; + } const resp = await fetch('/api/v1/chat/stream', { method: 'POST', - headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream' }, + headers, body: JSON.stringify(body), signal, }); diff --git a/web/src/api/chatWindow.ts b/web/src/api/chatWindow.ts index 991eb254..ff276079 100644 --- a/web/src/api/chatWindow.ts +++ b/web/src/api/chatWindow.ts @@ -62,6 +62,8 @@ export function mapRecordToMessage(r: RawChatMessageRecord): ChatMessage { const msgId = meta.client_id || (r.id ? `msg-${r.id}` : `msg-tmp-${Date.now()}-${Math.random().toString(36).substring(2, 6)}`); return { id: msgId, + clientTurnId: meta.client_turn_id || undefined, + clientMessageId: meta.client_id || undefined, serverId: r.id, sessionId: r.session_id, role: r.role, diff --git a/web/src/engine/messageIdentity.ts b/web/src/engine/messageIdentity.ts new file mode 100644 index 00000000..82df6ea3 --- /dev/null +++ b/web/src/engine/messageIdentity.ts @@ -0,0 +1,79 @@ +import type { ChatMessage } from '../types/workbench'; + +/** 收集消息的所有可对账身份键 */ +export function getMessageKeys(m: ChatMessage): string[] { + const keys: string[] = []; + if (m.clientTurnId) { + keys.push(`turn:${m.clientTurnId}:${m.role}`); + } + if (m.clientMessageId) { + keys.push(`client:${m.clientMessageId}`); + } + keys.push(`id:${m.id}`); + if (m.serverId) { + keys.push(`server:${m.serverId}`); + } + return keys; +} + +/** 判断两条消息是否代表同一条物理消息 */ +export function isSameMessage(a: ChatMessage, b: ChatMessage): boolean { + if (a.sessionId !== b.sessionId || a.role !== b.role) { + return false; + } + if (a.clientTurnId && b.clientTurnId && a.clientTurnId === b.clientTurnId) { + return true; + } + if (a.serverId && b.serverId && a.serverId === b.serverId) { + return true; + } + if (a.clientMessageId && b.clientMessageId && a.clientMessageId === b.clientMessageId) { + return true; + } + return false; +} + +/** 构建历史 bucket 中已有消息的身份键集合 */ +export function buildExistingMessageKeySet(messages: ChatMessage[]): Set { + const keys = new Set(); + for (const m of messages) { + for (const k of getMessageKeys(m)) { + keys.add(k); + } + } + return keys; +} + +/** live 消息是否已在历史中(按身份对账,而非仅 m.id) */ +export function isLiveMessageInHistory( + live: ChatMessage, + historyMessages: ChatMessage[], +): boolean { + return historyMessages.some((h) => isSameMessage(live, h)); +} + +/** 过滤出尚未出现在历史中的 live 消息 */ +export function filterPendingLiveMessages( + liveMessages: ChatMessage[], + historyMessages: ChatMessage[], +): ChatMessage[] { + return liveMessages.filter((m) => !isLiveMessageInHistory(m, historyMessages)); +} + +/** 用 live 流状态覆盖历史快照中的同一条消息 */ +export function mergeLiveOverHistory( + history: ChatMessage, + live: ChatMessage, +): ChatMessage { + if (!isSameMessage(history, live)) { + return history; + } + return { + ...history, + ...live, + id: history.id, + serverId: live.serverId ?? history.serverId, + clientTurnId: live.clientTurnId ?? history.clientTurnId, + clientMessageId: live.clientMessageId ?? history.clientMessageId, + }; +} diff --git a/web/src/hooks/useChatWindow.ts b/web/src/hooks/useChatWindow.ts index ac409b2e..2bdb4c03 100644 --- a/web/src/hooks/useChatWindow.ts +++ b/web/src/hooks/useChatWindow.ts @@ -11,6 +11,11 @@ import { saveSessionDraft, } from '../data/storage'; import { enrichMessagesWithCategories } from '../engine/categoryHelper'; +import { + filterPendingLiveMessages, + isLiveMessageInHistory, + mergeLiveOverHistory, +} from '../engine/messageIdentity'; export type WindowPhase = | 'loading_window' @@ -420,15 +425,8 @@ export function useChatWindow({ } const todayStr = getTodayKey(); - const existingMsgIds = new Set(); - for (const b of dayStack) { - for (const m of b.messages) { - existingMsgIds.add(m.id); - if (m.serverId) existingMsgIds.add(`msg-${m.serverId}`); - } - } - - const pendingLive = liveMessages.filter((m) => !existingMsgIds.has(m.id)); + const historyMessages = dayStack.flatMap((b) => b.messages); + const pendingLive = filterPendingLiveMessages(liveMessages, historyMessages); if (dayStack.length === 0) { return [ @@ -445,22 +443,22 @@ export function useChatWindow({ } if (pendingLive.length === 0) { - const liveMap = new Map(liveMessages.map((m) => [m.id, m])); let hasUpdate = false; const updatedBuckets = dayStack.map((b) => { let bucketChanged = false; const newMsgs = b.messages.map((m) => { - const liveM = liveMap.get(m.id); + const liveM = liveMessages.find((live) => isLiveMessageInHistory(live, [m])); if ( liveM && (liveM.content !== m.content || liveM.status !== m.status || liveM.livePhaseLabel !== m.livePhaseLabel || - liveM.errorInfo !== m.errorInfo) + liveM.errorInfo !== m.errorInfo || + liveM.serverId !== m.serverId) ) { bucketChanged = true; hasUpdate = true; - return { ...m, ...liveM }; + return mergeLiveOverHistory(m, liveM); } return m; }); diff --git a/web/src/types/workbench.ts b/web/src/types/workbench.ts index bccaa367..fe73c50f 100644 --- a/web/src/types/workbench.ts +++ b/web/src/types/workbench.ts @@ -121,6 +121,16 @@ export interface ChatErrorInfo { actions: Array<'retry' | 'continue' | 'restart'>; } +/** + * 任务输入材料(对应后端 bizdev.MaterialInput)。 + * content 为文件原始内容的 Base64 字符串(后端 JSON 反序列化 []byte 时自动解码)。 + * 仅文本类格式会填充 content;二进制格式 content 留空,仅传文件名作来源注记。 + */ +export type TaskRunMaterial = { + filename: string; + content: string; // Base64 编码的文件内容 +}; + /** 长任务运行上下文:触发 / 重试 / 重新开始时重发请求所需的最小参数集 */ export type TaskRunContext = { taskKey: string; @@ -131,6 +141,8 @@ export type TaskRunContext = { targetMarket?: string; userId?: number; projectId?: number; + /** 用户上传的参考文档材料,注入任务工作区 materials_text/ 目录 */ + materials?: TaskRunMaterial[]; }; export interface ChatMessage { @@ -163,6 +175,11 @@ export interface ChatMessage { /** P2: 最近一次任务成功/收尾时的工作区目录(done.workspace_id) */ taskWorkspaceId?: string; + /** 客户端回合 ID(一次发送的用户+助手共享) */ + clientTurnId?: string; + /** 客户端消息 ID(用户/助手各自稳定 ID) */ + clientMessageId?: string; + /** P0: 历史归档与导航扩展 */ serverId?: number; dayKey?: string;