weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(web): merge live and history by serverId and clientTurnId

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
WeiCHEN
2026-09-13 02:50:21 +00:00
committed by weichen
co-authored by Cursor
parent c4ef72506f
commit e1a9b302ba
6 changed files with 166 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/src/App.tsx
+42 -5
View File
@@ -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;
}
web/src/api/chatStream.ts
+15 -1
View File
@@ -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,
});
web/src/api/chatWindow.ts
+2
View File
@@ -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,
web/src/engine/messageIdentity.ts
+79
View File
@@ -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,
};
}
web/src/hooks/useChatWindow.ts
+11 -13
View File
@@ -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;
});
web/src/types/workbench.ts
+17
View File
@@ -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;