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

feat(web): send stable client_turn_id and resume turns after disconnect

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
WeiCHEN
2026-09-11 13:20:13 +08:00
co-authored by Cursor
parent 1438af18f7
commit 0dbfc32603
2 changed files with 123 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/src/App.tsx
+93 -35
View File
@@ -100,8 +100,6 @@ import {
saveSessions,
loadActiveSessionId,
saveActiveSessionId,
loadMessages,
saveMessages,
loadDeliverables,
saveDeliverables,
loadRoles,
@@ -118,6 +116,8 @@ import {
buildPrivateSessionId,
} from './data/storage';
import { useChatWindow } from './hooks/useChatWindow';
import { useTurnRuntime } from './hooks/useTurnRuntime';
import { fetchChatTurnStatus } from './api/chatTurn';
import { CategoryFilterBar } from './components/history/CategoryFilterBar';
import { HistoryCalendar } from './components/history/HistoryCalendar';
import { ChatSearchModal } from './components/history/ChatSearchModal';
@@ -1301,16 +1301,12 @@ export default function App() {
patchBlockId?: string;
} | null>(null);
// 消息流与思考脉冲 (支持 localStorage 持久化与项目会话严格隔离)
const [isThinking, setIsThinking] = useState(false);
const [messages, setMessages] = useState<ChatMessage[]>(() => loadMessages());
// 真实后端 SSE 流式对话：当前活跃请求的中止句柄 (支持停止与自动重试取消)
const streamAbortRef = useRef<AbortController | null>(null);
// 消息流：权威来源为服务端窗口；不再全量持久化到 localStorage
const [messages, setMessages] = useState<ChatMessage[]>([]);
// 自动持久化消息流
useEffect(() => {
saveMessages(messages);
}, [messages]);
void migrateAndClearLegacyMessages();
}, []);
// 基于当前激活的项目会话严格隔离对话记录
const currentSessionMessages = useMemo(() => {
@@ -1318,6 +1314,10 @@ export default function App() {
}, [messages, activeSessionId]);
// P0: 会话按日窗口状态机 (处理首屏拉取、瞬时落底、触顶加载补偿)
const turnRuntime = useTurnRuntime(activeSessionId);
const [taskThinking, setTaskThinking] = useState(false);
const isThinking = turnRuntime.isThinking || taskThinking;
const chatWindow = useChatWindow({
sessionId: activeSessionId,
liveMessages: currentSessionMessages,
@@ -1721,12 +1721,21 @@ export default function App() {
opts?: StreamAttemptOpts
) => {
const quote = opts?.quote;
// 同一时刻只允许一个活跃流：新尝试前中止上一个 (若存在)
streamAbortRef.current?.abort();
const controller = new AbortController();
streamAbortRef.current = controller;
const streamSessionId = activeSessionId;
if (opts?.clientTurnId) {
turnRuntime.registerTurn({
clientTurnId: opts.clientTurnId,
sessionId: streamSessionId,
userMsgId: opts.userMsgId || '',
asstId,
status: 'pending',
controller,
startedAt: Date.now(),
});
}
const currentProj = sessions.find((s) => s.sessionId === activeSessionId);
const currentProj = sessions.find((s) => s.sessionId === streamSessionId);
let accumulated = '';
let settled = false;
@@ -1738,13 +1747,21 @@ export default function App() {
setTimeout(() => {
if (controller.signal.aborted) {
updateAsst({ status: 'cancelled', livePhaseLabel: undefined });
setIsThinking(false);
if (opts?.clientTurnId) {
turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'cancelled');
}
return;
}
runStreamAttempt(userText, asstId, attemptIndex + 1, opts);
}, delayMs);
};
const finishTurn = (status: 'completed' | 'failed' | 'cancelled') => {
if (opts?.clientTurnId) {
turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, status);
}
};
const failOrRetry = (info: ChatErrorInfo) => {
settled = true;
if (shouldAutoRetryChat(info, attemptIndex)) {
@@ -1767,7 +1784,7 @@ export default function App() {
},
livePhaseLabel: undefined,
});
setIsThinking(false);
finishTurn('failed');
};
updateAsst({
@@ -1776,7 +1793,9 @@ export default function App() {
content: '',
errorInfo: undefined,
});
setIsThinking(true);
if (opts?.clientTurnId) {
turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'pending');
}
const bindUserServerId = (serverId: number) => {
if (!opts?.userMsgId) return;
@@ -1788,7 +1807,7 @@ export default function App() {
streamChat(
{
session_id: activeSessionId,
session_id: streamSessionId,
user_id: currentUser.id,
project_id: currentProj?.projectId || currentProjectId || 1,
project_name: currentProj?.title || '肌骨超声AI辅助诊断系统',
@@ -1815,11 +1834,17 @@ export default function App() {
case 'status': {
const label = ev.phase?.message || ev.status?.message;
updateAsst({ status: 'streaming_status', livePhaseLabel: label || '助手正在处理…' });
if (opts?.clientTurnId) {
turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'streaming');
}
break;
}
case 'delta': {
accumulated += ev.delta || '';
updateAsst({ status: 'streaming_status', content: accumulated });
if (opts?.clientTurnId) {
turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'streaming');
}
break;
}
case 'error': {
@@ -1843,7 +1868,7 @@ export default function App() {
serverId: typeof serverMsgId === 'number' ? serverMsgId : undefined,
exportCapability: exportCap,
});
setIsThinking(false);
finishTurn('completed');
break;
}
default:
@@ -1853,18 +1878,48 @@ export default function App() {
},
controller.signal,
opts?.clientTurnId ? { idempotencyKey: opts.clientTurnId } : undefined
).catch((_err) => {
).catch(async (_err) => {
if (settled) return;
if (controller.signal.aborted) {
updateAsst({ status: 'cancelled', livePhaseLabel: undefined });
setIsThinking(false);
finishTurn('cancelled');
return;
}
// fetch/网络层抛出：合成瞬态错误，走与后端 error 事件一致的自动重试路径
if (opts?.clientTurnId) {
try {
const turnStatus = await fetchChatTurnStatus(streamSessionId, opts.clientTurnId);
if (turnStatus?.status === 'completed' && turnStatus.assistantMessageId) {
settled = true;
if (turnStatus.userMessageId) bindUserServerId(turnStatus.userMessageId);
chatWindow.replaceMessageServerId(asstId, turnStatus.assistantMessageId);
updateAsst({
status: 'completed',
livePhaseLabel: undefined,
serverId: turnStatus.assistantMessageId,
});
finishTurn('completed');
return;
}
if (turnStatus?.status === 'processing') {
updateAsst({
status: 'streaming_status',
livePhaseLabel: '连接中断，服务端仍在处理…',
});
if (opts?.clientTurnId) {
turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'streaming');
}
return;
}
} catch {
// fall through to structured failure
}
}
failOrRetry(buildNetworkErrorInfo());
});
};
const taskStreamAbortRef = useRef<AbortController | null>(null);
// P2：长任务尝试模式。'run' 新任务 | 'resume' 续跑失败工作区 | 'restart' 按原意图新工作区
const runTaskAttempt = (
task: TaskRunContext,
@@ -1873,10 +1928,9 @@ export default function App() {
attemptIndex: number,
opts?: { phaseEvents?: PhaseEvent[]; workDir?: string }
) => {
// 同一时刻只允许一个活跃流：新尝试前中止上一个 (若存在)
streamAbortRef.current?.abort();
taskStreamAbortRef.current?.abort();
const controller = new AbortController();
streamAbortRef.current = controller;
taskStreamAbortRef.current = controller;
let accumulated = '';
let settled = false;
@@ -1891,7 +1945,7 @@ export default function App() {
setTimeout(() => {
if (controller.signal.aborted) {
patchAsst({ status: 'cancelled', livePhaseLabel: undefined });
setIsThinking(false);
setTaskThinking(false);
return;
}
runTaskAttempt(task, asstId, mode, attemptIndex + 1, opts);
@@ -1910,7 +1964,7 @@ export default function App() {
return;
}
patchAsst({ status: 'failed', errorInfo: info, livePhaseLabel: undefined });
setIsThinking(false);
setTaskThinking(false);
};
// 等待态：resume 保留已有 phaseEvents 续接；run/restart 重置正文与步骤条
@@ -1921,7 +1975,7 @@ export default function App() {
errorInfo: undefined,
...(mode === 'resume' ? {} : { content: '', phaseEvents: [], taskWorkspaceId: undefined }),
});
setIsThinking(true);
setTaskThinking(true);
const body =
mode === 'resume' && opts?.workDir
@@ -1973,7 +2027,7 @@ export default function App() {
livePhaseLabel: undefined,
taskWorkspaceId: ev.done?.workspace_id,
});
setIsThinking(false);
setTaskThinking(false);
break;
}
default:
@@ -1986,7 +2040,7 @@ export default function App() {
if (settled) return;
if (controller.signal.aborted) {
patchAsst({ status: 'cancelled', livePhaseLabel: undefined });
setIsThinking(false);
setTaskThinking(false);
return;
}
// fetch/网络层抛出：合成瞬态错误，走与后端 error 事件一致的自动重试路径
@@ -2024,12 +2078,16 @@ export default function App() {
: undefined;
const originalText = sourceMsg?.content?.trim();
if (!originalText) return;
runStreamAttempt(originalText, messageId, 0);
runStreamAttempt(originalText, messageId, 0, {
userMsgId: sourceMsg?.id,
clientTurnId: target.clientTurnId || sourceMsg?.clientTurnId,
});
};
// 用户主动停止：中止当前流式请求，级联取消后续自动重试
// 用户主动停止：仅中止当前 session 的活跃 turn
const handleStopStreaming = () => {
streamAbortRef.current?.abort();
turnRuntime.abortActiveSessionTurns();
taskStreamAbortRef.current?.abort();
};
// 发送消息与对话驱动双轨逻辑
@@ -2262,10 +2320,10 @@ export default function App() {
return;
}
setIsThinking(true);
setTaskThinking(true);
setTimeout(() => {
setIsThinking(false);
setTaskThinking(false);
// 1. 用户要求助手生成新文档/测算/推演/备忘 (默认自动归档至【我的私密生成】，保护隐私)
if (isGenDocShortcut) {
web/src/api/chatTurn.ts
+30
View File
@@ -0,0 +1,30 @@
export type ChatTurnStatus = {
sessionId: string;
clientTurnId: string;
status: string;
userMessageId?: number;
assistantMessageId?: number;
attemptCount?: number;
};
export async function fetchChatTurnStatus(
sessionId: string,
clientTurnId: string,
signal?: AbortSignal,
): Promise<ChatTurnStatus | null> {
const url = `/api/v1/chat/turns/${encodeURIComponent(clientTurnId)}?session_id=${encodeURIComponent(sessionId)}`;
const res = await fetch(url, { signal });
if (res.status === 404) return null;
if (!res.ok) {
throw new Error(`fetchChatTurnStatus failed: HTTP ${res.status}`);
}
const raw = await res.json();
return {
sessionId: raw.session_id,
clientTurnId: raw.client_turn_id,
status: raw.status,
userMessageId: raw.user_message_id,
assistantMessageId: raw.assistant_message_id,
attemptCount: raw.attempt_count,
};
}