diff --git a/apps/webclaw/src/screens/chat/chat-queries.ts b/apps/webclaw/src/screens/chat/chat-queries.ts index 2d08b5c..373c2b1 100644 --- a/apps/webclaw/src/screens/chat/chat-queries.ts +++ b/apps/webclaw/src/screens/chat/chat-queries.ts @@ -1,4 +1,4 @@ -import { normalizeSessions, readError } from './utils' +import { getMessageTimestamp, normalizeSessions, readError } from './utils' import type { QueryClient } from '@tanstack/react-query' import type { GatewayMessage, @@ -162,7 +162,7 @@ export function moveHistoryMessages( ) { const fromKey = chatQueryKeys.history(fromFriendlyId, fromSessionKey) const toKey = chatQueryKeys.history(toFriendlyId, toSessionKey) - const fromData = queryClient.getQueryData(fromKey) + const fromData = queryClient.getQueryData(fromKey) if (!fromData) return const messages = Array.isArray(fromData.messages) ? fromData.messages : [] queryClient.setQueryData(toKey, { @@ -179,19 +179,38 @@ export function updateSessionLastMessage( friendlyId: string, message: GatewayMessage, ) { + const messageUpdatedAt = getMessageTimestamp(message) queryClient.setQueryData( chatQueryKeys.sessions, function update(messages: unknown) { if (!Array.isArray(messages)) return messages - return (messages as Array).map((session) => { + const nextSessions = (messages as Array).map((session) => { if (session.key !== sessionKey && session.friendlyId !== friendlyId) { return session } return { ...session, lastMessage: message, + updatedAt: + typeof session.updatedAt === 'number' && + Number.isFinite(session.updatedAt) && + session.updatedAt > messageUpdatedAt + ? session.updatedAt + : messageUpdatedAt, } }) + + return [...nextSessions].sort((a, b) => { + const aUpdatedAt = + typeof a.updatedAt === 'number' && Number.isFinite(a.updatedAt) + ? a.updatedAt + : 0 + const bUpdatedAt = + typeof b.updatedAt === 'number' && Number.isFinite(b.updatedAt) + ? b.updatedAt + : 0 + return bUpdatedAt - aUpdatedAt + }) }, ) } diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts index 0a0aff7..f2282b1 100644 --- a/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts @@ -174,12 +174,19 @@ function mergeStreamingHistoryMessages( const streamingText = textFromMessage(streamingMessage) const serverText = textFromMessage(serverMessage) - if ( - streamingText && - streamingText !== serverText && - !serverText.startsWith(streamingText) - ) { - return false + if (streamingText && streamingText !== serverText) { + const normalizedStreamingText = normalizeAssistantTextForDedup(streamingText) + const normalizedServerText = normalizeAssistantTextForDedup(serverText) + const textLikelySameResponse = + normalizedStreamingText.length > 0 && + normalizedServerText.length > 0 && + (normalizedStreamingText === normalizedServerText || + normalizedStreamingText.includes(normalizedServerText) || + normalizedServerText.includes(normalizedStreamingText)) + + if (!textLikelySameResponse && !serverText.startsWith(streamingText)) { + return false + } } return messageCoversStreamingMessage(serverMessage, streamingMessage) @@ -224,6 +231,14 @@ function nonTextPartSignatures(message: GatewayMessage): Set { return signatures } +function normalizeAssistantTextForDedup(text: string): string { + return text + .replace(/\[\[reply_to:[^\]]*\]\]\s*/gi, '') + .replace(/[\s\S]*?<\/system-reminder>/gi, '') + .replace(/\s+/g, ' ') + .trim() +} + function mergeOptimisticHistoryMessages( serverMessages: Array, optimisticMessages: Array, @@ -232,7 +247,7 @@ function mergeOptimisticHistoryMessages( const merged = [...serverMessages] for (const optimisticMessage of optimisticMessages) { - const hasMatch = serverMessages.some((serverMessage) => { + const hasMatch = merged.some((serverMessage) => { if ( optimisticMessage.clientId && serverMessage.clientId && diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts index c7f158d..df8b4c0 100644 --- a/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts @@ -1,6 +1,6 @@ import { useCallback, useEffect, useRef } from 'react' -import { getMessageTimestamp } from '../utils' +import { getMessageTimestamp, textFromMessage } from '../utils' import { chatQueryKeys, updateHistoryMessages, @@ -50,6 +50,7 @@ export function useChatStream({ const streamRunStateVersionRef = useRef(new Map()) const streamRunSourceRef = useRef(new Map()) const streamSeenEventKeysRef = useRef(new Set()) + const streamSeenPayloadKeysRef = useRef(new Set()) const refreshHistoryRef = useRef(refreshHistory) refreshHistoryRef.current = refreshHistory @@ -66,6 +67,7 @@ export function useChatStream({ streamRunStateVersionRef.current.clear() streamRunSourceRef.current.clear() streamSeenEventKeysRef.current.clear() + streamSeenPayloadKeysRef.current.clear() }, []) useEffect(() => { @@ -126,11 +128,7 @@ export function useChatStream({ parsed.event === 'agent' ? 'agent' : 'chat' if (streamRunId) { const currentSource = streamRunSourceRef.current.get(streamRunId) - if ( - payloadSource === 'chat' && - currentSource === 'agent' && - payload.state === 'delta' - ) { + if (payloadSource === 'chat' && currentSource === 'agent') { continue } if (payloadSource === 'agent' || !currentSource) { @@ -166,6 +164,16 @@ export function useChatStream({ continue } + if ( + shouldSkipDuplicatePayload( + streamSeenPayloadKeysRef.current, + payloadSource, + payload, + ) + ) { + continue + } + if ( shouldSkipStaleRunEvent( streamRunId, @@ -220,6 +228,21 @@ export function useChatStream({ const resolvedLastUserIndex = lastUserIndex >= 0 ? messages.length - 1 - lastUserIndex : -1 + const nextId = getMessageId(nextMessage) + if (nextId) { + const existingById = messages.findIndex( + (message) => getMessageId(message) === nextId, + ) + if (existingById >= 0) { + const next = [...messages] + next[existingById] = mergeStreamMessage( + messages[existingById], + nextMessage, + ) + return next + } + } + if (streamRunId) { const index = findStreamMessageIndex( messages, @@ -244,7 +267,15 @@ export function useChatStream({ const target = messages.length - 1 - index if (target > resolvedLastUserIndex) { const targetTime = getMessageTimestamp(messages[target]) - if (Math.abs(nextTime - targetTime) <= 15000) { + const targetText = textFromMessage(messages[target]) + const nextText = textFromMessage(nextMessage) + const textMatches = shouldMergeAssistantByText( + targetText, + nextText, + ) + if ( + Math.abs(nextTime - targetTime) <= 15000 || textMatches + ) { const next = [...messages] next[target] = mergeStreamMessage(messages[target], nextMessage) return next @@ -490,6 +521,52 @@ function shouldSkipDuplicateEvent( return false } +function shouldSkipDuplicatePayload( + seen: Set, + source: 'agent' | 'chat', + payload: StreamChatPayload, +): boolean { + const runId = normalizeString(payload.runId) + const state = normalizeString(payload.state) + const sessionKey = normalizeString(payload.sessionKey) + const message = payload.message + const messageId = message ? getMessageId(message) : '' + const role = normalizeString(message?.role) + const toolCallId = normalizeString(message?.toolCallId) + const text = message ? textFromMessage(message).slice(0, 512) : '' + + if (!runId && !messageId && !text) return false + + const key = `${source}:${runId}:${state}:${sessionKey}:${role}:${messageId}:${toolCallId}:${text}` + if (seen.has(key)) return true + seen.add(key) + if (seen.size > 4000) { + seen.clear() + } + return false +} + +function shouldMergeAssistantByText(previousText: string, nextText: string): boolean { + if (!previousText || !nextText) return false + if (previousText === nextText) return true + + const previousNormalized = normalizeAssistantTextForDedup(previousText) + const nextNormalized = normalizeAssistantTextForDedup(nextText) + if (!previousNormalized || !nextNormalized) return false + if (previousNormalized === nextNormalized) return true + if (previousNormalized.includes(nextNormalized)) return true + if (nextNormalized.includes(previousNormalized)) return true + return false +} + +function normalizeAssistantTextForDedup(text: string): string { + return text + .replace(/\[\[reply_to:[^\]]*\]\]\s*/gi, '') + .replace(/[\s\S]*?<\/system-reminder>/gi, '') + .replace(/\s+/g, ' ') + .trim() +} + function extractChatPayloadsFromAgentPayload( payload: unknown, ): Array {