import { computed, nextTick } from 'vue'; import dayjs from 'dayjs'; import { message } from 'ant-design-vue'; import { unifiedStream, reviewPdfStream, getDetail, deleteMessage } from '/@/views/ventAI/manageAssistent/api'; import type { AttachedFile, Message, SseEvent, Task } from '/@/views/ventAI/manageAssistent/types/index'; import { getCssUrl, normalizeLineBreaks } from '/@/views/ventAI/manageAssistent/lib/index'; /** * 发送与流式管线:推送一条空 AI 消息 → 建 AbortController → 走 SSE(纯文本 / PDF 审查)→ 处理 * session_id 迁移与异常清理;另含停止生成、卸载断开、历史消息 id 回填、重新生成、编辑重发。 * * 输入框交互状态(草稿、待发附件、引用)留在组件里,这里只负责「消息落库 + 起流」;任务/消息读写、 * SSE 事件处理、滚动跟随、待办重置与审批 key 迁移等能力通过 deps 注入(一律延迟求值,避免 setup 期 TDZ)。 */ export interface ChatSendDeps { getTaskList: () => Task[]; getCurrentTaskId: () => string; setCurrentTaskId: (taskId: string) => void; getMessages: () => Message[]; setTaskList: (tasks: Task[]) => void; getEditMode: () => string; getPendingApproval: () => unknown; getChatMessages: () => { forceScrollToBottom?: () => void } | undefined; handleStreamEvent: (data: SseEvent, aiMsg: Message, taskId: string) => void; scrollToBottomThrottled: (force?: boolean) => void; resetTodos: () => void; migrateTaskApproval: (fromTaskId: string, toTaskId: string) => void; historyLoadedTasks: Set; transformHistoryToMessages: (data: Array>) => Message[]; } export const useChatSend = (deps: ChatSendDeps) => { const { getTaskList, getCurrentTaskId, setCurrentTaskId, getMessages, setTaskList, getEditMode, getPendingApproval, getChatMessages, handleStreamEvent, scrollToBottomThrottled, resetTodos, migrateTaskApproval, historyLoadedTasks, transformHistoryToMessages, } = deps; // 当前任务是否正在流式输出(原 streaming 别名与 loading 完全等价,统一用 loading) const loading = computed(() => { const currentTask = getTaskList().find((t) => t.id === getCurrentTaskId()); return currentTask?.isStreaming ?? false; }); // Non-reactive Map for AbortControllers - only used imperatively, no need for Vue reactivity const taskAbortControllers = new Map(); // 设置任务的AbortController const setTaskAbortController = (taskId: string, controller: AbortController | null) => { if (controller) { taskAbortControllers.set(taskId, controller); } else { taskAbortControllers.delete(taskId); } }; // 停止指定任务的流式传输 const stopTaskStreaming = (taskId: string) => { const controller = taskAbortControllers.get(taskId); controller?.abort(); }; const stopStreaming = () => { stopTaskStreaming(getCurrentTaskId()); }; // 组件卸载/清理时断开所有任务的流式连接 const abortAllStreams = () => { taskAbortControllers.forEach((controller) => controller.abort()); taskAbortControllers.clear(); }; const getSkillRefIconUrl = () => getCssUrl('--img-chat-options-skill-icon'); const getSessionRefIconUrl = () => getCssUrl('--img-chat-popup-session-icon'); // targetTaskId:回答写进哪个任务。默认当前任务;重新生成与编辑重发是在若干次 await 之后才调到这里, // 必须显式传入发起任务,否则请求期间切走后新回答会落到别的会话上。 const streamAiAnswer = async (userInput: string, file?: AttachedFile, targetTaskId?: string) => { const originTaskId = targetTaskId || getCurrentTaskId(); const originTask = getTaskList().find((t) => t.id === originTaskId); if (!originTask) return; const now = dayjs(); const newMsg: Message = { type: 'ai', content: '', time: now.format('HH:mm'), createdAt: now.toISOString(), isLoading: true, thinkingSteps: [], generateStartTime: Date.now(), }; originTask.messages.push(newMsg); // 从数组里读回这一条:任务状态是 reactive 的,push 进去的原始对象由代理包裹, // 之后必须通过代理读写才会触发视图更新——直接改上面的 newMsg,界面会一直停在“正在输入”。 const aiMsg = originTask.messages[originTask.messages.length - 1]; // 进度条属于「正在看的任务」,不要为别的任务清空它 if (getCurrentTaskId() === originTaskId) resetTodos(); originTask.isStreaming = true; const controller = new AbortController(); setTaskAbortController(originTaskId, controller); const onChunk = (data: SseEvent) => { handleStreamEvent(data, aiMsg, originTaskId); if (getCurrentTaskId() === originTaskId) scrollToBottomThrottled(); }; try { const originalFile = file?.originalFile; const result = originalFile ? await reviewPdfStream( { file: originalFile, session_id: originTask.sessionId || undefined, message: userInput || undefined, mode: getEditMode(), signal: controller.signal, }, onChunk ) : await unifiedStream( { message: userInput, session_id: originTask.sessionId || undefined, mode: getEditMode(), signal: controller.signal, }, onChunk ); if (result.session_id) { const originTask = getTaskList().find((t) => t.id === originTaskId); if (originTask && !originTask.sessionId) { originTask.sessionId = result.session_id; // 流式期间若列表已用后端标题归并过该任务,则保留后端标题;否则用首条消息命名 if (originTask.pendingSessionId !== result.session_id) { const taskName = userInput || (file ? `文件:${file.name}` : ''); originTask.name = taskName.substring(0, 20) + (taskName.length > 20 ? '...' : ''); } originTask.pendingSessionId = undefined; if (originTask.id !== result.session_id) { // 将 AbortController 从旧 ID 迁移到新 ID const controller = taskAbortControllers.get(originTaskId); if (controller) { setTaskAbortController(originTaskId, null); setTaskAbortController(result.session_id, controller); } setTaskList(getTaskList().filter((t) => t.id !== result.session_id)); originTask.id = result.session_id; historyLoadedTasks.add(result.session_id); // 迁移审批状态 key(任务 ID 变了,审批 Map 的 key 也要跟着变) migrateTaskApproval(originTaskId, result.session_id); } if (getCurrentTaskId() === originTaskId) { setCurrentTaskId(result.session_id); } } } } catch (error) { // 用户主动停止生成 if ((error as Error)?.name === 'AbortError') { // 目标消息直接用发起流式时创建的那个对象:数组可能已被删除消息重排/变短,按下标取会指错 aiMsg.isLoading = false; // 后端未返回耗时,冻结前端伪计时 aiMsg.durationMs = (aiMsg.baseDurationMs || 0) + (aiMsg.generateStartTime ? Date.now() - aiMsg.generateStartTime : 0); if (!aiMsg.content) { aiMsg.content = '已停止生成。'; } return; } throw error; } finally { // 任务在流结束时可能已被改名为后端 session_id,需要使用最终的 ID 清理状态 const finalTaskId = originTask?.id || originTaskId; setTaskAbortController(finalTaskId, null); if (originTask) originTask.isStreaming = false; } }; const sendTextOnly = async (userInput: string, displayContent?: string) => { const now = dayjs(); const userMsg: Message = { type: 'user', content: displayContent || userInput, rawInput: userInput, time: now.format('HH:mm'), createdAt: now.toISOString(), }; getMessages().push(userMsg); await nextTick(); getChatMessages()?.forceScrollToBottom?.(); await streamAiAnswer(userInput); }; const sendWithAttachment = async (file: AttachedFile, userInput: string, displayContent?: string) => { const originTaskId = getCurrentTaskId(); const currentTask = getTaskList().find((t) => t.id === originTaskId); if (currentTask) { currentTask.attachedFiles.push(file); } else { throw new Error('未找到当前任务'); } const now = dayjs(); const userMsg: Message = { type: 'user', content: displayContent || userInput || `上传了文件:${file.name}`, rawInput: userInput, time: now.format('HH:mm'), createdAt: now.toISOString(), attachedFile: file, }; getMessages().push(userMsg); await nextTick(); getChatMessages()?.forceScrollToBottom?.(); // 附件缺原始文件对象(如历史消息回填)时降级为纯文本重发 if (!file.originalFile) { await streamAiAnswer(userInput); return; } await streamAiAnswer(userInput, file); }; // 剥离用户消息里注入的技能/会话引用 HTML(rawInput 缺失时的兜底) const stripInjectedRefs = (content: string): string => (content || '') .replace(/\[]*>[^\]]*\]/g, '') .replace(/]*>/g, '') .trim(); // 补齐消息的数据库主键 id(刚流式生成的消息前端没有 id,但后端已持久化)。 // // sessionId 与 msgs 都由调用方在流程开始时锚定后传入:本函数内部有 await,如果这里再去读 // 「当前会话 / 当前任务」,请求期间用户切走就会把别的会话的消息当成目标,回填出的 id 会让 // 后续的删除打到另一条会话上。 // // 两边条数必须严格一致才回填。早期实现用「长度差」做尾部对齐,一旦本地存在仅前端的消息 // (例如发送失败的报错气泡,后端没有对应行),长度差为负,之后每条未回填的消息都会拿到 // 前一条的 id —— 删除按钮据此调 DELETE /messages/{id},删掉的是用户根本没想删的那一行。 const hydrateMessageIds = async (sessionId: string, msgs: Message[]) => { if (!sessionId || !msgs.length) return; try { const res = await getDetail(sessionId); const raw = res?.messages || (Array.isArray(res) ? res : res?.data); if (!Array.isArray(raw)) return; // 用 transformHistoryToMessages 归一化,避免 model_thinking 等记录导致条数与前端不对齐 const history = transformHistoryToMessages(raw); const ids = history.map((m) => m.id).filter((id): id is number => id != null); if (ids.length !== msgs.length) return; for (let i = 0; i < msgs.length; i++) { if (msgs[i].id == null) msgs[i].id = ids[i]; } } catch (e) { console.error('补齐消息 id 失败:', e); } }; // 重新生成后,删除后端因重发而产生的重复用户消息(保留第一次,形成「一次提问、多次回答」)。 // sessionId 同样由调用方锚定传入:若这里读「当前会话」,重新生成期间切走就会去别的会话里 // 按内容匹配并删除 user 记录,可能删掉一条与本次操作无关的正常消息。 const removeDuplicateRegenerateUserMessage = async (sessionId: string, expectedContent: string) => { if (!sessionId) return; try { const res = await getDetail(sessionId); const raw = res?.messages || (Array.isArray(res) ? res : res?.data); if (!Array.isArray(raw)) return; // 保留第一条 user(原始提问),删除其余与重发内容一致的重复 user,避免多次重新生成残留 const userMsgs = raw.filter((m: any) => m?.role === 'user' || m?.type === 'user_message'); const duplicates = userMsgs.filter( (m: any, idx: number) => idx > 0 && normalizeLineBreaks(m?.content || '').trim() === expectedContent.trim() && m?.id != null ); for (const dup of duplicates) { await deleteMessage(dup.id); } } catch (e) { console.error('删除重发产生的重复用户消息失败:', e); } }; // 重新生成最后一条回答 const handleRegenerate = async () => { const task = getTaskList().find((t) => t.id === getCurrentTaskId()); if (!task || task.isStreaming) return; if (getPendingApproval()) { message.warning('请先审批执行计划'); return; } const msgs = task.messages; const lastAiIdx = msgs.length - 1; if (lastAiIdx < 0 || msgs[lastAiIdx].type !== 'ai') return; // 找前一条用户消息作为重发内容 let userIdx = -1; for (let i = lastAiIdx - 1; i >= 0; i--) { if (msgs[i].type === 'user') { userIdx = i; break; } } if (userIdx < 0) return; const userMsg = msgs[userIdx]; const userInput = userMsg.rawInput || stripInjectedRefs(userMsg.content); const attachedFile = userMsg.attachedFile; // 不删除旧 AI 消息、不动用户提问,直接重发同一条用户输入重新生成。 // 以下三步都在 await 之后,一律显式锚定发起任务:用「当前任务/当前会话」会在切走后作用于别的会话 await streamAiAnswer(userInput, attachedFile, task.id); // 删除重发产生的重复用户消息,形成「一次提问、多次回答」 await removeDuplicateRegenerateUserMessage(task.sessionId, userInput); // 回填新生成消息的 id,使折叠组内各条消息都能正常显示删除按钮 await hydrateMessageIds(task.sessionId, task.messages); }; // 编辑用户消息并重发(截断该消息及其后所有消息) const handleEditResend = async (index: number, newContent: string) => { const task = getTaskList().find((t) => t.id === getCurrentTaskId()); if (!task || task.isStreaming) return; if (getPendingApproval()) { message.warning('请先审批执行计划'); return; } const msgs = task.messages; if (index < 0 || index >= msgs.length || msgs[index].type !== 'user') return; const trimmed = (newContent || '').trim(); if (!trimmed) { message.warning('消息内容不能为空'); return; } const attachedFile = msgs[index].attachedFile; // 全程锚定发起任务:这几步都在 await 之后,用「当前任务」会在切走后把别的会话的记录删掉 await hydrateMessageIds(task.sessionId, msgs); // 从后往前删除后端记录(有 id 的才调接口) const toDelete = msgs.slice(index).filter((m) => m.id != null); for (const m of toDelete) { try { await deleteMessage(m.id!); } catch (e) { console.error('删除消息失败:', e); } } // 本地截断到编辑位置(原地截断,始终操作同一个任务的消息数组) msgs.splice(index); const now = dayjs(); const userMsg: Message = { type: 'user', content: trimmed, rawInput: trimmed, time: now.format('HH:mm'), createdAt: now.toISOString(), attachedFile, }; msgs.push(userMsg); await nextTick(); if (getCurrentTaskId() === task.id) getChatMessages()?.forceScrollToBottom?.(); await streamAiAnswer(trimmed, attachedFile, task.id); }; return { loading, stopStreaming, setTaskAbortController, abortAllStreams, getSkillRefIconUrl, getSessionRefIconUrl, streamAiAnswer, sendTextOnly, sendWithAttachment, stripInjectedRefs, hydrateMessageIds, removeDuplicateRegenerateUserMessage, handleRegenerate, handleEditResend, }; };