fix(web): merge live and history by serverId and clientTurnId
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
+42
-5
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<void> {
|
||||
const headers: Record<string, string> = {
|
||||
'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,
|
||||
});
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string> {
|
||||
const keys = new Set<string>();
|
||||
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,
|
||||
};
|
||||
}
|
||||
@@ -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<string>();
|
||||
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;
|
||||
});
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user