|
|
|
@ -388,6 +388,136 @@ const getEventPayload = (event) => |
|
|
|
event?.params?.data ?? |
|
|
|
event; |
|
|
|
|
|
|
|
//[刷新回放]保留当前会话的消息快照，避免 hydrate 返回的旧快照覆盖 SSE 增量。
|
|
|
|
const mergeStreamMessageSnapshots = (previousMessages = [], currentMessages = []) => { |
|
|
|
const previous = Array.isArray(previousMessages) ? previousMessages : []; |
|
|
|
const current = Array.isArray(currentMessages) ? currentMessages : []; |
|
|
|
if (!previous.length) return current; |
|
|
|
if (!current.length) return previous; |
|
|
|
|
|
|
|
const currentById = new Map( |
|
|
|
current |
|
|
|
.filter((message) => message?.id) |
|
|
|
.map((message) => [String(message.id), message]) |
|
|
|
); |
|
|
|
const merged = []; |
|
|
|
const usedIds = new Set(); |
|
|
|
|
|
|
|
previous.forEach((message) => { |
|
|
|
const messageId = String(message?.id || ""); |
|
|
|
if (!messageId) return; |
|
|
|
merged.push(currentById.get(messageId) || message); |
|
|
|
usedIds.add(messageId); |
|
|
|
}); |
|
|
|
|
|
|
|
current.forEach((message) => { |
|
|
|
const messageId = String(message?.id || ""); |
|
|
|
if (!messageId || usedIds.has(messageId)) return; |
|
|
|
merged.push(message); |
|
|
|
}); |
|
|
|
|
|
|
|
return merged; |
|
|
|
}; |
|
|
|
|
|
|
|
//[刷新回放]直接消费 root messages 事件，保证网络层已有数据时页面同步更新。
|
|
|
|
const applyMessageEventToState = (currentState, event) => { |
|
|
|
const data = event?.params?.data || {}; |
|
|
|
const eventType = String(data.event || ""); |
|
|
|
const nextMessages = new Map(currentState.messages); |
|
|
|
let activeMessageId = currentState.activeMessageId; |
|
|
|
const messageId = String(data.id || activeMessageId || "").trim(); |
|
|
|
if (!messageId) return currentState; |
|
|
|
|
|
|
|
const previousMessage = nextMessages.get(messageId) || { |
|
|
|
id: messageId, |
|
|
|
type: "ai", |
|
|
|
role: "ai", |
|
|
|
content: "", |
|
|
|
tool_calls: [], |
|
|
|
}; |
|
|
|
const nextMessage = { |
|
|
|
...previousMessage, |
|
|
|
tool_calls: Array.isArray(previousMessage.tool_calls) |
|
|
|
? [...previousMessage.tool_calls] |
|
|
|
: [], |
|
|
|
}; |
|
|
|
|
|
|
|
if (eventType === "message-start") { |
|
|
|
activeMessageId = messageId; |
|
|
|
nextMessage.role = data.role || nextMessage.role || "ai"; |
|
|
|
nextMessage.type = nextMessage.role; |
|
|
|
} |
|
|
|
|
|
|
|
if (eventType === "content-block-delta") { |
|
|
|
const delta = data.delta || {}; |
|
|
|
if (delta.type === "text-delta") { |
|
|
|
nextMessage.content = `${String(nextMessage.content || "")}${String(delta.text || "")}`; |
|
|
|
} |
|
|
|
if (delta.type === "tool-call-delta") { |
|
|
|
const toolCallIndex = Number(data.index || 0); |
|
|
|
const previousToolCall = nextMessage.tool_calls[toolCallIndex] || {}; |
|
|
|
const deltaArgs = delta.args; |
|
|
|
const nextArgs = |
|
|
|
previousToolCall.args && |
|
|
|
typeof previousToolCall.args === "object" && |
|
|
|
deltaArgs && |
|
|
|
typeof deltaArgs === "object" |
|
|
|
? { ...previousToolCall.args, ...deltaArgs } |
|
|
|
: deltaArgs ?? previousToolCall.args; |
|
|
|
nextMessage.tool_calls[toolCallIndex] = { |
|
|
|
...previousToolCall, |
|
|
|
args: nextArgs, |
|
|
|
status: "running", |
|
|
|
}; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
if (eventType === "content-block-finish") { |
|
|
|
const content = data.content || {}; |
|
|
|
if (content.type === "text") { |
|
|
|
nextMessage.content = content.text || nextMessage.content || ""; |
|
|
|
} |
|
|
|
if (content.type === "tool_call") { |
|
|
|
const toolCallIndex = Number(data.index || 0); |
|
|
|
nextMessage.tool_calls[toolCallIndex] = { |
|
|
|
...(nextMessage.tool_calls[toolCallIndex] || {}), |
|
|
|
...content, |
|
|
|
status: "success", |
|
|
|
}; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
nextMessages.set(messageId, nextMessage); |
|
|
|
return { |
|
|
|
messages: nextMessages, |
|
|
|
activeMessageId, |
|
|
|
}; |
|
|
|
}; |
|
|
|
|
|
|
|
const RUNTIME_STATE_REFRESH_EVENT_TYPES = new Set([ |
|
|
|
"generation.completed", |
|
|
|
"generation.failed", |
|
|
|
"generation.cancelled", |
|
|
|
"generation.error", |
|
|
|
]); |
|
|
|
|
|
|
|
const shouldRefreshRuntimeState = ({ |
|
|
|
method = "", |
|
|
|
payload = {}, |
|
|
|
isResuming = false, |
|
|
|
} = {}) => { |
|
|
|
if (method === "messages") return isResuming; |
|
|
|
if (method === "custom") { |
|
|
|
return RUNTIME_STATE_REFRESH_EVENT_TYPES.has(String(payload.type || "")); |
|
|
|
} |
|
|
|
if (method === "lifecycle") { |
|
|
|
return ["completed", "failed", "aborted"].includes( |
|
|
|
String(payload.event || "") |
|
|
|
); |
|
|
|
} |
|
|
|
return false; |
|
|
|
}; |
|
|
|
|
|
|
|
const getEventDedupeKey = (event) => { |
|
|
|
if (event?.event_id) return String(event.event_id); |
|
|
|
const payload = getEventPayload(event) || {}; |
|
|
|
@ -431,12 +561,20 @@ const useSuperAgentStream = ({ |
|
|
|
const [runtimeCheckpointDeliveries, setRuntimeCheckpointDeliveries] = useState([]); |
|
|
|
const [runtimeInterrupts, setRuntimeInterrupts] = useState([]); |
|
|
|
const [dismissedInterruptIds, setDismissedInterruptIds] = useState(new Set()); |
|
|
|
const runtimeStateRefreshTimerRef = useRef(null); |
|
|
|
const runtimeStateRefreshPendingRef = useRef(false); |
|
|
|
const resolvedInterruptIdsRef = useRef(new Set()); |
|
|
|
const resolvedInterruptInteractionsRef = useRef(new Set()); |
|
|
|
const [currentInterruptId, setCurrentInterruptId] = useState(""); |
|
|
|
const [hydratedThreadId, setHydratedThreadId] = useState(""); |
|
|
|
const [transportStateRevision, setTransportStateRevision] = useState(0); |
|
|
|
const eventRequestControllersRef = useRef(new Map()); |
|
|
|
const eventMessageStateRef = useRef({ |
|
|
|
sessionId: "", |
|
|
|
activeMessageId: "", |
|
|
|
messages: new Map(), |
|
|
|
}); |
|
|
|
const [eventStreamMessages, setEventStreamMessages] = useState([]); |
|
|
|
const pendingSubmitThreadIdRef = useRef(""); |
|
|
|
const hydrationSessionRef = useRef({ |
|
|
|
sessionId: "", |
|
|
|
@ -529,6 +667,26 @@ const useSuperAgentStream = ({ |
|
|
|
transport: eventTransport, |
|
|
|
}); |
|
|
|
|
|
|
|
//[恢复中断后同步 thread state][避免 SSE 已收到但消息投影仍停留在旧快照]
|
|
|
|
const scheduleRuntimeStateRefresh = useCallback((delay = 120, force = false) => { |
|
|
|
if (runtimeStateRefreshTimerRef.current != null) return; |
|
|
|
if (!force && runtimeStateRefreshPendingRef.current) return; |
|
|
|
if (!force) runtimeStateRefreshPendingRef.current = true; |
|
|
|
|
|
|
|
runtimeStateRefreshTimerRef.current = setTimeout(async () => { |
|
|
|
runtimeStateRefreshTimerRef.current = null; |
|
|
|
if (!String(eventTransport.threadId || "").trim()) { |
|
|
|
runtimeStateRefreshPendingRef.current = false; |
|
|
|
return; |
|
|
|
} |
|
|
|
try { |
|
|
|
await eventTransport.getState().catch(() => undefined); |
|
|
|
} finally { |
|
|
|
runtimeStateRefreshPendingRef.current = false; |
|
|
|
} |
|
|
|
}, delay); |
|
|
|
}, [eventTransport]); |
|
|
|
|
|
|
|
const handleBusinessEvent = useCallback((event) => { |
|
|
|
const eventKey = getEventDedupeKey(event); |
|
|
|
if (handledEventKeysRef.current.has(eventKey)) return; |
|
|
|
@ -585,6 +743,11 @@ const useSuperAgentStream = ({ |
|
|
|
useEffect(() => { |
|
|
|
const sessionKey = String(activeSessionId || "").trim(); |
|
|
|
return () => { |
|
|
|
if (runtimeStateRefreshTimerRef.current != null) { |
|
|
|
clearTimeout(runtimeStateRefreshTimerRef.current); |
|
|
|
runtimeStateRefreshTimerRef.current = null; |
|
|
|
} |
|
|
|
runtimeStateRefreshPendingRef.current = false; |
|
|
|
if (sessionKey) { |
|
|
|
const sessionController = eventRequestControllersRef.current.get(sessionKey); |
|
|
|
sessionController?.abort(); |
|
|
|
@ -616,11 +779,18 @@ const useSuperAgentStream = ({ |
|
|
|
setIsBusinessRunning(false); |
|
|
|
setRuntimeCheckpointDeliveries([]); |
|
|
|
setRuntimeInterrupts([]); |
|
|
|
runtimeStateRefreshPendingRef.current = false; |
|
|
|
setDismissedInterruptIds(new Set()); |
|
|
|
resolvedInterruptIdsRef.current.clear(); |
|
|
|
resolvedInterruptInteractionsRef.current.clear(); |
|
|
|
setCurrentInterruptId(""); |
|
|
|
setHydratedThreadId(""); |
|
|
|
eventMessageStateRef.current = { |
|
|
|
sessionId: String(activeSessionId || ""), |
|
|
|
activeMessageId: "", |
|
|
|
messages: new Map(), |
|
|
|
}; |
|
|
|
setEventStreamMessages([]); |
|
|
|
}, [activeSessionId]); |
|
|
|
|
|
|
|
useEffect(() => { |
|
|
|
@ -649,6 +819,27 @@ const useSuperAgentStream = ({ |
|
|
|
transportStateRevision, |
|
|
|
]); |
|
|
|
|
|
|
|
const handleMessageEvent = useCallback((event) => { |
|
|
|
if (event?.method !== "messages") return; |
|
|
|
const namespace = event?.params?.namespace; |
|
|
|
if (Array.isArray(namespace) && namespace.length > 0) return; |
|
|
|
|
|
|
|
const sessionId = String(activeSessionIdRef.current || "").trim(); |
|
|
|
if (!sessionId) return; |
|
|
|
if (eventMessageStateRef.current.sessionId !== sessionId) { |
|
|
|
eventMessageStateRef.current = { |
|
|
|
sessionId, |
|
|
|
activeMessageId: "", |
|
|
|
messages: new Map(), |
|
|
|
}; |
|
|
|
} |
|
|
|
|
|
|
|
const nextState = applyMessageEventToState(eventMessageStateRef.current, event); |
|
|
|
if (nextState === eventMessageStateRef.current) return; |
|
|
|
eventMessageStateRef.current = nextState; |
|
|
|
setEventStreamMessages([...nextState.messages.values()]); |
|
|
|
}, []); |
|
|
|
|
|
|
|
const checkpointDeliveries = useMemo(() => { |
|
|
|
const deliveries = new Map(); |
|
|
|
const hydratedDeliveries = [ |
|
|
|
@ -707,6 +898,18 @@ const useSuperAgentStream = ({ |
|
|
|
// 切换会话后旧 Thread 的监听器可能在清理前再收到一个事件，
|
|
|
|
// 不能把旧会话的事件写入新会话的流式消息状态。
|
|
|
|
if (activeSessionIdRef.current !== listenerSessionId) return; |
|
|
|
handleMessageEvent(event); |
|
|
|
const eventPayload = getEventPayload(event) || {}; |
|
|
|
if (shouldRefreshRuntimeState({ |
|
|
|
method: event?.method, |
|
|
|
payload: eventPayload, |
|
|
|
isResuming: interruptResumeRef.current.active, |
|
|
|
})) { |
|
|
|
const isTerminalEvent = |
|
|
|
event?.method === "custom" || event?.method === "lifecycle"; |
|
|
|
const refreshDelay = event?.method === "messages" ? 160 : 0; |
|
|
|
scheduleRuntimeStateRefresh(refreshDelay, isTerminalEvent); |
|
|
|
} |
|
|
|
if (event?.method === "input.requested") { |
|
|
|
const data = event.params?.data || {}; |
|
|
|
const interruptId = String(data.interrupt_id || "").trim(); |
|
|
|
@ -786,7 +989,14 @@ const useSuperAgentStream = ({ |
|
|
|
if (!BUSINESS_CHANNELS.includes(event?.method)) return; |
|
|
|
handleBusinessEvent(event); |
|
|
|
}); |
|
|
|
}, [activeSessionId, handleBusinessEvent, stream.threadId]); |
|
|
|
}, [ |
|
|
|
activeSessionId, |
|
|
|
handleBusinessEvent, |
|
|
|
handleMessageEvent, |
|
|
|
scheduleRuntimeStateRefresh, |
|
|
|
stream.threadId, |
|
|
|
transportStateRevision, |
|
|
|
]); |
|
|
|
|
|
|
|
const prepareForSessionSwitch = useCallback(() => { |
|
|
|
// 先让旧 thread 的事件监听失效。activeSessionId 在 React 下一次渲染
|
|
|
|
@ -883,6 +1093,7 @@ const useSuperAgentStream = ({ |
|
|
|
setIsLocalRunActive(true); |
|
|
|
localRunHasSeenLoadingRef.current = false; |
|
|
|
interruptResumeRef.current = { active: true, running: false }; |
|
|
|
runtimeStateRefreshPendingRef.current = false; |
|
|
|
pendingInterruptRef.current = false; |
|
|
|
awaitingInterruptRef.current = false; |
|
|
|
setRunPhase("resuming"); |
|
|
|
@ -904,6 +1115,7 @@ const useSuperAgentStream = ({ |
|
|
|
interruptId: options.interruptId, |
|
|
|
namespace: options.namespace, |
|
|
|
}).then((result) => { |
|
|
|
scheduleRuntimeStateRefresh(100); |
|
|
|
if (interruptId) { |
|
|
|
resolvedInterruptIdsRef.current.add(interruptId); |
|
|
|
const interactionKey = getInterruptInteractionKey( |
|
|
|
@ -940,7 +1152,7 @@ const useSuperAgentStream = ({ |
|
|
|
setIsRespondingInterrupt(false); |
|
|
|
throw error; |
|
|
|
}); |
|
|
|
}, [stream]); |
|
|
|
}, [scheduleRuntimeStateRefresh, stream]); |
|
|
|
|
|
|
|
const stopRun = useCallback(async () => { |
|
|
|
localRunActiveRef.current = false; |
|
|
|
@ -1087,9 +1299,15 @@ const useSuperAgentStream = ({ |
|
|
|
!hydratedTailIds.every((messageId) => rawStreamMessageIds.has(messageId)))); |
|
|
|
const shouldPreferHydratedState = |
|
|
|
shouldPreferHydratedRuntimeState || shouldPreferHydratedMessages; |
|
|
|
const streamMessages = shouldPreferHydratedMessages |
|
|
|
? hydratedMessages |
|
|
|
: rawStreamMessages; |
|
|
|
const projectedMessages = shouldPreferHydratedMessages |
|
|
|
? hydratedMessages |
|
|
|
: rawStreamMessages; |
|
|
|
const streamMessages = mergeStreamMessageSnapshots( |
|
|
|
projectedMessages, |
|
|
|
isThreadBound && eventMessageStateRef.current.sessionId === currentSessionId |
|
|
|
? eventStreamMessages |
|
|
|
: [] |
|
|
|
); |
|
|
|
const streamToolCalls = isStreamProjectionReady ? stream.toolCalls : []; |
|
|
|
const streamValues = shouldPreferHydratedState |
|
|
|
? hydratedState?.values || {} |
|
|
|
|
Senor-Liu
1 month ago